标签: python-multiprocessing

我有两个并行运行的函数,每个函数都返回一个值.我需要等待两个函数完成,然后处理它们的返回.我怎么能在python中实现这一点.假设

def fun1():
   #do some calculation#
   return val1

def fun2():
   #do some calculation#
   return val2

我想fun1和fun2并行运行然后需要计算

valsum = val1+val2

我的最终目标是将stdout几个子进程重定向到一些队列，并将它们打印在某处（可能在一个小 GUI 中）。

第一步是将子类Queue化为一个行为与stdout. 但这就是我被卡住的地方。Queue在 Python v3.5 中子类化多处理似乎是不可能的。

# This is a Queue that behaves like stdout
# Unfortunately, doesn't work in Python 3.5   :-(
class StdoutQueue(Queue):
    def __init__(self,*args,**kwargs):
        Queue.__init__(self,*args,**kwargs, ctx='')

    def write(self,msg):
        self.put(msg)

    def flush(self):
        sys.__stdout__.flush()

我在下面的帖子中找到了这个片段（可能当时 Python 3.5 还不存在）：Python multiprocessing redirect stdout of a child process to a Tkinter Text

在 Python v3.5 中，您在继承多处理Queue类时会偶然发现奇怪的错误消息。我发现了两个描述该问题的错误报告：

https://bugs.python.org/issue21367

https://bugs.python.org/issue19895

我有两个问题：

1. 假设我想坚持使用 Python v3.5 - 使用以前的版本并不是一个真正的选择。我可以使用什么解决方法以某种方式对多处理队列进行子类化？
2. 如果我升级到 Python v3.6，错误是否仍然存在？

编辑 ： …

我有这个脚本来并行处理一些URL：

import multiprocessing
import time

list_of_urls = []

for i in range(1,1000):
    list_of_urls.append('http://example.com/page=' + str(i))

def process_url(url):
    page_processed = url.split('=')[1]
    print 'Processing page %s'% page_processed
    time.sleep(5)

pool = multiprocessing.Pool(processes=4)
pool.map(process_url, list_of_urls)

该列表是有序的，但是当我运行它时，脚本不会按顺序从列表中选择URL：

import multiprocessing
import time

list_of_urls = []

for i in range(1,1000):
    list_of_urls.append('http://example.com/page=' + str(i))

def process_url(url):
    page_processed = url.split('=')[1]
    print 'Processing page %s'% page_processed
    time.sleep(5)

pool = multiprocessing.Pool(processes=4)
pool.map(process_url, list_of_urls)

相反，我希望它首先处理第1,2,3,4页，然后继续按照列表中的顺序进行处理。是否可以选择执行此操作？

我面临以下问题.我正在尝试并行化一个更新文件的函数,但我无法启动Pool()因为一个OSError: [Errno 12] Cannot allocate memory.我开始在服务器上四处看看,这并不像我使用旧的,弱的/实际内存.见htop: 此外,free -m显示除了大约7GB的交换内存外,我还有足够的RAM: 而我正在尝试使用的文件也不是那么大.我将粘贴我的代码(和堆栈跟踪),其中,大小如下:

使用的predictionmatrix数据框占用大约.根据pandasdataframe.memory_usage() 文件geo.geojson为80MB 是2MB

我该如何调试呢？我可以检查什么以及如何检查？感谢您的任何提示/技巧!

码:

def parallelUpdateJSON(paramMatch, predictionmatrix, data):
    for feature in data['features']: 
        currentfeature = predictionmatrix[(predictionmatrix['SId']==feature['properties']['cellId']) & paramMatch]
        if (len(currentfeature) > 0):
            feature['properties'].update({"style": {"opacity": currentfeature.AllActivity.item()}})
        else:
            feature['properties'].update({"style": {"opacity": 0}})

def writeGeoJSON(weekdaytopredict, hourtopredict, predictionmatrix):
    with open('geo.geojson') as f:
        data = json.load(f)
    paramMatch = (predictionmatrix['Hour']==hourtopredict) & (predictionmatrix['Weekday']==weekdaytopredict)
    pool = Pool()
    func = partial(parallelUpdateJSON, paramMatch, predictionmatrix)
    pool.map(func, data)
    pool.close()
    pool.join()

    with …

我正在使用Python多重处理将任务拆分为多个进程。我的CPU有4个物理核和8个虚拟核。我希望所有内核的利用率均为100％，但是当使用4个子进程池时，我在1个内核上观察到100％，在3个内核上观察到9％multiprocessing.Manager Namespace。子进程仅从共享名称空间读取（而不是写入）。也许共享的命名空间不允许并发读取访问-我是否需要以某种方式指定不应进行锁定？

在我的实际代码中，共享命名空间很大，包括大列表和numpy数组，因此另一个原因可能是创建它所需的时间。共享命名空间是只创建一次，还是每次创建新的子进程时都创建（假设命名空间中的数据没有变化）？

示例代码：

import multiprocessing
import worker
mgr = multiprocessing.Manager()
multiProcShared = mgr.Namespace()
multiProcShared.a = 25
multiProcShared.b = 40
multiProcShared.c = 10
pool = multiprocessing.Pool(4)
allResults = []
max = xrange(100)
for segmentNum in max:
    allResults.append(pool.apply_async(worker.worker, args=(multiProcShared, segmentNum)))

更新：我怀疑由于我正在将大数据传递给子进程，因此数据切换可能会产生大量开销。我认为数据是先腌制然后再腌制？如果是这样，那么python多处理性能是否会随着传递给子进程的数据大小而降低？是否根据数据是通过共享名称空间还是通过函数参数传递来更改？

我正在使用Python的多处理模块来进行科学的并行处理。在我的代码中，我使用了多个繁重的工作流程以及一个将结果持久保存到磁盘的写入程序。要写入的数据通过队列从工作进程发送到写程序进程。数据本身非常简单，仅由一个包含文件名的元组和带有两个浮点数的列表组成。经过几个小时的处理，编写器进程通常会卡住。更准确地说，以下代码块

while (True):
    try:
        item = queue.get(timeout=60)
        break
    except Exception as error:
        logging.info("Writer: Timeout occurred {}".format(str(error)))

永远不会退出循环，我会收到连续的“超时”消息。

我还实现了一个日志记录过程，该过程除其他外输出队列的状态，即使我收到上面的超时错误消息，对qsize（）的调用也会不断返回完整的队列（在我的情况下为size = 48）。

我已经彻底检查了队列对象的文档，并且找不到关于为什么同时队列已满的get（）返回超时的可能解释。

有任何想法吗？

编辑：

我修改了代码以确保捕获到空队列异常：

while (True):
    try:
        item = queue.get(timeout=60)
        break
    except Empty as error:
        logging.info("Writer: Timeout occurred {}".format(str(error)))

我正在学习Python中的线程库。我不明白，如何并行运行两个线程？

这是我的python程序：

没有线程的程序（fibsimple.py）

def fib(n):
    if n < 2:
        return n
    else: 
        return fib(n-1) + fib(n-2)

fib(35)
fib(35)

print "Done"

运行时间：

$ time python fibsimple.py 
Done

real    0m7.935s
user    0m7.922s
sys 0m0.008s

带有线程的相同程序（fibthread.py）

from threading import Thread
def fib(n):
    if n < 2:
        return n
    else: 
        return fib(n-1) + fib(n-2)

t1 = Thread(target = fib, args = (35, ))
t1.start()

t2 = Thread(target = fib, args = (35, ))
t2.start()

t1.join()
t2.join()

print "Done"

运行时间：

$ …

众所周知,我们需要使用Python 来保护main()运行时代码.multiprocessingif __name__ == '__main__'

我理解在某些情况下这是必要的,以提供对main中定义的函数的访问,但我不明白为什么在这种情况下这是必要的:

file2.py

import numpy as np
from multiprocessing import Pool
class Something(object):
    def get_image(self):
        return np.random.rand(64,64)

    def mp(self):
        image = self.get_image()
        p = Pool(2)
        res1 = p.apply_async(np.sum, (image,))
        res2 = p.apply_async(np.mean, (image,))
        print(res1.get())
        print(res2.get())
        p.close()
        p.join()

main.py

from file2 import Something
s = Something()
s.mp()

Something工作所需的所有功能或进口都是其中的一部分file2.py.为什么子流程需要重新运行main.py？

我认为__name__解决方案不是很好,因为这阻止我分发代码,file2.py因为我不能确保他们保护他们的主要.Windows没有解决方法吗？如何解决这个问题(因为我从来没有遇到任何问题,没有用任何软件包来保护我的主程序 - 他们只是不使用多处理？)

编辑: 我知道这是因为fork()Windows中没有实现.我只是问是否有一个黑客让翻译开始,file2.py而不是main.py因为我可以肯定这file2.py是自给自足的

我正在尝试使用它的方法更改类字段,但是当放入方法时它不起作用.

来自多处理导入过程

class Multi:
    def __init__(self):
        self.x = 20


    def loop(self,):
        for i in range(1,100):
            self.x = i

M = Multi()

p = Process(target=M.loop)
p.start()

运行此程序后,Mx仍然是20.这怎么可能？

我正在尝试通过套接字连接打包和发送列数据。

为了加快速度，我考虑过将包装（struct.pack）分成多个过程。

为了避免两种方式的腌制，我认为最好由打包过程自己发送数据，因为据说可以从Python 3.4开始腌制套接字对象。

这是我正在工作的简化版本：

import socket
from multiprocessing import Pool
from struct import pack

# Start and connect a socket
s = socket.socket()
s.connect((ip, port))

# Data to be packed and sent in this order
data1 = 1, 2, 3, 4
data2 = 5, 6, 7, 8
data3 = 9, 10, 11, 12

# Top level column packer/sender for mp.pool
def send_column(column):
    return s.send(pack(f'{len(column)}i', *column))


pool = Pool()

# Will this necessarily send the data in order? …