标签: parallel-processing

OpenMP 文档中的 forschedule子句表示，当schedule(auto)指定时，有关调度的决策将委托给编译器或运行时系统。

编译器（例如，gcc）如何决定调度？它是从其中之一中选择static, dynamic, guided还是有自己的算法来选择时间表？

我正在尝试使用 C# 应用程序尽可能快地处理数字。我使用 aThread.Sleep()来模拟处理和随机数。我使用 3 种不同的技术。

这是我使用的测试代码：

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Test
{
    internal class Program
    {
        private static void Main()
        {
            var data = new int[500000];
            var random = new Random();

            for (int i = 0; i < 500000; i++)
            {
                data[i] = random.Next();
            }

            var partialTimes = new Dictionary<int, double>();
            var iterations = 5;

            for (int i = 1; i < iterations + 1; i++) …

我需要从旧数据库导入客户相关数据，并在此过程中执行多次转换。这意味着单个条目需要执行额外的“事件”（同步产品、创建发票等）。

我最初的解决方案是一种简单的并行方法。它工作正常，但有时会出现问题。如果当前处理的客户需要等待相同类型的事件，他们的处理队列可能会被卡住并最终超时，导致每个底层事件也失败（它们依赖于失败的事件）。这种情况并不总是发生，但还是很烦人。

于是我有了另一个想法，分批工作。我的意思是不仅限制同时处理的客户数量，还限制广播到队列的事件数量。在四处寻找想法时，我找到了这个答案，它指向TPL DataFlow。

我做了一个骨架来熟悉它。Complete()我设置了一个简单的管道，但我对和 waiting的用法有点困惑Completion()。

步骤如下

1. 制作一个数字列表（要导入的客户的 ID） - 这是导入逻辑之外的，它只是为了能够触发其余的逻辑
2. 创建一个BatchBlock（能够限制同时处理的客户数量）
3. MyClass1根据 id ( TransformBlock<int, MyClass1>)创建单个项目
4. MyClass2执行一些逻辑并生成( )的集合TransformManyBlock<MyClass1, MyClass2>- 例如，睡眠 1 秒
5. 对集合中的每个项目执行一些逻辑 ( ActionBlock<MyClass2>) - 例如，休眠 1 秒

这是完整的代码：

public static class Program
{
    private static void Main(string[] args)
    {
        var batchBlock = new BatchBlock<int>(2);
        for (var i = 1; i < 10; i++)
        {
            batchBlock.Post(i);
        }


        batchBlock.Complete();
        while (batchBlock.TryReceive(null, …

在 Java 中使用并行流而不是执行器服务被认为是不好的做法吗？为什么？

如您所知，将在幕后myList.parallelStream().map(e -> ...)使用。ForkJoinPool.common()因此，如果您同时使用至少两个并行流，您可能会在以下情况下遇到问题：

1. map函数被阻塞。但有ForkJoinPool.ManagedBlocker可以作为救援。
2. map函数可能非常占用 CPU 资源，这将导致其他并行流陷入饥饿。有什么办法可以设置RecursiveTasks之间或ForkJoinPools之间的优先级吗？

另一方面，您可以ForkJoinPool根据需要创建任意数量的 s。new ForkJoinPool(4).submit(() -> myList.parallelStream()...。在一个 JVM 上使用多个ForkJoinPools 是否被认为是性能明智的？

更新

使用或不使用并行流=使用或不使用ForkJoinPool，对吧？我发现这个和这个链接对于回答最后一个问题非常有用

从这个问题我了解到：

当您使用多处理打开第二个进程时，会创建一个全新的 Python 实例，具有自己的全局状态。该全局状态不共享，因此子进程对全局变量所做的更改对于父进程来说是不可见的。

为了验证这种行为，我制作了一个测试脚本：

import time
import multiprocessing as mp
from multiprocessing import Pool
x = [0]  # global
def worker(c):
    if c == 1:  # wait for proc 2 to finish; is global x overwritten by now?
        time.sleep(2)
    print('enter: x =', x, 'with id', id(x), 'in proc', mp.current_process())
    x[0] = c
    print('exit: x =', x, 'with id', id(x), 'in proc', mp.current_process())
    return x[0]

pool = Pool(processes=2)
x_vals = pool.map(worker, [1, 2])
print('parent: x =', x, 'with …

截至目前，我使用时间并行来并行运行脚本。示例...首先，我将转到脚本所在的目录。

cd $DIR

然后，执行脚本

time parallel ::: $script1 $script2 $script3

这效果很好。

但是如果脚本位于不同的目录中怎么办？

创建和销毁 CUDA 流的操作有多轻量？例如，对于 CPU 线程来说，这些操作很繁重，因此它们通常会池化 CPU 线程。我也应该池化 CUDA 流吗？或者每次需要时创建一个流然后销毁它是否很快？

我必须从已知的图连通分量并行计算中找到最佳算法。

以下是我的数据和计算机架构的简要概述：

• 我可以访问具有数千个处理器的计算集群（内存不共享，但我希望单个节点中应该有足够的内存来评估我对整个数据的需求）。
• 我的图的边数与顶点数之比相当小（大约 5）
• 我预计大多数连接组件都非常小（2-3 个顶点）
• 然而，将会存在具有数百万个顶点的非常大的组件（甚至占总顶点数的 10%）。

我读过有关计算图连通分量的并行算法。正如我所注意到的，其中一些基于序列化案例的经典 BFS 方法。老实说，我对这些算法的数量有点迷失了。谁能给我一些建议，哪种算法最适合我的目的？

Python v3.5、Windows 10

我正在使用多个进程并尝试捕获用户输入。input()搜索我看到的所有内容，在使用多个进程时会发生奇怪的事情。经过 8 个小时以上的尝试，我实施的任何措施都不起作用，我确信我做错了，但我一生都无法弄清楚。

以下是演示该问题的非常精简的程序。现在，当我在 PyCharm 中运行该程序时，它可以正常工作，但是当我用来pyinstaller创建单个可执行文件时，它会失败。该程序不断陷入循环，要求用户输入如下所示的内容：。

我非常确定这与 Windows 如何从我读过的内容中获取标准输入有关。我还尝试将用户输入变量作为Queue()项目传递给函数，但存在同样的问题。我读到你应该放入input()主要的 python 进程，所以我在下面这样做了if __name__ = '__main__':

from multiprocessing import Process
import time


def func_1(duration_1):
    while duration_1 >= 0:
        time.sleep(1)
        print('Duration_1: %d %s' % (duration_1, 's'))
        duration_1 -= 1


def func_2(duration_2):
    while duration_2 >= 0:
        time.sleep(1)
        print('Duration_2: %d %s' % (duration_2, 's'))
        duration_2 -= 1


if __name__ == '__main__':

    # func_1 user input
    while True:
        duration_1 = input('Enter a positive …

我想知道是否有人可以帮助使用 ScatterZipOutputStream 实现并行 Zip 创建。我已经搜索了很多，但没有找到相同的示例。

https://commons.apache.org/proper/commons-compress/zip.html

我尝试使用 ZipArchiveOutputStream 制作 Zip、压缩目录等。现在，我正在尝试同时做到这一点。

public static void makeZip(String filename) throws IOException,
        ArchiveException {
    File sourceFile = new File(filename);

    final OutputStream out = new FileOutputStream(filename.substring(0, filename.lastIndexOf('.')) + ".zip");
    ZipArchiveOutputStream os = new ZipArchiveOutputStream(out);
    os.setUseZip64(Zip64Mode.AsNeeded);

    os.putArchiveEntry(new ZipArchiveEntry(sourceFile.getName()));
    IOUtils.copy(new FileInputStream(sourceFile), os);
    os.closeArchiveEntry();
    os.close();
}

它应该能够作为线程处理单个文件，然后将其组合起来写入结果 zip。