标签: parallel-processing

我有一个用Erlang编写的神经网络,我刚买了一块带有240核GPU的GeForce GTX 260卡.使用CUDA作为粘合剂在显卡上运行它是否微不足道？

在Python中实现并行性有哪些选择？我想对一些非常大的栅格执行一堆CPU绑定计算,并希望将它们并行化.来自C背景,我熟悉三种并行方法:

1. 消息传递过程,可能分布在集群中,例如MPI.
2. 显式共享内存并行,使用pthreads或fork(),pipe()等.人
3. 隐式共享内存并行,使用OpenMP.

决定使用方法是权衡利弊.

在Python中,有哪些方法可用,它们的特征是什么？是否有可群集的MPI克隆？实现共享内存并行性的首选方法是什么？我听说过GIL的问题,以及对tasklet的引用.

简而言之,在选择它们之前,我需要了解Python中的不同并行化策略？

为什么parallel.ForEach循环使用OperationCancelledException退出,同时使用GetConsumableEnumerable？

//outside the function
static BlockingCollection<double> _collection = new     BlockingCollection<double>();


    var t = Task.Factory.StartNew(Producer);            
    Parallel.ForEach(_collection.GetConsumingEnumerable(),item => Console.WriteLine("Processed {0}", item));
    Console.WriteLine("FINISHED processing");



public static void Producer()
{
     var data = Enumerable.Range(1, 1000);
     foreach (var i in data)
     {
        _collection.Add(i);
        Console.WriteLine("Added {0}",i);
     }

     Console.WriteLine("Finished adding");
     _collection.CompleteAdding();
}

我正在构建一个必须处理大量数据的控制台应用程序.

基本上,应用程序从数据库中获取引用.对于每个引用,解析文件的内容并进行一些更改.这些文件是HTML文件,并且该过程正在使用RegEx替换进行繁重的工作(查找引用并将它们转换为链接).然后将结果存储在文件系统中并发送到外部系统.

如果我按顺序恢复该过程:

var refs = GetReferencesFromDB(); // ~5000 Datarow returned
foreach(var ref in refs)
{
    var filePath = GetFilePath(ref); // This method looks up in a previously loaded file list
    var html = File.ReadAllText(filePath); // Read html locally, or from a network drive
    var convertedHtml = ParseHtml(html);
    File.WriteAllText(destinationFilePath); // Copy the result locally, or a network drive
    SendToWs(ref, convertedHtml);
}

我的程序工作正常,但速度很慢.这就是为什么我想要并行化这个过程.

到现在为止,我做了一个简单的并行化添加AsParallel:

var refs = GetReferencesFromDB().AsParallel(); 
refs.ForAll(ref=>
{
    var filePath = GetFilePath(ref); 
    var html = File.ReadAllText(filePath); 
    var convertedHtml = …

我正在编写一个C#程序,通过FTP生成并上传50万个文件.我想并行处理4个文件,因为机器有4个核心,文件生成需要更长的时间.是否可以将以下Powershell示例转换为C#？或者是否有更好的框架,如C#中的Actor框架(如F#MailboxProcessor)？

Powershell的例子

$maxConcurrentJobs = 3;

# Read the input and queue it up
$jobInput = get-content .\input.txt
$queue = [System.Collections.Queue]::Synchronized( (New-Object System.Collections.Queue) )
foreach($item in $jobInput)
{
    $queue.Enqueue($item)
}

# Function that pops input off the queue and starts a job with it
function RunJobFromQueue
{
    if( $queue.Count -gt 0)
    {
        $j = Start-Job -ScriptBlock {param($x); Get-WinEvent -LogName $x} -ArgumentList $queue.Dequeue()
        Register-ObjectEvent -InputObject $j -EventName StateChanged -Action { RunJobFromQueue; Unregister-Event $eventsubscriber.SourceIdentifier; Remove-Job $eventsubscriber.SourceIdentifier } | Out-Null
    }
}

# Start …

今天我尝试做一些优化foreach声明,这可以继续XDocument.

优化前:

foreach (XElement elem in xDoc.Descendants("APSEvent").ToList())
{
    //some operations
}

优化后:

Parallel.ForEach(xDoc.Descendants("APSEvent").ToList(), elem =>
{
    //same operations
});

我看到.NET Parallel.ForEach(...)只打开一个线程!结果,时间跨度Parallel大于标准foreach.

为什么你认为.NET只开了1个线程？因为锁定文件？谢谢

我们目前正在使用TBB流程图,其中a)并行过滤器处理数组(与偏移并行)并将处理结果放入中间向量(在堆上分配;大多数向量将增长到8MB).然后将这些矢量传递给节点,然后节点根据它们的特性(在a中确定)对这些结果进行后处理.由于资源同步,每个特征只能有一个这样的节点.我们编写的原型在UMA架构上运行良好(在单CPU Ivy Bridge和Sandy Bridge架构上测试).但是,该应用程序无法在我们的NUMA架构(4 CPU Nehalem-EX)上扩展.我们将问题归结为内存分配并创建了一个最小的示例,其中我们有一个并行管道,它只是从堆中分配内存(通过8MB块的malloc,然后memset 8MB区域;类似于初始原型所做的)达到一定的记忆力.我们的发现是:

• 在UMA架构上,应用程序与管道使用的线程数呈线性关系(通过task_scheduler_init设置)

• 在NUMA架构上,当我们将应用程序固定到一个插槽(使用numactl)时,我们看到相同的线性放大

• 在我们使用多个套接字的NUMA架构中,我们的应用程序运行的时间随着套接字的数量而增加(负线性比例 - "向上")

对我们来说,这就像堆争用一样.我们到目前为止尝试的是将英特尔的TBB可扩展分配器替换为glibc分配器.但是,单个套接字上的初始性能比使用glibc更差,在多个套接字上性能不会变差但也没有变得更好.我们使用tcmalloc,hoard分配器和TBB的缓存对齐分配器获得了相同的效果.

问题是,是否有人遇到类似的问题.堆栈分配对我们来说不是一个选项,因为我们希望在管道运行后保持堆分配的向量.一个堆如何在多个线程的NUMA体系结构上有效地分配MB大小的内存区域？我们真的希望保持动态分配方法,而不是预先分配内存并在应用程序中管理内存.

我用numactl为各种执行附加了perf stats.Interleaving/localalloc无任何影响(QPI总线不是瓶颈;我们通过PCM验证,QPI链路负载为1%).我还添加了一个描绘glibc,tbbmalloc和tcmalloc结果的图表.

perf stat bin/prototype 598.867

'bin/prototype'的性能计数器统计信息:

  12965,118733 task-clock                #    7,779 CPUs utilized          
        10.973 context-switches          #    0,846 K/sec                  
         1.045 CPU-migrations            #    0,081 K/sec                  
       284.210 page-faults               #    0,022 M/sec                  
17.266.521.878 cycles                    #    1,332 GHz                     [82,84%]
15.286.104.871 stalled-cycles-frontend   #   88,53% frontend cycles idle    [82,84%]
10.719.958.132 stalled-cycles-backend    #   62,09% backend  cycles idle    [67,65%]
 3.744.397.009 instructions              #    0,22  insns per cycle        
                                         #    4,08  stalled cycles per insn [84,40%]
   745.386.453 branches …

这个问题适用于创建"宽"表,类似于您可以使用reshape2中的dcast创建的表.我知道之前已经讨论了很多次,但我的问题是如何使这个过程更有效率.我在下面提供了几个例子,这些例子可能会使问题看起来很冗长,但大多数只是用于基准测试的测试代码

从一个简单的例子开始,

> z <- data.table(col1=c(1,1,2,3,4), col2=c(10,10,20,20,30), 
                  col3=c(5,2,2.3,2.4,100), col4=c("a","a","b","c","a"))

> z
     col1 col2  col3 col4
1:    1   10   5.0    a      # col1 = 1, col2 = 10
2:    1   10   2.0    a      # col1 = 1, col2 = 10
3:    2   20   2.3    b
4:    3   20   2.4    c
5:    4   30 100.0    a

我们需要创建一个"宽"表,它将col4列的值作为列名,并将col1和col2的每个组合的和(col3)值组合在一起.

> ulist = unique(z$col4) # These will be the additional column names

# Create long table with sum
> z2 <- z[,list(sumcol=sum(col3)), by='col1,col2,col4']

# Pivot …

我是一个ETL进程,我从Spring Data Repository中检索了很多实体.然后我使用并行流将实体映射到不同的实体.我可以使用使用者将这些新实体逐个存储在另一个存储库中,或者将它们收集到List中并将其存储在单个批量操作中.第一种是昂贵的,而后者可能超过可用的内存.

有没有一种很好的方法来收集流中的一定数量的元素(如限制),消耗该块,并继续并行处理直到所有元素都被处理？

我在这里读了一些答案(例如),其中一些人说并行性不会提高性能(可能在读取IO中).

但是我创建了一些测试,表明WRITE操作也要快得多.

- 阅读测试:

我用伪数据创建了随机6000文件:

让我们尝试用w/o并行性来阅读它们:

var files =
    Directory.GetFiles("c:\\temp\\2\\", "*.*", SearchOption.TopDirectoryOnly).Take(1000).ToList();

    var sw = Stopwatch.StartNew();
    files.ForEach(f => ReadAllBytes(f).GetHashCode()); 
    sw.ElapsedMilliseconds.Dump("Run READ- Serial");
    sw.Stop(); 


    sw.Restart();
    files.AsParallel().ForAll(f => ReadAllBytes(f).GetHashCode()); 
    sw.ElapsedMilliseconds.Dump("Run READ- Parallel");
    sw.Stop();

结果1:

运行READ- Serial 595

运行READ-Parallel 193

结果2:

运行READ- Serial 316

运行READ-Parallel 192

- 写测试:

要创建1000个随机文件,每个文件为300K.(我从prev test中清空了目录)

var bytes = new byte[300000];
Random r = new Random();
r.NextBytes(bytes);
var list = Enumerable.Range(1, 1000).ToList();

sw.Restart();
list.ForEach((f) => WriteAllBytes(@"c:\\temp\\2\\" + Path.GetRandomFileName(), bytes)); 
sw.ElapsedMilliseconds.Dump("Run WRITE serial");
sw.Stop();

sw.Restart();
list.AsParallel().ForAll((f) => WriteAllBytes(@"c:\\temp\\2\\" …