标签: python-multiprocessing

我正在使用Qt开发GUI应用程序.

当我尝试QApplication使用multiprocessing以下方法创建另一个时出错:

RuntimeError:QApplication实例已存在

我有一个主窗口,其中包含一个用于生成新进程的按钮,以便我可以创建一个新的GUI实例.基本上是这样的:

from PySide.QtCore import *
from PySide.QtGui import *
import multiprocessing
import sys


class MainWindow(QMainWindow):
    def __init__(self, parent=None):
        super(MainWindow, self).__init__(parent)
        btn = QPushButton('run new instance')
        btn.clicked.connect(self.create_daemon)
        self.setCentralWidget(btn)

    def create_daemon(self):
        p = multiprocessing.Process(target=new_window)
        p.start()

def new_window():
    app=QApplication(sys.argv)
    ex = MainWindow()
    ex.show()
    sys.exit(app.exec_())

if __name__=="__main__":
    app=QApplication(sys.argv)
    ex = MainWindow()
    ex.show()
    sys.exit(app.exec_())

它适用于Windows但RuntimeError在Linux中提供.这是因为Windows和Linux之间的多处理机制存在差异吗？我怎么能在Linux中实现同样的东西？

如何找到multiprocessing.Pool实例的最佳块大小？

之前我用过这个来创建一个nsudoku对象的生成器:

processes = multiprocessing.cpu_count()
worker_pool = multiprocessing.Pool(processes)
sudokus = worker_pool.imap_unordered(create_sudoku, range(n), n // processes + 1)

为了测量时间,我time.time()在上面的片段之前使用,然后按照描述初始化池,然后我将生成器转换为list(list(sudokus))以触发生成项目(仅用于时间测量,我知道这在最终程序中是无意义的),然后我time.time()再次使用时间并输出差异.

我观察到每个对象的块大小n // processes + 1约为0.425毫秒.但我也观察到CPU只在整个过程的前半部分完全加载,最终使用率下降到25%(在具有2核和超线程的i3上).

如果我使用较小的块大小int(l // (processes**2) + 1),我会得到大约0.355毫秒的时间,并且CPU负载分布更好.它只有一些小的尖峰到ca. 75%,但在处理时间的长时间内保持高位,然后降至25%.

是否有更好的公式来计算块大小或更好的方法来使用CPU最有效？请帮助我提高这个多处理池的有效性.

我正在使用多处理来为我的Python应用程序创建一个子流程.我想在父进程和子进程之间共享数据.重要的是要提到我需要异步共享它,这意味着子进程和父进程将在代码运行期间更新数据.

什么是最好的方式来执行？

我想拉出当前队列中的所有项目.还有另一个线程不断地将项目放在另一端,每个时期我想要获取当前队列中的所有项目.

有什么理由可以选择:

res = []
while q.qsize > 0 :
    res.append(q.get())

要么

res = []
while True :
    try :
         res.append(q.get(block=False)) 
    except Queue.Empty :
         break

现在文档明确指出qsize()> 0不会阻止队列在get上阻塞,但只有在多个线程可以从输出获取的情况下才是这样吗？

队列.qsize()返回队列的大致大小.注意,qsize()> 0不保证后续的get()不会阻塞,qsize()<maxsize也不保证put()不会阻塞.

这是否意味着第二种形式应该始终是首选？EAFP和所有这些？另外,调用q.qsize()有任何成本吗？是否会阻塞队列的另一端以便计算？

我想我已经谈过使用第二种形式,但对我来说似乎不那么干净.

我有3个驱动程序(Firefox浏览器),我希望它们do something在网站列表中.

我有一个工人定义为:

def worker(browser, queue):
    while True:
        id_ = queue.get(True)
        obj = ReviewID(id_)
        obj.search(browser)
        if obj.exists(browser):
            print(obj.get_url(browser))
        else:
            print("Nothing")

因此,worker只会访问包含id的队列并使用浏览器执行某些操作.

我想拥有一个工作池,这样一旦工人完成了使用浏览器在id_定义的网站上做某事,那么它会立即开始使用相同的浏览器在队列中找到的下一个id_上做某事.我有这个:

pool = Pool(processes=3)  # I want to have 3 drivers
manager = Manager()
queue = manager.Queue()
# Define here my workers in the pool
for id_ in ids:
    queue.put(id_)
for i in range(3):
    queue.put(None)

我有一个问题,我不知道如何定义我的工人,以便他们在池中.对于每个驱动程序,我需要分配一个worker,并且所有worker都共享同一个id队列.这可能吗？我该怎么做？

我的另一个想法是创建一个浏览器队列,这样如果一个驱动程序什么也不做,它将由一个worker和一个来自队列的id_来执行,以便执行一个新进程.但我对多处理完全不熟悉,实际上不知道怎么写这个.

我感谢您的帮助.

我一直在使用Python的多处理模块分析一些代码('job'函数只是对数字进行平方).

data = range(100000000)
n=4
time1 = time.time()
processes = multiprocessing.Pool(processes=n)
results_list = processes.map(func=job, iterable=data, chunksize=10000)
processes.close()
time2 = time.time()
print(time2-time1)
print(results_list[0:10])

我发现奇怪的一件事是最佳的chunksize似乎是大约10k元素 - 这在我的计算机上花了16秒.如果我将chunksize增加到100k或200k,那么它会减慢到20秒.

这种差异可能是由于长时间列表中酸洗所需的时间更长吗？100个元素的块大小需要62秒,我假设是由于在不同进程之间来回传递块所需的额外时间.

我在尝试实现新的defaultdict代理对象时遇到了一些麻烦.该文档是有点恐慌,所以我不知道如何正确地去了解这一点.

我想defaultdict在Manager实例中添加一个类型列表.你不能Manager.register在股票上使用这个方法,multiprocessing.Manager所以我从中创建了自己的存根管理器multiprocessing.mangers.BaseManager

class Manager(BaseManager):
    pass

然后我创建了我的子类multiprocessing.managers.BaseProxy来容纳defaultdict(我最初尝试使用另一个子类defaultdict,BaseProxy但是它似乎没有用.但这是我现在拥有的:

class ProxyDefaultDict(BaseProxy):
    def __init__(self, default_factory=None, *args, **kwargs):
        self.__data = defaultdict(default_factory)
        super().__init__(*args, **kwargs)

    def _callmethod(self, methodname, args=(), kwds={}):
        return getattr(self.__data, methodname)(*args, **kwds)

    def _getvalue(self):
        return self.__data.copy()

    def __repr__(self):
        return self.__data.__repr__()

    def __str__(self):
        return self.__data.__str__()

Manager.register('defaultdict', ProxyDefaultDict)

最终目标是拥有一个共享字典,可以跨进程和线程安全地共享键控锁.以下是我将如何初始化图像的示例:

if __name__ == '__main__':
    m = Manager()
    d = m.defaultdict(m.Lock)
    with d['named_lock']:
        print('holding the lock')

但是,我遇到了一些问题:

1. BaseManager的子类似乎只能通过上下文管理器进行初始化,即 …

我试图让asyncio与子进程和限制一起工作.我已经以功能的方式实现了这一点,但是当我尝试在opp风格中实现相同的逻辑时,出现了几个问题.主要是不能腌制coroutine/generator错误.我跟踪了一些theese,但不是全部

import asyncio
from concurrent.futures import ProcessPoolExecutor
from itertools import islice
from random import randint

class async_runner(object):
    def __init__(self):
        self.futures = [] # container to store current futures
        self.futures_total = []
        self.loop = asyncio.get_event_loop() # main event_loop
        self.executor = ProcessPoolExecutor()
        self.limit = 1

    def run(self, func, *args):
        temp_loop = asyncio.new_event_loop()
        try:
            coro = func(*args)
            asyncio.set_event_loop(temp_loop)
            ret = temp_loop.run_until_complete(coro)
            return ret
        finally:
            temp_loop.close()
    def limit_futures(self, futures, limit):
        self.futures_total = iter(futures)
        self.futures = [future for future in islice(self.futures_total,0,limit)]
        async def first_to_finish():
            while …

据我了解，GIL使得不可能拥有分别利用内核的线程。

这是一个基本问题，但是threading库的意义是什么？如果线程代码的速度与普通程序相当，这似乎没有用。

我正在改进一个解析XML并对其子树进行分类和索引的程序.实际程序太大而无法显示在这里,所以我把它归结为一个最小的测试用例,显示我遇到的问题.

这个想法是:

1. 逐个处理目录中的XML文件
2. alpino_ds并行处理文件中的所有节点
3. 在此过程中,该进程需要对共享变量进行读/写访问,以便我们可以检查属性总共发生了多少次,或者跟踪文件句柄

请注意,在实际代码中还有一些注意事项:

• simply returning new values per process and then merging them in the main thread seems not advisable and presumably quite slow because the actual data structure are dicts of four levels deep consisting of dicts, sets, ints, and strings, as well as dict-to-filehandle, and Counter() objects;
• I tried using threads (with ThreadPoolExecutor) and even though there was some gain (I calculated around 5% improvement in speed), this …