标签: parallel-processing

我有一个功能代码,它将一个属性的字符串拆分为类的列表:Dataframe由string, string, string.

现在我声明一个空的Dataframe2(string,string[], string)并使用将项添加到列表中Add

class Program

{


    public static string[] SPString(string text)
    {
        string[] elements;
        elements = text.Split(' ');
        return elements;
    }

    //Structures
    public class Dataframe
    {

        public string Name { get; set; }
        public string Text { get; set; }
        public string Cat { get; set; }
    }

    public class Dataframe2
    {

        public string Name { get; set; }
        public string[] Text { get; set; }
        public string Cat { get; set; } …

我试图在主节点处从所有处理器(包括主节点)收集不同长度的不同字符串到单个字符串(字符数组).这是MPI_Gatherv的原型:

int MPI_Gatherv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
            void *recvbuf, const int *recvcounts, const int *displs,
            MPI_Datatype recvtype, int root, MPI_Comm comm)**.

我无法确定像一些参数recvbuf,recvcounts和displs.任何人都可以在C中提供源代码示例吗？

我正在使用Node.js运行服务器,需要从我正在运行的另一台服务器请求数据(localhost:3001).我需要向数据服务器发出许多请求(~200)并收集数据(响应大小从~20Kb到~20Mb不等).每个请求都是独立的,我想将响应保存为表单的一个巨大数组:

[{"urlAAA": responseAAA}, {"urlCCC": responseCCC}, {"urlBBB": responseBBB}, etc ]

请注意,项目的顺序并不重要,理想情况下,它们应按数据可用的顺序填充数组.

var express = require('express');
var router = express.Router();
var async = require("async");
var papa = require("papaparse");
var sync_request = require('sync-request');
var request = require("request");

var pinnacle_data = {};
var lookup_list = [];
for (var i = 0; i < 20; i++) {
    lookup_list.push(i);
}

function write_delayed_files(object, key, value) {
    object[key] = value;
    return;
}

var show_file = function (file_number) {
    var file_index = Math.round(Math.random() * 495) + 1; …

可以为Caffe(特别是pyCaffe)设置所有GPU吗？

就像是:

caffe train -solver examples/mnist/lenet_solver.prototxt -gpu all

尝试执行一些并行操作时，我收到“尝试读取或写入受保护的内存”的消息。我正在将AutoCad数据库读入内存以进行一些数据挖掘。我可以使用常规for循环执行此操作，但不能使用Parallel.ForEach。有任何想法吗？

Parallel.ForEach(_Files, (currentFile) =>
{
    var _File = currentFile;
    using (Database _Database = new Database(false, true))
    {
        _Database.ReadDwgFile(_File, FileOpenMode.OpenForReadAndAllShare, false, null);           
        _Database.CloseInput(true);
        // Do Stuff
    }
});

我目前正在处理一些大型数据集,因此并行化工作流程是唯一的方法.

我需要在开始时为每个线程加载一些包(即:for(this.thread in threads) { #load some packages }.

不幸的是,我不知道该怎么做.

下面的代码进一步说明我的问题,在这里我想使用管道操作者magrittr在%dopar%:

.

library(parallel)
library(doParallel)
library(foreach)
library(magrittr)


# Generate some random data and function :
# -----------------------------------------

randomData = runif(10^3)
randomFunction = function(x) {x * (2^x) } 

randomData[1] %>% randomFunction #Works



# And now ... The parallel part :
# --------------------------------

myCluster = makeCluster(6)
registerDoParallel(myCluster)


# Test that the do par is up and running: 
foreach(i = randomData) %dopar% { i }


# Use magrittr …

我正在做一些文件解析,这是一个CPU绑定任务.无论我在这个过程中抛出多少文件,它都使用不超过大约50MB的RAM.该任务是可并行的,我已将其设置为使用下面的并发期货来解析每个文件作为一个单独的过程:

    from concurrent import futures
    with futures.ProcessPoolExecutor(max_workers=6) as executor:
        # A dictionary which will contain a list the future info in the key, and the filename in the value
        jobs = {}

        # Loop through the files, and run the parse function for each file, sending the file-name to it.
        # The results of can come back in any order.
        for this_file in files_list:
            job = executor.submit(parse_function, this_file, **parser_variables)
            jobs[job] = this_file

        # Get the completed jobs whenever they are …

我一直在尝试优化一段涉及大型多维数组计算的python代码.我对numba的结果有违反直觉.我正在运行MBP,2015年中期,2.5 GHz i7 quadcore,OS 10.10.5,python 2.7.11.考虑以下:

 import numpy as np
 from numba import jit, vectorize, guvectorize
 import numexpr as ne
 import timeit

 def add_two_2ds_naive(A,B,res):
     for i in range(A.shape[0]):
         for j in range(B.shape[1]):
             res[i,j] = A[i,j]+B[i,j]

 @jit
 def add_two_2ds_jit(A,B,res):
     for i in range(A.shape[0]):
         for j in range(B.shape[1]):
             res[i,j] = A[i,j]+B[i,j]

 @guvectorize(['float64[:,:],float64[:,:],float64[:,:]'],
    '(n,m),(n,m)->(n,m)',target='cpu')
 def add_two_2ds_cpu(A,B,res):
     for i in range(A.shape[0]):
         for j in range(B.shape[1]):
             res[i,j] = A[i,j]+B[i,j]

 @guvectorize(['(float64[:,:],float64[:,:],float64[:,:])'],
    '(n,m),(n,m)->(n,m)',target='parallel')
 def add_two_2ds_parallel(A,B,res):
     for i in range(A.shape[0]):
         for j in range(B.shape[1]):
             res[i,j] = A[i,j]+B[i,j]

 def …

我正在BlockingCollection努力尝试更好地理解它们，但是我在努力理解为什么当我使用Parallel.For

我只是在上面加上一个数字（生产者？）：

var blockingCollection = new BlockingCollection<long>();

Task.Factory.StartNew(() =>
{
    while (count <= 10000)
    {
        blockingCollection.Add(count);
        count++;
    }
});

然后，我正在尝试处理（消费者？）：

Parallel.For(0, 5, x => 
{
    foreach (long value in blockingCollection.GetConsumingEnumerable())
    {
        total[x] += 1;
        Console.WriteLine("Worker {0}: {1}", x, value);
    }
});

但是，当完成所有数字的处理后，它就挂在那里了吗？我究竟做错了什么？

另外，当我将Parallel.For设置为5时，是否表示它正在5个单独的线程上处理数据？

我在这里读

如果没有提供值的份数执行(即,既不是"-np",也不设置在命令行上它的同义词),开放MPI将自动执行该程序的每个处理槽的副本(参见下面的"过程槽"的描述)

所以我期待

mpirun program

运行该程序的八个副本(实际上是一个简单的hello世界),因为我有一个英特尔®酷睿™i7-2630QM CPU @ 2.00GHz×8,但它没有:它只运行一个进程.