标签: python-multiprocessing

我试图从内存缓存(memcached)或网络加载HDF5数据,然后从多个Python进程查询它(只读),而不需要单独复制整个数据集.直观地说,我想将图像(就像它出现在磁盘上)映射到多个进程中,然后从Python中查询它.

我发现这很难实现,因此这个问题.指针/更正表示赞赏.

到目前为止我已经探索过的想法

• pytables - 这看起来最有前途,它支持用于查询HDF5数据的丰富界面,它(不像numpy)似乎可以处理数据而无需制作(进程本地)数据副本.它甚至支持一种File.get_file_image()似乎可以获取文件图像的方法.我没有看到如何从内存映像而不是磁盘文件构造新的File/FileNode.
• h5py - 另一种获取HDF5数据的方法,就像使用pytables一样,它似乎需要一个磁盘文件.选项driver ='core'看起来很有希望,但是我看不到如何在其中提供现有的mmap'd区域,而不是让它自己分配.
• numpy - 一种较低级别的方法,如果我通过mmap共享我的原始数据,那么我可能能够构建一个可以访问这些数据的numpy ndarray.但相关的构造函数ndarray.__new__(buffer=...)表示它将复制数据,而numpy视图似乎只能从现有的ndarray构建,而不是原始缓冲区.
• ctypes - 非常低级的方法(可能使用多处理的Value包装器来帮助一点).如果我直接使用ctypes,我可以毫无问题地读取我的mmap'd数据,但是我会丢失所有的结构信息,并从numpy/pandas/pytables中查询它.
• 分配磁盘空间 - 我可以只分配一个文件,写出所有数据,然后在我的所有进程中通过pytables共享它.我的理解是这将是内存效率,因为pytables不会复制(直到需要),显然这些进程将共享底层文件映像的操作系统磁盘缓存.我的反对意见是丑陋并将磁盘I/O带入了我想成为纯粹的内存系统.

这与此问题几乎相同,但那里给出的解决方案(调用freeze_support())对我不起作用.

我有以下脚本叫做start.py,我用它来构建一个带py2exe的独立可执行文件(版本0.9.2.2).我也在同一目录中有python.exe.

import multiprocessing

def main():
    print('Parent')
    p = multiprocessing.Process(target=new_process)
    multiprocessing.set_executable('python.exe')
    p.start()
    p.join()

def new_process():
    print('Child')

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

它作为纯python运行时工作得很好.但是,当打包为可执行文件时,这是我得到的错误:

Unknown option: --
usage: <path to start.exe> [option] ... [-c cmd | -m mod | file | -] [arg] ...
Try `python -h' for more information.

这显然是通过电话引起的

python.exe --multiprocessing-fork

如果我不调用set_executable()和freeze_support(),则子进程只启动exe并以__main__运行,导致无限链的新进程打印"Parent"而"Child"从不打印.

调用freeze_support()似乎唯一能做的就是如果我不调用multiprocessing.set_executable()会导致子进程引发以下错误

Traceback (most recent call last):
  File "start.py", line 17, in <module>
    multiprocessing.freeze_support()
  File "C:\Python34\Lib\multiprocessing\context.py", line 148, in freeze_support

    freeze_support()
  File "C:\Python34\Lib\multiprocessing\spawn.py", line 67, in …

我有这个使用模块中的Queue类的代码片段multiprocess.我很困惑,.empty()一个实例的方法Queue并没有像我期望的那样给我一个正确的值.这是我的代码:

from time import sleep
from multiprocessing import Queue, Lock

foo = Queue()
locker = Lock()

with locker:  # even with this, still True
    foo.put("bar")

print(foo.empty())  # True, obviously not
print(foo.empty())  # True
print(foo.empty())  # True
print(foo.qsize())  # 1L
print(foo.empty())  # True

但是,如果我使用sleep函数time,因为在执行中会导致按时间顺序延迟.有用.

from time import sleep
from multiprocessing import Queue, Lock

foo = Queue()
locker = Lock()

foo.put("bar")

sleep(0.01)

print(foo.empty())  # False
print(foo.empty())  # False
print(foo.empty())  # …

因此，我正在构建一个长期运行的查询Web应用程序以供内部使用。

我的目标是拥有一个带有守护进程的flask应用程序，该守护进程在服务器启动时启动，它将更新全局字典对象。

我不一定要发布任何示例代码，因为我尝试了许多方法，但都没有成功。

守护程序将创建一个线程池（multiprocessing.Pool）以遍历所有数据库实例，并对它们运行几个查询。

看来，无论我如何尝试实施（现在，使用flask开发服务器），它都将锁定应用程序，并且在运行时无法执行任何其他操作。我曾尝试阅读大量文档，但通常还是会假设很多其他知识，但最终我不知所措。

我想知道是否有人可以提供一些指导，即使是在我可以找到的地方，因为我到处搜索了“烧瓶启动例程”和类似的东西，但没有发现任何用处。看来，当我将其部署到服务器上时，也许可以在.wsgi文件中定义一些启动守护程序，但是在那之前还有什么方法可以在本地执行此操作？当我将其推出以用于一般用途时，这甚至是正确的方法吗？

否则，我只是在考虑建立一个cron作业，该作业连续运行一个python脚本来执行我需要的查询，然后转储到MongoDB实例或其他东西，以便客户端可以从中进行提取（就像执行所有查询一样）在Flask应用程序的服务器端只是锁定了服务器，因此无法执行其他操作-aka：无法对信息采取行动，杀死spids等）

任何帮助都将极大地帮助我，我的大脑已经旋转了好几天。

from flask import Flask
from celery import Celery

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'amqp://guest@localhost//'
app.config['CELERY_RESULT_BACKEND'] = 'amqp://guest@localhost//'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

output = 0

@app.before_first_request
def init():
    task = my_task.apply_async()

@app.route('/')
def hello_world():
    global output
    return 'Hello World! - ' + str(output)


@celery.task
def my_task():
    global output
    result = 0
    for i in range(100):
        result += i
        output = result




if __name__ == '__main__':
    app.run()

我想做的是每个过程都使用全局变量。但是我的过程没有采用全球价值观

import multiprocessing

count = 0 

def smile_detection(thread_name):
    global count

    for x in range(10):
        count +=1
        print thread_name,count

    return count    

x = multiprocessing.Process(target=smile_detection, args=("Thread1",))
y = multiprocessing.Process(target=smile_detection, args=("Thread2",))
x.start()
y.start()

我正在输出像

Thread1 1
Thread1 2
.
.
Thread1 9
Thread1 10
Thread2 1
Thread2 2
.
.
Thread2 9
Thread2 10

我想要的是

Thread1 1
Thread1 2
.
.
Thread1 9
Thread1 10
Thread2 11
Thread2 12
.
.
Thread2 19
Thread2 20

我必须要做些什么才能做到这一点？

这是一个完整简单的工作示例

import multiprocessing as mp
import time
import random


class Foo:
    def __init__(self):
        # some expensive set up function in the real code
        self.x = 2
        print('initializing')

    def run(self, y):
        time.sleep(random.random() / 10.)
        return self.x + y


def f(y):
    foo = Foo()
    return foo.run(y)


def main():
    pool = mp.Pool(4)
    for result in pool.map(f, range(10)):
        print(result)
    pool.close()
    pool.join()


if __name__ == '__main__':
    main()

我怎样才能修改它,所以Foo只被每个工人初始化一次,而不是每个任务？基本上我想要初始化调用4次,而不是10次.我使用的是python 3.5

考虑以下脚本,我在其中测试两种方法,对通过以下方式获得的生成器执行某些计算itertools.tee:

#!/usr/bin/env python3

from sys import argv
from itertools import tee
from multiprocessing import Process

def my_generator():
    for i in range(5):
        print(i)
        yield i

def double(x):
    return 2 * x

def compute_double_sum(iterable):
    s = sum(map(double, iterable))
    print(s)

def square(x):
    return x * x

def compute_square_sum(iterable):
    s = sum(map(square, iterable))
    print(s)

g1, g2 = tee(my_generator(), 2)

try:
    processing_type = argv[1]
except IndexError:
    processing_type = "no_multi"

if processing_type == "multi":
    p1 = Process(target=compute_double_sum, args=(g1,))
    p2 = Process(target=compute_square_sum, args=(g2,))
    print("p1 starts") …

我有两个独立的功能.每个都需要很长时间才能执行.

def function1(arg):
     do_some_stuff_here
     return result1

def function2(arg1, arg2, arg3):
     do_some_stuff_here
     return result2

我想并行启动它们,得到它们的结果(知道哪个是哪个)并在之后处理结果.根据我的理解,多处理比Python 2.7中的线程(GIL相关问题)更有效.但是我有点迷失是否更好地使用Process,Pool或Queue以及如何以正确的pythonic方式为我的用例实现它们.

任何帮助赞赏;)

我有以下情形：多个工作进程将有关其当前状态的事件发送到事件调度程序。然后，如果我们在主流程中，则此事件分发程序需要处理所有事件；如果在工作进程中，则该信号将需要主进程的事件分发程序发信号以处理这些事件。

这里的主要问题是事件处理也必须在主进程的主线程中，所以我不能只在线程中运行一会儿True循环并在那里等待工作进程的消息。

所以我有这个：

import asyncio
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import current_process, Process, Queue
from threading import current_thread
from time import sleep

def get_q(q):
    print("Waiting for the queue ({} / {})\n".format(current_thread().name, current_process().name))
    return q.get()

async def message_q(q):
    while True:
        f = loop.run_in_executor(None, get_q, q)

        await f

        if f.result() is None:
            print("Done")
            return;

        print("Got the result ({} / {})".format(current_thread().name, current_process().name))
        print("Result is: {}\n".format(f.result()))

async def something_else():
    while True:
        print("Something else\n")
        await asyncio.sleep(2)

def other_process(q):
    for i …

我正在尝试使用线程和多处理模块

import multiprocessing as mp
import threading
from threading import Thread
import time

QP = mp.Queue()
Lock = mp.Lock()
Input = 100
Checked = mp.Value("i",0)

class MultiThreading(threading.Thread):
    def __init__(self,QP,Checked):
        threading.Thread.__init__(self)
        self.QP = QP
        self.Checked = Checked
    def run(self):
        global Input
        global Lock
        QueueMode = self.QP.get()
        First,Second = QueueMode
        Lock.acquire()
        print ("{} {} - Has Successfully Done".format(First,Second))
        Checked.value += 1
        time.sleep(0.25)
        Lock.release()
        if not Input == Checked.value:
            t = MultiThreading(QP,Checked)
            t.setDaemon(False)
            t.start()

def MultiCall(QP,Checked):
    for i in range(10):
        t = MultiThreading(QP,Checked) …