标签: python-multiprocessing

我搭建了一个websocket服务器，它的简化版本如下所示：

import websockets, subprocess, asyncio, json, re, os, sys
from multiprocessing import Process

def docker_command(command_words):
    return subprocess.Popen(
        ["docker"] + command_words,
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)

async def check_submission(websocket:object, submission:dict):
    exercise=submission["exercise"]
    with docker_command(["exec", "-w", "badkan", "grade_exercise", exercise]) as proc:
        for line in proc.stdout:
            print("> " + line)
            await websocket.send(line)

async def run(websocket, path):
    submission_json = await websocket.recv()   # returns a string
    submission = json.loads(submission_json)   # converts the string to a python dict

    ####
    await check_submission(websocket, submission)


websocketserver = websockets.server.serve(run, '0.0.0.0', 8888, origins=None)
asyncio.get_event_loop().run_until_complete(websocketserver) …

Python 3.6.6

这是代码：

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


executor_processes = ProcessPoolExecutor(2)


def calculate():
    while True:
        print("while")
        time.sleep(1)


async def async_method():
    loop_ = asyncio.get_event_loop()
    loop_.run_in_executor(executor_processes, calculate)
    await asyncio.sleep(1)
    print("finish sleep")

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_method())
    print("main_thread is finished")

输出：

while
finish sleep
main_thread 已完成
while
while
...

我希望子进程将被终止，就像当使用守护进程属性生成进程时一样：

import asyncio
import time
import multiprocessing


def calculate():
    while True:
        print("while")
        time.sleep(1)


async def async_method():
    proc = multiprocessing.Process(target=calculate)
    proc.daemon = True
    proc.start()
    await asyncio.sleep(1)
    print("finish sleep")

if …

当在 Joblib 中使用处理全局变量的函数时，在 Linux 上无需任何副本即可从该函数访问全局变量。

我们可以在以下脚本中对此进行测试：

import joblib
import numpy as np

print("Initializing global")
# Let's create a global that is big, so it takes time to create it
my_global = np.random.uniform(0,100, size=(10**4, 10**4))
print("done")

# A simple function working on the global variable
def fun_with_global():
    return id(my_global)

print("starting // loop")
joblib.Parallel(n_jobs=3, backend="multiprocessing", verbose=100)((joblib.delayed(fun_with_global)() for i in range(1000)))
joblib.Parallel(n_jobs=3, backend="loky", verbose=100)((joblib.delayed(fun_with_global)() for i in range(1000)))
# We get that the two last parallel calls execute almost instantly, even for 1000 …

在调查 Python 代码库中的关键路径时，我们发现 ctypes 在延迟方面的行为是相当不可预测的。

我们的应用程序的更多背景。我们有很多进程，每个进程都通过共享内存进行通信。我们利用 python 库multiprocessing.RawValue，multiprocessing.RawArray它在内部用于ctypes数据管理。在生产环境中运行时，我们发现即使get()对这些共享数据类型进行简单的访问也需要大约 30-50 us，有时需要 100 us，而且速度相当慢。即使对于蟒蛇来说也是如此。

我创建了这个简单的示例，它创建了一个ctype结构并公开了get()方法

import ctypes
import sys
import time
import numpy as np
import random
from decimal import Decimal

def get_time_ns():
    return Decimal(str(time.time_ns()))

class Point(ctypes.Structure):
    _fields_ = [("x", ctypes.c_int),
                ("y", ctypes.c_int)]

    def __init__(self, x, y):
        return super().__init__(x, y)

    def get(self):
        return self.x
        #return str(self.x) + "," + str(self.y)

def benchmark(delay_mode):
    p = Point(10, 20)
    iters = 10
    while …

我有一个简单的python多处理脚本,它设置一个工作池,试图将工作输出附加到Manager列表.该脚本有3个调用堆栈: - 主调用f1,它产生几个调用另一个函数g1的工作进程.当一个人试图调试脚本时(偶然在Windows 7/64 bit/VS 2010/PyTools上),脚本会运行到嵌套的进程创建循环中,从而产生无数个进程.谁能确定原因？我确定我错过了很简单的事情.这是有问题的代码: -

import multiprocessing
import logging

manager = multiprocessing.Manager()
results = manager.list()

def g1(x):
    y = x*x
    print "processing: y = %s" % y
    results.append(y)

def f1():
    logger = multiprocessing.log_to_stderr()
    logger.setLevel(multiprocessing.SUBDEBUG)

    pool = multiprocessing.Pool(processes=4)
    for (i) in range(0,15):
        pool.apply_async(g1, [i])
    pool.close()
    pool.join()

def main():
    f1()

if __name__ == "__main__":
    main()

PS:尝试添加multiprocessing.freeze_support()到主要无济于事.

如果我有以下代码:

def f():
    print 'ok!'
    import sys
    sys.exit()

if __name__=='__main__':
    import billiard
    billiard.forking_enable(0)
    p = billiard.Process( target=f)
    p.start()
    while p.is_alive():
        pass

脚本按预期运行,打印"ok!" 并结束.但是,如果我省略该if __name__=='__main__':行并去除以下行,我的机器(OS X)会发疯,不断产生大量的Python进程直到我killall Python.知道这里发生了什么吗？

(对于那些标记为重复的人,请注意,虽然另一个问题if __name__=='__main__'通常会询问目的 ,但我特别询问为什么在此处使用它会导致意外行为)

我的原始问题是我正在尝试执行以下操作:

def submit_decoder_process(decoder, input_line):
    decoder.process_line(input_line)
    return decoder

self.pool = Pool(processes=num_of_processes)
self.pool.apply_async(submit_decoder_process, [decoder, input_line]).get()

解码器在这里有点介绍,但重要的是解码器是一个用调用setParseAction()的PyParsing表达式初始化的对象.这会使多处理使用的pickle失败,这反过来又失败了上面的代码.

现在,这是我已经孤立和简化的pickle/PyParsing问题.由于pickle失败,以下代码会产生错误消息.

import pickle
from pyparsing import *

def my_pa_func():
    pass

pickle.dumps(Word(nums).setParseAction(my_pa_func))

错误信息:

pickle.PicklingError: Can't pickle <function wrapper at 0x00000000026534A8>: it's not found as pyparsing.wrapper

现在,如果你删除调用.setParseAction(my_pa_func),它将没有问题:

pickle.dumps(Word(nums))

我怎么能绕过它呢？多处理使用泡菜,所以我猜不出它.据说使用dill的pathos包不够成熟,至少,我在Windows-64bit上安装它时遇到问题.我真的在这里摸不着头脑.

我正在启动几个python进程,在循环中下载调用这段代码:

startTimeMillis = int(round(time.time() * 1000))

for i in range(10):
        p = multiprocessing.Process(target=performCurl, args =("http://www.google.com/d", i, ))
        p.start()


endTimeMillis = int(round(time.time() * 1000))
totalTimeSeconds = (endTimeMillis - startTimeMillis)
print "The whole process took ", str(totalTimeSeconds)

我想检查所有进程完成所需的时间,那么如何让代码的最后一部分等待所有进程？

我正在尝试导入一个简单的模块,它使用多处理到我的主文件中.使用多处理的模块从a获得计算结果asyncresult.get().通过导入脚本调用此例程时,只会挂起并且不会继续.

这是一个小例子.

import_test.py(要导入的模块)

import sys
import multiprocessing as mp

if not hasattr(sys.stdin, 'close'):
    def dummy_close():
        pass
    sys.stdin.close = dummy_close

# simple test function to compute in parallel
def testf(x):
    return x*x

# multiprocessing code
pool = mp.Pool()
results = []
for i in range(0,2):
    results.append(pool.apply_async(testf, [i]))
for i in range(0,2):
    print results[i].get()
pool.close()
pool.join()

main.py(简单的脚本只导入代码,应该从import_test打印)

if __name__ == '__main__':
    import import_test

print "done"

运行脚本时,我可以看到导入发生,但第一次results[i].get()(asyncresult.get()例程)被调用整个脚本挂起并且不会继续而不会抛出任何错误.我在Mac OS X(El Capitan)和Windows 10下运行了两个明显不同的python设置.结果始终是上述行为.

如果我只是将导入模块中的代码放入其中,那么一切正常main.py.当然我的实际代码比这更复杂,我想在要导入的模块中保持并行计算.

我该如何解决这个问题？

我的代码没有产生2个进程，我也不知道为什么。我正在使用来自多处理python库的Process。

import os
from multiprocessing import Process

def dummy():
    while True:
        print(os.getpid())

p1 = Process(target=dummy())
p2 = Process(target=dummy())
p1.start()
p2.start()
p1.join()
p2.join()

预期结果：创建了2个进程-在Windows 7任务管理器中可见，并且我的代码打印了2个不同的进程ID

实际结果：仅运行一个python进程，仅输出一个进程ID。

3480
3480
3480