标签: python-multiprocessing

我有这个代码:

import os

pid = os.fork()

if pid == 0:
    os.environ['HOME'] = "rep1"
    external_function()
else:
    os.environ['HOME'] = "rep2"
    external_function()

而这段代码:

from multiprocessing import Process, Pipe

def f(conn):
    os.environ['HOME'] = "rep1"
    external_function()
    conn.send(some_data)
    conn.close()

if __name__ == '__main__':
    os.environ['HOME'] = "rep2"
    external_function()
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print parent_conn.recv()
    p.join()

将external_function通过建立在环境变量中找到的目录必要的子目录初始化一个外部程序HOME.此功能仅在每个过程中执行一次.

使用第一个示例,可以os.fork()按预期创建目录.但是使用第二个例子,multiprocessing只有rep2get创建的目录.

为什么没有第二个例子中创建两个目录rep1和rep2？

我正在使用Python多处理来产生一些工作者.它们中的每一个都应该返回一个大小为几MB的数组.

1. 是否正确,因为我的返回数组是在子进程中创建的,所以当进程结束时需要将它复制回父进程的内存中？(这似乎需要一段时间,但它可能是一个小问题)
2. 是否有一种机制允许父和子访问相同的内存中对象？(同步不是问题,因为只有一个孩子会访问每个对象)

我担心我在python如何实现多处理方面存在一些差距,并且试图说服pypy玩得很好并不会让事情变得更容易.谢谢!

做

import multiprocessing
import schedule


def worker():
     #do some stuff


def sched(argv):
    schedule.every(0.01).minutes.do(worker)          
    while True:
        schedule.run_pending()


processs = []
..
..
p = multiprocessing.Process(target=sched,args)
..
..
processs.append(p)

for p in processs:
    p.terminate()

优雅地杀死进程列表？

如果不是，最简单的方法是什么？

目标是将配置文件重新加载到内存中，所以我想杀死所有子进程并创建其他进程，后者将读取新的配置文件。

编辑：添加了更多代码来解释我正在运行一个while True循环

编辑：这是@dano建议后的新代码

def get_config(self):
        from ConfigParser import SafeConfigParser
..
        return argv

def sched(self, args, event):
#schedule instruction:
        schedule.every(0.01).minutes.do(self.worker,args)
        while not  event.is_set():
                schedule.run_pending()                                                                    

def dispatch_processs(self, conf):
        processs = []
        event = multiprocessing.Event()

        for conf in self.get_config():
                process = multiprocessing.Process(target=self.sched,args=( …

我正在编写一个使用 multiprocessing.managers.DictProxy 的程序。该程序遍历一个目录，并创建一个以用户名为关键字的字典，该关键字pwd.getpwuid(os.stat(file)[4])[0]对应的值将是一个包含该用户拥有的文件的列表。例如对于假设的数据结构：

{'root': ["/boot/vmlinuz", "/boot/grub"], 'testuser': ["/home/testuser", "/home/testuser/.bashrc"]}

我写的代码是

#!/usr/bin/python
import os
import multiprocessing
import sys
import pwd
import grp
manager_attributes = multiprocessing.Manager()
file_stats_user = manager_attributes.dict()    
def filestat(file):
  try:
      stat = os.stat(file)
      user = pwd.getpwuid(stat[4])[0]
      group = grp.getgrgid(stat[5])[0]
      if user not in file_stats_user:
          file_stats_user[user] = []
      file_stats_user[user].append(file)
  except OSError, e:
      print e

try:
    cores = (multiprocessing.cpu_count()*2)
except:
    cores = 8
print "Starting parallel execution with ", cores, "concurrency"
pool_get_attributes = multiprocessing.Pool(cores)
pool_get_attributes.map(filestat, files)
pool_get_attributes.close()
pool_get_attributes.join()

其中 files …

我想创建一个动态大小的共享数组。我想在另一个进程中为其分配一个未知大小的数组。

from multiprocessing import Process, Value, Array


def f(a):
    b=[3,5,7]
    #resize(a,len(b)) # How to resize "a" ???
    a[:]=b  # Only works when "a" is initialized with the same size like the "b"


arr = Array('d', 0) #Array with a size of 0

p = Process(target=f, args=(arr))
p.start()
p.join()

print arr[:]

我使用这个脚本（见最后的代码）来评估在父进程被 fork 时是否共享或复制了全局对象。

简而言之，脚本创建一个全局data对象，子进程迭代data. 该脚本还监视内存使用情况以评估对象是否在子进程中被复制。

结果如下：

1. data = np.ones((N,N)). 子进程中的操作： data.sum(). 结果：data被共享（无复制）
2. data = list(range(pow(10, 8))). 子进程中的操作：sum(data). 结果：data被复制。
3. data = list(range(pow(10, 8))). 子进程中的操作：for x in data: pass. 结果：data被复制。

由于写时复制，预期结果 1)。我对结果 2) 和 3) 有点困惑。为什么会被data复制？

脚本

来源

import multiprocessing as mp
import numpy as np
import logging
import os

logger = mp.log_to_stderr(logging.WARNING)

def …

根据我们客户的要求,我必须开发一个应该能够处理大量CSV文件的应用程序.文件大小可以在10 MB - 2GB的范围内.

根据大小,模块决定是Multiprocessing pool使用普通文件还是使用普通文件来读取文件CSV reader.但是从观察开始,在测试大小为100 MB的文件的两种模式时,multi processing需要比平时更长的时间CSV reading.

这是正确的行为吗？或者我做错了什么？

这是我的代码:

def set_file_processing_mode(self, fpath):
   """ """
   fsize = self.get_file_size(fpath)
   if fsize > FILE_SIZE_200MB:
      self.read_in_async_mode = True
   else:
      self.read_in_async_mode = False

def read_line_by_line(self, filepath):
    """Reads CSV line by line"""
    with open(filepath, 'rb') as csvin:
        csvin = csv.reader(csvin, delimiter=',')
        for row in iter(csvin):
          yield row

def read_huge_file(self, filepath):
    """Read file in chunks"""
    pool = mp.Pool(1)
    for chunk_number in range(self.chunks): #self.chunks = 20
        proc …

from multiprocessing import Pool

def op1(data):
    return [data[elem] + 1 for elem in range(len(data))]
data = [[elem for elem in range(20)] for elem in range(500000)]

import time

start_time = time.time()
re = []
for data_ in data:
    re.append(op1(data_))

print('--- %s seconds ---' % (time.time() - start_time))

start_time = time.time()
pool = Pool(processes=4)
data = pool.map(op1, data)

print('--- %s seconds ---' % (time.time() - start_time))

我使用池的运行时间比循环运行时慢得多.但是池不应该使用4个处理器并行进行计算吗？

我正在使用multiprocessingpython 库生成 4 个Process()对象来并行化 CPU 密集型任务。任务（来自这篇伟大文章的灵感和代码）是计算列表中每个整数的质因数。

主要.py：

import random
import multiprocessing
import sys

num_inputs  = 4000
num_procs   = 4
proc_inputs = num_inputs/num_procs
input_list  = [int(1000*random.random()) for i in xrange(num_inputs)]

output_queue = multiprocessing.Queue()
procs        = []
for p_i in xrange(num_procs):
  print "Process [%d]"%p_i
  proc_list = input_list[proc_inputs * p_i:proc_inputs * (p_i + 1)]
  print " - num inputs: [%d]"%len(proc_list)

  # Using target=worker1 HANGS on join
  p = multiprocessing.Process(target=worker1, args=(p_i, proc_list, output_queue))
  # Using target=worker2 RETURNS with …

我正在尝试在Windows 10 Intel Core i7-8550U处理器上使用Python 3.7.4进行python多处理。
我正在用两个函数测试多进程，一个函数使用基本sleep（），另一个函数使用sklearn的matthews_corrcoef。多重处理可使用睡眠功能，但不能使用sklearn功能。

import numpy as np
from sklearn.metrics import matthews_corrcoef
import time
import concurrent.futures
from multiprocessing import Process, Pool
from functools import partial
import warnings
import sys

class Runner():
  def sleeper(self, pred, man, thr = None):
    return time.sleep(2)

  def mcc_score(self, pred, man, thr = None):
    warnings.filterwarnings("ignore")
    return matthews_corrcoef(pred, man)

  def pool(self, func):
    t1 = time.perf_counter()
    p = Pool()
    meth = partial(func, pred, man)
    res = p.map(meth, thres)
    p.close()

    t2 = time.perf_counter()
    print(f'Pool {func.__name__} {round((t2-t1), 3)} …