[Python-Dev] futures API

[Thomas Nagy]

--- El vie, 10/12/10, Thomas Nagy escribió:
> --- El vie, 10/12/10, Brian Quinlan
> escribió:
> > On Dec 10, 2010, at 5:36 AM, Thomas Nagy wrote: 
> > > I have a process running for a long time, and
> which
> > may use futures of different max_workers count. I
> think it
> > is not too far-fetched to create a new futures object
> each
> > time. Yet, the execution becomes slower after each
> call, for
> > example with http://freehackers.org/~tnagy/futures_test.py:
> > > 
> > > """
> > > import concurrent.futures
> > > from queue import Queue
> > > import datetime
> > > 
> > > class counter(object):
> > >     def __init__(self, fut):
> > >         self.fut = fut
> > > 
> > >     def run(self):
> > >         def
> > look_busy(num, obj):
> > >         
> >    tot = 0
> > >         
> >    for x in range(num):
> > >             
> >    tot += x
> > >         
> >    obj.out_q.put(tot)
> > > 
> > >         start =
> > datetime.datetime.utcnow()
> > >         self.count = 0
> > >         self.out_q =
> > Queue(0)
> > >         for x in
> > range(1000):
> > >         
> >    self.count += 1
> > >         
> >    self.fut.submit(look_busy, self.count,
> > self)
> > > 
> > >         while
> > self.count:
> > >         
> >    self.count -= 1
> > >         
> >    self.out_q.get()
> > > 
> > >         delta =
> > datetime.datetime.utcnow() - start
> > >     
> >    print(delta.total_seconds())
> > > 
> > > fut =
> > concurrent.futures.ThreadPoolExecutor(max_workers=20)
> > > for x in range(100):
> > >     # comment the following line
> > >     fut =
> > concurrent.futures.ThreadPoolExecutor(max_workers=20)
> > >     c = counter(fut)
> > >     c.run()
> > > """
> > > 
> > > The runtime grows after each step:
> > > 0.216451
> > > 0.225186
> > > 0.223725
> > > 0.222274
> > > 0.230964
> > > 0.240531
> > > 0.24137
> > > 0.252393
> > > 0.249948
> > > 0.257153
> > > ...
> > > 
> > > Is there a mistake in this piece of code?
> > 
> > There is no mistake that I can see but I suspect that
> the
> > circular references that you are building are causing
> the
> > ThreadPoolExecutor to take a long time to be
> collected. Try
> > adding:
> > 
> >     c = counter(fut)
> >     c.run()
> > +    fut.shutdown()
> > 
> > Even if that fixes your problem, I still don't fully
> > understand this because I would expect the runtime to
> fall
> > after a while as ThreadPoolExecutors are collected.
> 
> The shutdown call is indeed a good fix :-) Here is the time
> response of the calls to counter() when shutdown is not
> called:
> http://www.freehackers.org/~tnagy/runtime_futures.png
> 
> After trying to stop the program by using CTRL+C, the
> following error may appear, after which the process cannot
> be interrupted:
> 
> """
> 19:18:12 /tmp/build> python3.2 futures_test.py
> 0.389657
> 0.417173
> 0.416513
> 0.421424
> 0.449666
> 0.482273
> ^CTraceback (most recent call last):
>   File "futures_test.py", line 36, in <module>
>     c.run()
>   File "futures_test.py", line 22, in run
>     self.fut.submit(look_busy, self.count, self)
>   File
> "/usr/local/lib/python3.2/concurrent/futures/thread.py",
> line 114, in submit
>     self._work_queue.put(w)
>   File "/usr/local/lib/python3.2/queue.py", line 135, in
> put
>     self.not_full.acquire()
> KeyboardInterrupt
> """
> 
> It is not expected, is it?

The problem also occurs when using a callback:
http://www.freehackers.org/~tnagy/futures_test2.py

If it is necessary to catch KeyboardInterrupt exceptions to cancel the futures execution, then how about adding this detail to the docs?

Thomas