标签: python-multiprocessing

我有一个 python 脚本，它应该并行运行多个作业。我将最大进程设置为 20，但我需要脚本在发送作业之间休眠 5 秒。这是我的示例代码：

#!/usr/bin/env python

import multiprocessing
import subprocess


def prcss(cmd):
    sbc = subprocess.call
    com = sbc(cmd, shell='True')
    return (com)


if __name__=='__main__':

    s = 'sleep 5'
    cmd= []
    for j in range(1,21):
        for i in range(10):
            sis = "nohup ~/mycodes/code > str(j)"+"/"+"out"+str(i)+".dat"
            cmd.append(sis)
            cmd.append(s)

    pool=multiprocessing.Pool(processes=20)
    pool.map(prcss,cmd)

尽管我在“sis”命令之间有睡眠 5 个时间，但当我运行脚本时，所有作业都会立即启动。我需要在“sis”命令之间休息一下，因为每个作业的输出取决于计算机时钟。因此，如果我运行 20 个作业，它们都以相同的系统时钟启动，因此它们都将具有相同的输出。

知道如何让我的脚本在“sis”命令之间休眠吗？

阿贝丁

我正在尝试使用 Python 2.7 和 PySerial 同时读取多个串行端口。
特点应该是：

1. 在主程序中，我获取所有打开的串行端口，打开它们并将串行对象附加到serialobjects
2. 我想在一个子进程中读取每个串口数据以进行并行化

最大的问题是：如何将串口对象传递给子进程？
或者：
是否存在另一种（也许更好）的解决方案？（也许是这样：如何将扭曲的串行端口应用于我的问题？）

编辑

我想我并不完全清楚我想要实现什么。
我想同时读取2个或更多串口。由于超时和读出时间的原因，不可能在一个进程中同时读出它们。
下面的方法

ser1 = serial.Serial(port="COM1",baudrate=9600)
ser2 = serial.Serial(port="COM2",baudrate=9600)

ser1.write('command for reading out device 1')
output1 = ser1.readline()

ser2.write('command for reading out device 2')
# now you have to wait at least 100ms for device 2 to respond
output2 = ser2.readline()

不能满足我的需求。

另一种方法是并行化子进程中的串行读取。

主要.py

import serial   # serial communication
from subprocess import Popen, PIPE

ports = ["COM1", "COM2"]
for port in ports:
    ser = serial.Serial() …

我有一个df交易者交易数据集。我有 2 个级别的 for 循环，如下所示：

smartTrader =[]

for asset in range(len(Assets)):
    df = df[df['Assets'] == asset]
    # I have some more calculations here
    for trader in range(len(df['TraderID'])):
        # I have some calculations here, If trader is successful, I add his ID  
        # to the list as follows
        smartTrader.append(df['TraderID'][trader])

    # some more calculations here which are related to the first for loop.

我想并行化 中每个资产的计算Assets，并且我还想并行化每个资产的每个交易者的计算。完成所有这些计算后，我想根据smartTrader.

这是我第一次尝试并行处理，所以请耐心等待，非常感谢您的帮助。

这是我的情况。该代码与文档中的示例几乎相同：

from multiprocessing import Pool
import numpy as np

def grad(x0, y): return 0 # does some computational-heavy work actually

if __name__ == '__main__':

    class UnrollArgs:
        def __init__(self, func):
            self.func = func

        def __call__(self, args):
            return self.func(*args)

    def batch_grad(x0, y, processes=4):
        g = Pool(processes).map(UnrollArgs(grad), [(x0, yi) for yi in y])
        return np.sum([gi for gi in g], axis=0) / len(y)

Iy传递给batch_grad有 50 个元素并Pool.map抛出错误：

错误：无法启动新线程

从谷歌我知道这通常是由于一个人试图启动太多线程这一事实引起的。也许只是我个人的看法，但我认为文档multiprocessing.Pool有点不完整。特别是，我不知道如何控制应该启动的线程数量。该类的文档中甚至没有提到术语“线程” Pool。

的整数参数multiprocessing.Pool是要启动的 …

我想使用多处理库并行处理字典。

我的问题可以简化为这段代码：

from multiprocessing import Manager,Pool

def modify_dictionary(dictionary):
    if((3,3) not in dictionary):
        dictionary[(3,3)]=0.
    for i in range(100):
        dictionary[(3,3)] = dictionary[(3,3)]+1
    return 0

if __name__ == "__main__":

    manager = Manager()

    dictionary = manager.dict(lock=True)
    jobargs = [(dictionary) for i in range(5)]

    p = Pool(5)
    t = p.map(modify_dictionary,jobargs)
    p.close()
    p.join()

    print dictionary[(3,3)]

我创建了一个由 5 个工作人员组成的池，每个工作人员应将字典[(3,3)] 增加 100 次。因此，如果锁定过程正常工作，我预计字典[(3,3)] 在脚本末尾为 500。

然而; 我的代码中的某些内容一定是错误的，因为这不是我得到的：锁定过程似乎没有被“激活”，并且字典[(3,3)]在脚本末尾总是有一个<500的值。

你可以帮帮我吗？

我有一个 celery 工作进程，它在celery.signals.worker_process_init. 如果无法建立连接并引发异常，则 celery Worker 将无法启动，这是预期的行为。

该套接字类还使用threading.Timer. 不幸的是，由于计时器线程没有特殊的异常处理，因此如果连接发生问题，我的工作线程不会退出。

我开始捕获套接字读写操作的异常，并尝试了很多方法来正常停止当前的工作进程：

sys.exit(1)  # Doesn't kill the worker process (because it's running in a thread?)

下一次尝试，我看到它在celery/billiard问题中的某处提到：

raise SystemExit("Socket error")  # Same, I guess

这有效：

os.kill(os.getpid(), signal.SIGHUP)

但是芹菜工作主管进程进入快速旋转，创建立即退出的芹菜工作进程，而不是在几次尝试后放弃并死亡。

有没有更好的方法从 celery 任务外部发出 celery 工作进程死亡信号？

场景 - 我有一个正在分配任务的进程池。但是，如果子进程在运行任务时被终止，则该AsyncResult对象永远不会被标记为就绪。我希望发生的情况是它会被标记为就绪但不成功。

要重现这一点：

>>> import multiprocessing
>>> import time
>>> p = multiprocessing.Pool(processes=1)
>>> result = p.apply_async(time.sleep, args=(1000,))
>>> result.ready()
False

在另一个 shell 中，找到进程 ID 并将其杀死。

>>> result.ready()
False
>>> result.wait(5) # Waits 5 seconds even though subprocess is dead

这是一个问题，因为我有一个线程正在等待作业完成，并且它通常有相当长的超时。如何让result.wait(timeout)通话结束而不需要等待超时？另外，我如何判断它已被放弃，而不仅仅是任务仍在运行但我们已达到超时？

我正在尝试将大约 1.46 亿行的巨大数据集写入 CSV。我试过这个：

def paramlist():
    for row in nodes.itertuples():
        l = []
        for row2 in ref_stops.itertuples():
            l.append((row[1], row[2], row[3], row2[1],
                     row2[2], row2[3], row2[4], haversine(row[3], row[2], row2[3], row2[2])))
        yield l

pool = multiprocessing.Pool()
pool.map(func, paramlist())

def func(params):
    with open(r'big_file.csv', 'a') as f:
        writer = csv.writer(f)
        for row in params:
            writer.writerow(row)

这段代码可以工作，但它会耗尽我所有的内存并中止。
我该如何优化它？

我有一个函数“func”，它接收一个包含名为“matrix_image”的图像行数的列表：

list_rows = range(N_rows)

然后，在 func 内部完成计算，我得到代表图像的结果矩阵的新行。

def func(list_rows):
    new_row = numpy.empty(N_columns)
    ....
    ....# some computation
    ....
    return new_row


matrix_image = pool.map(func, list_n_rows )

“new_row”在内部正确计算func，因为我在调试器中看到值，但生成的matrix_image具有形状：（list_n_rows，）。填充None处的所有值。

如何返回（1D numpy 数组）行并附func加到结果矩阵（2D numpy 数组）中？

使用多处理包时，我在单元测试中遇到此错误。
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

运行以下代码可以在常规 python 文件中运行，但是当放入 unittest 中时，它会引发我上面发布的酸洗错误。这是我可以举出的最简单的例子。

    # The following, works. 
    import multiprocessing

    def hello(number):
        print "hello"

    number_processes = 2
    pool = multiprocessing.Pool(number_processes)
    total_tasks = 2
    tasks = range(total_tasks)
    results = pool.map(hello, tasks)
    pool.close()
    pool.join()

下一个代码块：

     # This does not work, i'm running it via unittest runner
    import multiprocessing
    import unittest
    class Testing123(unittest.TestCase):
        test_1(self):

            def hello(number):
                print "hello"

            number_processes = 2
            pool = multiprocessing.Pool(number_processes)
            total_tasks = 2
            tasks = range(total_tasks) …