Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
Demonstrates a way to run blocking functions which require Ctrl-C handling capability on the main thread via queues from other threads in Python
from Queue import Queue, Empty
from functools import wraps
from threading import Thread, Semaphore
ctrlc_queue = Queue()
exit_sem = Semaphore(0) # Read: there are 0 ctrlc threads on start-up
def needs_ctrlc(fun):
def wrapped(*args, **kwargs):
result_queue = Queue()
ctrlc_queue.put((fun, args, kwargs, result_queue))
result, exception = result_queue.get()
if exception:
raise exception
return result
return wrapped
def process_input(threadno):
print raw_input("[%d] Write something: " % threadno)
def thread_loop(threadno):
exit_sem.release() # Read: add 1 to number of ctrlc threads
for _ in xrange(5):
except KeyboardInterrupt:
print "[%d] Caught KeyboardInterrupt, exiting thread" % threadno
print "[%d] Finished input, exiting thread" % threadno
exit_sem.acquire() # Read: subtract 1 from number of ctrlc threads
def ctrlc_threads_exist():
"""As soon as exit_sem.acquire() blocks, we know there are no mor threads"""
ctrlc_threads_exist = exit_sem.acquire(False)
if ctrlc_threads_exist:
return ctrlc_threads_exist
if __name__ == '__main__':
for i in xrange(5):
Thread(target=thread_loop, args=(i,)).start()
# you *may* need to sleep a millisecond or two here
while ctrlc_threads_exist():
fun, args, kwargs, result_queue = ctrlc_queue.get(timeout=1)
except Empty:
continue # must check exit_event regularly
except KeyboardInterrupt:
continue # unintentional Ctrl-C drugin Queue.get
result = fun(*args, **kwargs)
except BaseException, e:
result_queue.put((None, e))
result_queue.put((result, None))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment