Threading pool similar to the multiprocessing Pool

Question

Is there a Pool class for worker threads  similar to the multiprocessing module s Pool class   I like for example the easy way to parallelize a map function  def long running func p       c func no gil p   p   multiprocessing Pool 4  xs   p map long running func  range 100     however I would like to do it without the overhead of creating new processes   I know about the GIL  However  in my usecase  the function will be an IO-bound C function for which the python wrapper will release the GIL before the actual function call   Do I have to write my own threading pool

User · Answer

There is no built in thread based pool  However  it can be very quick to implement a producer consumer queue with the Queue class   From  https   docs python org 2 library queue html  from threading import Thread from Queue import Queue def worker        while True          item   q get           do work item          q task done    q   Queue   for i in range num worker threads        t   Thread target worker       t daemon   True      t start    for item in source        q put item   q join           block until all tasks are done

User · Answer

Yes  there is a threading pool similar to the multiprocessing Pool  however  it is hidden somewhat and not properly documented  You can import it by following way - from multiprocessing pool import ThreadPool  Just I show you simple example def test multithread stringio read csv self             see gh-11786         max row range   10000         num files   100          bytes to df                   n  join                     d  d  d     i  i  i  for i in range max row range                 encode   for j in range num files           files    BytesIO b  for b in bytes to df             read all files in many threads         pool   ThreadPool 8          results   pool map self read csv  files          first result   results 0           for result in results              tm assert frame equal first result  result

User · Answer

For something very simple and lightweight  slightly modified from here    from Queue import Queue from threading import Thread   class Worker Thread          Thread executing tasks from a given tasks queue        def   init   self  tasks           Thread   init   self          self tasks   tasks         self daemon   True         self start        def run self           while True              func  args  kargs   self tasks get               try                  func  args    kargs              except Exception  e                  print e             finally                  self tasks task done     class ThreadPool         Pool of threads consuming tasks from a queue        def   init   self  num threads           self tasks   Queue num threads          for   in range num threads               Worker self tasks       def add task self  func   args    kargs              Add a task to the queue            self tasks put  func  args  kargs        def wait completion self              Wait for completion of all the tasks in the queue            self tasks join    if   name         main         from random import randrange     from time import sleep      delays    randrange 1  10  for i in range 100        def wait delay d           print  sleeping for   d sec    d         sleep d       pool   ThreadPool 20       for i  d in enumerate delays           pool add task wait delay  d       pool wait completion     To support callbacks on task completion you can just add the callback to the task tuple

User · Answer

Here s the result I finally ended up using  It s a modified version of the classes by dgorissen above   File  threadpool py  from queue import Queue  Empty import threading from threading import Thread   class Worker Thread        TIMEOUT   2         Thread executing tasks from a given tasks queue  Thread is signalable           to exit             def   init   self  tasks  th num           Thread   init   self          self tasks   tasks         self daemon  self th num   True  th num         self done   threading Event           self start        def run self                  while not self done is set                try                  func  args  kwargs   self tasks get block True                                                     timeout self  TIMEOUT                  try                      func  args    kwargs                  except Exception as e                      print e                  finally                      self tasks task done               except Empty as e                  pass         return      def signal exit self               Signal to thread to exit             self done set     class ThreadPool         Pool of threads consuming tasks from a queue        def   init   self  num threads  tasks              self tasks   Queue num threads          self workers              self done   False         self  init workers num threads          for task in tasks              self tasks put task       def  init workers self  num threads           for i in range num threads               self workers append Worker self tasks  i        def add task self  func   args    kwargs              Add a task to the queue            self tasks put  func  args  kwargs        def  close all threads self               Signal all threads to exit and lose the references to them             for workr in self workers              workr signal exit           self workers           def wait completion self              Wait for completion of all the tasks in the queue            self tasks join        def   del   self           self  close all threads     def create task func   args    kwargs       return  func  args  kwargs    To use the pool  from random import randrange from time import sleep  delays    randrange 1  10  for i in range 30    def wait delay d       print  sleeping for   d sec    d      sleep d   pool   ThreadPool 20  for i  d in enumerate delays       pool add task wait delay  d  pool wait completion

User · Answer

another way can be adding the process to thethread queue pool import concurrent futures with concurrent futures ThreadPoolExecutor max workers cpus  as executor      for i in range 10           a   executor submit arg1  arg2

User · Answer

Yes  and it seems to have  more or less  the same API    import multiprocessing  def worker lnk                def start process                    if PROCESS       pool   multiprocessing Pool processes POOL SIZE  initializer start process  else      pool   multiprocessing pool ThreadPool processes POOL SIZE                                              initializer start process   pool map worker  inputs

User · Answer

In Python 3 you can use concurrent futures ThreadPoolExecutor  i e    executor   ThreadPoolExecutor max workers 10  a   executor submit my function    See the docs for more info and examples

User · Answer

Hi to use the thread pool in Python you can use this library    from multiprocessing dummy import Pool as ThreadPool   and then for use  this library do like that    pool   ThreadPool threads  results   pool map service  tasks  pool close   pool join   return results   The threads are the number of threads that you want and tasks are a list of task that most map to the service

User · Answer

I just found out that there actually  is a thread-based Pool interface in the multiprocessing module  however it is hidden somewhat and not properly documented   It can be imported via  from multiprocessing pool import ThreadPool   It is implemented using a dummy Process class wrapping a python thread   This thread-based Process class can be found in multiprocessing dummy which is mentioned briefly in the docs   This dummy module supposedly provides the whole multiprocessing interface based on threads

User · Answer

The overhead of creating the new processes is minimal  especially when it s just 4 of them  I doubt this is a performance hot spot of your application  Keep it simple  optimize where you have to and where profiling results point to

User · Answer

If you don t mind executing other s code  here s mine  Note  There is lot of extra code you may want to remove  added for better clarificaiton and demonstration how it works  Note  Python naming conventions were used for method names and variable names instead of camelCase  Working procedure   MultiThread class will initiate with no of instances of threads by sharing lock  work queue  exit flag and results  SingleThread will be started by MultiThread once it creates all instances  We can add works using MultiThread  It will take care of locking   SingleThreads will process work queue using a lock in middle  Once your work is done  you can destroy all threads with shared boolean value  Here  work can be anything  It can automatically import  uncomment import line  and process module using given arguments  Results will be added to results and we can get using get results  Code  import threading import queue   class SingleThread threading Thread       def   init   self  name  work queue  lock  exit flag  results           threading Thread   init   self          self name   name         self work queue   work queue         self lock   lock         self exit flag   exit flag         self results   results      def run self             print  quot Coming  s with parameters  s quot   self name  self exit flag          while not self exit flag                print self exit flag              self lock acquire               if not self work queue empty                    work   self work queue get                   module  operation  args  kwargs   work module  work operation  work args  work kwargs                 self lock release                   print  quot Processing    quot    operation    quot  with parameters  quot    str args     quot  and  quot    str kwargs     quot  by  quot    self name    quot  n quot                     module     import   module name                  result   str getattr module  operation   args    kwargs                   print  quot Result    quot    result    quot  for operation  quot    operation    quot  and input  quot    str args     quot   quot    str kwargs                   self results append result              else                  self lock release             process work queue self work queue   class MultiThread      def   init   self  no of threads           self exit flag   bool instance           self queue lock   threading Lock           self threads              self work queue   queue Queue           self results              for index in range 0  no of threads               thread   SingleThread  quot Thread quot    str index 1   self work queue  self queue lock  self exit flag  self results              thread start               self threads append thread       def add work self  work           self queue lock acquire           self work queue  put work          self queue lock release        def destroy self           self exit flag value   True         for thread in self threads              thread join        def get results self           return self results   class Work      def   init   self  module  operation  args  kwargs              self module   module         self operation   operation         self args   args         self kwargs   kwargs   class SimpleOperations      def sum self   args           return sum  int arg  for arg in args         staticmethod     def mul a  b  c 0           return int a    int b    int c    class bool instance      def   init   self  value False           self value   value      def   setattr   self  key  value           if key     quot value quot               raise AttributeError  quot Only value can be set  quot           if not isinstance value  bool               raise AttributeError  quot Only True False can be set  quot           self   dict   key    value           super   setattr   key  bool value        def   bool   self           return self value  if   name       quot   main   quot       multi thread   MultiThread 5      multi thread add work Work SimpleOperations     quot mul quot    2  3     quot c quot  4        while True          data input   input           if data input     quot  quot               pass         elif data input     quot break quot               break         else              work   data input split               multi thread add work Work SimpleOperations    work 0   work 1             multi thread destroy       print multi thread get results

[python] Threading pool similar to the multiprocessing Pool?

Examples related to python

Examples related to multithreading

Examples related to missing-features