标签: parallel-processing

最近在R2.14中增加了对并行计算的直接支持,这引发了一个问题.在R中创建集群有很多选项.我snow定期使用SOCK集群,但我知道还有其他方法,如MPI.我使用SOCK snow集群因为我不需要安装任何其他软件(我使用Fedora 13).

那么,我的具体问题:

1. 使用非SOCK群集时性能是否有所提升？
2. 使用非SOCK群集在多台计算机上创建群集更容易吗？

我在Erlang上花了一些时间,我想将TDD应用到我正在编写的代码中.

虽然标准库中的EUnit为测试常规样式代码提供了一个很好的传统单元测试框架,但似乎没有什么可以帮助测试并发代码,这在Erlang中使用了很多.

请注意,我们在这里讨论的是Erlang,它使用消息传递(而不是共享状态)来进行并发进程之间的通信,因此使用共享状态语言对并发代码进行单元测试的技术可能不适用.

有人找到了在Erlang中测试并发代码的好方法吗？

我的意思是:当您自由设计应用程序副作用等时,F#代码会自动分布在所有内核中吗？

我有4个线程,我试图设置线程1在CPU 1上运行,线程2在CPU 2上运行等等.但是,当我在下面运行我的代码时,亲和力掩码返回正确的值,但是当我这样做时sched_getcpu()在线程上,它们都返回它们在CPU 4上运行.

有人知道我的问题是什么吗？

提前致谢!

#define _GNU_SOURCE
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <sched.h>
#include <errno.h>

void *pthread_Message(char *message)
{
    printf("%s is running on CPU %d\n", message, sched_getcpu());
}

int main()
{
    pthread_t thread1, thread2, thread3, thread4;
    pthread_t threadArray[4];
    cpu_set_t cpu1, cpu2, cpu3, cpu4;
    char *thread1Msg = "Thread 1";
    char *thread2Msg = "Thread 2";
    char *thread3Msg = "Thread 3";
    char *thread4Msg = "Thread 4";
    int thread1Create, thread2Create, thread3Create, thread4Create, i, temp;

    CPU_ZERO(&cpu1);
    CPU_SET(1, &cpu1);
    temp = pthread_setaffinity_np(thread1, sizeof(cpu_set_t), …

并行流可能会产生与Java 8中的串行流不同的结果吗？根据我的信息,并行流与串行流相同,除了分成多个子流.这是一个速度问题.完成对元素的所有操作,并在最后组合子流的结果.最后,在我看来,对于并行和串行流,操作的结果应该是相同的.所以我的问题是,这段代码可能会给我一个不同的结果吗？如果有可能,为什么会发生？

int[] i = {1, 2, 5, 10, 9, 7, 25, 24, 26, 34, 21, 23, 23, 25, 27, 852, 654, 25, 58};
Double serial = Arrays.stream(i).filter(si -> {
    return si > 5;
}).mapToDouble(Double::new).map(NewClass::add).reduce(Math::atan2).getAsDouble();

Double parallel = Arrays.stream(i).filter(si -> {
    return si > 5;
}).parallel().mapToDouble(Double::new).map(NewClass::add).reduce(Math::atan2).getAsDouble();

System.out.println("serial: " + serial);
System.out.println("parallel: " + parallel);

public static double add(double i) {
    return i + 0.005;
}

结果是:

serial: 3.6971567726175894E-23

parallel: 0.779264049587662

通常,并不十分清楚并行流如何将输入分成块以及块连接的顺序.有没有办法可视化任何流源的整个过程,以更好地了解正在发生的事情？假设我创建了一个这样的流:

Stream<Integer> stream = IntStream.range(0, 100).boxed().parallel();

我想看到一些树状的结构:

             [0..99]
         _____/   \_____
        |               |
     [0..49]         [50..99]
    __/   \__        __/  \__
   |         |      |        |
[0..24]  [25..49] [50..74] [75..99]

这意味着整个输入范围[0..99]被拆分为范围,[0..49]而[50..99]范围又进一步分裂.当然这样的图应该反映Stream API的实际工作,所以如果我用这样的流执行一些实际操作,则应该以相同的方式执行拆分.

亚历山大·斯捷潘诺夫在A9的一篇精彩讲座(强烈推荐,顺便说一句)中指出,关联属性为我们提供了可并行性 - 这些日子是编译器,CPU和程序员自己可以利用的非常有用和重要的特性:

// expressions in parentheses can be done in parallel
// because matrix multiplication is associative
Matrix X = (A * B) * (C * D);

但是,交换性财产给我们带来了什么？重新排序？乱序执行？

以下代码并行化for循环.

import networkx as nx;
import numpy as np;
from joblib import Parallel, delayed;
import multiprocessing;

def core_func(repeat_index, G, numpy_arrary_2D):
  for u in G.nodes():
    numpy_arrary_2D[repeat_index][u] = 2;
  return;

if __name__ == "__main__":
  G = nx.erdos_renyi_graph(100000,0.99);
  nRepeat = 5000;
  numpy_array = np.zeros([nRepeat,G.number_of_nodes()]);
  Parallel(n_jobs=4)(delayed(core_func)(repeat_index, G, numpy_array) for repeat_index in range(nRepeat));
  print(np.mean(numpy_array));

可以看出,要打印的期望值是2.但是,当我在集群(多核,共享内存)上运行我的代码时,它返回0.0.

我认为问题是每个工作者都创建自己的numpy_array对象副本,并且不更新在main函数中创建的副本.如何修改代码numpy_array以便更新numpy数组？

我试图使用C++ 17标准中提出的新并行库功能,但我无法使其工作.我试着用的了最新版本的编译g++ 8.1.1和clang++-6.0和-std=c++17,但也似乎支持#include <execution>,std::execution::par或任何类似.

声称,在查看并行算法的cppreference时,有很多算法列表

技术规范提供以下69种算法的并行化版本algorithm,numeric并且memory:( ...长列表...)

听起来像算法已经准备好'在纸上',但尚未准备好使用？

在一年多前的SO问题中,答案声称这些功能尚未实现.但到现在为止,我希望看到某种实现方式.有什么我们可以使用的吗？

问题：我有大量的SQL查询（大约10k-20k），并且我想在50个（或更多）线程中异步运行它们。

我为此工作编写了一个powershell脚本，但是它非常慢（执行所有操作大约需要20个小时）。期望的结果是最多3-4小时。

问题：如何优化此Powershell脚本？我应该重新考虑和使用其他技术，如python或c#？

我认为这是powershell的问题，因为当我检查whoisactive查询时，它们执行得很快。创建，退出和卸载作业需要花费大量时间，因为为每个线程创建了单独的PS实例。

我的代码：

$NumberOfParallerThreads = 50;


$Arr_AllQueries = @('Exec [mystoredproc] @param1=1, @param2=2',
                    'Exec [mystoredproc] @param1=11, @param2=22',
                    'Exec [mystoredproc] @param1=111, @param2=222')

#Creating the batches
$counter = [pscustomobject] @{ Value = 0 };
$Batches_AllQueries = $Arr_AllQueries | Group-Object -Property { 
    [math]::Floor($counter.Value++ / $NumberOfParallerThreads) 
};

forEach ($item in $Batches_AllQueries) {
    $tmpBatch = $item.Group;

    $tmpBatch | % {

        $ScriptBlock = {
            # accept the loop variable across the job-context barrier
            param($query) 
            # Execute …