标签: parallel-processing

我开始学习multiprocessinginpython并且我注意到在主进程上执行相同的代码比在使用multiprocessing模块创建的进程中执行得快得多。

这是我的代码的简化示例，其中我首先执行代码main process并打印前 10 个计算的时间和总计算的时间。然后执行相同的代码new process（这是一个长时间运行的进程，我可以随时发送new_pattern）。

import multiprocessing
import random
import time


old_patterns = [[random.uniform(-1, 1) for _ in range(0, 10)] for _ in range(0, 2000)]
new_patterns = [[random.uniform(-1, 1) for _ in range(0, 10)] for _ in range(0, 100)]


new_pattern_for_processing = multiprocessing.Array('d', 10)
there_is_new_pattern = multiprocessing.Value('i', 0)
queue = multiprocessing.Queue()


def iterate_and_add(old_patterns, new_pattern):
    for each_pattern in old_patterns:
        sum = 0
        for count in range(0, 10):
            sum += …

我以前没有使用过分布式计算，但我正在尝试将 mpi4py 集成到程序中，以便在计算集群上并行化 for 循环。

这是我想要做的伪代码：

for file in directory: Initialize a class Run class methods Conglomerate results

我已经查看了堆栈溢出的所有内容，但找不到任何解决方案。有没有什么办法可以简单地使用mpi4py来做到这一点，或者有没有其他工具可以通过简单的安装和设置来做到这一点？

我试图了解如何在内存 sqlite3 中并行运行 django 测试。

我有具有该结构的 django 应用程序：

gbook
    order
        ...
        tests
            __init__.py
            test_a1.py
            test_b1.py
            utils.py

test_a1.py 和 test_b1.py 包含相同的代码：

import time
from order import models
from .utils import BackendTestCase


class ATestCase(BackendTestCase):
    def test_a(self):
        time.sleep(1)
        a = models.City.objects.count()
        self.assertEqual(a, a)

class BTestCase(BackendTestCase):
    def test_b(self):
        time.sleep(1)
        a = models.City.objects.count()
        self.assertEqual(a, a)

utils.py 是：

from django.test import TestCase, Client
from order import models
from django.conf import settings

from order.utils import to_hash


class BackendTestCase(TestCase):
    fixtures = ['City.json', 'Agency.json']

    def setUp(self):
        self.client = Client() …

我目前正在尝试实施Tensorflow 管道。事实上，我想用我的CPU加载数据，并使用我的GPU来运行图在同一时间。为了更好地理解正在发生的事情，我创建了一个非常简单的卷积网络：

import os
import h5py
import tensorflow as tf
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
sess= tf.InteractiveSession()
from tensorflow.python.client import timeline
import time



t1 = time.time()

class generator:
    def __init__(self, file):
        self.file = file

    def __call__(self):
        with h5py.File(self.file, 'r') as hf:
            for im in hf["data"]:
                yield tuple(im)


dataset = tf.data.Dataset().from_generator(generator('file.h5'),
                                           output_types= tf.float32,
                                           output_shapes=(tf.TensorShape([None,4,128,128,3])))



dataset = dataset.batch(batch_size=1000)
dataset = dataset.prefetch(10)


iter = dataset.make_initializable_iterator()
e1 = iter.get_next()
e1 = tf.reshape(e1, (-1, 128, 128, 3))

with tf.device('gpu'):
    output = tf.layers.conv2d(e1[:150],200,(5,5))
    output …

我使用 joblib 来并行化一个函数（使用多处理）。但是，这个函数返回 4 个值，但是当我从 Parallel 得到结果时，它只给了我 3 个值

from joblib import Parallel, delayed 
import numpy as np
from array import array
import time

def best_power_strategy():
    powerLoc = {0}
    speedLoc = {1}
    timeLoc = {2}
    previousSpeedLoc = {3}        
    return powerLoc,speedLoc,timeLoc,previousSpeedLoc

if __name__ == "__main__":
    realRiderName=['Rider 1', 'Rider 2', 'Rider 3']
    powerLoc = {}
    speedLoc = {}
    timeLoc = {}
    previousSpeedLoc = {}
    powerLoc,speedLoc,timeLoc,previousSpeedLoc = Parallel(n_jobs=3)(delayed(best_power_strategy)() for rider in realRiderName)
    print(powerLoc)
    print(speedLoc)
    print(timeLoc)
    print(previousSpeedLoc)

结果是：

ValueError: not enough values to unpack …

使用从命令行Rscript调用时，我间歇性地收到以下错误mclapply：

Error in sendMaster(try(lapply(X = S, FUN = FUN, ...), silent = TRUE)) : 
  write error, closing pipe to the master

如果我在 R Studio 或交互式 R 会话中运行完全相同的代码，则不会出现错误。这个错误会在非常大的作业的各种上下文中弹出，每个工作人员必须将非常大的对象返回给小作业。我也试过关闭prescheduling，但它仍然抛出错误。有时，如果我减少mc.cores参数中的线程数，它就会消失。我在 Ubuntu 18.04.1 上使用 Microsoft R Open。它也出现在 Ubuntu 16.04 上。我没有尝试过的一件事是在标准 R 而不是 MRO 中运行代码。

这是我的Rscript -e 'sessionInfo()'：

R version 3.5.1 (2018-07-02)
Platform: x86_64-pc-linux-gnu (64-bit)
Running under: Ubuntu 18.04.1 LTS

Matrix products: default
BLAS: /opt/microsoft/ropen/3.5.1/lib64/R/lib/libRblas.so
LAPACK: /opt/microsoft/ropen/3.5.1/lib64/R/lib/libRlapack.so

locale:
 [1] LC_CTYPE=en_US.UTF-8       LC_NUMERIC=C
 [3] LC_TIME=en_US.UTF-8        LC_COLLATE=en_US.UTF-8
 [5] …