标签: python-multiprocessing

我有一个由 管理的字典SyncManager，我正在使用.get()和.update()函数来检索和设置字典键和值。但是，该del dict[key]方法并未实施。有没有办法从托管字典键中删除键？

简单的例子：

mgr = multiprocessing.Manager()
shared_dict = mgr.dict()

shared_dict.update( {'key1' : 'value1'} )
print shared_dict.get('key1')

#Now let's delete key1
del shared_dict['key1']  #Nope!
shared_dict.__delitem__('key1')  #Also not allowed

有时将多处理池与管理器和 Python 3.4 结合使用，lock.acquire()会抛出一个奇怪的TypeError: an integer is required (got type NoneType).

有几次这出现在我的 Travis 测试套件上，我无法弄清楚它来自哪里以及它意味着什么。更糟糕的是，我无法可靠地重现它，它只是发生或不发生。通常它不是，而是每一百次运行一次左右它确实:-(。

我完全迷失了。也许有人以前遇到过这样的事情，并且可以提示在哪里寻找错误的来源。让我从完整的回溯开始：

Traceback (most recent call last):
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/site-packages/pypet-0.1b10.dev0-py3.4.egg/pypet/environment.py", line 150, in _single_run
    traj._store_final(store_data=store_data)
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/site-packages/pypet-0.1b10.dev0-py3.4.egg/pypet/trajectory.py", line 3271, in _store_final
    store_data=store_data)
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/site-packages/pypet-0.1b10.dev0-py3.4.egg/pypet/storageservice.py", line 295, in store
    self.acquire_lock()
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/site-packages/pypet-0.1b10.dev0-py3.4.egg/pypet/storageservice.py", line 284, in acquire_lock
    self._lock.acquire()
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/multiprocessing/managers.py", line 958, in acquire
    return self._callmethod('acquire', args)
  File "/home/travis/miniconda/envs/test-environment/lib/python3.4/multiprocessing/managers.py", line 731, in _callmethod …

这是我的代码。我正在尝试创建 n（此处为 10）个进程，只是想查看它们的 PIDS。当我打印它们并在终端中验证 PID 时，它们不一样。编辑：如果有帮助，我正在 Mac (Yosemite) 上运行它。

for proc in xrange(10):
        worker_process = Process (name="worker_process", target=worker_code, args=(proc, tree_space, self.simulator, mgr_nms))
        process_q.append(worker_process)
        worker_process.start()
        print worker_process.pid

我的输出：

60484
60485
60486
60487
60488
60489
60490
60491
60493
60494

终端输出（顶部）：

（图片）-> http://postimg.org/image/kkiboom6l/

关于为什么会这样的任何指示？（对不起，如果我遗漏了一些如此明显的东西。）

我有一个由经理列表组成的数组，例如：

[<ListProxy object, typeid 'list' at 0x177d350>, <ListProxy object, typeid 'list' at 0x177d390>, ...]

并尝试将其转换为纯列表。我的考虑是这样的：

y=[]
for i in my_manager_list_content:
    for j in i:
        y.append(float(j))
print y #This works

print [next(iter(i[:]))  for i in my_manager_list_content] #Why doesn't this work?

我怎么能在 oneliner 中做到这一点？

我有两个 python 类，它们共享一个基类，假设定义了一个 multiprocessing.Queue。

这些类中的每一个都将在单独的子进程上启动，并且需要通过队列共享信息。

现在，如果基类定义了一个队列，那么每个子类对象将实例化它自己的队列，从而无法共享队列元素。但话又说回来，我发现这记录

Queue objects should only be shared between processes through inheritance. 

那么在子进程之间共享队列的正确方法是什么，上面的句子甚至如何有意义？

请注意，我显然可以在子进程初始化时传递对队列的引用，但我想使用继承来解决这个问题

我有一个包含大量文本文件的文件夹。每一个都被压缩并重几个千兆字节。我写了一段代码来拆分每个 gzip 文件的内容：每个 gzip 文件都用 打开gzip，然后读取每个指定的行块并将其写入一个新的 gzip 文件。

这是文件中的代码file_compression.py：

import sys, os, file_manipulation as fm
import gzip


def splitGzipFile(fileName, dest=None, chunkPerSplit=100, linePerChunk=4, file_field_separator="_", zfill=3
                  , verbose=False, file_permission=None, execute=True):
    """
    Splits a gz file into chunk files.
    :param fileName:
    :param chunkPerSplit:
    :param linePerChunk:
    :return:
    """
    absPath = os.path.abspath(fileName)
    baseName = os.path.basename(absPath)
    dirName = os.path.dirname(absPath)
    destFolder = dirName if dest is None else dest


    ## Compute file fields
    rawBaseName, extensions = baseName.split(os.extsep, 1)

    if not str(extensions).startswith("."):
        extensions = …

我有一个利用多处理来提高效率的 python 程序，以及一个为每个进程创建一个记录器的函数。记录器函数如下所示：

import logging
import os

def create_logger(app_name):
    """Create a logging interface"""
    # create a logger
    if logging in os.environ:
        logging_string = os.environ["logging"]
        if logging_string == "DEBUG":
            logging_level = loggin.DEBUG
        else if logging_string == "INFO":
            logging_level = logging.INFO
        else if logging_string == "WARNING":
            logging_level = logging.WARNING
        else if logging_string == "ERROR":
            logging_level = logging.ERROR
        else if logging_string == "CRITICAL":
            logging_level = logging.CRITICAL
    else:
        logging_level = logging.INFO

    logger = logging.getLogger(app_name)
    logger.setLevel(logging_level)

    # Console handler for error output
    console_handler = logging.StreamHandler() …

在 Python 2.7 中，我实现了一个具有多个队列和消费者的多处理场景。简化的想法是，我有一个作业的生产者，这些作业被提供给消费者，处理作业和一个错误处理程序，它负责所有的日志记录。非常简化，看起来都可以与之媲美：

import multiprocessing as mp
import Queue

job_queue = mp.Queue()
error_queue = mp.Queue()
for i in range(10):
    job_queue.put(i)

def job_handler(job_queue, error_queue):
    print 'Job handler'
    while True:
        try: 
            element = job_queue.get_nowait()
            print element
        except:
# t1
            error_queue.put('Error')
            error_queue.close()
            error_queue.join_thread()
            job_queue.close()
            job_queue.join_thread()
# t2
            return 1

def error_handler(error_queue):
    result = error_queue.get()
    if result == 'Error':
        error_queue.close()
        error_queue.join_thread()

if __name__ == '__main__':
    print 'Starting'
    p1 = mp.Process(target = error_handler, args = (error_queue, ))
    p1.start()
    p2 = mp.Process(target = job_handler, args …

我有一个（大）名单，里面有男性和女性特工。

我想对每个应用不同的功能。

在这种情况下如何使用 Pool ？鉴于代理是相互独立的。

一个例子是：

males = ['a', 'b', 'c']
females = ['d', 'e', 'f']
for m in males:
    func_m(m)
for f in females:
    func_f(f)

我是这样开始的：

from multiprocessing import Pool
p = Pool(processes=2)
p.map() # Here is the problem

我想要一些类似的东西：

p.ZIP(func_f for f in females, func_m for m in males) # pseudocode

我有一个关于多处理和 tkinter 的问题。我在使我的进程与 tkinter GUI 并行运行时遇到了一些问题。我创建了一个简单的示例来练习，并且一直在阅读以了解多处理的基础知识。但是，将它们应用于 tkinter 时，一次只运行一个进程。（使用多处理模块更新 Tkinter GUI）此外，当我添加队列以在进程之间进行通信时，（如何在 Python 中使用多处理队列？），该进程甚至不会启动。

目标： 我想要一个进程倒计时并将值放入队列中，另一个进程在 1 秒后更新 tkinter 并向我显示值。

感谢所有建议

亲切的问候，S

编辑：我希望在调用 after 方法时数据可用。所以问题不在于after函数，而在于after函数调用的方法。每次完成计算需要0.5秒。因此，GUI 每秒无响应半秒。

EDIT2：根据反馈对代码进行了更正，但此代码尚未运行。

class Countdown(): 
    """Countdown prior to changing the settings of the flows"""

    def __init__(self,q):

        self.master = Tk()
        self.label = Label(self.master, text="", width=10)
        self.label.pack()
        self.counting(q)

    # Countdown()
    def counting(self, q):
        try:
            self.i = q.get()
        except:
            self.label.after(1000, self.counting, q)

        if int(self.i) <= 0:
            print("Go")
            self.master.destroy()

        else:
            self.label.configure(text="%d" % self.i)
            print(i)
            self.label.after(1000, self.counting, …