This class implements an interface similar to threading.Event using os.pipe() as the mechanism for signalling. This allows the class to be used to wake up select.select() and select.poll() loops without having to peridiodically resume execution to check the status of an event.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33
import select import os class WaitableEvent: """Provides an abstract object that can be used to resume select loops with indefinite waits from another thread or process. This mimics the standard threading.Event interface.""" def __init__(self): self._read_fd, self._write_fd = os.pipe() def wait(self, timeout=None): rfds, wfds, efds = select.select([self._read_fd], , , timeout) return self._read_fd in rfds def isSet(self): return self.wait(0) def clear(self): if self.isSet(): os.read(self._read_fd, 1) def set(self): if not self.isSet(): os.write(self._write_fd, '1') def fileno(self): """Return the FD number of the read side of the pipe, allows this object to be used with select.select().""" return self._read_fd def __del__(self): os.close(self._read_fd) os.close(self._write_fd)
This is good for many reasons, including allowing the CPU to stay in a low power mode (eg. in laptops/handhelds) rather than waking up once per second (or some other frequency) just to check if an event has been set.
The class can also be used to signal between pieces of code across a fork() - creating a WaitableEvent, forking, then setting the event in the child process wil result in the parent's event being set.