cndZddlmZddlZddlZddlZddlZddlZddlZ ddl Z n#e $rddl Z YnwxYwddl Z ddl mZddlmZejdZdZdZGdd Zd Zed kr edSdS) a/Worker ("slave") process used in computing distributed Latent Dirichlet Allocation (LDA, :class:`~gensim.models.ldamodel.LdaModel`). Run this script on every node in your cluster. If you wish, you may even run it multiple times on a single machine, to make better use of multiple cores (just beware that memory footprint increases accordingly). How to use distributed :class:`~gensim.models.ldamodel.LdaModel` ---------------------------------------------------------------- #. 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.lda_worker & #. Run dispatcher :: python -m gensim.models.lda_dispatcher & #. Run :class:`~gensim.models.ldamodel.LdaModel` in distributed mode : .. sourcecode:: pycon >>> from gensim.test.utils import common_corpus, common_dictionary >>> from gensim.models import LdaModel >>> >>> model = LdaModel(common_corpus, id2word=common_dictionary, distributed=True) Command line arguments ---------------------- .. program-output:: python -m gensim.models.lda_worker --help :ellipsis: 0, -7 )with_statementN)ldamodel)utilszgensim.models.lda_workerzgensim.lda_workerc|eZdZdZdZejdZejejdZ e j ddZ ejdZ eje j ddZeje j dd Zejd Zd S) WorkerzUsed as a Pyro4 class with exposed methods. Exposes every non-private method and property of the class automatically to be available for remote access. cd|_dS)zPartly initialize the model.N)modelselfs 8lib/python3.11/site-packages/gensim/models/lda_worker.py__init__zWorker.__init__Xs  c tj|_d|_||_||_d|_td|tj di||_ dS)aFully initialize the worker. Parameters ---------- myid : int An ID number used to identify this worker in the dispatcher object. dispatcher : :class:`~gensim.models.lda_dispatcher.Dispatcher` The dispatcher responsible for scheduling this worker. **model_params Keyword parameters to initialize the inner LDA model,see :class:`~gensim.models.ldamodel.LdaModel`. rFzinitializing worker #%sN) threadingLock lock_updatejobsdonemyid dispatcherfinishedloggerinforLdaModelr )r rr model_paramss r initializezWorker.initialize\sa%>++  $  -t444&6666 rc|jtdd}|F|js? |j|j}n#t j$rY;wxYw||j?|\t d|j|j | ||j |jdSt d|jdS)aRequest jobs from the dispatcher, in a perpetual loop until :meth:`gensim.models.lda_worker.Worker.getstate` is called. Raises ------ RuntimeError If `self.model` is None (i.e. worker non initialized). Nz0worker must be initialized before receiving jobszworker #%s received job #%iz#worker #%i stopping asking for jobs) r RuntimeErrorrrgetjobrQueueEmptyrrr processjobjobdone)r jobs r requestjobzWorker.requestjobss : SQRR R $-  o,,TY77;     $-   J KK5ty$- P P P OOC O # #DI . . . . . KK=ty I I I I IsAAArctd|j|j||xjdz c_t r^|jt zdkrKt jtj d}|j |t d|jdz dS)zIncrementally process the job and potentially logs progress. Parameters ---------- job : iterable of list of (int, float) Corpus in BoW format. zstarting to process job #%irzlda_worker.pklzfinished processing job #%iN) rdebugrr do_estep SAVE_DEBUGospathjointempfile gettempdirsaver)r r$fnames r r"zWorker.processjobs  2DMBBB C      #$-*49 #GLL!4!6!68HIIE JOOE " " " 14=13DEEEEErcdS)z"Test the connectivity with Worker.Trr s r pingz Worker.pings trctd|j|j|jj}t |tjsJ|j d|_ |S)zLog and get the LDA model's current state. Returns ------- result : :class:`~gensim.models.ldamodel.LdaState` The current state. z,worker #%i returning its state after %s jobsT) rrrrr state isinstancerLdaStateclearr)r results r getstatezWorker.getstatesb  BDIt}]]]!&("344444   rc|Jtd|j||j_|j|jjd|_dS)zReset the worker by setting sufficient stats to 0. Parameters ---------- state : :class:`~gensim.models.ldamodel.LdaState` Encapsulates information for distributed computation of LdaModel objects. Nzresetting worker #%iF)rrrr r5 sync_stateresetr)r r5s r r=z Worker.resetsg    *DI666        rcntd|jtjddS)zTerminate the worker.zterminating worker #%irN)rrrr+_exitr s r exitz Worker.exits,  ,di888  rN)__name__ __module__ __qualname____doc__r Pyro4exposeronewayr%r synchronousr"r3r:r=r@rrr rrQsF  \77\7, \ \JJ\\J6U}%%FF&%F" \\ \U}%%&%\  \U}%%&%\  \\rrc Vtjtddtj}|ddd|dddt |d d d d d |ddd|dddd dt jt j| }t j d|j t ddtj|j|j|j|jd}t+jt.t1d |t ddtjdS)Ni~) descriptionformatter_classz--hostz*Nameserver hostname (default: %(default)s))helpdefaultz--portz&Nameserver port (default: %(default)s))rLrMtypez--no-broadcastz(Disable broadcast (default: %(default)s) store_constTF)rLactionrMconstz--hmacz*Nameserver hmac key (default: %(default)s)z-vz --verbosez Verbose flagloglevel)rLrPdestrQrMz)%(asctime)s : %(levelname)s : %(message)s)formatlevelz running %s ) broadcasthostporthmac_key) random_suffixns_confzfinished running %s)argparseArgumentParserrDRawTextHelpFormatter add_argumentintloggingINFOWARNING parse_args basicConfigrRrrr-sysargv no_broadcastrXrYhmacr pyro_daemonLDA_WORKER_PREFIXr)parserargsr\s r mainros  $$QYQn o o oF 'S]abbb 'OY]dghhh IR_E 'S]abbb  k}:lGO     D JRVR_```` KK chhsx00111&  I G  'wWWWW KK%sxx'9'9:::::r__main__)rD __future__rr+rgrbrr.r]r ImportErrorqueuerE gensim.modelsrgensimr getLoggerrr*rlrrorArrr rwsO//b&%%%%% LLLL """"""  5 6 6 '~~~~~~~~B;;;8 z DFFFFF  s ' 33