start/stop fixes

alttch · Aug 10, 2019 · 391df51 · 391df51
1 parent b530d4a
commit 391df51
Show file tree

Hide file tree

Showing 10 changed files with 19 additions and 20 deletions.
diff --git a/atasker/__init__.py b/atasker/__init__.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 from atasker.supervisor import TaskSupervisor
 from atasker.supervisor import TASK_LOW

diff --git a/atasker/co.py b/atasker/co.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 from atasker import task_supervisor
 

diff --git a/atasker/f.py b/atasker/f.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 import traceback
 import threading

diff --git a/atasker/supervisor.py b/atasker/supervisor.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 import threading
 import multiprocessing
@@ -54,7 +54,7 @@ def __init__(self):
         self._active_mps = set()
         self._active = False
         self._main_loop_active = False
-        self._started = False
+        self._started = threading.Event()
         self._lock = threading.Lock()
         self._max_threads = {}
         self._max_mps = {}
@@ -107,7 +107,7 @@ def _higher_queues_busy(self, tt, task_priority):
             return False
 
     def put_task(self, task, priority=TASK_NORMAL, delay=None, tt=TT_THREAD):
-        if not self._started:
+        if not self._started.is_set():
             return False
         asyncio.run_coroutine_threadsafe(
             self._Q.put((RQ_TASK, (tt, task, priority, delay), time.time())),
@@ -122,7 +122,7 @@ def create_mp_pool(self, *args, **kwargs):
                 processes=multiprocessing.cpu_count())
 
     def register_scheduler(self, scheduler):
-        if not self._started:
+        if not self._started.is_set():
             return False
         asyncio.run_coroutine_threadsafe(
             self._Q.put((RQ_SCHEDULER, scheduler, time.time())),
@@ -248,15 +248,15 @@ def start(self):
         t = threading.Thread(
             name='supervisor_event_loop', target=self._start_event_loop)
         t.start()
-        while not self._started:
-            time.sleep(self.poll_delay)
+        self._started.wait()
 
     def block(self):
         while self._active:
             time.sleep(0.1)
 
     async def _main_loop(self):
         self._Q = asyncio.queues.Queue()
+        self._started.set()
         logger.info('supervisor event loop started')
         while self._main_loop_active:
             data = await self._Q.get()
@@ -292,7 +292,6 @@ def _start_event_loop(self):
                             self.thread_pool_size, self.thread_reserve_normal,
                             self.thread_reserve_high, mp))
             try:
-                self._started = True
                 self.event_loop.run_until_complete(self._main_loop())
             except CancelledError:
                 logger.warning('supervisor loop had active tasks')
@@ -348,5 +347,5 @@ def stop(self, wait=True, stop_schedulers=True, cancel_tasks=False):
                             break
                     if can_break: break
                 time.sleep(self.poll_delay)
-        self._started = False
+        self._started.clear()
         logger.info('supervisor stopped')
diff --git a/atasker/threads.py b/atasker/threads.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 import threading
 import time

diff --git a/atasker/workers.py b/atasker/workers.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 import threading
 import logging
@@ -379,8 +379,8 @@ def get_queue_obj(self):
 
 class BackgroundEventWorker(BackgroundAsyncWorker):
 
-    def trigger(self):
-        if self._current_executor:
+    def trigger(self, force=False):
+        if self._current_executor and not force:
             return
         asyncio.run_coroutine_threadsafe(
             self._set_event(), loop=self.supervisor.event_loop)
@@ -405,7 +405,7 @@ async def loop(self, *args, **kwargs):
 
     def send_stop_events(self, *args, **kwargs):
         try:
-            self.trigger()
+            self.trigger(force=True)
         except:
             pass
 

diff --git a/setup.py b/setup.py
@@ -1,4 +1,4 @@
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 import setuptools
 

diff --git a/tests/mp.py b/tests/mp.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 def test(*args, **kwargs):
     print('test mp method {} {}'.format(args, kwargs))

diff --git a/tests/mpworker.py b/tests/mpworker.py
@@ -1,7 +1,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 from atasker import BackgroundIntervalWorker
 

diff --git a/tests/test.py b/tests/test.py
@@ -3,7 +3,7 @@
 __author__ = "Altertech Group, https://www.altertech.com/"
 __copyright__ = "Copyright (C) 2018-2019 Altertech Group"
 __license__ = "Apache License 2.0"
-__version__ = "0.2.11"
+__version__ = "0.2.12"
 
 from pathlib import Path