Updated README

Author: codepr

README.md

@@ -22,12 +22,11 @@ production enviroments.
 
 ## Quickstart
 
-Starting a worker on a node, with debug flag set to true on configuration file
+Starting a worker on a node using Redis as backend
 
 ```
-$ tq redis
-Listening for jobs on 127.0.0.1:9000
-Response actor started
+$ tq redis-worker --log-level DEBUG
+2019-04-26 23:15:28 - tasq.remote.supervisor-17903: Worker type: Actor
 ```
 
 In a python shell
@@ -42,77 +41,59 @@ Warning: disable autoreload in ipython_config.py to improve performance.
 
 In [1]: from tasq.queue import TasqQueue
 
-In [2]: tq = TasqQueue(backend='redis://localhost:6379/0?name=test')
+In [2]: tq = TasqQueue(backend='redis://localhost:6379')
 
-In [3]: def dup(n):
-   ...:     return n * 2
+In [3]: def fib(n):
+   ...:     if n == 0:
+   ...:         return 0
+   ...:     a, b = 0, 1
+   ...:     for _ in range(n - 1):
+   ...:         a, b = b, a + b
+   ...:     return b
    ...:
 
-In [4]: fut = tq.put(dup, 5, name='task-01')
-
-In [5]: fut
-Out[5]: <TasqFuture at 0x7f2851826518 state=finished returned JobResult>
-
-In [6]: fut.unwrap()
-Out[6]: 10
-```
-
-**Lower-level TasqClient**
-
-```
-Python 3.6.5 (default, Apr 12 2018, 22:45:43)
-[GCC 7.3.1 20180312] on linux
-Type "help", "copyright", "credits" or "license" for more information.
->>> from tasq import TasqClient
->>> tc = TasqClient('127.0.0.1', 9000)
->>> tc.connect()
->>>
->>> def foo(num):
->>>     import time
->>>     import random
->>>     r = random.randint(0, 2)
->>>     time.sleep(r)
->>>     return f'Foo - {random.randint(0, num)}'
->>>
->>> fut = tc.schedule(foo, 5, name='Task-1')
->>> fut
->>> <Future at 0x7f7d6e048160 state=pending>
->>> fut.result
->>>
->>> # After some time, to let worker complete the job
->>> fut.result
->>> 'Foo - 2'
->>> tc.results
->>> {'Task-1': <Future at 0x7f7d6e048160 state=finished returned str>}
->>>
->>> tc.schedule_blocking(foo, 5, name='Task-2')
->>> 'Foo - 4'
->>>
->>> tc.results
->>> {'Task-1': <Future at 0x7f7d6e048160 state=finished returned str>,
->>>  'Task-2': <Future at 0x7f7d6e047268 state=finished returned str>}
-```
-
-Scheduling a job after a delay
-
-```
->>> fut = tc.schedule(foo, 5, name='Delayed-Task', delay=5)
->>> tc.results
->>> {'Delayed-Task': <Future at 0x7f7d6e044208 state=pending>}
->>> # Wait 5 seconds
->>> tc.results
->>> {'Delayed-Task': <Future at 0x7f7d6e044208 state=finished returned str>}
->>> fut.result()
->>> 'Foo - 2'
+In [4]: # Asynchronous execution
+In [5]: fut = tq.put(fib, 50, name='fib-async')
+
+In [6]: fut
+Out[6]: <TasqFuture at 0x7f2851826518 state=finished returned JobResult>
+
+In [7]: fut.unwrap()
+Out[7]: 12586269025
+
+In [8]: res = tq.put_blocking(fib, 50, name='fib-sync')
+
+In [9]: res.unwrap()
+Out[9]: 12586269025
 ```
 
-Scheduling a task to be executed continously in a defined interval
+Scheduling jobs after a delay
+```
+
+In [10]: fut = tc.schedule(fib, 5, name='fib-delayed', delay=5)
 
+In [11]: fut
+Out[11]: <TasqFuture at 0x7f2951856418 state=pending>
+
+In [12]: # wait 5 seconds
+
+In [13]: fut.unwrap()
+Out[13]: 5
+
+In [14] tq.results
+Out[14] {'fib-async': <TasqFuture at 0x7f2851826518 state=finished returned JobResult>,
+Out[14]  'fib-sync': <TasqFuture at 0x7f7d6e047268 state=finished returned JobResult>
+Out[14]  'fib-delayed': <TasqFuture at 0x7f2951856418 state=finished returned JobResult>}
 ```
->>> tc.schedule(foo, 5, name='8_seconds_interval_task', eta='8s')
->>> tc.schedule(foo, 5, name='2_hours_interval_task', eta='2h')
+
+Scheduling a task to be executed continously in a defined interval
+
 ```
+In [15] tq.put(fib, 5, name='8_seconds_interval_fib', eta='8s')
 
+In [16] tq.put(fib, 5, name='2_hours_interval_fib', eta='2h')
+
+```
 Delayed and interval tasks are supported even in blocking scheduling manner.
 
 Tasq also supports an optional static configuration file, in the `tasq.settings.py` module is
@@ -124,27 +105,18 @@ By setting the `-f` flag it is possible to also set a location of a configuratio
 filesystem
 
 ```
-$ tq --worker -f path/to/conf/conf.json
+$ tq worker -c path/to/conf/conf.json
 ```
 
 A worker can be started by specifying the type of sub worker we want:
 
 ```
-$ tq --worker --worker-type process
+$ tq rabbitmq-worker --worker-type process
 ```
 Using `process` type subworker it is possible to use a distributed queue for parallel execution,
 usefull when the majority of the jobs are CPU bound instead of I/O bound (actors are preferable in
 that case).
 
-Multiple workers can be started in the same node, this will start two worker process ready to
-receive jobs.
-
-```
-$ tq --workers 127.0.0.1:9000:9001, 127.0.0.1:9090:9091
-Listening for jobs on 127.0.0.1:9000
-Listening for jobs on 127.0.0.1:9090
-```
-
 If jobs are scheduled for execution on a disconnected client, or remote workers are not up at the
 time of the scheduling, all jobs will be enqeued for later execution. This means that there's no
 need to actually start workers before job scheduling, at the first worker up all jobs will be sent
@@ -201,15 +173,13 @@ See the [CHANGES](CHANGES.md) file.
 
 ## TODO:
 
-- [ ] Possibility of a broker to persist jobs (classic task queue celery like)
+- [x] Possibility of a broker to persist jobs (classic task queue celery like)
+- [x] Delayed tasks and scheduled cron tasks
+- [x] Configuration handling throughout the code
+- [x] Better explanation of the implementation and actors defined
+- [x] Improve CLI options
 - [ ] Check for pynacl for security on pickled data
 - [ ] Tests
-- [ ] A meaningful client pool
-- [x] Debugging multiprocessing start for more workers on the same node
 - [ ] Refactor of existing code and corner case handling (Still very basic implementation of even
       simple heuristics)
-- [x] Delayed tasks and scheduled cron tasks
-- [x] Configuration handling throughout the code
-- [x] Better explanation of the implementation and actors defined
-- [ ] Improve CLI options
 - [ ] Dockerfile

tasq/actors/routers.py

@@ -5,7 +5,8 @@
 remote calls.
 """
 
-from __future__ import absolute_import, division, print_function, unicode_literals
+from __future__ import (absolute_import, division,
+                        print_function, unicode_literals)
 
 from abc import ABCMeta, abstractmethod
 
@@ -25,6 +26,7 @@ class Router(metaclass=ABCMeta):
     :type func_name: str or 'submit'
     :param func_name: The name of the method that must be called after the
                       message has been routed.
+
     """
 
     def __init__(self, workers, func_name=u'submit', *args, **kwargs):
@@ -51,7 +53,7 @@ def _call_func(self, idx, msg):
     @abstractmethod
     def _route_message(self, msg):
         """To be defined on subclass"""
-        pass
+        raise NotImplementedError
 
     def route(self, msg):
         """Call `_route_message` private method, call function directly in case

tasq/cli/main.py

@@ -90,9 +90,11 @@ def start_worker(supervisor_type, worker_type, host, **kwargs):
         s_type = supervisors[worker_type][supervisor_type]
     except KeyError:
         raise UnknownSupervisorException()
-    else:
+    try:
         supervisor = supervisor_factory.create(s_type, host=host, **kwargs)
         supervisor.serve_forever()
+    except KeyboardInterrupt:
+        supervisor.stop()
 
 
 def main():

tasq/queue.py

@@ -77,6 +77,9 @@ def __init__(self, backend=u'zmq://localhost:9000',
         # Connect with the backend
         self._backend.connect()
 
+    def __len__(self):
+        return len(self.pending_jobs())
+
     def _store_results(self):
         while True:
             tasqfuture = self._results.get()

tasq/remote/actors.py

@@ -140,7 +140,8 @@ def __init__(self, name=u'', ctx=None, response_actor=None):
         self._response_actor = response_actor or self._ctx.actor_of(
             ResponseActor, 'TimedActor - ResponseActor'
         )
-        self._response_actor.start()
+        if isinstance(self._response_actor, Actor):
+            self._response_actor.start()
         super().__init__(name, ctx)
 
     def submit(self, job, eta):
@@ -205,7 +206,10 @@ def run(self):
             self._log.debug('%s - Timed job %s result = %s',
                             self.name, job.job_id, jobres)
             result.set_result(response)
-            self._response_actor.send(result)
+            if isinstance(self._response_actor, Actor):
+                self._response_actor.send(result)
+            else:
+                self._response_actor.route(result)
             self.submit(job, str(job.delay) + 's')
 
 

tasq/remote/sockets.py

@@ -219,3 +219,6 @@ def recv_result_data(self, timeout=None, unpickle=True, signkey=None):
         if zipped_result and unpickle:
             return decompress_and_unpickle(zipped_result)
         return zipped_result
+
+    def close(self):
+        self._backend.close()

tasq/remote/supervisor.py

@@ -8,7 +8,6 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import os
-import signal
 import asyncio
 from threading import Thread, Event
 from abc import ABCMeta, abstractmethod
@@ -114,8 +113,6 @@ def __init__(self, host, pull_port, push_port, num_workers=max_workers(),
         super().__init__(host, num_workers=num_workers, signkey=signkey)
         # Event loop
         self._loop = asyncio.get_event_loop()
-        # Handling loop exit
-        self._loop.add_signal_handler(signal.SIGINT, self.stop)
 
     @property
     def push_port(self):
@@ -298,7 +295,6 @@ def __init__(self, host, port, db, name, num_workers=max_workers(),
         self._response_thread = Thread(target=self._respond, daemon=True)
         self._response_thread.start()
         self._run = True
-        self._done = Event()
         self._log.info("Worker type: %s", worker_class.__name__)
 
     @property
@@ -314,7 +310,6 @@ def _init_server(self):
     def stop(self):
         self._log.info("\nStopping..")
         self._run = False
-        self._done.wait()
         # Use a poison pill to stop the loop
         self._jobqueue.shutdown()
         self._response_thread.join()
@@ -327,7 +322,6 @@ def serve_forever(self):
                 continue
             self._log.info("Received job")
             self._jobqueue.add_job(job)
-        self._done.set()
 
     def _respond(self):
         """Spin a loop and respond to client with whatever results arrive in
@@ -374,9 +368,6 @@ def __init__(self, host, port, db, name, num_workers=max_workers(),
             response_actor=self._responses
         )
         self._log.info("Worker type: Actor")
-        self._done = Event()
-        signal.signal(signal.SIGINT, self.stop)
-        signal.signal(signal.SIGTERM, self.stop)
 
     @property
     def name(self):
@@ -391,7 +382,6 @@ def _init_server(self):
     def stop(self):
         self._log.info("\nStopping..")
         self._run = False
-        self._done.wait()
 
     def serve_forever(self):
         """Receive jobs from clients with polling"""
@@ -403,7 +393,6 @@ def serve_forever(self):
             res = self._workers.route(job)
             self._responses.route(res)
             self._log.info("Routed")
-        self._done.set()
 
     @classmethod
     def create(cls, host, port, db, name, num_workers=max_workers(),
@@ -431,7 +420,6 @@ def __init__(self, host, port, name, num_workers=max_workers(),
         self._response_thread = Thread(target=self._respond, daemon=True)
         self._response_thread.start()
         self._run = True
-        self._done = Event()
         self._log.info("Worker type: %s", worker_class.__name__)
 
     @property
@@ -447,7 +435,6 @@ def _init_server(self):
     def stop(self):
         self._log.info("\nStopping..")
         self._run = False
-        self._done.wait()
         self._jobqueue.shutdown()
         self._response_thread.join()
 
@@ -459,7 +446,6 @@ def serve_forever(self):
                 continue
             self._log.info("Received job")
             self._jobqueue.add_job(job)
-        self._done.set()
 
     def _respond(self):
         """Spin a loop and respond to client with whatever results arrive in
@@ -505,9 +491,6 @@ def __init__(self, host, port, name, num_workers=max_workers(),
             response_actor=self._responses
         )
         self._log.info("Worker type: Actor")
-        self._done = Event()
-        signal.signal(signal.SIGINT, self.stop)
-        signal.signal(signal.SIGTERM, self.stop)
 
     @property
     def name(self):
@@ -523,7 +506,6 @@ def stop(self):
         self._log.info("\nStopping..")
         self._run = False
         self._server.close()
-        self._done.wait()
 
     def serve_forever(self):
         """Receive jobs from clients with polling"""
@@ -535,7 +517,6 @@ def serve_forever(self):
             res = self._workers.route(job)
             self._responses.route(res)
             self._log.info("Routed")
-        self._done.set()
 
     @classmethod
     def create(cls, host, port, name, num_workers=max_workers(),