标签: 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扩展的用例,有人会举例说明吗？

我有一个函数,它将图形的节点id作为输入并在图形中计算某些东西(不改变图形对象),然后将结果保存在文件系统中,我的代码如下所示:

...
# graph file is being loaded
g = loadGraph(gfile='data/graph.txt')
# list of nodeids is being loaded
nodeids = loadSeeds(sfile='data/seeds.txt')

import multiprocessing as mp
# parallel part of the code
print ("entering the parallel part ..")
num_workers = mp.cpu_count() # 4 on my machine
p = mp.Pool(num_workers)
# _myParallelFunction(nodeid) {calculate something for nodeid in g and save it into a file}
p.map(_myParallelFunction, nodeids)
p.close()
...

问题是当我将图形加载到Python中时需要大量内存(大约2G,它实际上是一个包含数千个节点的大图),但是当它开始进入代码的并行部分时(并行映射函数执行)似乎每个进程都有一个单独的g副本,我只是在我的机器上耗尽了内存(它有6G内存和3G交换),所以我想看到有一种方法可以为每个进程提供相同的副本g这样只需要存储器来容纳它的一个副本吗？任何建议都表示赞赏,并提前感谢.

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

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

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

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

我在Windows 7 64位上运行一个简单的"Hello World"程序,其规格如下:

Python 3.4.3 | Anaconda 2.3.0 (64-bit) | [MSC v.1600 64 bit (AMD64)] IPython 4.0.0

该程序:

from multiprocessing import Process, freeze_support

def f():
    print ('hello world!')

if __name__ == '__main__':
    #freeze_support()
    Process(target=f).start()

给出以下错误:

[I 15:02:23.855 NotebookApp] Saving file at /uhc/FeatureContributionToK-meansClu
sterWithPC.ipynb
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Anaconda3\lib\multiprocessing\spawn.py", line 106, in spawn_main
    exitcode = _main(fd)
  File "C:\Anaconda3\lib\multiprocessing\spawn.py", line 116, in _main
    self = pickle.load(from_parent)
AttributeError: Can't get attribute 'f' on module …

我有一个相当简单的嵌套for循环,迭代四个数组:

for a in a_grid:
    for b in b_grid:
        for c in c_grid:
            for d in d_grid:
                do_some_stuff(a,b,c,d)  # perform calculations and write to file

也许这不是开始在4D网格上执行计算的最有效方法.我知道joblib能够像这样并行化两个嵌套的for循环,但是我很难将它概括为四个嵌套循环.有任何想法吗？

根据 https://github.com/joblib/joblib/issues/180,有没有一种安全的方法从python中的线程创建子进程？ Python多处理模块不允许在线程内使用.这是真的？

我的理解是,从线程中分叉是好的,只要你没有持有线程.当你这样做时(在当前线程中？在进程中的任何地方？).但是,Python的文档没有说明在fork之后是否安全地共享threading.Lock对象.

还有这个:从日志记录模块共享的锁导致fork问题.https://bugs.python.org/issue6721

我不确定这个问题是怎么产生的.听起来当进程中的任何锁的状态都被复制到当前线程分叉的子进程中(这看起来像设计错误并且肯定会死锁).如果是这样,那么使用多真正提供对这种任何保护(因为我可以自由地创建我multiprocessing.Pool被创建和被其它线程进入后threading.Lock,经过线程已经开始,使用不是叉安全日志模块) - 多处理模块docs也没有说明是否应该在Locks之前分配multiprocessing.Pools.

用多处理替换threading.Lock.Lock到处避免这个问题并允许我们安全地组合线程和分支？

在运行Python 3.6的Windows 10系统上，当尝试multiprocessing.Process用于创建新rq工作线程时，

multiprocessing.Process(target=Worker(qs).work, kwargs={'burst': True}).start()     

我们遇到错误

TypeError：无法腌制_thread.lock对象

和

OSError：[WinError 87]参数不正确

rq.SimpleWorker之所以使用而不是rq.Worker因为Windows不支持所fork使用的功能rq.Worker。

问题：是什么原因导致此错误？我们该如何解决呢？

from rq import SimpleWorker, Connection

def needMoreWorkers():
    return True

if __name__ == '__main__':
    qs = sys.argv[1:] or ['default']
    with Connection(connection=my_redis_conn):
        if needMoreWorkers():
            multiprocessing.Process(target=SimpleWorker(qs).work, kwargs={'burst': True}).start()     

错误回溯

Traceback (most recent call last):
File "WorkerFactory.py", line 53, in <module>
    main(qs)
File "WorkerFactory.py", line 45, in main
    multiprocessing.Process(target=SimpleWorker(qs)).start()
File "C:\Users\x\AppData\Local\Continuum\anaconda3\envs\test\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
File …

我正在使用

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 …