互斥线程锁定，在互斥锁/锁释放时删除排队函数，在 Python 中？

Question

这是我遇到的问题：我使用的是 Python 2.7，并且我有一个在线程中运行的代码，该线程具有一个关键区域，当时只有一个线程应该执行。该代码目前没有互斥机制，所以我想询问我可以为我的特定用例使用什么，这涉及“删除”“排队”函数。我尝试使用以下最小工作示例来模拟这种行为：

useThreading=False # True

if useThreading:  from threading import Thread, Lock
else:             from multiprocessing import Process, Lock

mymutex = Lock()

import time
tstart = None

def processData(data):
  #~ mymutex.acquire()
  try:
    print('thread {0} [{1:.5f}] Do some stuff'.format(data, time.time()-tstart))
    time.sleep(0.5)
    print('thread {0} [{1:.5f}] 1000'.format(data, time.time()-tstart))
    time.sleep(0.5)
    print('thread {0} [{1:.5f}] done'.format(data, time.time()-tstart))
  finally:
    #~ mymutex.release()
    pass

# main:
tstart = time.time()
for ix in xrange(0,3):
  if useThreading: t = Thread(target = processData, args = (ix,))
  else: t = Process(target = processData, args = (ix,))
  t.start()
  time.sleep(0.001)

现在，如果你运行这段代码，你会得到这样的打印输出：

thread 0 [0.00173] Do some stuff
thread 1 [0.00403] Do some stuff
thread 2 [0.00642] Do some stuff
thread 0 [0.50261] 1000
thread 1 [0.50487] 1000
thread 2 [0.50728] 1000
thread 0 [1.00330] done
thread 1 [1.00556] done
thread 2 [1.00793] done

也就是说，三个线程很快就会一个接一个地“排队”（大约是 2-3 毫秒之后）。实际上，它们没有排队，它们只是在彼此之后 2-3 毫秒后开始并行执行。

现在，如果我启用 mymutex.acquire()/.release() 命令，我会得到预期的结果：

thread 0 [0.00174] Do some stuff
thread 0 [0.50263] 1000
thread 0 [1.00327] done
thread 1 [1.00350] Do some stuff
thread 1 [1.50462] 1000
thread 1 [2.00531] done
thread 2 [2.00547] Do some stuff
thread 2 [2.50638] 1000
thread 2 [3.00706] done

基本上，现在有了锁定，线程不会并行运行，而是由于锁定，它们一个接一个地运行——只要一个线程在工作，其他线程就会阻塞在.acquire()。但这也不是我想要实现的。

我想要实现的是：假设当.acquire() 首次由线程函数触发时，它会在队列中注册一个函数的 id（比如指向它的指针）。之后，行为与 Lock 基本相同 - 当一个线程工作时，其他线程在 .acquire() 处阻塞。当第一个线程完成后，它进入finally: 块——在这里，我想查看队列中有多少线程在等待；然后我想删除/删除最后一个线程的所有等待线程除了 - 最后，我要.release()锁；这意味着在此之后，队列中的最后一个线程将在接下来执行。我想，我想写类似下面的伪代码：

  ...
  finally:
    if (len(mymutex.queue) > 2): # more than this instance plus one other waiting:
      while (len(mymutex.queue) > 2):
        mymutex.queue.pop(1) # leave alone [0]=this instance, remove next element
    # at this point, there should be only queue[0]=this instance, and queue[1]= what was the last thread queued previously
    mymutex.release() # once we releace, queue[0] should be gone, and the next in the queue should acquire the mutex/lock..
    pass
  ...

这样，我希望得到这样的打印输出：

thread 0 [0.00174] Do some stuff
thread 0 [0.50263] 1000
thread 0 [1.00327] done
# here upon lock release, thread 1 would be deleted - and the last one in the queue, thread 2, would acquire the lock next:
thread 2 [1.00350] Do some stuff
thread 2 [1.50462] 1000
thread 2 [2.00531] done

在 Python 中实现此目的最直接的方法是什么？

Answer 1

您似乎想要一个类似队列的行为，那么为什么不使用Queue？

import threading
from  Queue import Queue
import time

# threads advertise to this queue when they're waiting
wait_queue = Queue()
# threads get their task from this queue
task_queue = Queue()

def do_stuff():
    print "%s doing stuff" % str(threading.current_thread())
    time.sleep(5)
def queue_thread(sleep_time):

    # advertise current thread waiting
    time.sleep(sleep_time)  
    wait_queue.put("waiting")  

    # wait for permission to pass
    message = task_queue.get()

    print "%s got task: %s" % (threading.current_thread(), message)
    # unregister current thread waiting
    wait_queue.get()

    if message == "proceed":
        do_stuff()
        # kill size-1 threads waiting
        for _ in range(wait_queue.qsize() - 1):
            task_queue.put("die")
        # release last
        task_queue.put("proceed")

    if message == "die":
        print "%s died without doing stuff" % threading.current_thread()
        pass

t1 = threading.Thread(target=queue_thread, args=(1, ))
t2 = threading.Thread(target=queue_thread, args=(2, ))
t3 = threading.Thread(target=queue_thread, args=(3, ))
t4 = threading.Thread(target=queue_thread, args=(4, ))

# allow first thread to pass
task_queue.put("proceed")

t1.start()
t2.start()
t3.start()
t4.start()

thread-1 首先到达并“获取”该部分，其他线程稍后会在队列中等待（并宣传他们正在等待）。然后，当线程 1 离开时，它通过告诉所有其他线程死亡并让最后一个线程继续进行，从而授予队列中最后一个线程的权限。

您可以使用不同的消息进行更精细的控制，典型的消息是wait_queue 中的线程ID（这样您就知道谁在等待，以及它到达的顺序）。

当您确定您的需要时，您可能可以利用非阻塞操作（queue.put(block=False) 和 queue.get(block=False)）。