我正在OSX上读一个网络摄像头,这个简单的脚本可以正常工作:
import cv2
camera = cv2.VideoCapture(0)
while True:
try:
(grabbed, frame) = camera.read() # grab the current frame
frame = cv2.resize(frame, (640, 480)) # resize the frame
cv2.imshow("Frame", frame) # show the frame to our screen
cv2.waitKey(1) # Display it at least one ms before going to the next frame
except KeyboardInterrupt:
# cleanup the camera and close any open windows
camera.release()
cv2.destroyAllWindows()
print "\n\nBye bye\n"
break
我现在想要在一个单独的进程中阅读视频,我的脚本更长,并且在Linux上的单独进程中正确地读取视频:
import numpy as np
import time
import ctypes
import argparse
from multiprocessing …我正在尝试使用Python的pathos将计算指定到单独的进程中,以便使用多核处理器加速它.我的代码组织如下:
class:
def foo(self,name):
...
setattr(self,name,something)
...
def boo(self):
for name in list:
self.foo(name)
由于我在multiprocessing.Pool中遇到酸洗问题,我决定尝试使用pathos.我尝试过,如前几个主题中所述:
import pathos.multiprocessing
但它导致错误:没有模块多处理 - 我在最新的病毒版本中找不到.
然后我尝试修改boo方法:
def boo(self):
import pathos
pathos.pp_map.pp_map(self.foo,list)
现在没有抛出错误,但是foo不起作用 - 我的类的实例没有新的属性.请帮助我,因为在花了一天之后,我不知道接下来要去哪里.
我遇到以下情况，我在 for 循环中创建一个池，如下所示（我知道这不是很优雅，但出于酸洗原因我必须这样做）。假设pathos.multiprocessing相当于 python 的multiprocessing库（因为它涉及到一些与此问题无关的细节）。我想执行以下代码：
self.pool = pathos.multiprocessing.ProcessingPool(number_processes)
for i in range(5):
all_responses = self.pool.map(wrapper_singlerun, range(self.no_of_restarts))
pool._clear()
现在我的问题：循环成功运行第一次迭代。然而，在第二次迭代时，算法突然停止（没有完成操作pool.map。我怀疑生成了僵尸进程，或者该进程以某种方式存在switched。下面您将找到我迄今为止尝试过的所有内容。
for i in range(5):
pool = pathos.multiprocessing.ProcessingPool(number_processes)
all_responses = self.pool.map(wrapper_singlerun, range(self.no_of_restarts))
pool._clear()
gc.collect()
for p in multiprocessing.active_children():
p.terminate()
gc.collect()
print("We have so many active children: ", multiprocessing.active_children()) # Returns []
上面的代码在我的 Mac 上运行得很好。但是，当我将其上传到具有以下规格的集群时，我收到错误，它在第一次迭代后卡住了：
DISTRIB_ID=Ubuntu
DISTRIB_RELEASE=18.04
DISTRIB_CODENAME=bionic
DISTRIB_DESCRIPTION="Ubuntu 18.04 LTS"
这是pathos的多处理库文件的链接
我目前正在使用python中的标准多处理来生成一系列无限期运行的进程.我并不特别关心表现; 每个线程只是在观察文件系统上的不同更改,并在修改文件时采取适当的操作.
目前,我有一个适合我的需求的解决方案,适用于Linux.我有一个函数和参数的字典,如下所示:
job_dict['func1'] = {'target': func1, 'args': (args,)}
对于每一个,我创建一个过程:
import multiprocessing
for k in job_dict.keys():
jobs[k] = multiprocessing.Process(target=job_dict[k]['target'],
args=job_dict[k]['args'])
有了这个,我可以跟踪每个正在运行的,并在必要时重新启动因任何原因崩溃的作业.
这在Windows中不起作用.我正在使用的许多函数都是包装器,使用各种functools函数,我得到的消息是无法序列化函数(请参阅多处理和dill可以一起做什么？).我还没弄清楚为什么我在Linux中没有得到这个错误,但在Windows中也是如此.
如果我dill在Windows中启动进程之前导入,则不会出现序列化错误.但是,这些过程实际上并没有做任何事情.我无法弄清楚为什么.
然后我切换到多处理实现pathos,但没有找到Process标准multiprocessing模块中的简单类的模拟.我能够使用每个作业生成线程pathos.pools.ThreadPool.这不是map的预期用途,我敢肯定,但它启动了所有线程,并且它们在Windows中运行:
import pathos
tp = pathos.pools.ThreadPool()
for k in job_dict.keys():
tp.uimap(job_dict[k]['target'], job_dict[k]['args'])
但是,现在我不确定如何监视一个线程是否仍处于活动状态,我正在寻找这样,以便我可以重新启动由于某种原因崩溃的线程.有什么建议？
是否可以将并发.futures（ProcessPoolExecutor）与pathos一起使用？
问这个是因为当我尝试使用时：
with concurrent.futures.ProcessPoolExecutor() as executor:
... 执行器.map()
我会得到：
PicklingError：无法 pickle：属性查找 内置.function 失败
我知道我们可以使用 pathos.multiprocess 来代替默认的多进程，并解决像这样的一些问题，问题是当我尝试使用 futures 实现并行处理时，我无法设置解决方案来修复 pickle 错误感伤。
任何想法 ？
谢谢。
我有一个类，它使用 GDAL 模块（https://pypi.org/project/GDAL/）打开一个大的光栅图像，并在多个位置从中提取小图像，由传递的坐标元组列表定义。我想使用 Pathos 处理大量坐标，就像我的代码的这个简化版本：
import gdal
import pathos.pools as pp
class MyClass:
def __init__(self, image):
self.image_object = gdal.Open(image)
def get_small_image(self, coord, size=100):
small_image = self.image_object.ReadAsArray(coord[0], coord[1], size, size)
return small_image
def run_multi(self, coords_in):
pool = pp.ProcessPool(2)
output = pool.map(self.get_small_image, coords_in)
return output
image = r'C:\path\to\image.JP2'
class_obj = MyClass(image)
coords = [(200, 200), (400, 400), (600, 600), (800, 800)]
results = class_obj.run_multi(coords)
print(results)
运行它会出现以下错误：
Traceback (most recent call last):
File "C:/Users/lharris/PycharmProjects/road_condition_py3/error_recreate.py", line 26, in <module>
results = class_obj.run_multi(coords) …我试图在循环中运行一些计算，每个计算都会创建、使用和关闭一个池。但计算只运行一次，然后抛出错误：“池未运行”。当然旧的没有运行，但新的不应该创建吗？
下面是一个简化的例子，类似于我的代码。更奇怪的是，在我的实际代码计算中，在崩溃之前运行了 7 次，所以我真的很困惑是什么问题。任何建议表示赞赏！
from pathos.multiprocessing import ProcessingPool as Pool
def add_two(number):
return (number + 2)
def parallel_function(numbers):
pool = Pool(10)
result = pool.imap(add_two, numbers)
pool.close()
pool.join()
return(result)
sets=[
[1, 2, 3],
[2, 3, 4],
[3, 4, 5]
]
for one_set in sets:
x = parallel_function(one_set)
for i in x:
print(i)
我正在使用以下测试代码：
from pathos.multiprocessing import ProcessingPool as Pool
import numpy
def foo(obj1, obj2):
a = obj1**2
b = numpy.asarray(range(1,5))
return obj1, b
if __name__ == '__main__':
p = Pool(5)
res = p.map(foo, [1,2,3], [4,5,6])
它给出了错误：
File "C:\Python27\lib\site-packages\multiprocess\pool.py", line 567, in get
raise self._value
NameError: global name 'numpy' is not defined
我在代码中做错了什么？
编辑：为什么这个问题被否决了两次？
我已经安装了 numpy 并且我的解释器一直在正确使用它，直到我尝试将其用于多处理。我已经用相同的安装编码了一段时间。
我在这里很困惑。我有一个受 CPU 限制的应用程序，因此我开始使用多进程实现并行化来克服 GIL 问题。
我第一次尝试使用multiprocessing，futures但遇到了酸洗问题，所以我去pathos使用哪个dill作为pickle替代品。
一切正常，但我想知道我是否使用的是最“面向未来”的解决方案。我也见过dask，但我不知道它是否会在酸洗类问题的情况下起作用（请参阅Python：（Pathos）多处理与类方法）。从文档来看，它使用了futures所以我假设它不会完成这项工作。
其次，我希望能够一次使用两台服务器，并且我已经看到这可以通过pathos(also dask) 实现，但我不明白它到底是如何工作的。这个答案/sf/answers/1886378091/仅显示如何使用一台服务器。使用 2 个或更多怎么样？我找不到任何关于此的示例，尽管包信息中描述的似乎是可能的。
感谢您的帮助！
我在使用运行图形保存功能的 pathos 多重处理时遇到了问题。
from pathos.multiprocessing import ProcessingPool as Pool
datasets = Pool().map(model_handler, analysisParams)
哪个抛出
File ".../lib/python3.6/site-packages/matplotlib/backends/backend_pdf.py", line 2029, in draw_text
font.set_text(s, 0.0, flags=LOAD_NO_HINTING)
RuntimeError: In set_text: could not load glyph
"""
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "wrapper.py", line 410, in <module>
datasets = Pool().map(model_handler, analysisParams)
File ".../lib/python3.6/site-packages/pathos/multiprocessing.py", line 137, in map
return _pool.map(star(f), zip(*args)) # chunksize
File ".../lib/python3.6/site-packages/multiprocess/pool.py", line 260, in map
return self._map_async(func, iterable, mapstar, chunksize).get()
File …