c&FdZddlZddlZddlZddlZddlZddlZddlmZddl Z ddl m Z ej e ZdZdZGddZe d krejd ej ejedd ej ZeddedeeZeddeje jdeejedej dSdS)aDispatcher process which orchestrates distributed :class:`~gensim.models.lsimodel.LsiModel` computations. Run this script only once, on any node in your cluster. Notes ----- The dispatcher expects to find worker scripts already running. Make sure you run as many workers as you like on your machines **before** launching the dispatcher. How to use distributed LSI -------------------------- #. Install needed dependencies (Pyro4) :: pip install gensim[distributed] #. Setup serialization (on each machine) :: export PYRO_SERIALIZERS_ACCEPTED=pickle export PYRO_SERIALIZER=pickle #. Run nameserver :: python -m Pyro4.naming -n 0.0.0.0 & #. Run workers (on each machine) :: python -m gensim.models.lsi_worker & #. Run dispatcher :: python -m gensim.models.lsi_dispatcher & #. Run :class:`~gensim.models.lsimodel.LsiModel` in distributed mode: .. sourcecode:: pycon >>> from gensim.test.utils import common_corpus, common_dictionary >>> from gensim.models import LsiModel >>> >>> model = LsiModel(common_corpus, id2word=common_dictionary, distributed=True) Command line arguments ---------------------- .. program-output:: python -m gensim.models.lsi_dispatcher --help :ellipsis: 0, -5 N)Queue)utils i3cfeZdZdZddZejdZejdZejdZ ejdZ ejdZ ejd Z ejej ejd d Zd Zej d ZdS) DispatcherzDispatcher object that communicates and coordinates individual workers. Warnings -------- There should never be more than one dispatcher running at any one time. rc0||_i|_d|_dS)aZPartly initialize the dispatcher. A full initialization (including initialization of the workers) requires a call to :meth:`~gensim.models.lsi_dispatcher.Dispatcher.initialize` Parameters ---------- maxsize : int, optional Maximum number of jobs to be kept pre-fetched in the queue. N)maxsizeworkerscallback)selfr s ++ []] $b!K(HIIDMWW,?W@@FFHH $ $ c $"[--F"4<00HKK @(CPPP%F%hYY4=YLYYY-3DL**|-$$$$$%bdghhhIIdOOOOO$ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $| hfgg g h hs9AE#A#DEAE E EEE"%E"cHd|jDS)zGet pyro URIs of all registered workers. Returns ------- list of URIs The pyro URIs for each worker. cg|] }|j S)_pyroUri).0r,s r z)Dispatcher.getworkers..sDDDFDDDr)r valuesr s r getworkerszDispatcher.getworkerss'EDdl.A.A.C.CDDDDrctd||jdd}td||j|S)aAtomically pop a job from the queue. Parameters ---------- worker_id : int The worker that requested the job. Returns ------- iterable of iterable of (int, float) The corpus in BoW format. zworker #%i requesting a new jobTblocktimeoutz"worker #%i got a new job (%i left))r r!rgetqsize)r worker_idjobs r getjobzDispatcher.getjobsX  5yAAAimm$m22 8)TY__EVEVWWW rc|xjdz c_|j|dttd|jdS)zAtomically add a job to the queue. Parameters ---------- job : iterable of list of (int, float) The corpus in BoW format. r8Tr9z%added a new job (len(queue)=%i items)N)rrput HUGE_TIMEOUTr r!r=)r r?s r putjobzDispatcher.putjobsY a c| <<< ;TY__=N=NOOOOOrctdtd|j|j|j|jkr$t jd|j|jk$tdt|jt|j }|dd }|ddD]G\}}td|| | Htd |S) zMerge projections from across all workers and get the final projection. Returns ------- :class:`~gensim.models.lsimodel.Projection` The current projection of the total model. z*end of input, assigning all remaining jobsz jobs done: %s, jobs received: %sg?zmerging states from %i workersrr8Nzpulling state from worker %szsending out merged projection) r r!debugrrtimesleeprr rrgetstatemerge)r r resultr-r,s r rIzDispatcher.getstates#  @AAA 7I[\\\nt11  JsOOOnt11   4c$,6G6GHHHt|))++,,A'')) '  , , Hf KK6 A A A LL** + + + + 3444 rc|jD]H\}}td|||Id|_d|_dS)z2Re-initialize all workers for a new decomposition.zresetting worker %srN)r rr r!reset requestjobrrr r-r,s r rMzDispatcher.resetsr!% 2 2 4 4  Hf KK-x 8 8 8 LLNNN       rrc|xjdz c_td||j|j|}|dS)aA worker has finished its job. Log this event and then asynchronously transfer control back to the worker. Callback used by workers to notify when their job is done. The job done event is logged and then control is asynchronously transfered back to the worker (who can then request another job). In this way, control flow basically oscillates between :meth:`gensim.models.lsi_dispatcher.Dispatcher.jobdone` and :meth:`gensim.models.lsi_worker.Worker.requestjob`. Parameters ---------- workerid : int The ID of the worker that finished the job (used for logging). r8zworker #%s finished job #%iN)rr r!r rNrOs r jobdonezDispatcher.jobdonesQ$ ! 18T^LLLh'rc|jS)zWrap :attr:`~gensim.models.lsi_dispatcher.Dispatcher._jobsdone`, needed for remote access through proxies. Returns ------- int Number of jobs already completed. )rr5s r jobsdonezDispatcher.jobsdones ~rc|jD]4\}}td||5tdt jddS)z9Terminate all registered workers and then the dispatcher.zterminating worker %szterminating dispatcherrN)r rr r!exitos_exitrOs r rUzDispatcher.exit sn!% 2 2 4 4   Hf KK/ : : : KKMMMM ,---  rN)r)__name__ __module__ __qualname____doc__rrexposer"r6r@rDrIrMonewayr synchronousrQrSrUr0rr rrTsW  \&h&h\&hP \ E E\ E \\& \ P P\ P \\6 \\ \ \U}%%&%\\(    \\rr__main__z)%(asctime)s - %(levelname)s - %(message)s)formatleveliy) descriptionformatter_classr ?z;Maximum number of jobs to be kept pre-fetched in the queue.)nargstypehelpdefaultz running %s zgensim.lsi_dispatcherrzfinished running %s)!r[rVsysloggingargparserrGqueuerrgensimr getLoggerrXr MAX_JOBS_QUEUErCr basicConfigINFOArgumentParserRawTextHelpFormatterparser add_argumentint parse_argsargsr!joinargv pyro_daemonr progr0rr r~s00d     8 $ $  " ~~~~~~~~B z4GJRYR^____ $X $$QYQn o o oF   J      D KK chhsx00111E-zz$,/O/O/OPPP KK%v{3333344r