Suspend two Python threads while the third does things (with locks?)

Question

I am new to parallel programming.

I would like to complete three tasks repeatedly. The first two should work all the time, the third should start every hour or so. The first two tasks can be performed in parallel, but I always want to pause them while the third task is being performed.

Here is the skeleton of what I tried:

import threading import time flock = threading.Lock() glock = threading.Lock() def f(): while True: with flock: print 'f' time.sleep(1) def g(): while True: with glock: print 'g' time.sleep(1) def h(): while True: with flock: with glock: print 'h' time.sleep(5) threading.Thread(target=f).start() threading.Thread(target=g).start() threading.Thread(target=h).start() 

I expect this code to print f and g every second, and h about every five seconds. However, when I run it, it takes about 12 f and 12 g before I start to see some h. It seems that the first two threads are constantly releasing and re-acquiring their locks, while the third thread remains out of the loop.

• Why? When a third thread tries to get a lock that is currently locked, and then it is released, should one immediately purchase an acquisition instead of the first / second thread immediately acquiring it again? I probably misunderstood something.
• What would be a good way to achieve what I want?

Note. Moving time.sleep(1) calls from a flock / glock block works for this simple example, but apparently not for my real application, where threads spend most of their time doing the actual operations. When the first two threads sleep for a second after each execution of the loop body, when the lock is released, the third task will never be completed anyway.

Answer 1

How to do this with threading.Events :

 import threading import time import logging logger=logging.getLogger(__name__) def f(resume,is_waiting,name): while True: if not resume.is_set(): is_waiting.set() logger.debug('{n} pausing...'.format(n=name)) resume.wait() is_waiting.clear() logger.info(name) time.sleep(1) def h(resume,waiters): while True: logger.debug('halt') resume.clear() for i,w in enumerate(waiters): logger.debug('{i}: wait for worker to pause'.format(i=i)) w.wait() logger.info('h begin') time.sleep(2) logger.info('h end') logger.debug('resume') resume.set() time.sleep(5) logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s %(threadName)s] %(message)s', datefmt='%H:%M:%S') # set means resume; clear means halt resume = threading.Event() resume.set() waiters=[] for name in 'fg': is_waiting=threading.Event() waiters.append(is_waiting) threading.Thread(target=f,args=(resume,is_waiting,name)).start() threading.Thread(target=h,args=(resume,waiters)).start() 

gives

 [07:28:55 Thread-1] f [07:28:55 Thread-2] g [07:28:55 Thread-3] halt [07:28:55 Thread-3] 0: wait for worker to pause [07:28:56 Thread-1] f pausing... [07:28:56 Thread-2] g pausing... [07:28:56 Thread-3] 1: wait for worker to pause [07:28:56 Thread-3] h begin [07:28:58 Thread-3] h end [07:28:58 Thread-3] resume [07:28:58 Thread-1] f [07:28:58 Thread-2] g [07:28:59 Thread-1] f [07:28:59 Thread-2] g [07:29:00 Thread-1] f [07:29:00 Thread-2] g [07:29:01 Thread-1] f [07:29:01 Thread-2] g [07:29:02 Thread-1] f [07:29:02 Thread-2] g [07:29:03 Thread-3] halt 

---

(In response to a question in the comments) This code tries to determine how long it takes for h -thread to get each lock from other worker threads.

It seems that even if h waiting for a lock, another worker thread may, with a rather high probability, release and re-lock the lock. There is no priority for h because he waited longer.

David Bezley introduced thread and GIL issues to PyCon. Here are the pdf slides . This is a fascinating read and can help explain it as well.

 import threading import time import logging logger=logging.getLogger(__name__) def f(lock,n): while True: with lock: logger.info(n) time.sleep(1) def h(locks): while True: t=time.time() for n,lock in enumerate(locks): lock.acquire() t2=time.time() logger.info('h acquired {n}: {d}'.format(n=n,d=t2-t)) t=t2 t2=time.time() logger.info('h {d}'.format(d=t2-t)) t=t2 for lock in locks: lock.release() time.sleep(5) logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s %(threadName)s] %(message)s', datefmt='%H:%M:%S') locks=[] N=5 for n in range(N): lock=threading.Lock() locks.append(lock) t=threading.Thread(target=f,args=(lock,n)) t.start() threading.Thread(target=h,args=(locks,)).start() 

Answer 2

Using communication for synchronization:

 #!/usr/bin/env python import threading import time from Queue import Empty, Queue def f(q, c): while True: try: q.get_nowait(); q.get() # get PAUSE signal except Empty: pass # no signal, do our thing else: q.get() # block until RESUME signal print c, time.sleep(1) def h(queues): while True: for q in queues: q.put_nowait(1); q.put(1) # block until PAUSE received print 'h' for q in queues: q.put(1) # put RESUME time.sleep(5) queues = [Queue(1) for _ in range(2)] threading.Thread(target=f, args=(queues[0], 'f')).start() threading.Thread(target=f, args=(queues[1], 'g')).start() threading.Thread(target=h, args=(queues,)).start() 

This may not be optimal in terms of performance, but it is much easier for me to follow.

Exit

 fg fgh fgfggffggfgffgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh fgfgfgfgfgfgfgh 

Answer 3

The easiest way to do this is with 3 Python processes. If you do this on Linux, then the hourly process may send a signal so that other tasks pause or you can even kill them, and then restart when the hourly task completes. No need for threads.

However, if you are configured to use streams, try to share NO data between streams, just send messages back and forth (you also know how to copy data, not exchange data). Threading is hard to get right.

But several processes make you not to share anything, and therefore it is much easier to do the right thing. If you use a library such as 0MQ http://www.zeromq.org to convey your message, it's easy to switch from a thread model to a multiprocess model.

Answer 4

What about semaphore initialized to 2? F and G wait and signal one block, H wait and signal 2 units.