Google
Edit File: task_queue_manager.cpython-39.pyc
a �)g�G � @ s� d dl mZmZmZ eZd dlZd dlZd dlZd dl Z d dl Z d dl Z d dl mZ d dl mZ d dlmZ d dlmZ d dlmZ d dlmZ d d lmZ d d lm Z m Z d d l m Z d d l m!Z! d d l"m#Z#m$Z$m%Z% d dl&m'Z' d dl(m)Z) d dl*m+Z+ d dl,m-Z- d dl.m/Z/ d dl0m1Z1 d dl2mZ3 dgZ4e/� Z5G dd� d�Z6G dd� d�Z7G dd� de j8j9�Z:G dd � d e;�Z<G d d� d�Z=dS ) � )�absolute_import�division�print_functionN)� constants)�context)� AnsibleError)� PlayIterator)�AggregateStats)� TaskResult)� string_types)�to_text� to_native)� PlayContext)�Task)�callback_loader�strategy_loader� module_loader)� CallbackBase)�Templar)�HostVars)�warn_if_reserved)�Display)�lock_decorator�TaskQueueManagerc @ s e Zd Zdd� ZdS )� CallbackSendc O s || _ || _|| _d S �N)� method_name�args�kwargs��selfr r r � r! �G/usr/lib/python3.9/site-packages/ansible/executor/task_queue_manager.py�__init__7 s zCallbackSend.__init__N��__name__� __module__� __qualname__r# r! r! r! r" r 6 s r c @ s e Zd Zdd� ZdS )� DisplaySendc O s || _ || _d S r )r r �r r r r! r! r" r# > s zDisplaySend.__init__Nr$ r! r! r! r" r( = s r( c s4 e Zd Z� fdd�Zdd� Zdd� Zdd� Z� ZS ) � FinalQueuec s" t |d< tt| �j|i |�� d S )N�ctx)�multiprocessing_context�superr* r# r) �� __class__r! r" r# D s zFinalQueue.__init__c O s$ | j t|g|�R i |��dd� d S �NF��block)�putr r r! r! r" � send_callbackH s �zFinalQueue.send_callbackc O s8 t |d t�r|d }nt|i |��}| j|dd� d S )Nr Fr1 )� isinstancer r3 )r r r �trr! r! r" �send_task_resultN s �zFinalQueue.send_task_resultc O s | j t|i |��dd� d S r0 )r3 r( r) r! r! r" � send_displayX s �zFinalQueue.send_display)r% r& r' r# r4 r7 r8 � __classcell__r! r! r. r" r* C s r* c @ s e Zd Zdd� ZdS )�AnsibleEndPlayc C s || _ d S r )�result)r r; r! r! r" r# ` s zAnsibleEndPlay.__init__Nr$ r! r! r! r" r: _ s r: c @ s� e Zd ZdZdZdZdZdZdZdZ d)d d �Z d d� Z dd� Z dd� Z dd� Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Zed%d&�d'd(� �ZdS )*r ao This class handles the multiprocessing requirements of Ansible by creating a pool of worker forks, a result handler fork, and a manager object with shared datastructures/queues for coordinating work between all processes. The queue manager is responsible for loading the play strategy plugin, which dispatches the Play's tasks to hosts. r � � � � � NTFc C s� || _ || _|| _t� | _|| _|| _|| _|| _|p8d| _ d| _ g | _ d| _ t j�dd�rzt jd D ]} | rft�| � qfd| _t� | _t� | _z t� | _W n4 ty� } z tdt| � ��W Y d } ~ n d } ~ 0 0 t� � | _ t � � | _ d S )N� FZ module_pathzXUnable to use multiprocessing, this is normally caused by lack of access to /dev/shm: %s)!� _inventory�_variable_manager�_loaderr Z_stats� passwords�_stdout_callback�_run_additional_callbacks� _run_tree�_forks�_callbacks_loaded�_callback_plugins�_start_at_doner �CLIARGS�getr Z add_directory� _terminated�dict� _failed_hosts�_unreachable_hostsr* �_final_q�OSErrorr r � threading�Lock�_callback_lock�tempfileZ TemporaryFile�_connection_lockfile) r � inventory�variable_manager�loaderrE Zstdout_callbackZrun_additional_callbacksZrun_treeZforks�path�er! r! r" r# w s2 & zTaskQueueManager.__init__c C s$ g | _ t|�D ]}| j �d � qd S r )�_workers�range�append)r Znum�ir! r! r" �_initialize_processes� s z&TaskQueueManager._initialize_processesc C sT | j r dS d}| jdu r tj| _t| jt�r2d}nLt| jt�rv| jtvrXtd| j ��q~t� | j�| _| j� � d}ntd��t tj dd��}tj D ]8}tj |dd�}|r�||vr�|�|� q�t�d| � q�|D �]t}t|dd �}t|d t|d d��}t|d g �}|�r|d } ntj�tj�|j��\} } t�d| � |dk�rn| | jk�sX|�rht�d| � q�d}n8| dk�r�| j�r�n$| jr�|�r�tj du s�| tj v�r�q�zX|� } | �r�| | jv�r�| � � | j�| � nt�d|| f � nt�d| � W q�W q� t �yF } z.t�d| t | �f � W Y d} ~ q�W Y d} ~ q�d} ~ 0 0 q�d| _ dS )z� Loads all available callbacks, with the exception of those which utilize the CALLBACK_TYPE option. When CALLBACK_TYPE is set to 'stdout', only one such callback plugin will be loaded. NFTz)Invalid callback for stdout specified: %szMcallback must be an instance of CallbackBase or the name of a callback plugin)Z class_onlyz-Skipping callback plugin '%s', unable to loadZ CALLBACK_TYPE� ZCALLBACK_NEEDS_ENABLEDZCALLBACK_NEEDS_WHITELISTZ_redirected_namesr z Attempting to use '%s' callback.�stdoutz=Skipping callback '%s', as we already have a stdout callback.Ztreez/Skipping callback '%s', already loaded as '%s'.zFSkipping callback '%s', as it does not create a valid plugin instance.z1Skipping callback '%s', unable to load due to: %s) rJ rF �CZDEFAULT_STDOUT_CALLBACKr5 r r r r rN Z set_options�list�allZCALLBACKS_ENABLEDra �display�warning�getattr�osr] �splitext�basenameZ_original_pathZvvvvvZvvrH rG rK � Exceptionr ) r Zstdout_callback_loadedZ callback_list�cZplugin�callback_pluginZ callback_typeZcallback_needs_enabledZcnamesZ callback_name�_Z callback_objr^ r! r! r" �load_callbacks� sn �� $z TaskQueueManager.load_callbacksc C s | j s| �� | jj|d�}t| j|d�}t||jj� � � |� � }|� |� |� � |j |_ t| j| j| jd�| _t|| j| j�� �}| jr�t| jd�r�| j�|� | jD ]}t|d�r�|�|� q�| �d|� t| j||| j|| jd�}| � t | j |j �� t �!|j"| �}|du �r,t#d|j" |j$d ��| j%� � D ] } | j�&| �} |�'| � �q6| j(� � D ]} |j)j*�+| � �q^| �,� t-j.�!d �du�r�|j/du �r�d | _z |�0||�} W |�1� | �2� n|�1� | �2� 0 |�3� D ]} d | j%| < �q�|j4�r�t5| ��| S ) al Iterates over the roles/tasks in a play, using the given (or default) strategy for queueing tasks. The default is the linear strategy, which operates like classic Ansible by keeping all hosts in lock-step with a given task (meaning no hosts move on to the next task until all hosts are done with the current task). )�play)r\ Z variables)rZ r[ r\ �set_play_contextZv2_playbook_on_play_start)rZ rt � play_contextr[ �all_varsZ start_at_doneNz#Invalid play strategy specified: %s)�obj� start_at_taskT)6rJ rs rC Zget_varsr rD r Z environment�globals�keys�copyZ post_validateZcompile_roles_handlers�handlersr rB Zhostvarsr rE rY �filenorF �hasattrru rK r4 r rL rc �minrI Z batch_sizer rN �strategyr Z_dsrQ Zget_hostZmark_host_failedrR Z_playZ_removed_hostsra �clear_failed_hostsr rM ry �run�cleanup�_cleanup_processesZget_failed_hostsZend_playr: ) r rt rw ZtemplarZnew_playrv rq �iteratorr� Z host_name�hostZ play_returnr! r! r" r� sj � � � � zTaskQueueManager.runc C s< t �d� | �� | j�� | �� tj�� tj �� d S )NzRUNNING CLEANUP) ri �debug� terminaterS �closer� �sysre �flush�stderr�r r! r! r" r� [ s zTaskQueueManager.cleanupc C s� t | d�r�ttjd dd�D ]8}tdd� | jD ��s8 qV|rJt�tj� q t � d� q | jD ]0}|r\|� � r\z |� � W q\ t y� Y q\0 q\d S )Nr_ r< ���c s s | ]}|o|� � V qd S r )�is_alive)�.0� worker_prcr! r! r" � <genexpr>h � z6TaskQueueManager._cleanup_processes.<locals>.<genexpr>zFOne or more worker processes are still running and will be terminated.)r r` rf ZWORKER_SHUTDOWN_POLL_COUNT�anyr_ �time�sleepZWORKER_SHUTDOWN_POLL_DELAYri rj r� r� �AttributeError)r Zattempts_remainingr� r! r! r" r� e s z#TaskQueueManager._cleanup_processesc C s t � | _d S r )rP rQ r� r! r! r" r� w s z#TaskQueueManager.clear_failed_hostsc C s | j S r )rB r� r! r! r" � get_inventoryz s z TaskQueueManager.get_inventoryc C s | j S r )rC r� r! r! r" �get_variable_manager} s z%TaskQueueManager.get_variable_managerc C s | j S r )rD r� r! r! r" � get_loader� s zTaskQueueManager.get_loaderc C s | j d d � S r )r_ r� r! r! r" � get_workers� s z TaskQueueManager.get_workersc C s d| _ d S )NT)rO r� r! r! r" r� � s zTaskQueueManager.terminatec C s$ d}| j D ]}t|dd �r d}q |S )NFZexitcodeT)r_ rk )r Zdefunct�xr! r! r" �has_dead_workers� s z!TaskQueueManager.has_dead_workersrW )�attrc O sl | j g| j D �]V}t|dd�r"qt|dd�}g }|dfD ]<}t||d �}|d u rdt||�d�d �}|d ur:|�|� q:g } d} |D ]<} t| t�r�| �| �� � n | �| � t| t�r�| j r�d} q�| r�|s�q|D ]�} z| | i |�� W q� t �yb } zbt � dt |�t |�t | �f � dd lm} dd lm} t �d d �||� d �� � W Y d } ~ q�d } ~ 0 0 q�qd S )NZdisabledF�wants_implicit_tasksZ v2_on_anyZv2_Tz5Failure using method (%s) in callback plugin (%s): %sr )� format_tb)�exc_infozCallback Exception: � r= )rF rK rk � removeprefixra r5 r Z clean_copyr Zimplicitro ri rj r � tracebackr� r� r� Zvvv�join)r r r r rq r� �methods�possible�gotit�new_argsZis_implicit_task�arg�methodr^ r� r� r! r! r" r4 � s: z TaskQueueManager.send_callback)NTFN)r% r& r' �__doc__ZRUN_OKZ RUN_ERRORZRUN_FAILED_HOSTSZRUN_UNREACHABLE_HOSTSZRUN_FAILED_BREAK_PLAYZRUN_UNKNOWN_ERRORr# rc rs r� r� r� r� r� r� r� r� r� r� r r4 r! r! r! r" r d s, (]Y )>Z __future__r r r �typeZ __metaclass__rl r� rX rU r� Zmultiprocessing.queuesZmultiprocessingZansibler rf r Zansible.errorsr Z ansible.executor.play_iteratorr Zansible.executor.statsr Z ansible.executor.task_resultr Zansible.module_utils.sixr Zansible.module_utils._textr r Z ansible.playbook.play_contextr Zansible.playbook.taskr Zansible.plugins.loaderr r r Zansible.plugins.callbackr Zansible.templater Zansible.vars.hostvarsr Zansible.vars.reservedr Zansible.utils.displayr Zansible.utils.lockr Z ansible.utils.multiprocessingr, �__all__ri r r( ZqueuesZQueuer* ro r: r r! r! r! r" �<module> s@