【问题标题】:Preventing thread duplication in a multiprocessed Python app防止多处理 Python 应用程序中的线程重复
【发布时间】:2014-04-25 23:29:55
【问题描述】:

我有一个 Python 应用程序,它在 multiprocessing.Process 启动的子进程中运行多个作业。父应用程序还启动一个线程来向数据库报告进度。但是,我注意到,如果任何作业启动它们自己的子进程,它们就会复制这个线程,从而导致数据库中的数据损坏。例如当一个孙子子进程完成时,它的线程在数据库中将父作业标记为完成,因为它认为它父进程,即使父进程仍在运行。

如何使用multiprocess.Process 使其不复制任何当前正在运行的线程?在我的线程中记录原始 PID 是最简单的选择,如果“当前”PID 与此值不匹配,则立即退出?

我在去年看到了这个similar question,但它似乎被忽略了。

【问题讨论】:

  • 简单的解决方案是在产生子进程后创建监控线程。这不可行吗?
  • 就我而言,这并不可行,因为我的父进程不断为新作业生成子进程。
  • @Cerin:您可以使用 fork-server(一个没有派生新子线程的线程的专用进程)。它内置在 Python 3.4 中,请参阅multiprocessing.set_start_method()

标签: python multithreading multiprocessing python-multithreading


【解决方案1】:

您对问题的描述表明父进程中的后台线程继续存在并在子进程中执行。那是不可能的;至少,在 POSIX 系统上是不可能的。在你的情况下发生的事情是另一回事。我将在下面对此进行一些推测,然后提出如何避免该问题的建议。依次取这些点...

1.只有一个线程在分叉中幸存下来。

只有调用fork()的线程在fork后还活着。下面是一个小例子,展示了其他线程不会在子进程中继续执行:

def output():
    time.sleep(3)
    print "Thread executing in process: %d" % os.getpid()

thread = threading.Thread(target=output)
thread.start()
os.fork()
print "Pid: %d" % os.getpid()

您会看到父进程和子进程都将它们的 pid 打印到标准输出,但第二个线程只会在父进程中产生输出。

所以让监视器线程检查它的 pid 或以其他方式检查它正在运行的进程不会有任何影响;该线程只在一个进程中执行。

2。分叉的某些方式可能会导致您所看到的问题。

分叉会以各种方式导致程序状态损坏。例如:

  • 由于分叉而死的线程中引用的对象可能会超出范围,从而调用其终结器。如果这样的对象表示网络资源并且调用其del 方法导致连接的一端意外关闭,这可能会导致问题。
  • 任何缓冲的 IO 都会导致问题,因为缓冲区在子进程中是重复的。

请注意,第二点甚至不需要线程。考虑以下几点:

f = open("testfile", "w", 1024)
f.write("a")
os.fork()

我们向testfile 写了一个字符,我们在分叉之前在父代中这样做了。但是我们在内容没有刷新的时候分叉了,所以:

alp:~ $ wc -c testfile
      2 testfile

该文件包含两个字符,因为输出缓冲区已复制到子级,并且父级和子级最终都刷新了它们的缓冲区。

我怀疑您的问题是由第二个问题引起的(尽管我很高兴地承认这纯粹是猜测)。

3.重新架构以避免此类问题。

您在评论中提到,您在生成工作人员后无法启动监控线程,因为您需要重复创建新工作人员。重组你正在做的事情以避免这种情况可能比你想象的要容易。与其为每个新的工作单元生成一个进程,不如创建一组由控制进程管理的长寿工人:控制器向队列提供需要处理的作业的规范;它在闲暇时这样做。每个工作人员无限循环,当它们到达并执行它们时从队列中提取作业。 (multiprocessing 的队列实现将保证每个作业描述仅由一个工作人员绘制。)因此您只需要在早期生成一次工作人员,并且可以在所有分叉完成后创建监控线程。

这是这种组织的示意图:

from multiprocessing import Process, Queue

def work(q):
    while True:
        job = q.get()
        if job is None:
            # We've been signaled to stop.
            break
        do_something_with(job)

queue = Queue()
NUM_WORKERS = 3
NUM_JOBS = 20

# Start workers.
for _ in range(NUM_WORKERS):
    p = Process(target=work, args=(queue,))
    p.start()

# Create your monitor thread here.

# Put work in the queue.  This continues as long as you want.
for i in range(NUM_JOBS):
    queue.put(i)

# When there's no more work, put sentinel values in the queue so workers
# know to gracefully exit.
for _ in range(NUM_WORKERS):
    queue.put(None)

【讨论】:

猜你喜欢
  • 2015-04-24
  • 2019-01-29
  • 2016-11-24
  • 1970-01-01
  • 2020-04-24
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多