How to break time.sleep() in a python concurrent.futures

Question

I am playing around with concurrent.futures.

Currently my future calls time.sleep(secs).

It seems that Future.cancel() does less than I thought.

If the future is already executing, then time.sleep() does not get cancel by it.

The same for the timeout parameter for wait(). It does not cancel my time.sleep().

How to cancel time.sleep() which gets executed in a concurrent.futures?

For testing I use the ThreadPoolExecutor.

Answer 1

If you submit a function to a ThreadPoolExecutor, the executor will run the function in a thread and store its return value in the Future object. Since the number of concurrent threads is limited, you have the option to cancel the pending execution of a future, but once control in the worker thread has been passed to the callable, there's no way to stop execution.

Consider this code:

import concurrent.futures as f
import time

T = f.ThreadPoolExecutor(1) # Run at most one function concurrently
def block5():
    time.sleep(5)
    return 1
q = T.submit(block5)
m = T.submit(block5)

print q.cancel()  # Will fail, because q is already running
print m.cancel()  # Will work, because q is blocking the only thread, so m is still queued

In general, whenever you want to have something cancellable you yourself are responsible for making sure that it is.

There are some off-the-shelf options available though. E.g., consider using asyncio, they also have an example using sleep. The concept circumvents the issue by, whenever any potentially blocking operation is to be called, instead returning control to a control loop running in the outer-most context, together with a note that execution should be continued whenever the result is available - or, in your case, after n seconds have passed.

Answer 2

I do not know much about concurrent.futures, but you can use this logic to break the time. Use a loop instead of sleep.time() or wait()

for i in range(sec):
    sleep(1)

interrupt or break can be used to come out of loop.

Answer 3

I figured it out.

Here is a example:

from concurrent.futures import ThreadPoolExecutor
import queue
import time

class Runner:
    def __init__(self):
        self.q = queue.Queue()
        self.exec = ThreadPoolExecutor(max_workers=2)

    def task(self):
        while True:
            try:
                self.q.get(block=True, timeout=1)
                break
            except queue.Empty:
                pass
            print('running')

    def run(self):
        self.exec.submit(self.task)

    def stop(self):
        self.q.put(None)
        self.exec.shutdown(wait=False,cancel_futures=True)

r = Runner()
r.run()
time.sleep(5)
r.stop()

Answer 4

As it is written in its link, You can use a with statement to ensure threads are cleaned up promptly, like the below example:

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

Answer 5

I've faced this same problem recently. I had 2 tasks to run concurrently and one of them had to sleep from time to time. In the code below, suppose task2 is the one that sleeps.

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=2)
executor.submit(task1)
executor.submit(task2)

executor.shutdown(wait=True)

In order to avoid the endless sleep I've extracted task2 to run synchronously. I don't whether it's a good practice, but it's simple and fit perfectly in my scenario.

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=1)
executor.submit(task1)

task2()

executor.shutdown(wait=True)

Maybe it's useful to someone else.