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

Question

这是我遇到的问题：我正在使用 Python 2.7，并且我有一个代码运行在一个线程中，它有一个临界区，只有一个线程应该执行当时。该代码目前没有互斥机制，所以我想询问我可以为我的特定用例使用什么，它涉及 "dropping" 的 "queued" 函数。我尝试使用以下最小工作示例来模拟该行为：

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

也就是说，三个线程很快就一个接一个地得到"queued"（大约相隔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: 块 - 在这里，我想检查队列中有多少线程在等待；然后我想 delete/drop 所有等待线程 除了最后一个线程 - 最后，我 .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 首先到达，"acquires" 该部分，其他线程稍后到达队列中等待（并通告它们正在等待）。然后，当 thread-1 离开时，它通过告诉所有其他线程停止并让最后一个线程继续执行来为队列中的最后一个线程授予权限。

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

当您根据需要设置时，您可能可以利用非阻塞操作 (queue.put(block=False) and queue.get(block=False))。

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

Mutual exclusion thread locking, with dropping of queued functions upon mutex/lock release, in Python?

python

multithreading

mutex

critical-section