标签: parallel-processing

从optim调用包含foreach%dopar%构造的函数会导致错误:

> workers <- startWorkers(6) # 6 cores
> 
> registerDoSMP(workers)
> 
> t0 <- Sys.time() 
>
> optim(w,maxProb2,control=list(fnscale=-1))
> 
> Error in { : task 1 failed - "unused argument(s) (isPrebuilt = TRUE)"
> 
> Sys.time()-t0
>
> Time difference of 2.032 secs
> 
> stopWorkers(workers)

被调用的函数看起来像这样:

> maxProb2 <- function(wp) {
>   
>   r <- foreach (i=s0:s1, .combine=c) %dopar% { pf(i,x[i,5],wp,isPrebuilt=TRUE) }
>   
>   cat("w=",wp,"max=",sum(r),"\n")
>   
>   sum(r)
>   
> }

pf是一些其他函数,x是预先计算元素的静态表.

同样调用要优化的函数只会导致同样的错误:

> workers <- startWorkers(6) # …

我正在寻找一个python库或命令行工具,用于并行下载多个文件.我目前的解决方案是按顺序下载文件,这很慢.我知道你可以很容易地在python中编写一个半成熟的线程解决方案,但是在使用线程时我总是遇到烦人的问题.它用于从网站轮询大量xml提要.

我对解决方案的要求是:

1. 应该是可以打断的.Ctrl + C应立即终止所有下载.
2. 即使主程序崩溃或抛出异常,也不应该使用kill手动杀死剩余的进程.
3. 它也适用于Linux和Windows.
4. 它应该重试下载,对网络错误具有弹性并且应该正常超时.
5. 应该明智的做法是不要同时下载100多个同一台服务器,但要以理智的方式对它们进行排队.
6. 它应该处理重要的http状态代码,如301,302和304.这意味着对于每个文件,它应该将Last-Modified值作为输入,并且只有在自上次更改后才下载.
7. 优选地,它应该具有进度条,或者应该容易为其编写进度条以监视所有文件的下载进度.
8. 优选地,它应该利用http keep-alive来最大化传输速度.

请不要建议我如何实施上述要求.我正在寻找一个现成的,经过实战考验的解决方案.

我想我应该描述我想要的东西......我有大约300种不同的数据源,因为xml格式的文件来自50个数据提供者.每个文件的大小在100kb到5mb之间.我需要经常轮询它们(如每隔几分钟一次)以确定它们是否有我需要处理的新数据.因此,下载程序使用http缓存来最小化要获取的数据量非常重要.它显然也使用gzip压缩.

那么最大的问题是如何在不超越任何边界的情况下以尽可能高效的方式使用带宽.例如,如果您打开20个同时连接到其数据源的连接,则一个数据提供商可能会认为它被滥用.相反,最好使用一个或两个重用于多个文件的连接.或者您自己的连接可能会以奇怪的方式受到限制.我的isp限制了您可以执行的dns查找次数,因此某种dns缓存会很好.

我想使用OpenMP以并行方式遍历std :: list中的所有元素.循环应该能够改变列表的元素.有一个简单的解决方案吗？当迭代器是随机访问迭代器时,似乎OpenMP 3.0支持并行for循环,但不是其他.无论如何,我更喜欢使用OpenMP 2.0,因为我无法完全控制哪些编译器可供我使用.

如果我的容器是矢量,我可能会使用:

#pragma omp parallel for
for (auto it = v.begin(); it != v.end(); ++it) {
    it->process();
}

我知道我可以将列表复制到矢量中,执行循环,然后将所有内容复制回来.但是,如果可能的话,我想避免这种复杂性和开销.

我在为我的应用程序决定使用python多处理或celery或pp时遇到了一些麻烦.

我的应用程序非常重CPU,但目前只使用一个cpu,所以我需要将它扩展到所有可用的cpus(这使我看到python的多处理库)但我读到如果需要,这个库不会扩展到其他机器.现在我不确定我是否需要多个服务器才能运行我的代码,但我想在本地运行celery然后扩展只需要添加新的服务器而不是重构代码(就像我使用的那样)多).

我的问题:这个逻辑是否正确？在本地使用芹菜是否存在任何负面影响(如果事实证明,具有多个核心的单个服务器可以完成我的任务)？或者是否更多建议使用多处理并在以后将其扩展到其他内容？

谢谢!

ps这是一个个人学习项目,但我可能有一天会喜欢在一家公司担任开发人员,并希望了解专业人士如何做到这一点.

我开始学习CUDA,我认为计算pi的长数字将是一个很好的介绍性项目.

我已经实现了简单的蒙特卡罗方法,该方法很容易并行化.我只是让每个线程在单位正方形上随机生成点,计算单位圆内有多少点,并使用缩小操作计算结果.

但这当然不是计算常数的最快算法.以前,当我在单线程CPU上进行此练习时,我使用类似Machin的公式来进行计算,以便更快地收敛.对于那些感兴趣的人,这涉及将pi表示为反复数组的总和并使用泰勒级数来评估表达式.

这样一个公式的一个例子:

不幸的是,我发现将这种技术并行化到数千个GPU线程并不容易.问题是大多数操作只是在进行高精度数学运算,而不是对长数据向量进行浮点运算.

所以我想知道,在GPU上计算pi的任意长数字的最有效方法是什么？

如何使multiprocessing.pool.map按数字顺序分配进程？

更多信息:
我有一个程序可以处理几千个数据文件,并绘制每个文件的图.我正在使用a multiprocessing.pool.map将每个文件分发到处理器,它运行良好.有时这需要很长时间,在程序运行时查看输出图像会很好.如果地图进程按顺序分发快照,这将会容易得多; 相反,对于我刚刚执行的特定运行,分析的前8个快照是: 0, 78, 156, 234, 312, 390, 468, 546.有没有办法让它按数字顺序更紧密地分配它们？

示例:
这是一个包含相同键元素的示例代码,并显示相同的基本结果:

import sys
from multiprocessing import Pool
import time

num_proc  = 4; num_calls = 20; sleeper   = 0.1

def SomeFunc(arg):
    time.sleep(sleeper)
    print "%5d" % (arg),
    sys.stdout.flush()     # otherwise doesn't print properly on single line

proc_pool = Pool(num_proc)
proc_pool.map( SomeFunc, range(num_calls) )

产量:

   0  4  2  6   1   5   3   7   8  10  12  14  13  11   9  15  16  18  17  19

回答:

来自@Hayden:使用'chunksize'参数, …

我计划在我的公司内部进行Java 8中的新功能和概念的内部演示.

我想关注的是新集合库的并行处理功能.

无论我在哪里阅读Java 8以及对集合库的更多功能样式迭代器的需求,都会提到这将有助于利用当前正常的多核服务器.但很少有人提到这是如何实现的,以及这是否是一个普遍的事实,更不用说任何有关性能的基准.

正如我公司中经验丰富的开发人员声称了解线程并不知道实际线程如何在较低级别工作,我正在尝试收集这方面的一些知识.基于阅读几篇博客等,我做了一系列以下断言.

我会感谢以下几点的反馈(真/假) ..

1. 线程是操作系统中最低的调度单位(是基本的东西,但不是所有应用程序员都知道这个;-))

2. 单线程程序一次只能在一个核心上运行.因此,在四核CPU中,例如不使用75%的CPU.

3. 现有Java集合迭代器的问题在于它是一个外部迭代器,并且不可能(至少开箱即用)将庞大的集合迭代分发给多个线程.新的集合库操作使得可以在不需要处理低级并发问题的情况下实现并发

4. Java 8使用增强的集合库可以使用内部迭代器并行化迭代

   而不是Java 7

   for (Shape s : shapes) {if (s.getColor() == RED)s.setColor(BLUE); }

   我们有Java 8

   shapes.forEach(s -> { if (s.getColor() == RED) s.setColor(BLUE); })

5. 但为了平行上述迭代,必须明确使用parallel()方法Stream API

   private static void printUsingCoolLambda (final List<String> names) { names.parallelStream().forEach(s -> System.out.println(s)); System.out.println("Printed using printUsingCoolLambda"); }

   但即便如此,也不能保证操作将并行完成,因为Javadoc parallelStream()说下面的"返回一个可能并行的{@code Stream},并将此集合作为其源.此方法允许返回顺序流"

6. 最终,无法保证所有核心都将被利用,因为线程调度不是JVM的责任,而是由OS决定.

编辑

我最难得到第5点和第6点.正如各种Java 8博客所说的那样"使用这个新的parallelStream()并且您将获得开箱即用的并行处理(免费,并且您作为应用程序员免于担心这一点)",我的问题用一句话本来应该是真的正确吗？

有没有一种简单的方法来跟踪joblib.Parallel执行的整体进度？

我有一个由数千个作业组成的长期执行,我想跟踪并记录在数据库中.但是,要做到这一点,每当Parallel完成任务时,我都需要它来执行回调,报告剩余的剩余作业数.

我之前使用Python的stdlib multiprocessing.Pool完成了类似的任务,通过启动一个记录Pool的作业列表中待处理作业数量的线程.

看看代码,Parallel继承了Pool,所以我认为我可以使用相同的技巧,但它似乎没有使用这些列表,我也无法弄清楚如何"读取"它的内部任何其他方式的状态.

我有要并行处理的元素集合.当我使用a时List,并行性有效.但是,当我使用a时Set,它并不是并行运行的.

我写了一个显示问题的代码示例:

public static void main(String[] args) {
    ParallelTest test = new ParallelTest();

    List<Integer> list = Arrays.asList(1,2);
    Set<Integer> set = new HashSet<>(list);

    ForkJoinPool forkJoinPool = new ForkJoinPool(4);

    System.out.println("set print");
    try {
        forkJoinPool.submit(() ->
            set.parallelStream().forEach(test::print)
        ).get();
    } catch (Exception e) {
        return;
    }

    System.out.println("\n\nlist print");
    try {
        forkJoinPool.submit(() ->
            list.parallelStream().forEach(test::print)
        ).get();
    } catch (Exception e) {
        return;
    }   
}

private void print(int i){
    System.out.println("start: " + i);
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
    }
    System.out.println("end: …

我遇到了以下问题:

我有一个名为TrainModel的函数,它在一个线程上运行很长时间.当它完成计算时,它返回一个函数作为输出参数,让我们称之为f.当我询问这个f的类型时,朱莉娅回归:

(1种方法的通用功能)

(我不确定这最后一条信息对于阅读此内容的人有用)

现在,在第二步中,我需要在非常大的值数组上应用函数f.这是我想要并行化的一个步骤.已经启动了具有多个流程的Julia,例如

julia -p 4

理想情况下,我会使用:

pmap(f, my_values)

也许:

aux = @parallel (hcat) for ii=1:100000000
        f(my_values[ii])
      end

不幸的是,这不起作用.朱莉娅抱怨工人不知道功能f,即我得到一个消息:

错误:过程2中未定义函数f

我怎样才能使功能˚F提供给所有工人呢？显然,一个"脏"的解决方案是对所有工人运行耗时的函数TrainModel,这可能是:

@everywhere f = TrainModel( ... )

但是当我想要的只是结果f可供所有工作人员使用时,这将浪费cpu .

虽然我搜索了类似问题的帖子,但到目前为止我找不到答案......

提前致谢!最好,

N.