标签: parallel-processing

我正在测试该类cv::ParallelLoopBody的图像处理代码。

我首先开始实现归一化，在那里我必须为每个通道划分具有特定值的所有像素，这是一个简单的并行代码。

但是，在测试它时，我没有看到任何区别。

我在这里做错了吗？

这是我的课：

class Parallel_process : public cv::ParallelLoopBody
    {

    private:
        cv::Mat img; //my image to normalize
        std::vector<int> A;
        int diff;

    public:
        Parallel_process(cv::Mat inputImage, std::vector<int> AA, int diffVal)
                           : img(inputImage), A(AA), diff(diffVal){}

        virtual void operator()(const cv::Range& range) const
        {
            for(int i = range.start; i < range.end; i++)
            {
              //in is a patch of my original image
               cv::Mat in(img, cv::Rect(0, (img.rows/diff)*i, img.cols, img.rows/diff));
               std::vector<int> AAA (A);
               in.forEach<cv::Vec3f>
                (
                  [&AAA](cv::Vec3f &pixel, const int* po) -> void
                  {
                    pixel[0]/=AAA[0];
                    pixel[1]/=AAA[1]; …

我正在尝试使用 最小化成本函数scipy.optimize.minimize，但速度非常慢。我的函数有接近 5000 个变量，所以 scipy 很慢也就不足为奇了。但是，如果有它的并行版本scipy.optimize.minimize可能会有很大帮助。

我想知道这样的版本是否scipy.optimize.minimize存在，或者是否有任何其他 scipy/numpy 工具可用于执行这种量级的最小化。我真的很感激任何和所有的帮助。

谢谢大家的意见。这是使用 SLSQP 求解器的约束最小化。我已经花了很多时间来确保成本函数计算得到优化，所以问题一定是在计算梯度或由于约束。换句话说，花在函数评估上的时间只占最小化总时间的一小部分。

我有两个函数func1()，func2()它们相互独立，可以在两个线程中运行。我正在使用threading库python3来运行这两个线程。不过里面func1()，我也运行for使用并行循环joblib。因此，使用线程以及 joblib 会给出以下警告 -

/usr/local/lib/python3.6/dist-packages/joblib/parallel.py:547: UserWarning: Multiprocessing-backed parallel loops cannot be nested below threads, setting n_jobs=1,

然后里面的 for 循环func1()只是按顺序运行。

以下是代码片段的示例：

from joblib import Parallel, delayed
import threading
from math import sqrt

class MyThread(threading.Thread):
    def __init__(self, sample, type):
        threading.Thread.__init__(self)
        self.type = type
        self.sample = sample
    def run(self):
        if self.type=='func1':
            self.sample.func1()
        else:
            self.sample.func2()

class Sample:
   def func1(self):
       print('this function runs for …

我的应用程序中有 3 种不同的方法。都回来了CompletableFuture<SomeType>。我想并行执行方法 1 和方法 2。完成方法 1 和方法 2 后，我想使用方法 1 和方法 2 返回值的参数触发方法 3。

代码示例：

CompletableFuture<Request> future1 = RequestConverter.Convert(requestDto);
CompletableFuture<String> future2 = tokenProvider.getAuthToken();

CompletableFuture<CompletableFuture<String>> future3 = 
future1.thenCombine(future2,(request,token) -> 
requestProcessor.Process(request,token)); 

但是，与上面的代码的问题是，我得到一个CompletableFuture的CompletableFuture。我想避免这种情况并获得简单CompletableFuture<String>而不阻塞。这可能吗？

我几天来一直在寻找这个问题的答案，但没有结果。我可能只是不理解那些漂浮在外面的部分，并且该multiprocessing模块的 Python 文档相当大，对我来说不清楚。

假设您有以下 for 循环：

import timeit


numbers = []

start = timeit.default_timer()

for num in range(100000000):
    numbers.append(num)

end = timeit.default_timer()

print('TIME: {} seconds'.format(end - start))
print('SUM:', sum(numbers))

输出：

TIME: 23.965870224497916 seconds
SUM: 4999999950000000

对于此示例，假设您有一个 4 核处理器。有没有办法总共创建 4 个进程，其中每个进程都在单独的 CPU 核心上运行，并且完成速度大约快 4 倍，因此 24 秒/4 个进程 = 约 6 秒？

以某种方式将 for 循环分成 4 个相等的块，然后将这 4 个块添加到数字列表中以等于相同的总和？有一个 stackoverflow 线程：Parallel Simple For Loop但我不明白。谢谢大家。

考虑以下张量流代码片段：

import time
import numpy as np
import tensorflow as tf

def fn(i):
    # do some junk work
    for _ in range(100):
        i ** 2
    return i

n = 1000
n_jobs = 8
stuff = np.arange(1, n + 1)
eager = False
t0 = time.time()
if eager:
    tf.enable_eager_execution()
res = tf.map_fn(fn, stuff, parallel_iterations=n_jobs)
if not eager:
    with tf.Session() as sess:
        res = sess.run(res)
        print(sum(res))
else:
    print(sum(res))
dt = time.time() - t0
print("(eager=%s) Took %ims" % (eager, dt * 1000))

如果使用 …

我想并行化一个 for 循环，其中更新 unordered_map 的值：

unordered_map<string,double> umap {{"foo", 0}, {"bar", 0}};

#pragma omp parallel for reduction(my_reduction:umap)
for (int i = 0; i < 100; ++i)
{
    // some_string(i) would return either "foo" or "bar"
    umap[some_string(i)] += some_double(i);
}

因此，unordered_map 中不会创建新条目，只会更新现有条目的总和。

在这个答案中，用户声明的归约是针对向量的情况定义的。在 unordered_map 的情况下，用户声明的归约是否可以类似地定义？

我正在读取一个视频文件，每 20 帧我将第一帧存储在输入队列中。一旦我在输入队列中获得了所有必需的帧，我就会运行多个进程来对这些帧执行一些操作并将结果存储在输出队列中。但代码总是卡在 join 处，我尝试了针对此类问题提出的不同解决方案，但似乎都不起作用。

import numpy as np
import cv2
import timeit
import face_recognition
from multiprocessing import Process, Queue, Pool
import multiprocessing
import os

s = timeit.default_timer()

def alternative_process_target_func(input_queue, output_queue):

    while not output_queue.full():
        frame_no, small_frame, face_loc = input_queue.get()
        print('Frame_no: ', frame_no, 'Process ID: ', os.getpid(), '----', multiprocessing.current_process())
        #canny_frame(frame_no, small_frame, face_loc)

        #I am just storing frame no for now but will perform something else later
        output_queue.put((frame_no, frame_no)) 

        if output_queue.full():
            print('Its Full ---------------------------------------------------------------------------------------')
        else:
            print('Not Full')

    print(timeit.default_timer() - s, ' seconds.') …

我的场景是这样的。

1. 我有一个由大量报告组成的向量，需要使用 REST API 调用发送。
2. 我正在使用 Futures.traverse（1中提到的向量）
3. 由于向量太大，因此会因超出最大打开请求数而失败。

我能想到的一个初步解决方案是增加 max-open-requests 设置。但这里的问题是我不知道需要提前发送多少报告。

有人可以建议一个替代解决方案，例如限制通过 Futures.traverse 发生的并行性

据我所知，水晶会通过 io 循环光纤，这意味着如果一根光纤正在等待 io，水晶将切换到另一根光纤。

如果我们生成两个纤程，但其中一个在没有 io 的情况下进行持续计算/循环，该怎么办？

例如，使用下面的代码，服务器不会响应任何 http 请求

spawn do
  Kemal.run
end

spawn do
  # constant computation/loop with no IO
  some_func
end

Fiber.yield
# or sleep