标签: multiprocess

如何使用 start_http_server 公开多进程应用程序中的指标

我在互联网上找到了很多gunicorn的例子，但我想使用start_http_server

我应该如何处理下面的代码才能使其正常工作？

from multiprocessing import Process
import time, os
from prometheus_client import start_http_server, multiprocess, CollectorRegistry, Counter


MY_COUNTER = Counter('my_counter', 'Description of my counter')
os.environ["PROMETHEUS_MULTIPROC_DIR"] = "tmp"

def f():
    print("+1")
    MY_COUNTER.inc()

if __name__ == '__main__':
    start_http_server(8000)
    p = Process(target=f, args=())
    a = p.start()
    p2 = Process(target=f, args=())
    p2.start()
    time.sleep(1)
    print("collect")
    registry = CollectorRegistry()
    data = multiprocess.MultiProcessCollector(registry)
    while True:
        time.sleep(1)

我试图明确地找出 SciPy/NumPy 中的哪些函数在多个处理器上运行。例如，我可以在 SciPy 参考手册中读到 SciPy 使用它，但我对哪些函数确实运行并行计算更感兴趣，因为并非所有函数都这样做。理想的场景当然是在您键入 help(SciPy.foo) 时包含它，但情况似乎并非如此。

任何帮助都感激不尽。

最好的事物，

马蒂亚斯

我是一个多处理新手,

我对线程有所了解,但我需要提高计算速度,希望通过多处理:

示例描述:将字符串发送到线程,更改字符串+基准测试,将结果发送回打印.

from threading import Thread

class Alter(Thread):
    def __init__(self, word):
        Thread.__init__(self)
        self.word = word
        self.word2 = ''

    def run(self):
        # Alter string + test processing speed
        for i in range(80000):
            self.word2 = self.word2 + self.word

# Send a string to be altered
thread1 = Alter('foo')
thread2 = Alter('bar')
thread1.start()
thread2.start()

#wait for both to finish
while thread1.is_alive() == True: pass
while thread2.is_alive() == True: pass


print(thread1.word2)
print(thread2.word2)

Run Code Online (Sandbox Code Playgroud)

目前这需要大约6秒钟,我需要它更快.
我一直在研究多处理,找不到与上面代码相​​同的东西.我认为我所追求的是汇集,但我发现的例子很难理解.我想利用所有核心(8核),multiprocessing.cpu_count()但我真的只是有关多处理的有用信息,而不足以复制上述代码.如果有人能指出我正确的方向或更好的方向,请提供一个非常感谢的例子.Python 3请

我试图用Python实现多处理.它可以在汇集非常快速的任务时工作,但是当汇集更长的任务时会冻结.请参阅下面的示例:

from multiprocessing import Pool
import math
import time

def iter_count(addition):
    print "starting ", addition
    for i in range(1,99999999+addition):
        if i==99999999:  
            print "completed ", addition
            break

if __name__ == '__main__':
    print "starting pooling "
    pool = Pool(processes=2)
    time_start = time.time()
    possibleFactors = range(1,3)   

    try: 
        pool.map( iter_count, possibleFactors)
    except:
        print "exception"

    pool.close()
    pool.join()      

    #iter_count(1)
    #iter_count(2)
    time_end = time.time()
    print "total loading time is : ", round(time_end-time_start, 4)," seconds"

在这个例子中,如果我在for循环中使用较小的数字(类似9999999)它可以工作.但是当它运行99999999时会冻结.我尝试按顺序运行两个进程(iter_count(1)和iter_count(2)),大约需要28秒,所以不是一个大任务.但是当我把它们汇集起来时它会冻结.我知道python中存在一些已知的多处理错误,但在我的情况下,相同的代码适用于较小的子任务,但冻结较大的子任务.

我有一个 Android 应用程序，它扩展了Application该类，并且有许多应用程序组件（服务、活动和广播接收器）在三个不同的进程中运行。

每个进程将在Application启动后立即实例化该类。所以我一直在寻找如何检查Application类代码，如果实例归主应用程序进程所有，但我找不到任何东西。

我的清单看起来像这样：

<application
    android:allowBackup="true"
    android:icon="@drawable/ic_launcher"
    android:label="@string/app_name"
    android:name=".MyApplication"
    android:theme="@style/AppTheme" >

    <activity
        android:name="...">
        ...
    </activity>

    <activity
        android:name="...">
        ...
    </activity>

    <service android:name="..." >
    </service>

    <service android:name="..."
        android:process=":SecondProcess" >
    </service>

    <service android:name="..."
        android:process=":ThirdProcess" >
    </service>

接下来是我的Application类实现：

public class MyApplication extends Application {

    @Override
    public void onCreate() {
        super.onCreate();

        if (is this instance being ran on the main process?) {
            // Do something
        }
    }

}

有谁知道如何检查实例是否在主进程上运行？

我对 Python 相当陌生，我的经验是通过 Siemens PSS/e 中提供的 API 在 Powerflow 建模中使用它的。我有一个脚本，我已经使用了好几年了，它在大型数据集上运行一些模拟。

为了快速完成，我通常将输入分成多个部分，然后在 IDLE 中运行脚本的多个实例。我最近为输入添加了一个 GUI，并改进了代码以更加面向对象，创建了一个类，GUI 将输入传递给它，然后像原始脚本一样工作。

我的问题是如何从程序本身内部拆分过程而不是制作多个副本？我已经阅读了一些关于 mutliprocess 模块的内容，但我不确定如何将它应用于我的情况。本质上，我希望能够实例化 N 个相同的对象，每个对象都并行运行。

我现在拥有的类（称为 Bot）传递了一组参数并在它运行时创建一个 csv 输出，直到它完成。我有一个单独的代码块，可以在最后将各个部分组合在一起，但现在我只需要了解在我的 GUI 中点击运行后踢出多个 Bot 对象的最佳方法。GUI 中有输入来指定要使用的 N 段的数量。

如果我的问题相当模糊，我会提前道歉。感谢您提供任何信息，因为我有点卡住了，不知道在哪里寻找更好的答案。

我有一个包含多个“记录”的大型熊猫数据框，其中包含 2 个或更多行项目。我正在尝试使用多处理有效地对每条记录执行 CPU 密集型计算。这是一个简单的例子，它的函数只是向每条记录添加一个随机数：

import pandas as pd
from random import randrange
from multiprocessing import Pool

#Trivial example function 
def my_func(record): 
    df.loc[((df.Record == record), 'Result')] = randrange(0,100)
    print (df)

d = {'Record': ['A', 'A', 'B', 'B'], 'Values': [100, 200, 50, 70]}
df = pd.DataFrame(d)
all_records = df['Record'].unique()

if __name__ == '__main__':
    pool = Pool(processes=2)
    pool.map(my_func,all_records)
    df.to_csv('output.csv')

所需的输出是带有标题为“结果”的新列的原始数据框，其中包含每个记录的随机数。例如：

    Record  Values  Result
0      A     100    63.0
1      A     200    63.0
2      B      50    22.0
3      B      70    22.0

实际结果是我的 CSV 输出没有用结果列更新。我可以通过函数中的打印语句判断进程正在运行。根据我的研究，这些进程对 …

我正在尝试通过我的应用程序的多重处理进行数据库连接。我假设我调用函数的地方发生了错误。这几天我一直在为多处理而苦苦挣扎。我从https://engineering.talentpair.com/django-multiprocessing-153dbcf51dab/找到了代码

我的代码如下

视图.py

from src.conf.multiproces import MultiProcess

class GridSearchAPI(generics.GenericAPIView):
    @transaction.atomic
    def post(self, request):
        with MultiProcess() as mp:
            mp.map(request)
            returnData = mp.results()

            if returnData['result']:
                return Response(returnData, status=status.HTTP_200_OK)
            else:
                returnData["message"] = "no data found"
                return Response(result, status=status.HTTP_400_BAD_REQUEST)

多进程.py

from __future__ import print_function, absolute_import, unicode_literals
import time
try:
    from Queue import Empty
except ImportError:
    from queue import Empty

# from pathos import 
from multiprocessing import Process,cpu_count, Manager
import logging
import traceback

from django import db
from django.conf import settings
from django.core.cache …

我正在尝试使用forkserver并NameError: name 'xxx' is not defined在工作进程中遇到。

我使用的是 Python 3.6.4，但文档应该是相同的，来自https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods它说：

fork 服务器进程是单线程的，因此使用 os.fork() 是安全的。没有不必要的资源被继承。

此外，它说：

继承比pickle/unpickle更好

使用 spawn 或forkserver启动方法时，多处理中的许多类型需要可picklable，以便子进程可以使用它们。但是，通常应该避免使用管道或队列将共享对象发送到其他进程。相反，您应该安排程序，以便需要访问在别处创建的共享资源的进程可以从祖先进程继承它。

很明显，我的工作进程需要处理的关键对象没有被服务器进程继承，然后传递给工作人员，为什么会发生这种情况？我想知道 forkserver 进程究竟从父进程继承了什么？

这是我的代码的样子：

import multiprocessing
import (a bunch of other modules)

def worker_func(nameList):
    global largeObject
    for item in nameList:
        # get some info from largeObject using item as index
        # do some calculation
        return [item, info]

if __name__ == '__main__':
    result = []
    largeObject # This is my large object, …

我正在尝试通过多进程提高某些代码的速度。我注意到速度没有按预期增加。我知道子进程的产生有开销，父进程和子进程之间的数据传输也有开销。然而，即使在我最小化开销之后，多进程的性能仍然没有达到我的预期。于是我写了一个简单的测试代码：

import multiprocessing
import numpy as np
import time

def test_function():
    start_time = time.time()
    n = 1000
    x = np.random.rand(n,n)
    p = np.random.rand(n,n)
    y = 0
    for i in range(n):
        for j in range(n):
            y += np.power(x[i][j], p[i][j])

    print ("= Running time:",time.time()-start_time)
    return 

def main():
    procs = [1,2,3,4,5,6]
    for proc in procs:
        print("Number of process:", proc)
        pool = multiprocessing.Pool(processes=proc)
        para = [(),] * proc
        pool.starmap(test_function,para)
        pool.close()
        pool.join()

if __name__ == '__main__':
    main()

您可以看到测试函数只有两个循环和一些数学计算。主进程和子进程之间没有数据传输，并且时间是在子进程内部计算的，因此不会包含任何开销。这是输出：

Number of process: 1
= Running time: …