A class which works as a FIFO 'gate' for threads. By default the gate is open and any thread calling on the 'enter' method returns immediately. A thread can 'close' the gate by calling the method 'close'. This thread becomes the owner of the gate. Any other thread calling 'ente
| 5 | pass |
| 6 | |
| 7 | class ThreadGate(object): |
| 8 | """ A class which works as a FIFO 'gate' for threads. By |
| 9 | default the gate is open and any thread calling on the |
| 10 | 'enter' method returns immediately. |
| 11 | |
| 12 | A thread can 'close' the gate by calling the method 'close'. |
| 13 | This thread becomes the owner of the gate. Any other thread |
| 14 | calling 'enter' after this is automatically blocked till |
| 15 | the owner calls reopens the gate by calling 'open'. |
| 16 | |
| 17 | The gate requires a certain number of threads to block |
| 18 | before the owner exits from the 'close' method. Otherwise, |
| 19 | the owner waits for a timeout before returning from the 'close' |
| 20 | method, without actually closing the gate. |
| 21 | |
| 22 | The gate class can be used to block running threads for a |
| 23 | particular operation and making sure that they resume after |
| 24 | the operation is complete, for a fixed number of threads. |
| 25 | """ |
| 26 | |
| 27 | def __init__(self, numthreads, timeout=0): |
| 28 | self.lock = threading.Lock() |
| 29 | self.sem = threading.BoundedSemaphore(1) |
| 30 | self.evt = threading.Event() |
| 31 | self.count = 0 |
| 32 | self.owner_timeout = timeout |
| 33 | self.owner = None |
| 34 | self.nthreads = numthreads |
| 35 | # Open by default |
| 36 | self.position = 1 |
| 37 | |
| 38 | def close(self): |
| 39 | """ Close the gate. The calling thread |
| 40 | becomes the owner of the gate and blocks |
| 41 | till the requisite number of threads block |
| 42 | on the gate or a timeout occurs, whichever |
| 43 | is first. |
| 44 | |
| 45 | It is an error if the gate is already closed """ |
| 46 | |
| 47 | if self.position == 0: |
| 48 | # Already closed |
| 49 | raise ThreadGateException,"trying to close an already closed gate" |
| 50 | |
| 51 | try: |
| 52 | self.lock.acquire() |
| 53 | self.position = 0 |
| 54 | self.owner = threading.currentThread() |
| 55 | self.sem.acquire() |
| 56 | finally: |
| 57 | self.lock.release() |
| 58 | |
| 59 | # Wait on the event till timeout |
| 60 | self.evt.clear() |
| 61 | self.evt.wait(self.owner_timeout) |
| 62 | |
| 63 | # If event was set, requisite number off |
| 64 | # threads have blocked, else reset the gate |