标签: parallel-processing

在我的应用程序中,我有几个Builder类,负责从外部API请求获取数据并构建/保存资源到数据库.我正在处理大量数据并实现了Parallel gem,以通过使用多个进程来加快速度.

但是,我发现使用Parallel的方法的任何测试都会失败并出现相同的错误:

 ActiveRecord::StatementInvalid:
   PG::ConnectionBad: PQconsumeInput() server closed the connection unexpectedly
    This probably means the server terminated abnormally
    before or while processing the request.

以下是正在测试的代码示例:

class AirportBuilder < Resource
  def build_from_collection
    Parallel.each(object_producer, in_processes: 24) do |params|
      instance = Airport.find_or_initialize_by(fsid: params[:fs])
      build!(instance, params)
    end
  end
end

我已经对此进行了一些搜索,但谷歌的所有结果都与使用多个线程/进程有关,使测试套件运行得更快,这是一个不同的问题.

有关如何在不引起PG错误的情况下有效测试此问题的任何想法？我意识到我可能需要将某些东西排除在外但不太确定要存根什么但仍然有一个有意义的测试.

提前感谢任何可能提供帮助的人!

我正在尝试并行化(OpenMP)一些科学的C++代码,其中大部分(> 95%)的CPU时间用于计算令人讨厌(且不可避免)的O(N ^ 2)交互,以便订购N~200个不同的粒子.该计算重复1e10个时间步长.我已尝试使用OpenMP进行各种不同的配置,每个配置比串行代码慢一些(至少数量级),并且随着附加内核的增加而缩放不良.

下面是相关代码的草图,具有代表性的虚拟数据层次结构Tree->Branch->Leaf.每个Leaf对象存储其自身的位置和速度,用于当前和之前的三个时间步骤等.每个Branch然后存储的集合Leaf对象和每个Tree存储的集合Branch对象.这种数据结构非常适用于复杂但CPU密集度较低的计算,这些计算也必须在每个时间步骤执行(需要数月才能完善).

#include <omp.h>

#pragma omp parallel num_threads(16) // also tried 2, 4 etc - little difference - hoping that placing this line here spawns the thread pool at the onset rather than at every step
{
while(i < t){
    #pragma omp master
    {
       /* do other calculations on single core, output etc.  */
       Tree.PreProcessing() 
       /* PreProcessing can drastically change data for certain conditions, but …

我正在尝试计算一些大数字.为了加快计算速度,我想利用多线程.每个线程都应该计算一个数字,最后计算一个总和.

我曾经看过一些与a SumThread和a Collector有关的东西,如下所示:

public BigInteger compute(int p) {
    Collector c = new Collector(p);

    for(T element : Collection<T> bigCollection) {
        new SumThread(c) {

            @Override
            protected void doTheJob() {
                long big = someVeryComplexCalculation(element, ...); //n!
                receive(BigInteger.valueOf(big));
            }

        }
    }

    if(collector.isReady())
        return collector.getResult();

    return null;
}

public class Collector {

    private int numberOfProcesses;
    private int numberOfAllowedProcesses;
    private BigInteger result;

    public Collector(int n) {
        numberOfAllowedProcesses = n;
        numberOfProcesses = 0;
        result = BigInteger.ZERO;
    }

    synchronized public void enter() throws InterruptedException { …

在我的Spark应用程序中，我尝试从RDBMS读取多个表，进行一些数据处理，然后将多个表写入另一个RDBMS，如下所示（在Scala中）：

val reading1 = sqlContext.load("jdbc", Map("url" -> myurl1, "dbtable" -> mytable1))
val reading2 = sqlContext.load("jdbc", Map("url" -> myurl1, "dbtable" -> mytable2))
val reading3 = sqlContext.load("jdbc", Map("url" -> myurl1, "dbtable" -> mytable3))

// data processing
// ..............

myDF1.write.mode("append").jdbc(myurl2, outtable1, new java.util.Properties)
myDF2.write.mode("append").jdbc(myurl2, outtable2, new java.util.Properties)
myDF3.write.mode("append").jdbc(myurl2, outtable3, new java.util.Properties)

我了解可以使用分区并行读取一个表。但是，read1，read2，read3的读操作似乎是顺序的，myDF1，myDF2，myDF3的写操作也是如此。

如何并行读取多个表（mytable1，mytable2，mytable3）？并且还并行写入多个表（我认为逻辑相同）？

我有一个简单的问题,即将所有元素相互比较.比较本身是对称的,因此,它不必进行两次.

以下代码示例通过显示所访问元素的索引来显示我要查找的内容:

int n = 5;
for (int i = 0; i < n; i++)
{
    for (int j = i + 1; j < n; j++)
    {
        printf("%d %d\n", i,j);
    }
}

输出是:

0 1
0 2
0 3
0 4
1 2
1 3
1 4
2 3
2 4
3 4

因此每个元素相互比较一次.当我想并行化这段代码时,我遇到的问题是首先我必须坚持动态调度,因为每次迭代的计算时间确实变化很大而且我不能使用崩溃,因为嵌套迭代是索引 - 依赖于外循环.

使用#pragma omp parallel for schedule(dynamic, 3)对于外环可导致在最后单个核心执行而使用此用于内部循环可能导致外循环的每次迭代内,处决.

是否有更复杂的做/并行化方式？

先决条件:

• 并行引擎:OpenMP 3.1+(如果需要可以是OpenMP 4.0)
• 并行结构:OpenMP任务
• 编译器:gcc 4.9.x(支持OpenMP 4.0)

输入:

• 带循环的C代码
• 循环具有交叉迭代数据依赖性:"i + 1"迭代需要来自"i"迭代的数据(只有这种依赖,没有别的)
• 循环体可以部分依赖
• 循环不能分为两个循环; 循环体应保持稳固
• 任何合理的东西都可以添加到循环或循环体函数定义中

代码示例:

(这里conf/config/configData变量仅用于说明目的,主要的兴趣在于value/valueData变量.)

void loopFunc(const char* config, int* value)
{
    int conf;
    conf = prepare(config);         // independent, does not change “config”
    *value = process(conf, *value); // dependent, takes prev., produce next
    return;
}

int main()
{
    int N = 100;
    char* configData;           // never changes
    int valueData = 0;          // initial value
    …
    for (int i = 0; i < N; i++)
    {
        loopFunc(configData, …

我正在努力提高我的java技能,但有点不确定如何处理这个多线程应用程序.基本上,程序读取文本文件并找到最大的数字.我在我的搜索算法中添加了一个for循环来创建10个线程,但我不确定它是否实际创建了10个线程.这个想法是为了改善执行时间,或者至少是我认为应该发生的事情.反正有没有检查我是否正确执行了,是否确实改善了执行时间？

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class ProcessDataFile {

    public static void main(String[] args) throws IOException {

        int max = Integer.MIN_VALUE;
        int i = 0;
        int[] numbers = new int[100000];
        String datafile = "dataset529.txt"; //string which contains datafile
        String line; //current line of text file

        try (BufferedReader br = new BufferedReader(new FileReader(datafile))) { //reads in the datafile
            while ((line = br.readLine()) != null) { //reads through each line
                numbers[i++] = Integer.parseInt(line); //pulls out the number of each line …

使用Spring批处理,我希望我的步骤可以跨节点分布,并为给定的作业执行它们.我有一个用例,其中一个作业有多个步骤,每个步骤可以在托管应用程序的多个节点中运行.有人试过吗？任何想法都将受到高度赞赏!

我正在尝试同时学习Perl 6和并行/并发.

对于一个简单的学习练习,我有一个550'.htm'文件的文件夹,我想要所有这些文件中的代码行总和.到目前为止,我有这个:

use v6;

my $start_time = now;
my $exception;
my $total_lines = 0;

my @files = "c:/testdir".IO.dir(test => / '.' htm $/);
for @files -> $file {
    $total_lines += $file.lines.elems;
    CATCH {
        default { $exception = $_; } #some of the files error out for malformed utf-8
    }
}
say $total_lines;
say now - $start_time;

这在大约3秒内得到577,449的总和.

我如何重写它以利用Perl 6并行思想？我意识到节省的时间不会太多,但它可以作为概念的证明.

我是新手,并且误解了Gunicorn + Flask的工作方式.

当我使用4名工作人员运行Gunicorn时,它会创建我的Flask应用程序的4个实例,或者它将创建4个处理来自Nginx和Flask应用程序实例的Web请求的进程？

如果我在我的应用程序中简单地实现内存缓存(例如字典),那么gunicorn会创建多个app实例,因此会创建多个缓存实例吗？