python - 多处理队列，存储列表列表

我试图将进程中的列表列表（在下面名为proc的函数中）排队，然后在调用event.set()后让进程自行终止。从打印结果来看，我的功能总是完成的，但过程本身仍在进行如果我将调用proc时排队的列表数设置为较低（put变量）（或使每个嵌套列表的大小变小），我就可以实现这一点。

import multiprocessing as mp
import queue
import numpy as np
import time

def main():
    trainbatch_q = mp.Queue(10)

    batchperq = 50
    event = mp.Event()

    tl1 = mp.Process(target=proc,
                            args=( trainbatch_q, 20, batchperq, event))
    tl1.start()
    time.sleep(3)
    event.set()
    tl1.join()
    print("Never printed..")

def proc(batch_q, batch_size, batchperentry, the_event):
    nrow = 100000
    i0 = 0
    to_q = []
    while i0 < nrow:
        rowend = min(i0 + batch_size,nrow)
        somerows = np.random.randint(0,5,(rowend-i0,2))
        to_q.append(somerows.tolist())
        if len(to_q) == batchperentry:
            print("adding..", i0, len(to_q))
            while not the_event.is_set():
                try:
                    batch_q.put(to_q, block=False)
                    to_q = []
                    break
                except queue.Full:
                    time.sleep(1)
        i0 += batch_size
    print("proc finishes")

当我做键盘中断时，我得到下面的轨迹它想要的“锁”是什么跟排队有关系吗？

Traceback (most recent call last):
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/multiprocessing/process.py", line 252, in _bootstrap
    util._exit_function()
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/multiprocessing/util.py", line 322, in _exit_function
    _run_finalizers()
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/multiprocessing/util.py", line 262, in _run_finalizers
    finalizer()
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/multiprocessing/util.py", line 186, in __call__
    res = self._callback(*self._args, **self._kwargs)
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/multiprocessing/queues.py", line 198, in _finalize_join
    thread.join()
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/threading.py", line 1056, in join
    self._wait_for_tstate_lock()
  File "/software/Anaconda3-5.0.0.1-el7-x86_64/lib/python3.6/threading.py", line 1072, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt

最佳答案

你的进程从不退出的原因是因为你从来没有告诉过它退出。我在函数的末尾添加了一个return，您的进程现在似乎正确退出。

def proc(batch_q, batch_size, batchperentry, the_event):
    nrow = 100000
    i0 = 0
    to_q = []
    while i0 < nrow:
        rowend = min(i0 + batch_size,nrow)
        somerows = np.random.randint(0,5,(rowend-i0,2))
        to_q.append(somerows.tolist())
        if len(to_q) == batchperentry:
            print("adding..", i0, len(to_q))
            while not the_event.is_set():
                try:
                    batch_q.put(to_q, block=False)
                    to_q = []
                    break
                except queue.Full:
                    time.sleep(1)
        i0 += batch_size
    print("proc finishes")
    return            # Added this line, You can have it return whatever is most relevant to you.

这里是我运行的完整程序，包括我的更改，使它成功退出。

import multiprocessing as mp
import queue
import numpy as np
import random
import time

def main():
    trainbatch_q = mp.Queue(10)
    batchperq = 50
    event = mp.Event()

    tl1 = mp.Process(target=proc,
                            args=( trainbatch_q, 20, batchperq, event))
    print("Starting")
    tl1.start()
    time.sleep(3)
    event.set()
    tl1.join()
    print("Never printed..")

def proc(batch_q, batch_size, batchperentry, the_event):
    nrow = 100000
    i0 = 0
    to_q = []
    while i0 < nrow:
        rowend = min(i0 + batch_size,nrow)
        somerows = np.random.randint(0,5,(rowend-i0,2))
        to_q.append(somerows.tolist())
        if len(to_q) == batchperentry:
            print("adding..", i0, len(to_q))
            while not the_event.is_set():
                try:
                    batch_q.put(to_q, block=False)
                    to_q = []
                    break
                except queue.Full:
                    time.sleep(1)
        i0 += batch_size
    print("proc finishes")
    return      # Added this line, You can have it return whatever is most relevant to you.

if __name__ == "__main__":
  main()

希望这有帮助。

关于python - 多处理队列，存储列表列表，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/50479274/