We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
lsi_dispatcher is not working from command-line when not specifying maxsize argument
#### Problem description

When running `lsi_dispatcher` from the command-line, if you don't specify the `maxsize` argument explicitly, you get an error for the missing positional argument:

```
usage: lsi_dispatcher.py [-h] maxsize
lsi_dispatcher.py: error: the following arguments are required: maxsize
```

According to the documentation, this argument should be optional.

The issue seems to be that the nargs argument to `add_argument` is missing:

```python
    parser.add_argument(
        'maxsize', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE
    )
 ```
In order to make this argument optional, this should be:

```python
    parser.add_argument(
        'maxsize', nargs='?', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE
    )
```

#### Steps/code/corpus to reproduce

Include full tracebacks, logs and datasets if necessary. Please keep the examples minimal ("minimal reproducible example").

If your problem is with a specific Gensim model (word2vec, lsimodel, doc2vec, fasttext, ldamodel etc), include the following:

```python
$ python3 -m gensim.models.lsi_dispatcher
usage: lsi_dispatcher.py [-h] maxsize
lsi_dispatcher.py: error: the following arguments are required: maxsize
```

#### Versions

```python
Linux-5.4.0-67-generic-x86_64-with-glibc2.2
Python 3.8.5 (default, Jan 27 2021, 15:41:15) 
[GCC 9.3.0]
Bits 64
NumPy 1.19.4
SciPy 1.6.0
gensim 4.0.1
FAST_VERSION 1
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `gensim/models/lsi_dispatcher.py`
Content:
```
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 #
4 # Copyright (C) 2010 Radim Rehurek <radimrehurek@seznam.cz>
5 # Licensed under the GNU LGPL v2.1 - http://www.gnu.org/licenses/lgpl.html
6 
7 """Dispatcher process which orchestrates distributed :class:`~gensim.models.lsimodel.LsiModel` computations.
8 Run this script only once, on any node in your cluster.
9 
10 Notes
11 -----
12 The dispatcher expects to find worker scripts already running. Make sure you run as many workers as you like on
13 your machines **before** launching the dispatcher.
14 
15 
16 How to use distributed LSI
17 --------------------------
18 
19 #. Install needed dependencies (Pyro4) ::
20 
21     pip install gensim[distributed]
22 
23 #. Setup serialization (on each machine) ::
24 
25     export PYRO_SERIALIZERS_ACCEPTED=pickle
26     export PYRO_SERIALIZER=pickle
27 
28 #. Run nameserver ::
29 
30     python -m Pyro4.naming -n 0.0.0.0 &
31 
32 #. Run workers (on each machine) ::
33 
34     python -m gensim.models.lsi_worker &
35 
36 #. Run dispatcher ::
37 
38     python -m gensim.models.lsi_dispatcher &
39 
40 #. Run :class:`~gensim.models.lsimodel.LsiModel` in distributed mode:
41 
42     .. sourcecode:: pycon
43 
44         >>> from gensim.test.utils import common_corpus, common_dictionary
45         >>> from gensim.models import LsiModel
46         >>>
47         >>> model = LsiModel(common_corpus, id2word=common_dictionary, distributed=True)
48 
49 Command line arguments
50 ----------------------
51 
52 .. program-output:: python -m gensim.models.lsi_dispatcher --help
53    :ellipsis: 0, -5
54 
55 """
56 
57 import os
58 import sys
59 import logging
60 import argparse
61 import threading
62 import time
63 from queue import Queue
64 
65 import Pyro4
66 
67 from gensim import utils
68 
69 
70 logger = logging.getLogger(__name__)
71 
72 # How many jobs (=chunks of N documents) to keep "pre-fetched" in a queue?
73 # A small number is usually enough, unless iteration over the corpus is very very
74 # slow (slower than the actual computation of LSI), in which case you can override
75 # this value from command line. ie. run "python ./lsi_dispatcher.py 100"
76 MAX_JOBS_QUEUE = 10
77 
78 # timeout for the Queue object put/get blocking methods.
79 # it should really be infinity, but then keyboard interrupts don't work.
80 # so this is really just a hack, see http://bugs.python.org/issue1360
81 HUGE_TIMEOUT = 365 * 24 * 60 * 60  # one year
82 
83 
84 class Dispatcher:
85     """Dispatcher object that communicates and coordinates individual workers.
86 
87     Warnings
88     --------
89     There should never be more than one dispatcher running at any one time.
90 
91     """
92     def __init__(self, maxsize=0):
93         """Partly initialize the dispatcher.
94 
95         A full initialization (including initialization of the workers) requires a call to
96         :meth:`~gensim.models.lsi_dispatcher.Dispatcher.initialize`
97 
98         Parameters
99         ----------
100         maxsize : int, optional
101             Maximum number of jobs to be kept pre-fetched in the queue.
102 
103         """
104         self.maxsize = maxsize
105         self.workers = {}
106         self.callback = None  # a pyro proxy to this object (unknown at init time, but will be set later)
107 
108     @Pyro4.expose
109     def initialize(self, **model_params):
110         """Fully initialize the dispatcher and all its workers.
111 
112         Parameters
113         ----------
114         **model_params
115             Keyword parameters used to initialize individual workers
116             (gets handed all the way down to :meth:`gensim.models.lsi_worker.Worker.initialize`).
117             See :class:`~gensim.models.lsimodel.LsiModel`.
118 
119         Raises
120         ------
121         RuntimeError
122             When no workers are found (the :mod:`gensim.model.lsi_worker` script must be ran beforehand).
123 
124         """
125         self.jobs = Queue(maxsize=self.maxsize)
126         self.lock_update = threading.Lock()
127         self._jobsdone = 0
128         self._jobsreceived = 0
129 
130         # locate all available workers and store their proxies, for subsequent RMI calls
131         self.workers = {}
132         with utils.getNS() as ns:
133             self.callback = Pyro4.Proxy('PYRONAME:gensim.lsi_dispatcher')  # = self
134             for name, uri in ns.list(prefix='gensim.lsi_worker').items():
135                 try:
136                     worker = Pyro4.Proxy(uri)
137                     workerid = len(self.workers)
138                     # make time consuming methods work asynchronously
139                     logger.info("registering worker #%i from %s", workerid, uri)
140                     worker.initialize(workerid, dispatcher=self.callback, **model_params)
141                     self.workers[workerid] = worker
142                 except Pyro4.errors.PyroError:
143                     logger.exception("unresponsive worker at %s, deleting it from the name server", uri)
144                     ns.remove(name)
145 
146         if not self.workers:
147             raise RuntimeError('no workers found; run some lsi_worker scripts on your machines first!')
148 
149     @Pyro4.expose
150     def getworkers(self):
151         """Get pyro URIs of all registered workers.
152 
153         Returns
154         -------
155         list of URIs
156             The pyro URIs for each worker.
157 
158         """
159         return [worker._pyroUri for worker in self.workers.values()]
160 
161     @Pyro4.expose
162     def getjob(self, worker_id):
163         """Atomically pop a job from the queue.
164 
165         Parameters
166         ----------
167         worker_id : int
168             The worker that requested the job.
169 
170         Returns
171         -------
172         iterable of iterable of (int, float)
173             The corpus in BoW format.
174 
175         """
176         logger.info("worker #%i requesting a new job", worker_id)
177         job = self.jobs.get(block=True, timeout=1)
178         logger.info("worker #%i got a new job (%i left)", worker_id, self.jobs.qsize())
179         return job
180 
181     @Pyro4.expose
182     def putjob(self, job):
183         """Atomically add a job to the queue.
184 
185         Parameters
186         ----------
187         job : iterable of list of (int, float)
188             The corpus in BoW format.
189 
190         """
191         self._jobsreceived += 1
192         self.jobs.put(job, block=True, timeout=HUGE_TIMEOUT)
193         logger.info("added a new job (len(queue)=%i items)", self.jobs.qsize())
194 
195     @Pyro4.expose
196     def getstate(self):
197         """Merge projections from across all workers and get the final projection.
198 
199         Returns
200         -------
201         :class:`~gensim.models.lsimodel.Projection`
202             The current projection of the total model.
203 
204         """
205         logger.info("end of input, assigning all remaining jobs")
206         logger.debug("jobs done: %s, jobs received: %s", self._jobsdone, self._jobsreceived)
207         while self._jobsdone < self._jobsreceived:
208             time.sleep(0.5)  # check every half a second
209 
210         # TODO: merge in parallel, so that we're done in `log_2(workers)` merges,
211         # and not `workers - 1` merges!
212         # but merging only takes place once, after all input data has been processed,
213         # so the overall effect would be small... compared to the amount of coding :-)
214         logger.info("merging states from %i workers", len(self.workers))
215         workers = list(self.workers.items())
216         result = workers[0][1].getstate()
217         for workerid, worker in workers[1:]:
218             logger.info("pulling state from worker %s", workerid)
219             result.merge(worker.getstate())
220         logger.info("sending out merged projection")
221         return result
222 
223     @Pyro4.expose
224     def reset(self):
225         """Re-initialize all workers for a new decomposition."""
226         for workerid, worker in self.workers.items():
227             logger.info("resetting worker %s", workerid)
228             worker.reset()
229             worker.requestjob()
230         self._jobsdone = 0
231         self._jobsreceived = 0
232 
233     @Pyro4.expose
234     @Pyro4.oneway
235     @utils.synchronous('lock_update')
236     def jobdone(self, workerid):
237         """A worker has finished its job. Log this event and then asynchronously transfer control back to the worker.
238 
239         Callback used by workers to notify when their job is done.
240 
241         The job done event is logged and then control is asynchronously transfered back to the worker
242         (who can then request another job). In this way, control flow basically oscillates between
243         :meth:`gensim.models.lsi_dispatcher.Dispatcher.jobdone` and :meth:`gensim.models.lsi_worker.Worker.requestjob`.
244 
245         Parameters
246         ----------
247         workerid : int
248             The ID of the worker that finished the job (used for logging).
249 
250         """
251         self._jobsdone += 1
252         logger.info("worker #%s finished job #%i", workerid, self._jobsdone)
253         worker = self.workers[workerid]
254         worker.requestjob()  # tell the worker to ask for another job, asynchronously (one-way)
255 
256     def jobsdone(self):
257         """Wrap :attr:`~gensim.models.lsi_dispatcher.Dispatcher._jobsdone`, needed for remote access through proxies.
258 
259         Returns
260         -------
261         int
262             Number of jobs already completed.
263 
264         """
265         return self._jobsdone
266 
267     @Pyro4.oneway
268     def exit(self):
269         """Terminate all registered workers and then the dispatcher."""
270         for workerid, worker in self.workers.items():
271             logger.info("terminating worker %s", workerid)
272             worker.exit()
273         logger.info("terminating dispatcher")
274         os._exit(0)  # exit the whole process (not just this thread ala sys.exit())
275 
276 
277 if __name__ == '__main__':
278     logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
279     parser = argparse.ArgumentParser(description=__doc__[:-135], formatter_class=argparse.RawTextHelpFormatter)
280     parser.add_argument(
281         'maxsize', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE
282     )
283     args = parser.parse_args()
284 
285     logger.info("running %s", " ".join(sys.argv))
286     utils.pyro_daemon('gensim.lsi_dispatcher', Dispatcher(maxsize=args.maxsize))
287     logger.info("finished running %s", parser.prog)
288
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/gensim/models/lsi_dispatcher.py b/gensim/models/lsi_dispatcher.py
--- a/gensim/models/lsi_dispatcher.py
+++ b/gensim/models/lsi_dispatcher.py
@@ -278,7 +278,11 @@
     logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
     parser = argparse.ArgumentParser(description=__doc__[:-135], formatter_class=argparse.RawTextHelpFormatter)
     parser.add_argument(
-        'maxsize', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE
+        'maxsize',
+        nargs='?',
+        type=int,
+        help='Maximum number of jobs to be kept pre-fetched in the queue.',
+        default=MAX_JOBS_QUEUE,
     )
     args = parser.parse_args()

{"golden_diff": "diff --git a/gensim/models/lsi_dispatcher.py b/gensim/models/lsi_dispatcher.py\n--- a/gensim/models/lsi_dispatcher.py\n+++ b/gensim/models/lsi_dispatcher.py\n@@ -278,7 +278,11 @@\n     logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)\n     parser = argparse.ArgumentParser(description=__doc__[:-135], formatter_class=argparse.RawTextHelpFormatter)\n     parser.add_argument(\n-        'maxsize', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE\n+        'maxsize',\n+        nargs='?',\n+        type=int,\n+        help='Maximum number of jobs to be kept pre-fetched in the queue.',\n+        default=MAX_JOBS_QUEUE,\n     )\n     args = parser.parse_args()\n", "issue": "lsi_dispatcher is not working from command-line when not specifying maxsize argument\n#### Problem description\r\n\r\nWhen running `lsi_dispatcher` from the command-line, if you don't specify the `maxsize` argument explicitly, you get an error for the missing positional argument:\r\n\r\n```\r\nusage: lsi_dispatcher.py [-h] maxsize\r\nlsi_dispatcher.py: error: the following arguments are required: maxsize\r\n```\r\n\r\nAccording to the documentation, this argument should be optional.\r\n\r\nThe issue seems to be that the nargs argument to `add_argument` is missing:\r\n\r\n```python\r\n    parser.add_argument(\r\n        'maxsize', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE\r\n    )\r\n ```\r\nIn order to make this argument optional, this should be:\r\n\r\n```python\r\n    parser.add_argument(\r\n        'maxsize', nargs='?', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE\r\n    )\r\n```\r\n\r\n#### Steps/code/corpus to reproduce\r\n\r\nInclude full tracebacks, logs and datasets if necessary. Please keep the examples minimal (\"minimal reproducible example\").\r\n\r\nIf your problem is with a specific Gensim model (word2vec, lsimodel, doc2vec, fasttext, ldamodel etc), include the following:\r\n\r\n```python\r\n$ python3 -m gensim.models.lsi_dispatcher\r\nusage: lsi_dispatcher.py [-h] maxsize\r\nlsi_dispatcher.py: error: the following arguments are required: maxsize\r\n```\r\n\r\n#### Versions\r\n\r\n```python\r\nLinux-5.4.0-67-generic-x86_64-with-glibc2.2\r\nPython 3.8.5 (default, Jan 27 2021, 15:41:15) \r\n[GCC 9.3.0]\r\nBits 64\r\nNumPy 1.19.4\r\nSciPy 1.6.0\r\ngensim 4.0.1\r\nFAST_VERSION 1\r\n```\r\n\n", "before_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n#\n# Copyright (C) 2010 Radim Rehurek <radimrehurek@seznam.cz>\n# Licensed under the GNU LGPL v2.1 - http://www.gnu.org/licenses/lgpl.html\n\n\"\"\"Dispatcher process which orchestrates distributed :class:`~gensim.models.lsimodel.LsiModel` computations.\nRun this script only once, on any node in your cluster.\n\nNotes\n-----\nThe dispatcher expects to find worker scripts already running. Make sure you run as many workers as you like on\nyour machines **before** launching the dispatcher.\n\n\nHow to use distributed LSI\n--------------------------\n\n#. Install needed dependencies (Pyro4) ::\n\n    pip install gensim[distributed]\n\n#. Setup serialization (on each machine) ::\n\n    export PYRO_SERIALIZERS_ACCEPTED=pickle\n    export PYRO_SERIALIZER=pickle\n\n#. Run nameserver ::\n\n    python -m Pyro4.naming -n 0.0.0.0 &\n\n#. Run workers (on each machine) ::\n\n    python -m gensim.models.lsi_worker &\n\n#. Run dispatcher ::\n\n    python -m gensim.models.lsi_dispatcher &\n\n#. Run :class:`~gensim.models.lsimodel.LsiModel` in distributed mode:\n\n    .. sourcecode:: pycon\n\n        >>> from gensim.test.utils import common_corpus, common_dictionary\n        >>> from gensim.models import LsiModel\n        >>>\n        >>> model = LsiModel(common_corpus, id2word=common_dictionary, distributed=True)\n\nCommand line arguments\n----------------------\n\n.. program-output:: python -m gensim.models.lsi_dispatcher --help\n   :ellipsis: 0, -5\n\n\"\"\"\n\nimport os\nimport sys\nimport logging\nimport argparse\nimport threading\nimport time\nfrom queue import Queue\n\nimport Pyro4\n\nfrom gensim import utils\n\n\nlogger = logging.getLogger(__name__)\n\n# How many jobs (=chunks of N documents) to keep \"pre-fetched\" in a queue?\n# A small number is usually enough, unless iteration over the corpus is very very\n# slow (slower than the actual computation of LSI), in which case you can override\n# this value from command line. ie. run \"python ./lsi_dispatcher.py 100\"\nMAX_JOBS_QUEUE = 10\n\n# timeout for the Queue object put/get blocking methods.\n# it should really be infinity, but then keyboard interrupts don't work.\n# so this is really just a hack, see http://bugs.python.org/issue1360\nHUGE_TIMEOUT = 365 * 24 * 60 * 60  # one year\n\n\nclass Dispatcher:\n    \"\"\"Dispatcher object that communicates and coordinates individual workers.\n\n    Warnings\n    --------\n    There should never be more than one dispatcher running at any one time.\n\n    \"\"\"\n    def __init__(self, maxsize=0):\n        \"\"\"Partly initialize the dispatcher.\n\n        A full initialization (including initialization of the workers) requires a call to\n        :meth:`~gensim.models.lsi_dispatcher.Dispatcher.initialize`\n\n        Parameters\n        ----------\n        maxsize : int, optional\n            Maximum number of jobs to be kept pre-fetched in the queue.\n\n        \"\"\"\n        self.maxsize = maxsize\n        self.workers = {}\n        self.callback = None  # a pyro proxy to this object (unknown at init time, but will be set later)\n\n    @Pyro4.expose\n    def initialize(self, **model_params):\n        \"\"\"Fully initialize the dispatcher and all its workers.\n\n        Parameters\n        ----------\n        **model_params\n            Keyword parameters used to initialize individual workers\n            (gets handed all the way down to :meth:`gensim.models.lsi_worker.Worker.initialize`).\n            See :class:`~gensim.models.lsimodel.LsiModel`.\n\n        Raises\n        ------\n        RuntimeError\n            When no workers are found (the :mod:`gensim.model.lsi_worker` script must be ran beforehand).\n\n        \"\"\"\n        self.jobs = Queue(maxsize=self.maxsize)\n        self.lock_update = threading.Lock()\n        self._jobsdone = 0\n        self._jobsreceived = 0\n\n        # locate all available workers and store their proxies, for subsequent RMI calls\n        self.workers = {}\n        with utils.getNS() as ns:\n            self.callback = Pyro4.Proxy('PYRONAME:gensim.lsi_dispatcher')  # = self\n            for name, uri in ns.list(prefix='gensim.lsi_worker').items():\n                try:\n                    worker = Pyro4.Proxy(uri)\n                    workerid = len(self.workers)\n                    # make time consuming methods work asynchronously\n                    logger.info(\"registering worker #%i from %s\", workerid, uri)\n                    worker.initialize(workerid, dispatcher=self.callback, **model_params)\n                    self.workers[workerid] = worker\n                except Pyro4.errors.PyroError:\n                    logger.exception(\"unresponsive worker at %s, deleting it from the name server\", uri)\n                    ns.remove(name)\n\n        if not self.workers:\n            raise RuntimeError('no workers found; run some lsi_worker scripts on your machines first!')\n\n    @Pyro4.expose\n    def getworkers(self):\n        \"\"\"Get pyro URIs of all registered workers.\n\n        Returns\n        -------\n        list of URIs\n            The pyro URIs for each worker.\n\n        \"\"\"\n        return [worker._pyroUri for worker in self.workers.values()]\n\n    @Pyro4.expose\n    def getjob(self, worker_id):\n        \"\"\"Atomically pop a job from the queue.\n\n        Parameters\n        ----------\n        worker_id : int\n            The worker that requested the job.\n\n        Returns\n        -------\n        iterable of iterable of (int, float)\n            The corpus in BoW format.\n\n        \"\"\"\n        logger.info(\"worker #%i requesting a new job\", worker_id)\n        job = self.jobs.get(block=True, timeout=1)\n        logger.info(\"worker #%i got a new job (%i left)\", worker_id, self.jobs.qsize())\n        return job\n\n    @Pyro4.expose\n    def putjob(self, job):\n        \"\"\"Atomically add a job to the queue.\n\n        Parameters\n        ----------\n        job : iterable of list of (int, float)\n            The corpus in BoW format.\n\n        \"\"\"\n        self._jobsreceived += 1\n        self.jobs.put(job, block=True, timeout=HUGE_TIMEOUT)\n        logger.info(\"added a new job (len(queue)=%i items)\", self.jobs.qsize())\n\n    @Pyro4.expose\n    def getstate(self):\n        \"\"\"Merge projections from across all workers and get the final projection.\n\n        Returns\n        -------\n        :class:`~gensim.models.lsimodel.Projection`\n            The current projection of the total model.\n\n        \"\"\"\n        logger.info(\"end of input, assigning all remaining jobs\")\n        logger.debug(\"jobs done: %s, jobs received: %s\", self._jobsdone, self._jobsreceived)\n        while self._jobsdone < self._jobsreceived:\n            time.sleep(0.5)  # check every half a second\n\n        # TODO: merge in parallel, so that we're done in `log_2(workers)` merges,\n        # and not `workers - 1` merges!\n        # but merging only takes place once, after all input data has been processed,\n        # so the overall effect would be small... compared to the amount of coding :-)\n        logger.info(\"merging states from %i workers\", len(self.workers))\n        workers = list(self.workers.items())\n        result = workers[0][1].getstate()\n        for workerid, worker in workers[1:]:\n            logger.info(\"pulling state from worker %s\", workerid)\n            result.merge(worker.getstate())\n        logger.info(\"sending out merged projection\")\n        return result\n\n    @Pyro4.expose\n    def reset(self):\n        \"\"\"Re-initialize all workers for a new decomposition.\"\"\"\n        for workerid, worker in self.workers.items():\n            logger.info(\"resetting worker %s\", workerid)\n            worker.reset()\n            worker.requestjob()\n        self._jobsdone = 0\n        self._jobsreceived = 0\n\n    @Pyro4.expose\n    @Pyro4.oneway\n    @utils.synchronous('lock_update')\n    def jobdone(self, workerid):\n        \"\"\"A worker has finished its job. Log this event and then asynchronously transfer control back to the worker.\n\n        Callback used by workers to notify when their job is done.\n\n        The job done event is logged and then control is asynchronously transfered back to the worker\n        (who can then request another job). In this way, control flow basically oscillates between\n        :meth:`gensim.models.lsi_dispatcher.Dispatcher.jobdone` and :meth:`gensim.models.lsi_worker.Worker.requestjob`.\n\n        Parameters\n        ----------\n        workerid : int\n            The ID of the worker that finished the job (used for logging).\n\n        \"\"\"\n        self._jobsdone += 1\n        logger.info(\"worker #%s finished job #%i\", workerid, self._jobsdone)\n        worker = self.workers[workerid]\n        worker.requestjob()  # tell the worker to ask for another job, asynchronously (one-way)\n\n    def jobsdone(self):\n        \"\"\"Wrap :attr:`~gensim.models.lsi_dispatcher.Dispatcher._jobsdone`, needed for remote access through proxies.\n\n        Returns\n        -------\n        int\n            Number of jobs already completed.\n\n        \"\"\"\n        return self._jobsdone\n\n    @Pyro4.oneway\n    def exit(self):\n        \"\"\"Terminate all registered workers and then the dispatcher.\"\"\"\n        for workerid, worker in self.workers.items():\n            logger.info(\"terminating worker %s\", workerid)\n            worker.exit()\n        logger.info(\"terminating dispatcher\")\n        os._exit(0)  # exit the whole process (not just this thread ala sys.exit())\n\n\nif __name__ == '__main__':\n    logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)\n    parser = argparse.ArgumentParser(description=__doc__[:-135], formatter_class=argparse.RawTextHelpFormatter)\n    parser.add_argument(\n        'maxsize', type=int, help='Maximum number of jobs to be kept pre-fetched in the queue.', default=MAX_JOBS_QUEUE\n    )\n    args = parser.parse_args()\n\n    logger.info(\"running %s\", \" \".join(sys.argv))\n    utils.pyro_daemon('gensim.lsi_dispatcher', Dispatcher(maxsize=args.maxsize))\n    logger.info(\"finished running %s\", parser.prog)\n", "path": "gensim/models/lsi_dispatcher.py"}], "after_files": [{"content": "#!/usr/bin/env python\n# -*- coding: utf-8 -*-\n#\n# Copyright (C) 2010 Radim Rehurek <radimrehurek@seznam.cz>\n# Licensed under the GNU LGPL v2.1 - http://www.gnu.org/licenses/lgpl.html\n\n\"\"\"Dispatcher process which orchestrates distributed :class:`~gensim.models.lsimodel.LsiModel` computations.\nRun this script only once, on any node in your cluster.\n\nNotes\n-----\nThe dispatcher expects to find worker scripts already running. Make sure you run as many workers as you like on\nyour machines **before** launching the dispatcher.\n\n\nHow to use distributed LSI\n--------------------------\n\n#. Install needed dependencies (Pyro4) ::\n\n    pip install gensim[distributed]\n\n#. Setup serialization (on each machine) ::\n\n    export PYRO_SERIALIZERS_ACCEPTED=pickle\n    export PYRO_SERIALIZER=pickle\n\n#. Run nameserver ::\n\n    python -m Pyro4.naming -n 0.0.0.0 &\n\n#. Run workers (on each machine) ::\n\n    python -m gensim.models.lsi_worker &\n\n#. Run dispatcher ::\n\n    python -m gensim.models.lsi_dispatcher &\n\n#. Run :class:`~gensim.models.lsimodel.LsiModel` in distributed mode:\n\n    .. sourcecode:: pycon\n\n        >>> from gensim.test.utils import common_corpus, common_dictionary\n        >>> from gensim.models import LsiModel\n        >>>\n        >>> model = LsiModel(common_corpus, id2word=common_dictionary, distributed=True)\n\nCommand line arguments\n----------------------\n\n.. program-output:: python -m gensim.models.lsi_dispatcher --help\n   :ellipsis: 0, -5\n\n\"\"\"\n\nimport os\nimport sys\nimport logging\nimport argparse\nimport threading\nimport time\nfrom queue import Queue\n\nimport Pyro4\n\nfrom gensim import utils\n\n\nlogger = logging.getLogger(__name__)\n\n# How many jobs (=chunks of N documents) to keep \"pre-fetched\" in a queue?\n# A small number is usually enough, unless iteration over the corpus is very very\n# slow (slower than the actual computation of LSI), in which case you can override\n# this value from command line. ie. run \"python ./lsi_dispatcher.py 100\"\nMAX_JOBS_QUEUE = 10\n\n# timeout for the Queue object put/get blocking methods.\n# it should really be infinity, but then keyboard interrupts don't work.\n# so this is really just a hack, see http://bugs.python.org/issue1360\nHUGE_TIMEOUT = 365 * 24 * 60 * 60  # one year\n\n\nclass Dispatcher:\n    \"\"\"Dispatcher object that communicates and coordinates individual workers.\n\n    Warnings\n    --------\n    There should never be more than one dispatcher running at any one time.\n\n    \"\"\"\n    def __init__(self, maxsize=0):\n        \"\"\"Partly initialize the dispatcher.\n\n        A full initialization (including initialization of the workers) requires a call to\n        :meth:`~gensim.models.lsi_dispatcher.Dispatcher.initialize`\n\n        Parameters\n        ----------\n        maxsize : int, optional\n            Maximum number of jobs to be kept pre-fetched in the queue.\n\n        \"\"\"\n        self.maxsize = maxsize\n        self.workers = {}\n        self.callback = None  # a pyro proxy to this object (unknown at init time, but will be set later)\n\n    @Pyro4.expose\n    def initialize(self, **model_params):\n        \"\"\"Fully initialize the dispatcher and all its workers.\n\n        Parameters\n        ----------\n        **model_params\n            Keyword parameters used to initialize individual workers\n            (gets handed all the way down to :meth:`gensim.models.lsi_worker.Worker.initialize`).\n            See :class:`~gensim.models.lsimodel.LsiModel`.\n\n        Raises\n        ------\n        RuntimeError\n            When no workers are found (the :mod:`gensim.model.lsi_worker` script must be ran beforehand).\n\n        \"\"\"\n        self.jobs = Queue(maxsize=self.maxsize)\n        self.lock_update = threading.Lock()\n        self._jobsdone = 0\n        self._jobsreceived = 0\n\n        # locate all available workers and store their proxies, for subsequent RMI calls\n        self.workers = {}\n        with utils.getNS() as ns:\n            self.callback = Pyro4.Proxy('PYRONAME:gensim.lsi_dispatcher')  # = self\n            for name, uri in ns.list(prefix='gensim.lsi_worker').items():\n                try:\n                    worker = Pyro4.Proxy(uri)\n                    workerid = len(self.workers)\n                    # make time consuming methods work asynchronously\n                    logger.info(\"registering worker #%i from %s\", workerid, uri)\n                    worker.initialize(workerid, dispatcher=self.callback, **model_params)\n                    self.workers[workerid] = worker\n                except Pyro4.errors.PyroError:\n                    logger.exception(\"unresponsive worker at %s, deleting it from the name server\", uri)\n                    ns.remove(name)\n\n        if not self.workers:\n            raise RuntimeError('no workers found; run some lsi_worker scripts on your machines first!')\n\n    @Pyro4.expose\n    def getworkers(self):\n        \"\"\"Get pyro URIs of all registered workers.\n\n        Returns\n        -------\n        list of URIs\n            The pyro URIs for each worker.\n\n        \"\"\"\n        return [worker._pyroUri for worker in self.workers.values()]\n\n    @Pyro4.expose\n    def getjob(self, worker_id):\n        \"\"\"Atomically pop a job from the queue.\n\n        Parameters\n        ----------\n        worker_id : int\n            The worker that requested the job.\n\n        Returns\n        -------\n        iterable of iterable of (int, float)\n            The corpus in BoW format.\n\n        \"\"\"\n        logger.info(\"worker #%i requesting a new job\", worker_id)\n        job = self.jobs.get(block=True, timeout=1)\n        logger.info(\"worker #%i got a new job (%i left)\", worker_id, self.jobs.qsize())\n        return job\n\n    @Pyro4.expose\n    def putjob(self, job):\n        \"\"\"Atomically add a job to the queue.\n\n        Parameters\n        ----------\n        job : iterable of list of (int, float)\n            The corpus in BoW format.\n\n        \"\"\"\n        self._jobsreceived += 1\n        self.jobs.put(job, block=True, timeout=HUGE_TIMEOUT)\n        logger.info(\"added a new job (len(queue)=%i items)\", self.jobs.qsize())\n\n    @Pyro4.expose\n    def getstate(self):\n        \"\"\"Merge projections from across all workers and get the final projection.\n\n        Returns\n        -------\n        :class:`~gensim.models.lsimodel.Projection`\n            The current projection of the total model.\n\n        \"\"\"\n        logger.info(\"end of input, assigning all remaining jobs\")\n        logger.debug(\"jobs done: %s, jobs received: %s\", self._jobsdone, self._jobsreceived)\n        while self._jobsdone < self._jobsreceived:\n            time.sleep(0.5)  # check every half a second\n\n        # TODO: merge in parallel, so that we're done in `log_2(workers)` merges,\n        # and not `workers - 1` merges!\n        # but merging only takes place once, after all input data has been processed,\n        # so the overall effect would be small... compared to the amount of coding :-)\n        logger.info(\"merging states from %i workers\", len(self.workers))\n        workers = list(self.workers.items())\n        result = workers[0][1].getstate()\n        for workerid, worker in workers[1:]:\n            logger.info(\"pulling state from worker %s\", workerid)\n            result.merge(worker.getstate())\n        logger.info(\"sending out merged projection\")\n        return result\n\n    @Pyro4.expose\n    def reset(self):\n        \"\"\"Re-initialize all workers for a new decomposition.\"\"\"\n        for workerid, worker in self.workers.items():\n            logger.info(\"resetting worker %s\", workerid)\n            worker.reset()\n            worker.requestjob()\n        self._jobsdone = 0\n        self._jobsreceived = 0\n\n    @Pyro4.expose\n    @Pyro4.oneway\n    @utils.synchronous('lock_update')\n    def jobdone(self, workerid):\n        \"\"\"A worker has finished its job. Log this event and then asynchronously transfer control back to the worker.\n\n        Callback used by workers to notify when their job is done.\n\n        The job done event is logged and then control is asynchronously transfered back to the worker\n        (who can then request another job). In this way, control flow basically oscillates between\n        :meth:`gensim.models.lsi_dispatcher.Dispatcher.jobdone` and :meth:`gensim.models.lsi_worker.Worker.requestjob`.\n\n        Parameters\n        ----------\n        workerid : int\n            The ID of the worker that finished the job (used for logging).\n\n        \"\"\"\n        self._jobsdone += 1\n        logger.info(\"worker #%s finished job #%i\", workerid, self._jobsdone)\n        worker = self.workers[workerid]\n        worker.requestjob()  # tell the worker to ask for another job, asynchronously (one-way)\n\n    def jobsdone(self):\n        \"\"\"Wrap :attr:`~gensim.models.lsi_dispatcher.Dispatcher._jobsdone`, needed for remote access through proxies.\n\n        Returns\n        -------\n        int\n            Number of jobs already completed.\n\n        \"\"\"\n        return self._jobsdone\n\n    @Pyro4.oneway\n    def exit(self):\n        \"\"\"Terminate all registered workers and then the dispatcher.\"\"\"\n        for workerid, worker in self.workers.items():\n            logger.info(\"terminating worker %s\", workerid)\n            worker.exit()\n        logger.info(\"terminating dispatcher\")\n        os._exit(0)  # exit the whole process (not just this thread ala sys.exit())\n\n\nif __name__ == '__main__':\n    logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)\n    parser = argparse.ArgumentParser(description=__doc__[:-135], formatter_class=argparse.RawTextHelpFormatter)\n    parser.add_argument(\n        'maxsize',\n        nargs='?',\n        type=int,\n        help='Maximum number of jobs to be kept pre-fetched in the queue.',\n        default=MAX_JOBS_QUEUE,\n    )\n    args = parser.parse_args()\n\n    logger.info(\"running %s\", \" \".join(sys.argv))\n    utils.pyro_daemon('gensim.lsi_dispatcher', Dispatcher(maxsize=args.maxsize))\n    logger.info(\"finished running %s\", parser.prog)\n", "path": "gensim/models/lsi_dispatcher.py"}]}