Python Threading: Can I Sleep On Two Threading.event()s Simultaneously?

If I have two threading.Event() objects, and wish to sleep until either one of them is set, is there an efficient way to do that in python? Clearly I could do something with polli

Solution 1:

Here is a non-polling non-excessive thread solution: modify the existing Events to fire a callback whenever they change, and handle setting a new event in that callback:

import threading

defor_set(self):
    self._set()
    self.changed()

defor_clear(self):
    self._clear()
    self.changed()

deforify(e, changed_callback):
    e._set = e.set
    e._clear = e.clear
    e.changed = changed_callback
    e.set = lambda: or_set(e)
    e.clear = lambda: or_clear(e)

defOrEvent(*events):
    or_event = threading.Event()
    defchanged():
        bools = [e.is_set() for e in events]
        ifany(bools):
            or_event.set()
        else:
            or_event.clear()
    for e in events:
        orify(e, changed)
    changed()
    return or_event

Sample usage:

def wait_on(name, e):
    print "Waiting on %s..." % (name,)
    e.wait()
    print "%s fired!" % (name,)

def test():
    import time

    e1 = threading.Event()
    e2 = threading.Event()

    or_e = OrEvent(e1, e2)

    threading.Thread(target=wait_on, args=('e1', e1)).start()
    time.sleep(0.05)
    threading.Thread(target=wait_on, args=('e2', e2)).start()
    time.sleep(0.05)
    threading.Thread(target=wait_on, args=('or_e', or_e)).start()
    time.sleep(0.05)

    print "Firing e1 in 2 seconds..."
    time.sleep(2)
    e1.set()
    time.sleep(0.05)

    print "Firing e2 in 2 seconds..."
    time.sleep(2)
    e2.set()
    time.sleep(0.05)

The result of which was:

Waiting on e1...
Waiting on e2...
Waiting on or_e...
Firing e1 in2 seconds...
e1 fired!or_e fired!

Firing e2 in2 seconds...
e2 fired!

This should be thread-safe. Any comments are welcome.

EDIT: Oh and here is your wait_for_either function, though the way I wrote the code, it's best to make and pass around an or_event. Note that the or_event shouldn't be set or cleared manually.

def wait_for_either(e1, e2):
    OrEvent(e1, e2).wait()

Solution 2:

I think the standard library provides a pretty canonical solution to this problem that I don't see brought up in this question: condition variables. You have your main thread wait on a condition variable, and poll the set of events each time it is notified. It is only notified when one of the events is updated, so there is no wasteful polling. Here is a Python 3 example:

from threading import Thread, Event, Condition
from time import sleep
from random import random

event1 = Event()
event2 = Event()
cond = Condition()

defthread_func(event, i):
    delay = random()
    print("Thread {} sleeping for {}s".format(i, delay))
    sleep(delay)

    event.set()
    with cond:
        cond.notify()

    print("Thread {} done".format(i))

with cond:
    Thread(target=thread_func, args=(event1, 1)).start()
    Thread(target=thread_func, args=(event2, 2)).start()
    print("Threads started")

    whilenot (event1.is_set() or event2.is_set()):
        print("Entering cond.wait")
        cond.wait()
        print("Exited cond.wait ({}, {})".format(event1.is_set(), event2.is_set()))

    print("Main thread done")

Example output:

Thread 1 sleeping for 0.31569427100177794s
Thread 2 sleeping for 0.486548134317051s
Threads started
Entering cond.wait
Thread 1 done
Exited cond.wait (True, False)
Main thread done
Thread 2 done

Note that wit no extra threads or unnecessary polling, you can wait for an arbitrary predicate to become true (e.g. for any particular subset of the events to be set). There's also a wait_for wrapper for the while (pred): cond.wait() pattern, which can make your code a bit easier to read.

Solution 3:

One solution (with polling) would be to do sequential waits on each Event in a loop

defwait_for_either(a, b):
    whileTrue:
        if a.wait(tunable_timeout):
            breakif b.wait(tunable_timeout):
            break

I think that if you tune the timeout well enough the results would be OK.

---

The best non-polling I can think of is to wait for each one in a different thread and set a shared Event whom you will wait after in the main thread.

def repeat_trigger(waiter, trigger):
    waiter.wait()
    trigger.set()

def wait_for_either(a, b):
    trigger = threading.Event()
    ta = threading.Thread(target=repeat_trigger, args=(a, trigger))
    tb = threading.Thread(target=repeat_trigger, args=(b, trigger))
    ta.start()
    tb.start()
    # Now do the union waiting
    trigger.wait()

---

Pretty interesting, so I wrote an OOP version of the previous solution:

classEventUnion(object):
    """Register Event objects and wait for release when any of them is set"""def__init__(self, ev_list=None):
        self._trigger = Event()
        if ev_list:
            # Make a list of threads, one for each Event
            self._t_list = [
                Thread(target=self._triggerer, args=(ev, ))
                for ev in ev_list
            ]
        else:
            self._t_list = []

    defregister(self, ev):
        """Register a new Event"""
        self._t_list.append(Thread(target=self._triggerer, args=(ev, )))

    defwait(self, timeout=None):
        """Start waiting until any one of the registred Event is set"""# Start all the threadsmap(lambda t: t.start(), self._t_list)
        # Now do the union waitingreturn self._trigger.wait(timeout)

    def_triggerer(self, ev):
        ev.wait()
        self._trigger.set()

Solution 4:

Starting extra threads seems a clear solution, not very effecient though. Function wait_events will block util any one of events is set.

def wait_events(*events):
    event_share = Event()

    def set_event_share(event):
        event.wait()
        event.clear()
        event_share.set()
    foreventin events:
        Thread(target=set_event_share(event)).start()

    event_share.wait()

wait_events(event1, event2, event3)

Solution 5:

Extending Claudiu's answer where you can either wait for:

1. event 1 OR event 2
2. event 1 AND even 2

---

from threading import Thread, Event, _Event

classConditionalEvent(_Event):
    def__init__(self, events_list, condition):
        _Event.__init__(self)

        self.event_list = events_list
        self.condition = condition

        for e in events_list:
            self._setup(e, self._state_changed)

        self._state_changed()

    def_state_changed(self):
        bools = [e.is_set() for e in self.event_list]
        if self.condition == 'or':                
            ifany(bools):
                self.set()
            else:
                self.clear()

        elif self.condition == 'and':                 
            ifall(bools):
                self.set()
            else:
                self.clear()

    def_custom_set(self,e):
        e._set()
        e._state_changed()

    def_custom_clear(self,e):
        e._clear()
        e._state_changed()

    def_setup(self, e, changed_callback):
        e._set = e.set
        e._clear = e.clear
        e._state_changed = changed_callback
        e.set = lambda: self._custom_set(e)
        e.clear = lambda: self._custom_clear(e)

Example usage will be very similar as before

import time

e1 = Event()
e2 = Event()

# Example to wait for triggering ofevent1ORevent2
or_e = ConditionalEvent([e1, e2], 'or')

# Example to wait for triggering ofevent1ANDevent2
and_e = ConditionalEvent([e1, e2], 'and')