我正在为D编程语言开发并行化库.现在我对基本原语(并行foreach,map,reduce和tasks/futures)非常满意,我开始考虑一些更高级别的并行算法.并行化的更明显的候选者之一是排序.

我的第一个问题是,在现实世界中有用的排序算法的并行版本,还是主要是学术性的？如果它们有用,它们在哪里有用？我个人很少在我的工作中使用它们,仅仅是因为我通常使用比单一sort()调用更粗糙的并行度来将100%的所有内核挂起.

其次,对于大型阵列来说,似乎快速排序几乎是令人尴尬的并行,但我不能得到接近线性的加速,我相信我应该得到.对于快速排序,唯一固有的串行部分是第一个分区.我尝试并行化快速排序,在每个分区之后,并行排序两个子阵列.在简化的伪代码中:

// I tweaked this number a bunch.  Anything smaller than this and the 
// overhead is smaller than the parallelization gains.
const  smallestToParallelize = 500; 

void quickSort(T)(T[] array) {
    if(array.length < someConstant) {
        insertionSort(array);
        return;
    }

    size_t pivotPosition = partition(array);

    if(array.length >= smallestToParallelize) {
        // Sort left subarray in a task pool thread.
        auto myTask = taskPool.execute(quickSort(array[0..pivotPosition]));
        quickSort(array[pivotPosition + 1..$]);
        myTask.workWait();
    } else {
        // Regular serial quick sort.
        quickSort(array[0..pivotPosition]);
        quickSort(array[pivotPosition + 1..$]);
    }
}

即使对于非常大的阵列,第一个分区所花费的时间可以忽略不计,与纯粹的串行版本的算法相比,我只能在双核上获得大约30%的加速.我猜测瓶颈是共享内存访问.有关如何消除这个瓶颈或瓶颈可能是什么的任何见解？

编辑:我的任务池具有固定数量的线程,等于系统中的核心数减1(因为主线程也起作用).此外,我正在使用的等待类型是工作等待,即如果任务已启动但尚未完成,则线程调用会 …

我需要并行化一个方法,该方法对列表中的元素进行详尽的成对比较.串行实现很简单:

foreach (var element1 in list)
    foreach (var element2 in list)
        foo(element1, element2);

在这种情况下,foo不会改变element1或element2的状态.我知道简单地执行嵌套的Parallel.ForEach语句是不安全的:

Parallel.ForEach(list, delegate(A element1)
{
    Parallel.ForEach(list, delegate(A element2)
    {
        foo(element1, element2);
    });
});

使用并行任务库实现此目的的理想方法是什么？

我之前问过一个相关但非常普遍的问题(尤其是这个回答).

这个问题非常具体.这是我关心的所有代码:

result = {}
for line in open('input.txt'):
  key, value = parse(line)
  result[key] = value

该函数parse是完全独立的(即,不使用任何共享资源).

我有Intel i7-920 CPU(4核,8个线程;我认为线程更相关,但我不确定).

我该怎么做才能使我的程序使用该CPU的所有并行功能？

我假设我可以打开这个文件,在8个不同的线程中读取而没有太多的性能损失,因为磁盘访问时间相对于总时间来说很小.

以下代码解释了我的问题.我知道列表不是线程安全的.但是这个潜在的"真正"原因是什么？

    class Program
{
    static void Main(string[] args)
    {
        List<string> strCol = new List<string>();

        for (int i = 0; i < 10; i++)
        {
            int id = i;
            Task.Factory.StartNew(() =>
            {
                AddElements(strCol);
            }).ContinueWith((t) => { WriteCount(strCol, id.ToString()); });
        }

        Console.ReadLine();
    }

    private static void WriteCount(List<string> strCol, string id)
    {
        Console.WriteLine(string.Format("Task {0} is done. Count: {1}. Thread ID: {2}", id, strCol.Count, Thread.CurrentThread.ManagedThreadId));
    }

    private static void AddElements(List<string> strCol)
    {
        for (int i = 0; i < 20000; i++)
        {
            strCol.Add(i.ToString());
        } …

我之前的评论(特别是@Zboson)之后我编辑了我的问题,以提高可读性

我一直采取行动并观察传统观点,即openmp线程的数量应与机器上的超线程数大致匹配,以获得最佳性能.但是,我观察到我的新笔记本电脑采用Intel Core i7 4960HQ,4核 - 8线程的奇怪行为.(请参阅此处的英特尔文档)

这是我的测试代码:

#include <math.h>
#include <stdlib.h>
#include <stdio.h>
#include <omp.h>

int main() {
    const int n = 256*8192*100;
    double *A, *B;
    posix_memalign((void**)&A, 64, n*sizeof(double));
    posix_memalign((void**)&B, 64, n*sizeof(double));
    for (int i = 0; i < n; ++i) {
        A[i] = 0.1;
        B[i] = 0.0;
    }
    double start = omp_get_wtime();
    #pragma omp parallel for
    for (int i = 0; i < n; ++i) {
        B[i] = exp(A[i]) + sin(B[i]);
    }
    double end = omp_get_wtime(); …

我正在使用concurrent.futures来实现多处理.我得到一个队列.全错误,这是奇怪的,因为我只分配10个工作.

A_list = [np.random.rand(2000, 2000) for i in range(10)]

with ProcessPoolExecutor() as pool:
    pool.map(np.linalg.svd, A_list)

错误:

Exception in thread Thread-9:
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/threading.py", line 921, in _bootstrap_inner
    self.run()
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/threading.py", line 869, in run
    self._target(*self._args, **self._kwargs)
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/concurrent/futures/process.py", line 251, in _queue_management_worker
    shutdown_worker()
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/concurrent/futures/process.py", line 209, in shutdown_worker
    call_queue.put_nowait(None)
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/queues.py", line 131, in put_nowait
    return self.put(obj, False)
  File "/Library/Frameworks/Python.framework/Versions/3.4/lib/python3.4/multiprocessing/queues.py", line 82, in put
    raise Full
queue.Full

我一直在尝试使用多台计算机构建一个集群三天,并且失败了.所以现在我要试着吮吸一大堆你为我解决问题.如果一切顺利的话,我希望我们能够生成一个循序渐进的指南,以便将来作为参考来使用,因为到目前为止,我还没有找到一个合适的参考来设置它(也许这太具体了？)

在我的例子中,让我们假设Windows 7,PuTTY作为SSH客户端,'localhost'将作为主服务器.

此外,我们现在假设同一网络上只有两台计算机.我想这个过程很容易概括,如果我可以让它在两台计算机上运行,​​我可以让它在三台计算机上工作.因此,我们会努力的localhost和remote-computer.

这是我到目前为止收集的内容(底部有引用链接)

1. 安装PuTTY localhost.
2. 安装PuTTY remote-computer
3. 安装SSH服务器 remote-computer
4. 为它分配一个端口来监听？(我不确定这一步)
5. 安装R on localhost
6. 安装相同版本的R on remote-computer
7. R添加到PATH环境变量上都localhost和remote-computer
8. 运行下面的R代码 localhost

码:

library(parallel)
cl <- makePSOCKcluster(c(rep("localhost", 2),
                         rep("remote-computer", 2)))

到目前为止,我已经完成了步骤1-3,不确定我是否需要执行4,完成5-7,而步骤8的代码只是无限期挂起.

当我检查我的SSH服务器日志时,似乎我没有点击SSH服务器localhost.所以看来我的第一个问题是正确配置SSH.有没有人成功做到这一点,你愿意分享你的专业知识吗？

编辑哎呀:参考 http://www.milanor.net/blog/wp-content/uploads/2013/10/03.FirstStepinParallelComputing.pdf

R并行 - 连接到远程核心

https://stat.ethz.ch/pipermail/r-sig-hpc/2010-October/000780.html

我有30家子公司,每家公司都实施了他们的网络服务(使用不同的技术).

我需要实现一个Web服务来聚合它们,例如,所有子公司Web服务都有一个带有名称的Web方法,GetUserPoint(int nationalCode)我需要实现我的Web服务,它将调用所有这些并收集所有响应(例如总和)分数).

这是我的基类:

public abstract class BaseClass
{ // all same attributes and methods
  public long GetPoint(int nationalCode);
}

对于每个子公司Web服务,我实现了一个继承此基类并定义自己的GetPoint方法的类.

public class Company1
{
  //implement own GetPoint method (call a web service).
}

至

public class CompanyN
{
  //implement own GetPoint method (call a web service).
}

所以,这是我的网络方法:

        [WebMethod]
        public long MyCollector(string nationalCode)
        {

           BaseClass[] Clients = new BaseClass[]  { new Company1(),//... ,new Company1()}

           long Result = 0;
           foreach (var item in Clients)
           {
                long …

我想帮助理解我所做的事情/为什么我的代码没有像我期望的那样运行.

我已经开始使用joblib来尝试通过并行运行(大)循环来加速我的代码.

我这样使用它:

from joblib import Parallel, delayed
def frame(indeces, image_pad, m):

    XY_Patches = np.float32(image_pad[indeces[0]:indeces[0]+m, indeces[1]:indeces[1]+m,  indeces[2]])
    XZ_Patches = np.float32(image_pad[indeces[0]:indeces[0]+m, indeces[1],                  indeces[2]:indeces[2]+m])
    YZ_Patches = np.float32(image_pad[indeces[0],                 indeces[1]:indeces[1]+m,  indeces[2]:indeces[2]+m])

    return XY_Patches, XZ_Patches, YZ_Patches


def Patch_triplanar_para(image_path, patch_size):

    Image, Label, indeces =  Sampling(image_path)

    n = (patch_size -1)/2
    m = patch_size

    image_pad = np.pad(Image, pad_width=n, mode='constant', constant_values = 0)

    A = Parallel(n_jobs= 1)(delayed(frame)(i, image_pad, m) for i in indeces)
    A = np.array(A)
    Label = np.float32(Label.reshape(len(Label), 1))
    R, T, Y =  np.hsplit(A, 3)

    return R, T, …

我看到两种指定超时的方法concurrent.futures.

• as_completed()
• wait()

两种方法都处理N运行期货.

我想为每个未来指定一个单独的超时.

使用案例:

• 从数据库获取数据的未来超时为0.5秒.
• 从HTTP服务器获取数据的未来超时为1.2秒.

我该如何处理concurrent.futures？或者这个库不是正确的工具吗？

结论

• AFAIK mdurant的解决方案是一个很好的解决方案.
• 我想下次我会使用不同的库.也许asyncio对此有更好的支持.请参阅:https://docs.python.org/3/library/asyncio-task.html#asyncio.sleep