a )gG@sddlmZmZmZeZddlZddlZddlZddl Z ddl Z ddl Z ddl mZddl mZddlmZddlmZddlmZddlmZdd lmZdd lmZmZdd lmZdd l m!Z!dd l"m#Z#m$Z$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1ddl2mZ3dgZ4e/Z5GdddZ6GdddZ7Gddde j8j9Z:Gddde;ZszDisplaySend.__init__Nr$r!r!r!r"r(=sr(cs4eZdZfddZddZddZddZZS) FinalQueuecs"t|d<tt|j|i|dS)Nctx)multiprocessing_contextsuperr*r#r) __class__r!r"r#DszFinalQueue.__init__cOs$|jt|g|Ri|dddSNFblock)putrrr!r!r" send_callbackHszFinalQueue.send_callbackcOs8t|dtr|d}nt|i|}|j|dddS)NrFr1) isinstancer r3)r rrtrr!r!r"send_task_resultNs zFinalQueue.send_task_resultcOs|jt|i|dddSr0)r3r(r)r!r!r" send_displayXs zFinalQueue.send_display)r%r&r'r#r4r7r8 __classcell__r!r!r.r"r*Cs  r*c@seZdZddZdS)AnsibleEndPlaycCs ||_dSr)result)r r;r!r!r"r#`szAnsibleEndPlay.__init__Nr$r!r!r!r"r:_sr:c@seZdZdZdZdZdZdZdZdZ d)d d Z d dZ ddZ ddZ ddZddZddZddZddZddZdd Zd!d"Zd#d$Zed%d&d'd(ZdS)*rao 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. rNTFc Cs||_||_||_t|_||_||_||_||_|p8d|_ d|_ g|_ d|_ t jddrzt jdD]} | rft| qfd|_t|_t|_z t|_Wn4ty} ztdt| WYd} ~ n d} ~ 00t|_t|_ dS)NFZ 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_donerCLIARGSgetrZ add_directory _terminateddict _failed_hosts_unreachable_hostsr*_final_qOSErrorrr threadingLock_callback_locktempfileZ TemporaryFile_connection_lockfile) r inventoryvariable_managerloaderrEZstdout_callbackZrun_additional_callbacksZrun_treeZforkspather!r!r"r#ws2   & zTaskQueueManager.__init__cCs$g|_t|D]}|jdqdSr)_workersrangeappend)r Znumir!r!r"_initialize_processess z&TaskQueueManager._initialize_processesc CsT|jr dSd}|jdur tj|_t|jtr2d}nLt|jtrv|jtvrXtd|jq~t |j|_|j d}ntdt tj dd}tj D]8}tj |dd}|r||vr||qtd|q|D]t}t|dd }t|d t|d d}t|d g}|r|d } ntjtj|j\} } td| |dkrn| |jksX|rhtd| qd}n8| dkr|jrn$|jr|rtj dus| tj vrqzX|} | r| |jvr| |j| ntd|| fntd| WqWqtyF} z.td| t| fWYd} ~ qWYd} ~ qd} ~ 00qd|_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_TYPEZCALLBACK_NEEDS_ENABLEDZCALLBACK_NEEDS_WHITELISTZ_redirected_namesrz 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)rJrFCZDEFAULT_STDOUT_CALLBACKr5rr rrrNZ set_optionslistallZCALLBACKS_ENABLEDradisplaywarninggetattrosr]splitextbasenameZ_original_pathZvvvvvZvvrHrGrK Exceptionr ) r Zstdout_callback_loadedZ callback_listcZplugincallback_pluginZ callback_typeZcallback_needs_enabledZcnamesZ callback_name_Z callback_objr^r!r!r"load_callbackssn              $zTaskQueueManager.load_callbacksc Cs|js||jj|d}t|j|d}t||jj | }| || |j |_ t|j|j|jd|_t||j|j}|jrt|jdr|j||jD]}t|dr||q|d|t|j|||j||jd}|t|j|jt !|j"|}|dur,t#d|j"|j$d |j% D]} |j&| } |'| q6|j( D]} |j)j*+| q^|,t-j.!d dur|j/durd |_z |0||} W|1|2n|1|20|3D]} d |j%| <q|j4rt5| | 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)rZr[r\set_play_contextZv2_playbook_on_play_start)rZrt play_contextr[all_varsZ start_at_doneNz#Invalid play strategy specified: %s)obj start_at_taskT)6rJrsrCZget_varsrrDrZ environmentglobalskeyscopyZ post_validateZcompile_roles_handlershandlersrrBZhostvarsrrErYfilenorFhasattrrurKr4rrLrcminrIZ batch_sizerrNstrategyrZ_dsrQZget_hostZmark_host_failedrRZ_playZ_removed_hostsraclear_failed_hostsrrMryruncleanup_cleanup_processesZget_failed_hostsZend_playr:) r rtrwZtemplarZnew_playrvrqiteratorrZ host_namehostZ play_returnr!r!r"rsj              zTaskQueueManager.runcCs<td||j|tjtj dS)NzRUNNING CLEANUP) ridebug terminaterSclosersysreflushstderrr r!r!r"r[s    zTaskQueueManager.cleanupc Cst|drttjdddD]8}tdd|jDs8qV|rJttjqt dq|jD]0}|r\| r\z | Wq\t yYq\0q\dS)Nr_r<css|]}|o|VqdSr)is_alive).0 worker_prcr!r!r" hz6TaskQueueManager._cleanup_processes..zFOne or more worker processes are still running and will be terminated.)rr`rfZWORKER_SHUTDOWN_POLL_COUNTanyr_timesleepZWORKER_SHUTDOWN_POLL_DELAYrirjrrAttributeError)r Zattempts_remainingrr!r!r"res      z#TaskQueueManager._cleanup_processescCs t|_dSr)rPrQrr!r!r"rwsz#TaskQueueManager.clear_failed_hostscCs|jSr)rBrr!r!r" get_inventoryzszTaskQueueManager.get_inventorycCs|jSr)rCrr!r!r"get_variable_manager}sz%TaskQueueManager.get_variable_managercCs|jSr)rDrr!r!r" get_loaderszTaskQueueManager.get_loadercCs|jddSr)r_rr!r!r" get_workersszTaskQueueManager.get_workerscCs d|_dS)NT)rOrr!r!r"rszTaskQueueManager.terminatecCs$d}|jD]}t|ddr d}q |S)NFZexitcodeT)r_rk)r Zdefunctxr!r!r"has_dead_workerss   z!TaskQueueManager.has_dead_workersrW)attrc Osl|jg|jD]V}t|ddr"qt|dd}g}|dfD]<}t||d}|durdt||dd}|dur:||q:g} d} |D]<} t| tr| | n | | t| tr| j rd} q| r|sq|D]} z| | i|Wqt yb} zbt dt |t |t | fdd lm}dd lm}t d d ||d WYd} ~ qd} ~ 00qqdS)NZdisabledFwants_implicit_tasksZ v2_on_anyZv2_Tz5Failure using method (%s) in callback plugin (%s): %sr) format_tb)exc_infozCallback Exception:  r=)rFrKrk removeprefixrar5r Z clean_copyrZimplicitrorirjr tracebackrrrZvvvjoin)r rrrrqrmethodspossiblegotitnew_argsZis_implicit_taskargmethodr^rrr!r!r"r4s:          zTaskQueueManager.send_callback)NTFN)r%r&r'__doc__ZRUN_OKZ RUN_ERRORZRUN_FAILED_HOSTSZRUN_UNREACHABLE_HOSTSZRUN_FAILED_BREAK_PLAYZRUN_UNKNOWN_ERRORr#rcrsrrrrrrrrrrrr4r!r!r!r"rds,  (]Y  )>Z __future__rrrtypeZ __metaclass__rlrrXrUrZmultiprocessing.queuesZmultiprocessingZansiblerrfrZansible.errorsrZansible.executor.play_iteratorrZansible.executor.statsr Zansible.executor.task_resultr Zansible.module_utils.sixr Zansible.module_utils._textr r Zansible.playbook.play_contextrZansible.playbook.taskrZansible.plugins.loaderrrrZansible.plugins.callbackrZansible.templaterZansible.vars.hostvarsrZansible.vars.reservedrZansible.utils.displayrZansible.utils.lockrZansible.utils.multiprocessingr,__all__rirr(ZqueuesZQueuer*ror:rr!r!r!r"s@