标签: python-multiprocessing

我想训练具有不同随机状态的多个LinearSVC模型,但我更愿意并行执行.sklearn中是否有支持此功能的机制？我知道Gridsearch或者一些集合方法正在暗中进行,但是幕后的内容是什么？

我有一个脚本,我正在加载一个文件,这需要一段时间,因为有很多数据要读取,并防止用户终止我希望显示某种加载指示的过程.我认为这是学习如何使用多处理模块的好机会,所以我写了这个例子来测试模块:

import time, multiprocessing

def progress():
    delay = 0.5
    while True:
        print "Loading.",
        time.sleep(delay)
        print "\b.",
        time.sleep(delay)
        print "\b.",
        time.sleep(delay)
        print "\r                        \r",

    return

def loader(filename, con):
    # Dummy loader
    time.sleep(5)
    con.send(filename)
    con.close()
    return

if __name__ == "__main__":
    parrent_con, child_con = multiprocessing.Pipe()
    filename = "main.key"

    p1 = multiprocessing.Process(target=progress)
    p2 = multiprocessing.Process(target=loader, args=(filename, child_con))

    p1.start()
    p2.start()

    data = parrent_con.recv()
    p1.terminate()
    print "\n", data

当我在windows cmd中运行它时,它可以正常工作,它会打印"正在加载"并按顺序添加点,直到加载程序完成.但是在我需要它工作的unix中,我没有从进度函数,进程p1获得任何输出.

我正在创建一个字典如下:

y=[(1,2),(2,3),(1,2),(5,6)]

dict={}

for tup in y:
    tup=tuple(sorted(tup))
    if tup in dict.keys():
        dict[tup]=dict[tup]+1
    else:
        dict[tup]=1

但是我的实际y包含大约4000万个元组,有没有办法使用多处理来加速这个过程？

谢谢

我正在尝试更新多处理模块的manager.dict（）的嵌套字典中的键，但无法执行此操作。它不会更新值，也不会引发任何错误。

码：

import time
import random
from multiprocessing import Pool, Manager

def spammer_task(d, token, repeat):
    success = 0
    fail = 0
    while success+fail<repeat:
        time.sleep(random.random()*2.0)
        if (random.random()*100)>98.0:
            fail+=1
        else:
            success+=1
        d[token] = {
            'status': 'ongoing',
            'fail': fail,
            'success': success,
            'repeat': repeat
        }
    print d[token]['status']
    d[token]['status'] = 'complete'
    return

p = Pool()
m = Manager()
d = m.dict()

p.apply_async(spammer_task (d, 'abc', 5))
print d

输出：

进行中

{'abc'：{'status'：'进行中'，'fail'：0，'repeat'：5，'success'：5}}

我的期望是，当while循环结束时，它应该使d ['abc'] ['status'] =完成。但是在最终打印中，它仅将状态打印为“进行中”。

我有一个宠物项目,具有以下逻辑:

import asyncio, multiprocessing

async def sub_main():
    print('Hello from subprocess')

def sub_loop():
    asyncio.get_event_loop().run_until_complete(sub_main())

def start():
    multiprocessing.Process(target=sub_loop).start()

start()

如果你运行它,你会看到:

Hello from subprocess

那很好.但我要做的就是制作start()协程:

async def start():
    multiprocessing.Process(target=sub_loop).start()

要运行它,我必须做这样的事情:

asyncio.get_event_loop().run_until_complete(start())

这是问题:当创建子进程时,它会克隆整个Python环境,因此事件循环已经在那里运行:

Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/lib/python3.5/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "test.py", line 7, in sub_loop
    asyncio.get_event_loop().run_until_complete(sub_main())
  File "/usr/lib/python3.5/asyncio/base_events.py", line 361, in run_until_complete
    self.run_forever()
  File "/usr/lib/python3.5/asyncio/base_events.py", line 326, in run_forever
    raise RuntimeError('Event loop is running.')
RuntimeError: Event …

我需要执行下面的代码(我在Python 3.5中的真实代码库的简化版本):

import multiprocessing
def forever(do_something=None):
    while True:
        do_something()

p = multiprocessing.Process(target=forever, args=(lambda: print("do  something"),))
p.start()

为了创建新进程,Python需要pickle函数和作为目标传递的lambda.不幸的是,pickle无法序列化lambda,输出如下:

_pickle.PicklingError: Can't pickle <function <lambda> at 0x00C0D4B0>: attribute lookup <lambda> on __main__ failed

我发现了cloudpickle,可以使用相同的pickle接口序列化和反序列化lambda和闭包.

如何强制Python多处理模块使用cloudpickle而不是pickle？

显然黑客攻击标准的lib多处理代码不是一个选择!

谢谢

查理

我正在使用多处理库来生成两个子进程.我想确保只要父进程处于活动状态,如果子进程死掉(接收SIGKILL或SIGTERM),它们就会自动重启.另一方面,如果父进程收到SIGTERM/SIGINT,我希望它终止所有子进程然后退出.

这就是我解决问题的方法:

import sys
import time
from signal import signal, SIGINT, SIGTERM, SIGQUIT, SIGCHLD, SIG_IGN
from functools import partial
import multiprocessing
import setproctitle

class HelloWorld(multiprocessing.Process):
    def __init__(self):
        super(HelloWorld, self).__init__()

        # ignore, let parent handle it
        signal(SIGTERM, SIG_IGN)

    def run(self):

        setproctitle.setproctitle("helloProcess")

        while True:
            print "Hello World"
            time.sleep(1)

class Counter(multiprocessing.Process):
    def __init__(self):
        super(Counter, self).__init__()

        self.counter = 1

        # ignore, let parent handle it
        signal(SIGTERM, SIG_IGN)

    def run(self):

        setproctitle.setproctitle("counterProcess")

        while True:
            print self.counter
            time.sleep(1)
            self.counter += 1


def signal_handler(helloProcess, counterProcess, signum, frame):

    print …

我正在尝试在docker容器中测试python的多处理功能，但是即使成功创建了进程（我有8个CPU并创建了8个进程），它们也始终只占用一个物理CPU。这是我的代码：

from sklearn.externals.joblib.parallel import Parallel, delayed
import multiprocessing
import pandas
import numpy
from scipy.stats import linregress
import random
import logging

def applyParallel(dfGrouped, func):
    retLst = Parallel(n_jobs=multiprocessing.cpu_count())(delayed(func)(group) for name, group in dfGrouped)
    return pandas.concat(retLst)

def compute_regression(df):
    result = {}

    (slope,intercept,rvalue,pvalue,stderr) = linregress(df.date,df.value)
    result["slope"] = [slope]
    result["intercept"] = [intercept]

    return pandas.DataFrame(result)

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    logging.info("start")
    random_list = []
    for i in range(1,10000):
        for j in range(1,100):
            random_list.append({"id":i,"date":j,"value":random.random()})

    df = pandas.DataFrame(random_list)

    df = applyParallel(df.groupby('id'), …

如何配置使用多重处理（multiprocessing.Pool.map）的python模块，以便每个生成的进程也逐行进行配置。

目前，我使用line_profiler进行性能分析，但它不支持多处理。有手动方法吗？或者也许使用其他工具？

我有以下for循环：

for j in range(len(a_nested_list_of_ints)):
    arr_1_, arr_2_, arr_3_ = foo(a_nested_list_of_ints[j])
    arr_1[j,:] = arr_1_.data.numpy()
    arr_2[j,:] = arr_2_.data.numpy()
    arr_3[j,:] = arr_3_.data.numpy()

a_nested_list_of_ints嵌套的整数列表在哪里。但是，这需要很多时间才能完成。如何通过多处理对其进行优化？到目前为止，我尝试使用multiprocessing

p = Pool(5)
for j in range(len(a_nested_list_of_ints)):
    arr_1_, arr_2_, arr_3_ = p.map(foo,a_nested_list_of_ints[j])
    arr_1[j,:] = arr_1_.data.numpy()
    arr_2[j,:] = arr_2_.data.numpy()
    arr_3[j,:] = arr_3_.data.numpy()

但是，我得到：

ValueError: not enough values to unpack (expected 3, got 2)

这里：

    arr_1_, arr_2_, arr_3_ = p.map(foo,a_nested_list_of_ints[j])

是否知道如何使上述操作更快？我什至也尝试过使用starmap，但它不能正常工作。