多处理中的外部与内部__main__变量定义答案

【问题标题】：outside vs inside __main__ variable definition in multiprocessing多处理中的外部与内部__main__变量定义
【发布时间】：2015-09-03 00:31:19
【问题描述】：

我有以下代码：

import multiprocessing
import time
import os

# WHEN SEMAPHORE IS DEFINED HERE THEN IT IT WORKS
semaphore = multiprocessing.Semaphore(1)

def producer(num, output):
  semaphore.acquire()
  time.sleep(1)
  element = "PROCESS: %d PID: %d PPID: %d" % (num, os.getpid(), os.getppid())
  print "WRITE -> " + element
  output.put(element)
  time.sleep(1)
  semaphore.release()

if __name__ == '__main__':
    """
    Reads elements as soon as they are are put inside queue
    """

    output    = multiprocessing.Manager().Queue()
    pool      = multiprocessing.Pool(4)
    lst       = range(40)

    # WHEN SEMAPHORE IS DEFINED HERE THEN IT DOES NOT WORKS
    # semaphore = multiprocessing.Semaphore(1)

    for i in lst:
        pool.apply_async(producer, (i, output))
        # print "%d Do not wait!" % i
        # res.get()

    counter = 0
    while True:
      try:
        print "READ  <- " + output.get_nowait()
        counter += 1
        if (counter == len(lst)):
          print "Break"
          break
      except:
        print "READ  <- NOTHING IN BUFFER"  
        pass
      time.sleep(1)

此代码按预期工作并打印：

READ  <- NOTHING IN BUFFER
WRITE -> PROCESS: 0 PID: 15803 PPID: 15798
READ  <- NOTHING IN BUFFER
READ  <- PROCESS: 0 PID: 15803 PPID: 15798
READ  <- NOTHING IN BUFFER
WRITE -> PROCESS: 1 PID: 15806 PPID: 15798
READ  <- PROCESS: 1 PID: 15806 PPID: 15798
...

然后我有这个版本不工作（它与第一个基本相同，除了信号量的定义在另一个地方）：

import multiprocessing
import time
import os

# WHEN SEMAPHORE IS DEFINED HERE THEN IT IT WORKS
# semaphore = multiprocessing.Semaphore(1)

def producer(num, output):
  print hex(id(semaphore))
  semaphore.acquire()
  time.sleep(1)
  element = "PROCESS: %d PID: %d PPID: %d" % (num, os.getpid(), os.getppid())
  print "WRITE -> " + element
  output.put(element)
  time.sleep(1)
  semaphore.release()

if __name__ == '__main__':
    """
    Reads elements as soon as they are are put inside queue
    """

    output    = multiprocessing.Manager().Queue()
    pool      = multiprocessing.Pool(4)
    lst       = range(40)

    # WHEN SEMAPHORE IS DEFINED HERE THEN IT DOES NOT WORKS
    semaphore = multiprocessing.Semaphore(1)

    for i in lst:
        pool.apply_async(producer, (i, output))
        # print "%d Do not wait!" % i
        # res.get()

    counter = 0
    while True:
      try:
        print "READ  <- " + output.get_nowait()
        counter += 1
        if (counter == len(lst)):
          print "Break"
          break
      except:
        print "READ  <- NOTHING IN BUFFER"  
        pass
      time.sleep(1)

此版本打印：

READ  <- NOTHING IN BUFFER
READ  <- NOTHING IN BUFFER
READ  <- NOTHING IN BUFFER
READ  <- NOTHING IN BUFFER
READ  <- NOTHING IN BUFFER
READ  <- NOTHING IN BUFFER
READ  <- NOTHING IN BUFFER
...

似乎producer 从不向队列写入任何内容。我在某处读到apply_sync 不打印错误消息。所以我在第二个代码中将pool.apply_async(producer, (i, output)) 更改为pool.apply(producer, (i, output))，看看发生了什么。 semaphore 好像没有定义，输出如下：

Traceback (most recent call last):
  File "glob_var_wrong.py", line 31, in <module>
    pool.apply(producer, (i, output))
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 244, in apply
    return self.apply_async(func, args, kwds).get()
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 567, in get
    raise self._value
NameError: global name 'semaphore' is not defined

但是以下代码运行正确并打印 10（在 __main__ 中定义的值）：

global_var = 20

def print_global_var():
    print global_var

if __name__ == '__main__':
    global_var = 10
    print_global_var()

似乎在这段代码中可以在__main__ 中定义全局变量，而在以前的代码中这是不可能的。首先，我假设在__main__ 中定义的变量不会在进程之间共享，但它只影响semaphore 而不会影响output、pool、lst。为什么会这样？

【问题讨论】：

与__main__无关。您必须在创建池之前定义信号量。如果您在创建池之前将行 semaphore = ... 向上移动，您的代码将起作用。
您的意思是在multiprocessing.Pool(4) 之前？这有什么关系？
@mehtunguh 谢谢你这样做了。您能否发表您的评论作为答案并解释背景细节（我仍然不明白为什么会这样），以便我接受。谢谢

标签： python multithreading multiprocessing python-multiprocessing

【解决方案1】：

当您使用Multiprocessing.Process（Pool 在后台使用）创建新进程时，它会复制本地范围，对其进行腌制，然后将其发送到新进程进行评估。

因为您在调用Pool(4) 之前没有定义变量semaphore，所以该变量未定义（在那些对代码进行评估的其他进程中）并且函数producer 将引发异常。

要查看此内容，请更改定义

def producer(num, output):
    print hex(id(semaphore))
    try:
        semaphore.acquire()
    except Exception as e:
        print e
    time.sleep(1)
    element = "PROCESS: %d PID: %d PPID: %d" % (num, os.getpid(), os.getppid())
    print "WRITE -> " + element
    output.put(element)
    time.sleep(1)
    semaphore.release()

现在您的失败代码将打印出一堆 (40) 看起来像的错误

global name 'semaphore' is not defined

这就是为什么在调用Pool之前必须定义信号量的原因

【讨论】：