标签: parallel-processing

我需要快速遍历一棵树,我想并行完成.我宁愿使用并行扩展而不是手动旋转一堆线程.

我当前的代码看起来像这样:

   public void Traverse(Node root)
    {
        var nodeQueue = new Queue<Node>();
        nodeQueue.Enqueue(root);
        while (nodeQueue.Count!=0)
        {
            var node = nodeQueue.Dequeue();
            if (node.Property = someValue) DoSomething(node);
            foreach (var node in node.Children)
            {
                nodeQueue.Enqueue(node);
            }
        }
    }

我真的希望Parallel.ForEach有一个Parallel.While模拟.我遇到了Stephen Toub关于使用Parallel.ForEach实现Parallels Parallel的文章.如果正确读取它仍然无法工作,因为我正在改变我试图迭代的队列.

我是否需要使用任务工厂和递归(这有风险吗？)？还是有一些我忽略的简单解决方案？

编辑:@svick

该树有超过250,000个节点.现在最大深度是14个节点,包括根.

根目录下有大约500个节点,之后的平衡具有相当随机的分布.我很快就会得到更好的分布统计数据.

@Enigmativity:

是的,许多用户同时修改了树,但我通常会为树或子树提供共​​享读锁,或允许脏读.

对node.Children的调用可以被认为是原子的.

DoSomething实际上是几个代理之一,对于一些昂贵的操作,我可能会收集节点的快照列表并在遍历之外处理它们.

我意识到我应该看一般情况(遍历的子树而不是整个树.)为此,我在树的每个节点上运行遍历并查看总时间.

我为每个遍历算法使用了Parallel.ForEach(nodes,Traverse),其中节点包含所有~250k节点.这模拟(某种程度上)许多用户同时请求许多不同的节点.

00256ms广度优先顺序

00323ms广度优先连续工作(我将静态计数器增加为"工作")

01495ms Kirks第一个回答

01143ms Svicks第二个答案

00000ms Recursive Single Threaded在60s后没有完成

00000ms电子书的答案在60年代后没有完成

@Enigma,我想我可能会以某种方式搞砸你的算法,因为它似乎应该更快.

结果令我惊讶的是至少可以说.为了让自己相信编译器并没有神奇地优化遍历,我不得不在广度第一顺序中添加一些工作.

对于头部的单次遍历,并行化第一级仅具有最佳性能.但几乎没有,这个数字有所改善,因为我向第二级添加了更多节点(2000而不是500).

有没有办法fork()在Perl中实现非阻塞/异步执行(没有'ing)？

我曾经是一名Python开发人员多年...... Python拥有非常棒的'Twisted'框架允许这样做(使用DEFERREDs.当我运行搜索以查看Perl中是否有任何内容可以执行相同操作时,我遇到了POE框架 - 看起来与我正在搜索的内容"相近".但是......花了一些时间阅读文档并"玩"代码后,我反对"墙" - 这是限制性的(来自POE) ::会话文档):

回调不是先发制人的.只要一个人正在运行,就不会派遣其他人.这称为协作式多任务处理.每个会话必须通过返回中央调度内核进行协作.

这种限制基本上违背了异步/并行/非阻塞执行的目的 - 通过限制在任何给定时刻执行的只有一个回调(代码块).当另一个回调已经在运行时,没有其他回调可以开始运行!

所以......在Perl中有没有办法实现多任务(并行,非阻塞,异步执行代码)而不用fork()- 类似于Python中的DEFERREDs？

我理解使用subprocess是调用外部命令的首选方式.

但是如果我想在parall中运行几个命令,但是限制生成的进程数呢？困扰我的是我无法阻止子进程.例如,如果我打电话

subprocess.Popen(cmd, stderr=outputfile, stdout=outputfile)

然后该过程将继续,无需等待cmd完成.因此,我无法将其包装在multiprocessing图书馆的工作人员中.

例如,如果我这样做:

def worker(cmd): 
    subprocess.Popen(cmd, stderr=outputfile, stdout=outputfile);

pool = Pool( processes = 10 );
results =[pool.apply_async(worker, [cmd]) for cmd in cmd_list];
ans = [res.get() for res in results];

然后每个工人将在产生子流程后完成并返回.所以我无法真正限制subprocess使用生成的进程数Pool.

什么是限制子过程数量的正确方法？

我有一个包含我想要运行的命令行的文件.该文件包含大约2,000行.

我有8个核心可用.是否可以解析文件并启动8个进程,然后在其中一个程序完成时从文件中执行另一个进程？我希望这一直持续到文件结束.

我正在使用 ASP.NET Core 2.1 版中的托管服务功能对新的后台任务进行一些测试，更具体地说是使用排队后台任务，我想到了一个关于并行性的问题。

我目前严格遵循 Microsoft 提供的教程，并且在尝试模拟来自同一用户的多个请求以将任务排入队列时的工作负载时，我注意到所有工作项都是按顺序执行的，因此没有并行性。

我的问题是，这种行为是预期的吗？如果是这样，为了使请求执行并行，是否可以触发并忘记，而不是等待 workItem 完成？

我在没有运气的情况下搜索了几天有关此特定场景的信息，因此如果有人提供任何指南或示例，我会非常高兴。

编辑：教程中的代码很长，所以它的链接是https://docs.microsoft.com/en-us/aspnet/core/fundamentals/host/hosted-services?view=aspnetcore-2.1#queued -背景任务

执行工作项的方法是这样的：

public class QueuedHostedService : IHostedService
{
    ...

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Queued Hosted Service is starting.");

        _backgroundTask = Task.Run(BackgroundProceessing);

        return Task.CompletedTask;
    }

    private async Task BackgroundProceessing()
    {
        while (!_shutdown.IsCancellationRequested)
        {
            var workItem = 
                await TaskQueue.DequeueAsync(_shutdown.Token);

            try
            {
                await workItem(_shutdown.Token);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, 
                    $"Error occurred executing {nameof(workItem)}.");
            }
        }
    }

    ...
}

问题的重点是要知道是否有人可以分享有关如何使用此特定技术同时执行多个工作项的知识，因为服务器可以处理此工作负载。

我在执行工作项时尝试了即发即弃的方法，它按照我的预期工作，同时并行执行多个任务，我不确定这是否是一个好的做法，或者是否有处理这种情况的更好或正确的方法。

我有 4 个 GPU（0,1,2,3），我想在 GPU 2 上运行一个 Jupyter notebook，在 GPU 0 上运行另一个。因此，在执行之后，

 export CUDA_VISIBLE_DEVICES=0,1,2,3

对于我做的 GPU 2 笔记本，

device = torch.device( f'cuda:{2}' if torch.cuda.is_available() else 'cpu')
device, torch.cuda.device_count(), torch.cuda.is_available(), torch.cuda.current_device(), torch.cuda.get_device_properties(1)

在创建新模型或加载一个模型后，

model = nn.DataParallel( model, device_ids = [ 0, 1, 2, 3])
model = model.to( device)

然后，当我开始训练模型时，我得到，

RuntimeError                              Traceback (most recent call last)
<ipython-input-18-849ffcb53e16> in <module>
 46             with torch.set_grad_enabled( phase == 'train'):
 47                 # [N, Nclass, H, W]
 ---> 48                 prediction = model(X)
 49                 # print( prediction.shape, y.shape)
 50 …

我希望将一个适度昂贵的函数映射到一个大的懒惰seq并行.pmap很棒,但我对上下文切换很不满意.我想我需要增加传递给每个线程的工作块的大小.

我写了一个函数来将seq分解为块并将函数pmap到每个块上并重新组合它们.这"有效",但结果并不壮观.原始代码基本上如下所示:

(pmap eval-polynomial (range x) coificients)

我怎么能在保持懒惰的同时真正挤压它？

我怎么能告诉gnu不要并行构建一些配方.假设我有以下makefile:

sources = a.xxx b.xxx c.xxx
target  = program

all : $(target)

$(target) : $(patsubst %.xxx,%.o,$(sources))
    $(CXX) -o $@ $<

%.o : %.cpp
    $(CXX) -c -o $@ $<

%.cpp : %.xxx
    my-pre-processor -o $@ $<

但是,该my-pre-processor命令创建具有固定名称的临时文件(我无法更改此名称).如果我只使用没有-j参数的make,这工作正常.但是,如果使用该-j选项,则构建有时会失败,因为两次并发调用会my-pre-processor覆盖其临时文件.

我想知道是否有办法告诉make它必须不构建并行化%.cpp : %.xxx配方执行的尝试.

我需要知道我的应用程序通过OpenMP生成的线程总数.不幸的是,该omp_get_num_threads()功能也不会在这里,因为它只产生的线程在目前球队数量的工作.

但是,我的代码以递归方式运行(基本上是分而治之),只要仍有空闲的处理器,我想生成新的线程,但不会更多.

有没有办法绕过限制omp_get_num_threads并获得正在运行的线程总数？

如果需要更多细节,请考虑以下伪代码,它们非常接近地模拟我的工作流程:

function divide_and_conquer(Job job, int total_num_threads):
  if job.is_leaf(): # Recurrence base case.
    job.process()
    return

  left, right = job.divide()

  current_num_threads = omp_get_num_threads()
  if current_num_threads < total_num_threads: # (1)
    #pragma omp parallel num_threads(2)
      #pragma omp section
        divide_and_conquer(left, total_num_threads)
      #pragma omp section
        divide_and_conquer(right, total_num_threads)

  else:
    divide_and_conquer(left, total_num_threads)
    divide_and_conquer(right, total_num_threads)

  job = merge(left, right)

如果我使用total_num_threads值4来调用此代码,则条件注释(1)将始终求值true(因为每个线程团队最多包含两个线程),因此代码将始终生成两个新线程,无论已经运行了多少线程在更高的层次上.

我正在寻找一种独立于平台的方法来确定我的应用程序中当前运行的线程总数.

我正在玩C#,想加快一个程序.我做了改变,并且能够这样做.但是,我需要帮助理解为什么变化使它变得更快.

我试图将代码简化为更容易理解的问题.Score1和Report1是较慢的方式.Score2和Report2是更快的方式.第一种方法首先并行地在结构中存储字符串和int.接下来,在串行循环中,它循环遍历这些结构的数组并将其数据写入缓冲区.第二种方法首先将数据并行写入字符串缓冲区.接下来,在串行循环中,它将字符串数据写入缓冲区.以下是一些示例运行时间:

运行1总平均时间= 0.492087秒运行2总平均时间= 0.273619秒

当我使用早期的非并行版本时,时间几乎相同.为什么与并行版本的区别？

即使我减少Report1中的循环以将单行输出写入缓冲区,它仍然较慢(总时间约为.42秒).

这是简化的代码:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using System.Threading.Tasks;
using System.IO;

namespace OptimizationQuestion
{
    class Program
    {
        struct ValidWord
        { 
            public string word;
            public int score;
        }
        ValidWord[] valid;
        StringBuilder output;
        int total; 

        public void Score1(string[] words)
        {
            valid = new ValidWord[words.Length];

            for (int i = 0; i < words.Length; i++)
            {
                StringBuilder builder = new StringBuilder();

                foreach (char c in words[i])
                {
                    if (c != 'U')
                        builder.Append(c);
                } …