Python线程/多处理不需要Mutex？

Question

对于类似管理程序的项目,我使用线程库来管理一些子进程.在某些时候,用户可以提示命令将指令发送到进程管理线程.这些命令存储在主进程和进程管理线程之间共享的Queue对象中.我以为我需要互斥来解决并发问题,所以我做了一个小脚本试试,但没有互斥,首先要确保我得到预期的并发问题.

我希望从脚本中每秒打印一个凌乱的int列表:

import threading
import time

def longer(l, mutex=None):
    while 1:
        last_val = l[-1]
        l.append(last_val + 1)
        time.sleep(1)
    return

dalist = [0]
t = threading.Thread(target=longer, args=(dalist,))
t.daemon = True
t.start()

while 1:
    last_val = dalist[-1]
    dalist.append(last_val + 1)
    print dalist
    time.sleep(1)

但实际上它打印了一个很好的以下int列表,如下所示:

[0, 1, 2]
[0, 1, 2, 3]
[0, 1, 2, 3, 4, 5, 6]

从这个答案在另一篇文章我认为它来自线程库,所以我也与多LIB一样的:

import multiprocessing as mp
import time

def longer(l, mutex=None):
    while 1:
        last_val = l[-1]
        l.append(last_val + 1)
        time.sleep(1)
    return

dalist = [0]
t = mp.Process(target=longer, args=(dalist,))
t.start()

while 1:
    last_val = dalist[-1]
    dalist.append(last_val + 1)
    print dalist
    time.sleep(1)

但我得到了相同的结果,有点"慢":

[0, 1]
[0, 1, 2]
[0, 1, 2, 3]
[0, 1, 2, 3, 4]

所以我想知道我真的需要mutex来管理线程之间共享的类似Queue的对象吗??? 而且,从上面的代码之一,我怎样才能有效地重现我搜索的预期并发问题？

谢谢阅读

编辑1: 从user4815162342的评论中我改变了第一个片段,并通过在值检索和列表追加之间的'较长'函数内移动睡眠调用来设置某种竞争条件:

import threading
import time

def longer(l, mutex=None):
    while 1:
        last_val = l[-1]
        time.sleep(1)
        l.append(last_val + 1)
    return 

dalist = [0]
t = threading.Thread(target=longer, args=(dalist,))
t.daemon = True
t.start()

while 1:
    last_val = dalist[-1]
    dalist.append(last_val + 1)
    print dalist
    time.sleep(1)

给我这样的东西:

[0, 1]
[0, 1, 1, 2]
[0, 1, 1, 2, 2, 3]
[0, 1, 1, 2, 2, 3, 3, 4]

我设法使用线程锁解决我的人为问题:

import threading
import time

def longer(l, mutex=None):
    while 1:
        if mutex is not None:
            mutex.acquire()
        last_val = l[-1]
        time.sleep(1)
        l.append(last_val + 1)
        if mutex is not None:
            mutex.release()
    return

dalist = [0]
mutex = threading.Lock()
t = threading.Thread(target=longer, args=(dalist, mutex))
t.daemon = True
t.start()

while 1:
    if mutex is not None:
        mutex.acquire()
    last_val = dalist[-1]
    dalist.append(last_val + 1)
    if mutex is not None:
        mutex.release()
    print dalist
    time.sleep(1)

然后产生:

[0, 1, 2]
[0, 1, 2, 3, 4, 5]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

Answer 1

您的第一个代码片段包含竞争条件并且确实需要互斥体。全局解释器锁使得竞争条件很少见，因为在任何给定时间都有一个线程在运行。然而，每执行几个字节码指令，当前线程就会放弃全局解释器锁的所有权，以便为其他线程提供运行的机会。所以，给定你的代码：

last_val = dalist[-1]
dalist.append(last_val + 1)

如果字节码切换发生在执行第一行之后，另一个线程将选取相同的字节码last_val并将其附加到列表中。将控制权交还给初始线程后，存储的值last_val将第二次追加到列表中。互斥体会以明显的方式阻止竞争：列表访问和追加之间的上下文切换会将控制权交给另一个线程，但它会立即在互斥体中被阻塞，并将控制权交还给原始线程。

您的第二个示例仅“有效”，因为这两个进程具有单独的列表实例。修改一个列表不会影响另一个列表，因此另一个进程也可能没有运行。尽管multiprocessing有一个直接替代 API threading，但底层概念有很大不同，从一种切换到另一种时需要考虑这一点。