File: //proc/1233/cwd/usr/lib/python3/dist-packages/compose/__pycache__/parallel.cpython-310.pyc
o
S �`,- � @ s� d dl Zd dlZd dlZd dlZd dlmZ d dlmZ d dlm Z d dlm
Z
d dlm Z d dl m
Z
d dl mZ d d lmZ d d
lmZ d d lmZ d d lmZ d d
lmZ d dlmZ d dlmZ d dlmZ d dlmZ e� e �Z e � Z G dd� d�Z!dd� Z"d2dd�Z#dd� Z$G dd� d�Z%G d d � d �Z&d d � Z'd d!� Z(d"d#� Z)G d$d%� d%e*�Z+G d&d'� d'�Z,d(d)� Z-d*d+� Z.d,d-� Z/d.d/� Z0d0d1� Z1dS )3� N)�Empty)�Queue)�Lock)� Semaphore)�Thread)�APIError)�
ImageNotFound)�AnsiMode)�green)�red)�ShutdownException)�PARALLEL_LIMIT)�CompletedUnsuccessfully)�HealthCheckFailed)�NoHealthCheckConfigured)�OperationFailedErrorc @ s$ e Zd ZdZee�Zedd� �ZdS )� GlobalLimitz�Simple class to hold a global semaphore limiter for a project. This class
should be treated as a singleton that is instantiated when the project is.
c C s |d u rt }t|�| _d S �N)r
r �global_limiter)�cls�value� r �2/usr/lib/python3/dist-packages/compose/parallel.py�set_global_limit% s z GlobalLimit.set_global_limitN) �__name__�
__module__� __qualname__�__doc__r r
r � classmethodr r r r r r s
r c C s, d}| D ]�\}} }
|
du r0|dur ||�r |� |||�dt� n
|� |||�dt� |�| � qt|
t�rI|
j|||�< |� |||�dt� |
}qt|
t�r`|
j|||�< |� |||�dt� qt|
tt t
t f�r{|
j |||�< |� |||�dt� qt|
t
�r�|� |||�dt� q|
|||�< |
}q|S )z Watch events from a parallel execution, update status and fill errors and results.
Returns exception to re-raise.
N�failed�done�error)�writer r
�append�
isinstancer � explanationr r r r r �msg�
UpstreamError) �events�writer�errors�resultsr& �get_name�
fail_check�error_to_reraise�obj�result� exceptionr r r �parallel_execute_watch, s0
�
r2 c C s� t | �} tj}t�t|��}| D ]
} |�||| �� q| D ]
} |�||| �� q t| |||�}
i } g } t|
|| | |||�}
| � � D ]
\}}|�
d� ||�� qA|
rS|
�| | fS )a� Runs func on objects in parallel while ensuring that func is
ran on object only after it is ran on all its dependencies.
get_deps called on object must return a collection with its dependencies.
get_name called on object must return its name.
fail_check is an additional failure check for cases that should display as a failure
in the CLI logs, but don't raise an exception (such as attempting to start 0 containers)
z
ERROR: for {} {}
) �list�sys�stderr�ParallelStreamWriter�get_or_assign_instance�
add_object�
write_initial�parallel_execute_iterr2 �itemsr" �format)�objects�funcr, r& �get_deps�limitr- �streamr) r/ r( r* r+ r. �obj_namer! r r r �parallel_executeM s$ �rC c C s g S r r )�xr r r �_no_depsq � rE c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �Statez�
Holds the state of a partially-complete parallel operation.
state.started: objects being processed
state.finished: objects which have been processed
state.failed: objects which either failed or whose dependencies failed
c C s" || _ t� | _t� | _t� | _d S r )r= �set�started�finishedr )�selfr= r r r �__init__} s zState.__init__c C s t | j�t | j� t | j�kS r )�lenrJ r r= �rK r r r �is_done� s z
State.is_donec C s t | j�| j | j | j S r )rH r= rI rJ r rN r r r �pending� s z
State.pendingN)r r r r rL rO rP r r r r rG u s
rG c @ s e Zd Zdd� Zdd� ZdS )�NoLimitc C � d S r r rN r r r � __enter__� rF zNoLimit.__enter__c G rR r r )rK �exr r r �__exit__� rF zNoLimit.__exit__N)r r r rS rU r r r r rQ � s rQ c c s� � |du rt }|du rt� }nt|�}t� }t| �} t| |||||� z|jdd�}W n ty5 Y q tj y? t
� �w |t u rFdS |\}} }
|
du r^t �
d�|�� |j�|� nt �
d�|�� |j�|� |V q)a�
Runs func on objects in parallel while ensuring that func is
ran on object only after it is ran on all its dependencies.
Returns an iterator of tuples which look like:
# if func returned normally when run on object
(object, result, None)
# if func raised an exception when run on object
(object, None, exception)
# if func raised an exception when run on one of object's dependencies
(object, None, UpstreamError())
NTg�������?)�timeoutzFinished processing: {}z
Failed: {})rE rQ r r rG �
feed_queue�getr �threadr! r �STOP�log�debugr< rJ �addr ) r= r> r? r@ �limiterr+ �state�eventr/ �_r1 r r r r: � s6 � �
�r: c C s� |�M t j�1 z|| �}|�| |df� W n ty/ } z|�| d|f� W Y d}~nd}~ww W d � n1 s:w Y W d � dS W d � dS 1 sRw Y dS )z�
The entry point for a producer thread which runs func on a single object.
Places a tuple on the results queue once func has either returned or raised.
N)r r �put� Exception)r/ r> r+ r^ r0 �er r r �producer� s ���P�re c sn �� � }t�d�|�� |D ]�}||�}zPt�fdd�|D ��r8t�d�|�� |�|dt� f� �j�|� n+t � �fdd�|D ��rct�d�|�� t
t ||||fd�} d | _ | �
� �j�|� W q ttfy� }
zt�d
�|�� |�|d|
f� W Y d}
~
qd}
~
w ty� }
zt�d �|�� |�|d|
f� W Y d}
~
qd}
~
ww ��� r�|�t� dS dS ) a)
Starts producer threads for any objects which are ready to be processed
(i.e. they have no dependencies which haven't been successfully processed).
Shortcuts any objects whose dependencies have failed and places an
(object, None, UpstreamError()) tuple on the results queue.
z Pending: {}c 3 s � | ]
}|d � j v V qdS )r N)r )�.0�dep)r_ r r � <genexpr>� s � z feed_queue.<locals>.<genexpr>z'{} has upstream errors - not processingNc 3 s4 � | ]\}}|� vp|�j v o| p||�V qd S r )rJ )rf rg � ready_check�r= r_ r r rh � s � �
�z Starting producer thread for {})�target�argsTzAHealthcheck for service(s) upstream of {} failed - not processingzIService(s) upstream of {} did not completed successfully - not processing)rP r[ r\ r< �anyrb r' r r] �allr re �daemon�startrI r r r rO rZ ) r= r> r? r+ r_ r^ rP r/ �deps�trd r rj r rW � sH � �� �� ���rW c @ s e Zd ZdS )r' N)r r r r r r r r' s r' c @ s| e Zd ZdZejZe� ZdZ e� Z
e dd� �Z e dd� �Z
e dd� �Zdd d
�Zd d � Zd
d� Zdd� Zdd� Zdd� ZdS )r6 z�Write out messages for operations happening in parallel.
Each operation has its own line, and ANSI code characters are used
to jump to the correct line, and write over the line.
Nc C s | j S r )�instance)r r r r � get_instance s z!ParallelStreamWriter.get_instancec C s: | j �� z| jd u r|| _| jW | j �� S | j �� w r )�
instance_lock�acquirers �release)r r) r r r r7 s
z+ParallelStreamWriter.get_or_assign_instancec C s
|| _ d S r )�default_ansi_mode)r � ansi_moder r r �set_default_ansi_mode s
z*ParallelStreamWriter.set_default_ansi_modec C s0 |d u r| j }|| _|�|�| _g | _d| _d S )Nr )rx rA �use_ansi_codes�lines�width)rK rA ry r r r rL $ s
z ParallelStreamWriter.__init__c C s: |d u rd S | j �|| � t| jt|d | ��| _d S )N� )r| r# �maxr} rM �rK r& � obj_indexr r r r8 , s z ParallelStreamWriter.add_objectc C s |d u rd S | � ||d�S )N� )�
_write_noansir� r r r r9 2 s z"ParallelStreamWriter.write_initialc C s� | j �� | j�|| �}t| j�| }| j�dd|f � | j�dd � | j�dj|d | || jd�� | j�dd|f � | j� � | j �
� d S )Nz%c[%dA� z%c[2K
z{:<{width}} ... {}
r~ �r} z%c[%dB) �
write_lockrv r| �indexrM rA r" r< r} �flushrw )rK r& r� �status�position�diffr r r � _write_ansi7 s
�
z ParallelStreamWriter._write_ansic C s0 | j �dj|d | || jd�� | j �� d S )Nz{:<{width}} ... {}
r~ r� )rA r" r< r} r� )rK r& r� r� r r r r� F s ��z"ParallelStreamWriter._write_noansic C s: |d u rd S | j r| �||||�� d S | �|||� d S r )r{ r� r� )rK r& r� r� �
color_funcr r r r" N s
zParallelStreamWriter.writer )r r r r r �AUTOrx r r� rs ru r rt r7 rz rL r8 r9 r� r� r" r r r r r6 s$
r6 c C s& t | tj|fi |��t�d�|� d S )N�name)rC �operator� methodcaller�
attrgetter)�
containers� operation�options�messager r r �parallel_operationW s �r� c C s dd� | D �}t |d|d� d S )Nc S s g | ]}|j s|�qS r )�
is_running)rf �cr r r �
<listcomp>a s z#parallel_remove.<locals>.<listcomp>�remove�Removing�r� )r� r� �stopped_containersr r r �parallel_remove` s r� c C � t | d|d� d S )N�pause�Pausingr� �r� r� r r r �parallel_pausee � r� c C r� )N�unpause� Unpausingr� r� r r r �parallel_unpausei r� r� c C r� )N�kill�Killingr� r� r r r �
parallel_killm r� r� )NNN)2�_threadrY �loggingr� r4 �queuer r � threadingr r r �
docker.errorsr r �compose.cli.colorsr r
r �compose.cli.signalsr �
compose.constr
�compose.errorsr r r r � getLoggerr r[ �objectrZ r r2 rC rE rG rQ r: re rW rc r' r6 r� r� r� r� r� r r r r �<module> sL
!$4
-R