comments, lock thread pool, while 1 and snake_case everywhere

telegram/ext/dispatcher.py

@@ -36,7 +36,7 @@ logging.getLogger(__name__).addHandler(NullHandler())
 ASYNC_QUEUE = Queue()
 ASYNC_THREADS = set()
 """:type: set[Thread]"""
-ASYNC_LOCK = Lock()
+ASYNC_LOCK = Lock()  # guards ASYNC_THREADS
 DEFAULT_GROUP = 0
 
 
@@ -48,16 +48,17 @@ def _pooled():
         try:
             func, args, kwargs = ASYNC_QUEUE.get()
 
+        # If unpacking fails, the thread pool is being closed from Updater._join_async_threads
         except TypeError:
-            logging.debug("Closing run_async thread %s/%d" %
-                          (current_thread().getName(), len(ASYNC_THREADS)))
+            logging.getLogger(__name__).debug("Closing run_async thread %s/%d" %
+                                              (current_thread().getName(), len(ASYNC_THREADS)))
             break
 
         try:
             func(*args, **kwargs)
 
         except:
-            logging.exception("Async function raised exception")
+            logging.getLogger(__name__).exception("run_async function raised exception")
 
 
 def run_async(func):
@@ -110,17 +111,18 @@ class Dispatcher(object):
         self.__stop_event = Event()
         self.__exception_event = exception_event or Event()
 
-        if not len(ASYNC_THREADS):
-            if request.is_con_pool_initialized():
-                raise RuntimeError('Connection Pool already initialized')
+        with ASYNC_LOCK:
+            if not ASYNC_THREADS:
+                if request.is_con_pool_initialized():
+                    raise RuntimeError('Connection Pool already initialized')
 
-            request.CON_POOL_SIZE = workers + 3
-            for i in range(workers):
-                thread = Thread(target=_pooled, name=str(i))
-                ASYNC_THREADS.add(thread)
-                thread.start()
-        else:
-            self.logger.debug('Thread pool already initialized, skipping.')
+                request.CON_POOL_SIZE = workers + 3
+                for i in range(workers):
+                    thread = Thread(target=_pooled, name=str(i))
+                    ASYNC_THREADS.add(thread)
+                    thread.start()
+            else:
+                self.logger.debug('Thread pool already initialized, skipping.')
 
     def start(self):
         """
@@ -140,7 +142,7 @@ class Dispatcher(object):
         self.running = True
         self.logger.debug('Dispatcher started')
 
-        while True:
+        while 1:
             try:
                 # Pop update from update queue.
                 update = self.update_queue.get(True, 1)
@@ -154,7 +156,7 @@ class Dispatcher(object):
                 continue
 
             self.logger.debug('Processing Update: %s' % update)
-            self.processUpdate(update)
+            self.process_update(update)
 
         self.running = False
         self.logger.debug('Dispatcher thread stopped')
@@ -169,7 +171,7 @@ class Dispatcher(object):
                 sleep(0.1)
             self.__stop_event.clear()
 
-    def processUpdate(self, update):
+    def process_update(self, update):
         """
         Processes a single update.
 
@@ -179,7 +181,7 @@ class Dispatcher(object):
 
         # An error happened while polling
         if isinstance(update, TelegramError):
-            self.dispatchError(None, update)
+            self.dispatch_error(None, update)
 
         else:
             for group in self.groups:
@@ -194,7 +196,7 @@ class Dispatcher(object):
                                          'Update.')
 
                         try:
-                            self.dispatchError(update, te)
+                            self.dispatch_error(update, te)
                         except Exception:
                             self.logger.exception('An uncaught error was raised while '
                                                   'handling the error')
@@ -280,7 +282,7 @@ class Dispatcher(object):
         if callback in self.error_handlers:
             self.error_handlers.remove(callback)
 
-    def dispatchError(self, update, error):
+    def dispatch_error(self, update, error):
         """
         Dispatches an error.
 

telegram/ext/updater.py

@@ -285,9 +285,10 @@ class Updater(object):
     def _check_ssl_cert(self, cert, key):
         # Check SSL-Certificate with openssl, if possible
         try:
-            exit_code = subprocess.call(["openssl", "x509", "-text", "-noout", "-in", cert],
-                                        stdout=open(os.devnull, 'wb'),
-                                        stderr=subprocess.STDOUT)
+            exit_code = subprocess.call(
+                ["openssl", "x509", "-text", "-noout", "-in", cert],
+                stdout=open(os.devnull, 'wb'),
+                stderr=subprocess.STDOUT)
         except OSError:
             exit_code = 0
         if exit_code is 0:
@@ -308,7 +309,7 @@ class Updater(object):
 
     def _bootstrap(self, max_retries, clean, webhook_url, cert=None):
         retries = 0
-        while True:
+        while 1:
 
             try:
                 if clean:
@@ -353,9 +354,8 @@ class Updater(object):
                 self._stop_httpd()
                 self._stop_dispatcher()
                 self._join_threads()
-                # async threads must be join()ed only after the dispatcher
-                # thread was joined, otherwise we can still have new async
-                # threads dispatched
+                # async threads must be join()ed only after the dispatcher thread was joined,
+                # otherwise we can still have new async threads dispatched
                 self._join_async_threads()
 
     def _stop_httpd(self):
@@ -373,16 +373,17 @@ class Updater(object):
     def _join_async_threads(self):
         with dispatcher.ASYNC_LOCK:
             threads = list(dispatcher.ASYNC_THREADS)
-        total = len(threads)
+            total = len(threads)
 
-        for i in range(total):
-            dispatcher.ASYNC_QUEUE.put(0)
+            # Stop all threads in the thread pool by put()ting one non-tuple per thread
+            for i in range(total):
+                dispatcher.ASYNC_QUEUE.put(None)
 
-        for i, thr in enumerate(threads):
-            self.logger.debug('Waiting for async thread {0}/{1} to end'.format(i + 1, total))
-            thr.join()
-            dispatcher.ASYNC_THREADS.remove(thr)
-            self.logger.debug('async thread {0}/{1} has ended'.format(i + 1, total))
+            for i, thr in enumerate(threads):
+                self.logger.debug('Waiting for async thread {0}/{1} to end'.format(i + 1, total))
+                thr.join()
+                dispatcher.ASYNC_THREADS.remove(thr)
+                self.logger.debug('async thread {0}/{1} has ended'.format(i + 1, total))
 
     def _join_threads(self):
         for thr in self.__threads: