Commits

Anonymous committed be8e564

ThreadPool: replace __len__() with threads() for getting number of threads in thread pool. More explicit meaning, so clearer and less change of unexpected behavior. Renamed get_idle_threads() to idle_threads(), get_queue_size() to qsize().
Reactor unit test checkes that thread pool is being used.

Comments (0)

Files changed (5)

asyncthreads/reactor.py

         threading.Thread.__init__(self, None, None, None)
         self._call_queue = queue.Queue()
         self.idle_wake_sec = None
-        if thread_pool:
+        if thread_pool is not None:
             self._thread_pool = thread_pool
         else:
             self._thread_pool = None

asyncthreads/threadpool.py

             # Pool is manually resizeable only.
             self._max_size = min_size
 
-
-    def __len__(self):
-        """Return the number of threads in the pool."""
-        return self._threads
-
     def submit(self, function, args=None, callback=None):
         """Insert a task function into the queue.
 
             self._task_queue.put((function, args, callback), False)
             self._adjust_thread_count()
 
-    def get_queue_size(self):
+    def threads(self):
+        """Return the number of threads in the pool."""
+        return self._threads
+
+    def idle_threads(self):
+        return int(self._idle_counter)
+
+    def qsize(self):
         """Return the number of items currently in task queue."""
         return self._task_queue.qsize()
 
-    def get_idle_threads(self):
-        return int(self._idle_counter)
-
     def shutdown(self, wait_for_threads=True, drop_queued_tasks=False):
         """Shutdown the thread pool.
 
 def main():
     setup(
         name='asyncthreads',
-        version= '1.4.3',
+        version= '1.4.4',
         author='Andrew Gillis',
         author_email='gillis.andrewj@gmail.com',
         url='http://bitbucket.org/agillis/asyncthreads',

test/test_reactor.py

     def setup_class(cls):
         gc.set_debug(gc.DEBUG_UNCOLLECTABLE)
 
-        cls.rktr = Reactor(ThreadPool(MIN_THREADS, MAX_THREADS))
+        cls.tp = ThreadPool(MIN_THREADS, MAX_THREADS)
+        cls.rktr = Reactor(cls.tp)
         cls.called = False
         cls.cb_event = threading.Event()
         cls.cb_rsp = None
         assert ret == 'hello-world'
         assert self.cb_rsp == 'hello-world'
 
+        print 'Threads in pool:', TestReactor.tp.threads()
+        assert TestReactor.tp.thread_count() > 0
+
     def test_call_scheduled(self):
         """Run scheduled in main thread"""
         print('Calling function in main thread in 3 seconds')

test/test_threadpool.py

         self.pool.submit(sort_task, (205, 200005), task_callback)
 
         for _ in range(3):
-            print('\n---> queued items: %s' % self.pool.get_queue_size())
-            print('---> idle threads: %s' % self.pool.get_idle_threads())
-            print('---> thread count: %s' % len(self.pool))
+            print('\n---> queued items: %s' % self.pool.qsize())
+            print('---> idle threads: %s' % self.pool.idle_threads())
+            print('---> thread count: %s' % self.pool.threads())
             time.sleep(2)
 
+        assert self.pool.threads() <= POOL_MAX
+
         self.pool.resize_pool(80)
-        assert len(self.pool) == 80
+        assert self.pool.threads() == 80
 
-        while self.pool.get_queue_size():
-            print('\n---> queued items: %s' % self.pool.get_queue_size())
-            print('---> idle threads: %s' % self.pool.get_idle_threads())
-            print('---> thread count: %s' % len(self.pool))
+        while self.pool.qsize():
+            print('\n---> queued items: %s' % self.pool.qsize())
+            print('---> idle threads: %s' % self.pool.idle_threads())
+            print('---> thread count: %s' % self.pool.threads())
             time.sleep(5)
 
-        print('\n---> queued items: %s' % self.pool.get_queue_size())
-        print('---> idle threads: %s' % self.pool.get_idle_threads())
-        print('---> thread count: %s' % len(self.pool))
+        print('\n---> queued items: %s' % self.pool.qsize())
+        print('---> idle threads: %s' % self.pool.idle_threads())
+        print('---> thread count: %s' % self.pool.threads())
 
-        while len(self.pool) > POOL_MIN:
+        while self.pool.threads() > POOL_MIN:
             self.pool.submit(range, (0,1))
-            print('\n---> queued items: %s' % self.pool.get_queue_size())
-            print('---> idle threads: %s' % self.pool.get_idle_threads())
-            print('---> thread count: %s' % len(self.pool))
+            print('\n---> queued items: %s' % self.pool.qsize())
+            print('---> idle threads: %s' % self.pool.idle_threads())
+            print('---> thread count: %s' % self.pool.threads())
             time.sleep(0.25)
 
         print('finish')
         self.pool.shutdown(False, False)
 
         while True:
-            print('\n---> queued items: %s' % self.pool.get_queue_size())
-            print('---> idle threads: %s' % self.pool.get_idle_threads())
-            print('---> thread count: %s' % len(self.pool))
+            print('\n---> queued items: %s' % self.pool.qsize())
+            print('---> idle threads: %s' % self.pool.idle_threads())
+            print('---> thread count: %s' % self.pool.threads())
             time.sleep(5)
-            if not self.pool.get_queue_size():
+            if not self.pool.qsize():
                 break
 
     def test_memory_leaks(self):