标签: multiprocessing

所以我想我终于要发帖了；管理Process工人的正确方法是什么？我尝试使用 a Pool，但我注意到我无法获得每个已完成进程的返回值。我尝试使用回调，但也没有按预期工作。我应该自己管理它们active_children ()吗？

我的池代码：

from multiprocessing import *                                                                                      
import time
import random

SOME_LIST = []

def myfunc():
    a = random.randint(0,3)
    time.sleep(a)
    return a

def cb(retval):
    SOME_LIST.append(retval)

print("Starting...")

p = Pool(processes=8)
p.apply_async(myfunc, callback=cb)
p.close()
p.join()

print("Stopping...")
print(SOME_LIST)

我希望有一个值列表；但我得到的只是工人作业中要完成的最后一项：

$ python multi.py 
Starting...
Stopping...
[3]

注意：答案不应使用threading模块；原因如下：

在 CPython 中，由于全局解释器锁，一次只有一个线程可以执行 Python 代码（即使某些面向性能的库可能会克服这一限制）。如果您希望您的应用程序更好地利用多核机器的计算资源，建议您使用多处理。

我有一个程序，它读取一些输入文本文件并将它们全部写入一个名为的列表中ListOutput，该列表是我的程序中使用的两个进程之间的共享内存（我使用了两个进程，因此我的程序运行得更快！）我还有一个共享内存变量调用processedFiles它存储任何进程已读取的输入文件的名称，因此当前进程不会再次读取它们！

1. 确保两个进程不会同时检查“processedFiles”（例如在开始时，它们可能同时得出“processedFiles”为空的结论，因此它们读取相同文件），因此，我Lock在检查部分周围添加了一个，processedFiles以便一个进程应该在另一个进程签入该部分之前完成并释放锁定的部分！

   我的问题是该Lock功能似乎不起作用，当我ProcessName在锁定部分内打印电流时，它表明两个进程都在锁定部分内。我不知道我的代码有什么问题？（请参阅下面的输出。）

2. 由于我的主程序不仅是关于读取输入文本文件并将它们打印在列表中，而且在将它们打印到列表之前它必须对输入文件进行非常复杂的操作，我应该使用Pool而不是，Process为什么？

代码：

import glob
from multiprocessing import Process, Manager
from threading import *
import timeit
import os

os.chdir("files")

def print_content(ProcessName,processedFiles,ListOutput,lock):

   for file in glob.glob("*.txt"):
      newfile=0

      lock.acquire()

      print "\n Current Process:",ProcessName

      if file not in processedFiles:
         print "\n", file, " not in ", processedFiles," for ",ProcessName
         processedFiles.append(file)
         newfile=1#it is a new file

      lock.release()
      #if it is a new file
      if newfile==1:
         f …

今天，我遇到了一个关于共享数据进程间的奇怪问题。我声明MainActivity运行在另一个进程中，将TestApplication中的共享数据写为1，然后启动SubActivity来展示共享数据。不幸的是，SubActivity中显示的值仍然为0，因此我们得出结论，在两个进程中填充了两个TestApplication实例，并且共享数据的读写是相互独立的。实际上，共享数据不再在进程间共享。我的问题是在新进程中开始的活动与原始活动之间的另一个区别是什么，例如关于内存？这是我的代码：

<application
        android:name=".TestApplication"
        android:allowBackup="true"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/AppTheme">
<activity android:name=".MainActivity"
            android:process="com.rlk.miaoxinli.hellokitty"
            >
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />

                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
        <activity android:name=".SubActivity">

        </activity>
    </application>


public class TestApplication extends Application {

    public int mValue = 0;
}


public class MainActivity extends AppCompatActivity implements View.OnClickListener {

    private TextView mTextView;
    private TestApplication mApplication;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        mApplication = (TestApplication) getApplication();

        setContentView(R.layout.activity_main);
        mTextView = (TextView)findViewById(R.id.first);
        mApplication.mValue = 1;

        mTextView.setClickable(true);
        mTextView.setOnClickListener(this);
    }

    @Override
    protected void onResume() {
        super.onResume();
        mTextView.setText("" …

基于这个非常有用的教程,我尝试了一个简单的Python实现multiprocessing来衡量它的有效性.这些模块multi1,multi2,multi3包含ODE的集成,并在CSV输出的计算值(它并不重要,他们在这里为一个脚本来执行的东西).

import multiprocessing
import multi1
import multi2
import multi3
import time

t0 = time.time()

if __name__ == '__main__':

    p1 = multiprocessing.Process(target = multi1.main(), args=())
    p2 = multiprocessing.Process(target = multi2.main(), args=())
    p3 = multiprocessing.Process(target = multi3.main(), args=())

    p1.start()
    p2.start()
    p3.start()

    p1.join()
    p2.join()
    p3.join()

t1 = time.time()

multi1.main()
multi2.main()
multi3.main()

t2 = time.time()

print t1-t0
print t2-t1

问题是打印时间相等,所以multiprocessing没有加快过程.为什么？

我有这个非常简单的python代码,我希望通过并行化来加速它.然而,无论我做什么,multiprocessing.Pool.map都不会在标准地图上获得任何东西.

我已经读过其他线程,其中人们使用这个非常小的函数,这些函数不能很好地并行化并导致过多的开销,但我认为这不应该是这种情况.

难道我做错了什么？

这是一个例子

#!/usr/bin/python

import numpy, time

def AddNoise(sample):
    #time.sleep(0.001)
    return sample + numpy.random.randint(0,9,sample.shape)
    #return sample + numpy.ones(sample.shape)

n=100
m=10000
start = time.time()
A = list([ numpy.random.randint(0,9,(n,n)) for i in range(m) ])
print("creating %d numpy arrays of %d x %d took %.2f seconds"%(m,n,n,time.time()-start))

for i in range(3):
    start = time.time()
    A = list(map(AddNoise, A))
    print("adding numpy arrays took %.2f seconds"%(time.time()-start))

for i in range(3):
    import multiprocessing
    start = time.time()
    with multiprocessing.Pool(processes=2) as pool:
        A = list(pool.map(AddNoise, A, …

假设我在多核 CPU（比如 4）上运行多个 python 进程（不是线程）。GIL 是流程级别，因此特定流程中的 GIL 不会影响其他流程。

我的问题是，一个进程中的 GIL 是只占用 4 个内核中的一个内核，还是会占用所有 4 个内核？

如果一个进程同时锁定所有内核，那么多处理应该不会比 Python 中的多线程更好。如果不是，如何将内核分配给各种进程？

作为观察，在我的 8 核系统中（由于超线程是 4*2），当我运行单个 CPU 绑定进程时，8 个核中有 4 个的 CPU 使用率上升。

简化这个：

在 4 核 CPU 上运行的 4 个 python 线程（在一个进程中）将比执行相同工作的单线程花费更多时间（考虑到工作完全受 CPU 限制）。4 个不同的进程做这么多的工作会减少近 4 倍的时间吗？

我在一台机器上有64个内核,运行排序总计1GB的数据.它们每个排序156,250项,不应共享任何数据结构(即总共有64个单独的数组被排序).但是,我运行的核心越多,每个核心在其自己的排序任务中的速度就越慢.

时间测量正在这样做:

void sort_ranges(std::vector<std::vector<std::vector<int> > > & range_partitions, int num_workers, std::string filename, std::string outfile)
{
  #pragma omp parallel default(none) shared(range_partitions, outfile, num_workers)
  {
    int i = omp_get_thread_num();
    std::vector<int> data_vec; //Data copied into separate data structure for each thread
    for(int x = 0; x < num_workers; x ++) {
      data_vec.reserve(data_vec.size() + (range_partitions[x][i]).size());
      data_vec.insert(data_vec.end(), range_partitions[x][i].begin(), range_partitions[x][i].end());
    }
    int n = data_vec.size();
    int * data = &data_vec[0];
    double start = omp_get_wtime();
    std::sort(data, data + n); //Measure sort function call
    double sort_done = omp_get_wtime() …

我有 4 个不同的 Python 自定义对象和一个事件队列。每个 obect 都有一个方法，允许它从共享事件队列中检索事件，如果类型是所需的，则处理它，然后将新事件放在同一个事件队列中，允许其他进程处理它。

这是一个例子。

import multiprocessing as mp

class CustomObject:

    def __init__(events_queue: mp.Queue) -> None:
        self.events_queue = event_queue

    def process_events_queue() -> None:
        event = self.events_queue.get()
        if type(event) == SpecificEventDataTypeForThisClass:
            # do something and create a new_event
            self.events_queue.put(new_event)
        else:
            self.events_queue.put(event)

    # there are other methods specific to each object

这 4 个对象有特定的任务要做，但它们都共享相同的结构。由于我需要“模拟”生产条件，我希望它们同时运行，彼此独立。

如果可能的话，这只是我想做的一个例子。

import multiprocessing as mp
import CustomObject

if __name__ == '__main__':

    events_queue = mp.Queue()

    data_provider = mp.Process(target=CustomObject, args=(events_queue,))
    portfolio = mp.Process(target=CustomObject, args=(events_queue,))
    engine = mp.Process(target=CustomObject, …

I'm Currently building an application which relies heavily on running multithreaded. I'm Currently spawning as many threads as the system has cores using the environment.ProcessorCount constant, but when I run the application on our server which has 2 separate CPUs (Intel Xeon E5-2620) all threads work on the first CPU while the second CPU idles. Is it possible to utilize the second CPU in one process or do I need to split the process and force the utilisation with Processor …

我有这个功能（在一个带有信号量的长程序中，如有必要，我会附上任何代码）：

void signalHandler(int signumber){
    /* for SIGINT */
    if(signumber == SIGINT){    
        printf(" Pressed - Stop the timer & start cleaning process...\n");      
        flag = 0; //relevant for the rest of the program
    }
}

当我运行程序并按下CTRL+C停止它时，而不是获得以下输出：

^C Pressed - Stop the timer & start cleaning process...

*****Timer finished / stopped - clean the queue*****
....

我用双打印得到这个输出：

^C Pressed - Stop the timer & start cleaning process...
 Pressed - Stop the timer & start cleaning process...

*****Timer finished / stopped - …