标签: python-multiprocessing

我知道Python线程一次只能执行一个字节码,那么为什么线程库会提供锁？我假设如果一次只执行一个线程,则不会发生竞争条件.

该库提供锁,条件和信号量.这是同步执行的唯一目的吗？

更新:

我做了一个小实验:

from threading import Thread
from multiprocessing import Process

num = 0

def f():
    global num
    num += 1

def thread(func):
    # return Process(target=func)
    return Thread(target=func)


if __name__ == '__main__':
    t_list = []
    for i in xrange(1, 100000):
        t = thread(f)
        t.start()
        t_list.append(t)

    for t in t_list:
        t.join()

    print num

基本上我应该启动100k线程并递增1.返回的结果是99993.

a)如果有GIL同步和避免竞争条件,结果如何不是99999？b)甚至可以启动100k OS线程吗？

看到答案后更新2:

如果GIL没有真正提供一种方法来执行简单的操作,例如原子递增,那么将它放在那里的目的是什么？它对于讨厌的并发问题没有帮助,那么它为什么要到位呢？我听说过C扩展的用例,有人会举例说明吗？

我正在尝试使用Python的multiprocessing.Pool对象来理解最佳实践.

在我的程序中,我经常使用Pool.imap.通常每次我并行启动任务时,我都会创建一个新的池对象,然后在完成后关闭它.

我最近遇到了挂起,其中提交到池的任务数量少于进程数.奇怪的是它只发生在我的测试管道中,它之前运行了很多东西.作为独立运行测试不会导致手.我认为它与制作多个池有关.

我真的很想找到一些资源来帮助我理解使用Python多处理的最佳实践.具体来说,我目前正在尝试了解制作多个池对象与仅使用一个池对象的含义.

我的工作对股票项目chrome browsers使用启动selenium package多个进程.由于这些浏览器(存储为列表对象)是使用a在进程间共享的Manager.list(),因此我还希望共享跨进程的相应列表,Manager.Lock()以便进程lock在通过它访问URL之前获取浏览器.但是我lock在示例代码中收到如下序列化错误:

>>> from multiprocessing import Manager
>>> manager = Manager()
>>> lock_list = manager.list()
>>> lock_list.append(manager.Lock())
>>> print type(lock_list[0])
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<string>", line 2, in __getitem__
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 774, in _callmethod
    raise convert_to_error(kind, result)
multiprocessing.managers.RemoteError: 
---------------------------------------------------------------------------
Unserializable message: ('#RETURN', <thread.lock object at 0x7fa2115452b0>)
---------------------------------------------------------------------------
>>> 

如果有人可以帮助我解决这个问题,那真的很棒.谢谢.

我正在使用

Zope - 2.13.19
Python - 2.6.8

下面的代码在手动运行时有效,但在外部方法时则无效.它会引发以下错误.我在做概念错误的事情吗？

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/opt/python2.6/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/opt/python2.6/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/opt/python2.6/lib/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
TypeError: expected string or Unicode object, NoneType found


import time
from multiprocessing import Pool
import logging


def func(name):
    print 'hello %s,' % name
    time.sleep(5)
    print 'nice to meet you.'


def get_data():
    pool = Pool(processes=2)
    pool.map(func, ('frank', 'justin', 'osi', 'thomas'))

我正在做一些并行处理，如下所示：

with mp.Pool(8) as tmpPool:
        results = tmpPool.starmap(my_function, inputs)

输入看起来像：[（1,0.2312），（5,0.52）...]，即一个int和float的元组。

该代码运行良好，但我似乎无法将其包裹在加载栏（tqdm）上，例如可以使用imap方法完成，如下所示：

tqdm.tqdm(mp.imap(some_function,some_inputs))

星图也可以做到吗？

谢谢！

与 python 的多处理相关的序列化和反序列化似乎限制了并行处理数据的好处。

在下面的示例中，我创建一个返回 numpy 数组的自定义迭代。随着numpy数组大小的增加，数据获取过程成为瓶颈。这是预料之中的。然而，我预计会增加num_worker并prefetch_factor通过提前准备批次来减少这一瓶颈。但我在下面的示例中没有看到这种行为。

我测试了两种MyIterable返回的情况

1. 小物体np.array((10, 150))
2. 大物体np.array((1000, 150))

两种情况下处理一个批次的平均时间如下：

# small np object
avg time per batch for num workers=0: 0.47068126868714444
avg time per batch for num workers=2: 0.20982365206225495
avg time per batch for num workers=4: 0.10560789656221914
avg time per batch for num workers=6: 0.07202646931250456
avg time per batch for num workers=8: 0.05311137337469063

# large np object
avg time per batch for num workers=0: 0.6090951558124971
avg time per batch …

我的最终目标是将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，错误是否仍然存在？

编辑 ： …

如何更改 Pythonmultiprocessing库使用的序列化方法？特别是，默认序列化方法使用pickle具有该版本 Python 的默认 pickle 协议版本的库。默认的 pickle 协议在 Python 2.7 中是版本 2，在 Python 3.6 中是版本 3。如何在 Python 3.6 中将协议版本设置为 2，以便我可以使用库中的一些类（如Client和Listener）在multiprocessingPython 2.7 运行的服务器处理和 Python 3.6 运行的客户端进程之间进行通信？

（旁注：作为测试，我修改了第 206 行multiprocessing/connection.py，添加protocol=2到dump()调用中以强制协议版本为 2，并且我的客户端/服务器进程在我有限的测试中工作，服务器由 2.7 运行，客户端由 3.6 运行）。

在 Python 3.6 中，合并了一个补丁来设置序列化程序，但是该补丁没有记录，我还没有弄清楚如何使用它。这是我尝试使用它的方式（我也将其发布到我链接到的 Python 票证上）：

pickle2reducer.py：

from multiprocessing.reduction import ForkingPickler, AbstractReducer

class ForkingPickler2(ForkingPickler):
    def __init__(self, *args):
        if len(args) > 1:
            args[1] = 2
        else:
            args.append(2)
        super().__init__(*args)

    @classmethod …

我有一个相当简单的类，类似于：

class Tick:
    def __init__():
        self.val = Value('i', 0)
        self.lock = Lock()

    def tick(self):
        self.lock().acquire()
        try:
            self.val.value = self.val.value  + 1
        finally:
            self.lock.release()

我想将它与星图一起使用：

def mp_tick(a, t):
    t.tick()

tick = Tick()

with Pool(processes=4) as pool:
    pool.starmap(mp_tick, [(i, tick) for i in range(100)]

抛出：

RuntimeError: Synchronized objects should only be shared between processes through inheritance

然而：

RuntimeError: Synchronized objects should only be shared between processes through inheritance

才不是。我知道为什么第二个有效。我不明白为什么第一个失败。

理想情况下，我希望有一个Tick可以传递给多个线程的实例，并Tick处理锁定和设置自己的值。我不希望函数必须假设存在全局变量。

我寻找了其他问题，这个未接受的答案问题是我能找到的唯一一个以某种方式涵盖了这个问题并且没有真正帮助的问题。另外，我需要它来处理进程，而不是线程。

因此，我从头开始编写了一个示例程序来展示我的问题，您应该能够粘贴它并且它将运行：

import multiprocessing
import time 

class Apple:
   def __init__(self, color):
      self.color = color

def thinkAboutApple(apple):
   while True:
      print(apple.color)
      time.sleep(2)

my_apple = Apple("red")
new_process = multiprocessing.Process(target=thinkAboutApple, args=(my_apple,))
new_process.start()
time.sleep(4)
print("new: brown")
my_apple.color = "brown"

#so that the program doesn't exit after time.sleep(4)
while True:
    pass

# actual output | # wanted output
red             | red
red             | red
new: brown      | new: brown
red             | brown
red             | brown

这告诉我，要么苹果处于一个奇怪的假设，它同时有两种颜色，要么 new_process' 苹果位于 ram 中的另一个位置，并与主进程中的苹果分开。

所以问题是：有没有办法让进程中的苹果指针指向同一个苹果，或者什么是Pythonic方法来保持所有进程中苹果的所有实例相同？如果我在许多进程中有相同的苹果，甚至更多进程没有苹果，我如何确保它们始终相同？