标签: python-multiprocessing

我有一个作为守护进程运行的Python脚本.在启动时,它会生成5个进程,每个进程都连接到Postgres数据库.现在,为了减少数据库连接的数量(最终会变得非常大),我试图找到一种跨多个进程共享单个连接的方法.为此我正在寻找multiprocessing.sharedctypes.ValueAPI.但是,我不确定如何psycopg2.connection跨进程使用此API 传递对象.谁能告诉我怎么做？

我也愿意接受其他想法以解决这个问题.

我没有考虑将连接作为构造函数的一部分传递给5个进程的原因是互斥处理.如果我遵循这种方法,我不确定如何防止多个进程访问连接.谁能告诉我这是否是正确的做法？

我一直在尝试以下方法：

from multiprocessing import Pool

def f(some_list):
    some_list.append(4)
    print 'Child process: new list = ' + str(some_list)
    return True

if __name__ == '__main__':

    my_list = [1, 2, 3]
    pool = Pool(processes=4)
    result = pool.apply_async(f, [my_list])
    result.get()

    print 'Parent process: new list = ' + str(my_list)

我得到的是：

Child process: new list = [1, 2, 3, 4]
Parent process: new list = [1, 2, 3]

因此，这意味着my_list是按值传递的，因为它没有突变。那么，是否将规则传递给另一个进程时确实按值传递规则呢？谢谢。

我正在尝试学习使用Python的多处理模块.作为第一次测试,我想我会同时运行四个15秒的过程.我写了这个模块,我称之为"multiPtest.py"::

import time
import timeit
import multiprocessing


def sleepyMe(napTime):
    time.sleep(napTime)
    print "Slept %d secs" % napTime

def tester(numTests):
    #Launch 'numTests' processes using multiProcessing module
    for _ in range(numTests):
        p = multiprocessing.Process(target=sleepyMe(15))
        p.start() #Launch an 'independent' process
        #p.join() ##Results identical with or without join

def multiTester():
    #Time running of 4 processes
    totTime = timeit.Timer('tester(4)', setup = 'from multiPtest import     tester').repeat(1,1)
    print "Total = ", totTime[0]

但是,当我跑步时,我得到了这些结果:

Python 2.7.6 (default, Mar 22 2014, 22:59:56) 
[GCC 4.8.2] on linux2
Type "help", "copyright", "credits" …

我正在使用multiprocessingpython模块.我有大约20-25个任务同时运行.每个任务将创建一个pandas.DataFrame~20k行的对象.问题是,所有任务都执行得很好,但是当谈到"加入"流程时,它就会停止.我尝试过"小"DataFrames并且效果很好.为了说明我的观点,我在下面创建了代码.

import pandas
import multiprocessing as mp

def task(arg, queue):
    DF = pandas.DataFrame({"hello":range(10)}) # try range(1000) or range(10000)
    queue.put(DF)
    print("DF %d stored" %arg)

listArgs = range(20)
queue = mp.Queue()
processes = [mp.Process(target=task,args=(arg,queue)) for arg in listArgs]

for p in processes:
    p.start()

for i,p in enumerate(processes):
    print("joining %d" %i)
    p.join()

results = [queue.get() for p in processes]

编辑:

使用DF = pandas.DataFrame({"hello":range(10)})我有一切正确:"DF 0存储"直到"DF 19存储",与"加入0"到"加入19"相同.

然而,DF = pandas.DataFrame({"hello":range(1000)})问题出现了:当它存储DF时,加入步骤在"加入3"之后停止.

感谢有用的提示:)

我有一个清单的理解：

thingie=[f(a,x,c) for x in some_list]

我正在并行化如下：

from multiprocessing import Pool

pool=Pool(processes=4)

thingie=pool.map(lambda x: f(a,x,c), some_list)

但我收到以下错误：

_pickle.PicklingError: Can't pickle <function <lambda> at 0x7f60b3b0e9d8>:
attribute lookup <lambda> on __main__ failed

我尝试安装pathos显然可以解决此问题的软件包，但是当我尝试导入它时，出现错误：

ImportError: No module named 'pathos'

我想获得一些帮助，以使用multiprocessing模块并行运行多个python测试用例。我创建了一个FooTest类，其中包含10个测试用例（testA，testB ...）。类之外有一个测试用例test_foo。

如何使用python多处理模块并行运行所有测试用例？谢谢您的帮助。

import unittest
import time

def setUp():
    print "module setup"

def test_foo():
    pass

class FooTest(unittest.TestCase):

    # preparing to test
    def setUp(self):
        """ Setting up for the test """
        print "FooTest:setUp_:begin"

    # ending the test
    def tearDown(self):
        """Cleaning up after the test"""
        print "FooTest:tearDown_:begin"

    # test routine A
    def testA(self):
        """Test routine A"""
        print "FooTest:testA"
        time.sleep(2)

    # test routine B
    def testB(self):
        """Test routine B"""
        print "FooTest:testB"

    # test routine C
    def testC(self):
        """Test routine C"""
        print "FooTest:testC"

    # …

从pickle加载multiprocessing.manager.dictionary时遇到错误.我确定pickle文件存在于我运行python的目录中.

Python 2.7.6 (default, Jun 22 2015, 17:58:13) 
[GCC 4.8.2] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import pickle
>>> import multiprocessing
>>> a = open("test.pkl", "rb")
>>> pickle.load(a)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/python2.7/pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "/usr/lib/python2.7/pickle.py", line 858, in load
    dispatch[key](self)
  File "/usr/lib/python2.7/pickle.py", line 1133, in load_reduce
    value = func(*args)
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 879, in RebuildProxy
    return func(token, serializer, incref=incref, **kwds) …

在python 3中,我试图同时运行多个参数的相同函数.我在Windows 7中使用Python 3.5.2,Anaconda 4.1.1(64位)中的多处理.我收到有关spawn.py的以下错误:

在当前进程完成其自举阶段之前,已尝试启动新进程.

这可能意味着您没有使用fork来启动子进程,而您忘记在主模块中使用正确的习惯用法:

if __name__ == '__main__':
    freeze_support()
    ...

如果程序不会被冻结以生成可执行文件,则可以省略"freeze_support()"行.

我的代码是:

from multiprocessing import Process

#Function to be run
def savedatetime(password,rangestart,rangeend):
    print((password,rangestart,rangeend))

# Inputs to the function
passwords=['p0','p1','p2']
ranges=[(1,10),(10,20),(20,30)]

# Creating the list of processes to append
fileind=0
proc=[]
for rangeind in range(0,len(passwords)):    
    password=passwords[fileind]
    rangestart=ranges[fileind][0]
    rangeend=ranges[fileind][1]
    p = Process(target=savedatetime,args=(password,rangestart,rangeend))            
    proc.append(p)
    fileind=fileind+1
    # running sequentially to check that the function works
    savedatetime(password,rangestart,rangeend)

print(proc)

# Attempting to run simultaneously. This is where the error occurs:
for p in …

在我的项目中，我使用argprse传递参数，而在脚本中的某个地方使用多重处理来完成其余的计算。如果我从命令提示符处调用脚本，则脚本工作正常。

“ python complete_script.py --arg1=xy --arg2=yz”。

但是在使用Pyinstaller使用命令“ pyinstaller --onefile complete_script.py”将其转换为exe后，它将抛出

错误

“错误：无法识别的参数：--multiprocessing-fork 1448”

任何建议，我如何使这项工作。或任何其他选择。我的目标是创建一个exe应用程序，可以在未安装Python的其他系统中调用该应用程序。

以下是我的工作站的详细信息：

Platform: Windows 10
Python : 2.7.13 <installed using Anaconda>
multiprocessing : 0.70a1
argparse:   1.1

从评论中复制：

def main():
     main_parser = argparse.ArgumentParser()
     < added up arguments here>
    all_inputs = main_parser.parse_args()
    wrap_function(all_inputs)


def wrap_function(all_inputs):
    <Some calculation here >
   distribute_function(<input array for multiprocessing>)

def distribute_function(<input array>):
    pool = Pool(process = cpu_count)
    jobs = [pool.apply_async(target_functions, args = (i,) for i in input_array)]
    pool.close()

我有一个非常简单的vanilla实现concurrent.futures.ProcessPoolExecutor- 类似于(使用Python 3.6):

files = get_files()
processor = get_processor_instance()
with concurrent.futures.ProcessPoolExecutor() as executor:
    list(executor.map(processor.process, files))

虽然它processor是许多可用处理器类中的任何一个的实例,但它们都共享该process方法,其大致如下所示:

def process(self, file):
    log.debug(f"Processing source file {file.name}.")
    with DBConnection(self.db_url) as session:
        file = session.merge(file)
        session.refresh(file)
        self._set_file(file)
        timer = perf_counter()
        try:
            self.records = self._get_records()
            self._save_output()
        except Exception as ex:
            log.warning(f"Failed to process source file {file.ORIGINAL_NAME}: {ex}")
            self.error_time = time.time()
            self.records = None
        else:
            process_duration = perf_counter() - timer
            log.info(f'File {file.name} processed in {process_duration:.6f} seconds.')
            file.process_duration = process_duration
        session.commit()

每个类的实现 …