abortable sleep() in Python

Question

I need a sleep() method which can be aborted (as described here or here).

My approach is to let a threading.Event.wait() timeout at the specified duration:

def abortable_sleep(secs, abort_event):
    abort_event.wait(timeout=secs)
    abort_event.clear()

After calling abortable_sleep(10, _abort) I can now (from another thread) call _event.set(_abort) to let abortable_sleep() terminate before the 10 seconds.

Example:

def sleeping_thread():
    _start = time.perf_counter()
    print("%f thread started" % (time.perf_counter() - _start))
    abortable_sleep(5, _abort)
    print("%f thread stopped" % (time.perf_counter() - _start))

if __name__ == '__main__':

    _abort = threading.Event()
    while True:
        threading.Thread(target=sleeping_thread).start()
        time.sleep(3)
        _abort.set()
        time.sleep(1)

Output:

0.000001 thread started
3.002668 thread stopped
0.000002 thread started
3.003014 thread stopped
0.000001 thread started
3.002928 thread stopped
0.000001 thread started

This code is working as expected but I still have some questions:

• isn't there an easier way to have s.th. likea sleep() which can be aborted?
• can this be done more elegant? E.g. this way I have to be careful with the Event instance which is not bound to an instance of abortable_sleep()
• do I have to expect performance issues with high frequency loops like while True: abortable_sleep(0.0001)? How is the wait()-timeout implemented?

Answer 1

I have a wrapper class which basically slaps some sleep semantics on top of an Event. The nice thing is that you only have to pass around a Sleep object, which you can call sleep() on several times if you like (sleep() is not thread safe though) and that you can wake() from another thread.

from threading import Event

class Sleep(object):
    def __init__(self, seconds, immediate=True):
        self.seconds = seconds
        self.event = Event()
        if immediate:
            self.sleep()

    def sleep(self, seconds=None):
        if seconds is None:
            seconds = self.seconds
        self.event.clear()
        self.event.wait(timeout=seconds)

    def wake(self):
        self.event.set()

Usage example:

if __name__ == '__main__':
    from threading import Thread
    import time
    import logging

    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(created)d - %(message)s')
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    logger.info("sleep")
    s = Sleep(3)
    logger.info("awake")

    def wake_it(sleeper):
        time.sleep(1)
        logger.info("wakeup!")
        sleeper.wake()

    logger.info("sleeping again")
    s = Sleep(60, immediate=False)
    Thread(target=wake_it, args=[s]).start()
    s.sleep()
    logger.info("awake again")

The above might output something like this:

1423750549 - sleep
1423750552 - awake
1423750552 - sleeping again
1423750553 - wakeup!
1423750553 - awake again

Exactly what you did, but encapsulated in a class.

Answer 2

Due to race conditions, your solution is not always perfectly correct. You should use a threading.BoundedSemaphore() instead. Call aquire() immediately after creating it. When you want to sleep, call acquire() with a timeout, then call release() if the acquire() returned true. To abort the sleep early, call release() from a different thread; this will raise ValueError if there is no sleep in progress.

Using an event instead is problematic if the other thread calls set() at the wrong time (i.e. at any time other than when you are actually waiting on the event).

Answer 3

I'd wrap the sleep/abort function up in a new class:

class AbortableSleep():
    def __init__(self):
        self._condition = threading.Condition()

    def __call__(self, secs):
        with self._condition:
            self._aborted = False
            self._condition.wait(timeout=secs)
            return not self._aborted

    def abort(self):
        with self._condition:
            self._condition.notify()
            self._aborted = True

I'd then also supply a Thread subclass to manage the sharing of the wakeup routine on a per-thread basis:

class ThreadWithWakeup(threading.Thread):
    def __init__(self, *args, **kwargs):
        self.abortable_sleep = AbortableSleep()
        super(ThreadWithWakeup, self).__init__(*args, **kwargs)

    def wakeup(self):
        self.abortable_sleep.abort()

Any other thread with access to this thread can call wakeup() to abort the current abortable_sleep() (if one is in progress).

---

Using ThreadWithWakeup

You can create threads using the ThreadWithWakeup class, and use it like this:

class MyThread(ThreadWithWakeup):
    def run(self):
        print "Sleeper: sleeping for 10"
        if self.abortable_sleep(10):
            print "Sleeper: awoke naturally"
        else:
            print "Sleeper: rudely awoken"

t = MyThread()
t.start()
print "Main: sleeping for 5"
for i in range(5):
    time.sleep(1)
    print i + 1 
print "Main: waking thread"
t.wakeup()

The output of which looks like:

Sleeper: sleeping for 10
Main: sleeping for 5
1
2
3
4
5
Main: waking thread
Sleeper: rudely awoken

---

Using AbortableSleep on its own

You can also use the AbortableSleep class on its own, which is handy if you can't use the ThreadWithWakeup class for some reason (maybe you're in the main thread, maybe something else creates the threads for you, etc.):

abortable_sleep = AbortableSleep()
def run():
    print "Sleeper: sleeping for 10"
    if abortable_sleep(10):
        print "Sleeper: awoke naturally"
    else:
        print "Sleeper: rudely awoken"
threading.Thread(target=run).start()

print "Main: sleeping for 5"
for i in range(5):
    time.sleep(1)
    print i + 1
print "Main: aborting"
abortable_sleep.abort()