标签: parallel-processing

我正在考虑对我们的项目使用并行单元测试,并且想知道实际编写这种并行单元测试的任何最佳实践.

我在使用C#4.0 Task.Factory.StartNew时看到的几乎所有文档都指出,为了等待Task完成,你需要一个Wait.但我的初步测试表明这是不必要的.其他人可以给我确认吗？我很好奇为什么这么多在线和印刷的参考文献说你应该打电话给Wait.

这是一个简单的控制台应用程序,显示我不需要Wait语句,所以我评论了它.无论我是否注释掉tsk.Wait(),输出都是一样的.

所有情况下的预期产出如下:

Main thread starting.
After running MyTask. The result is True
After running SumIt. The result is 1
Main thread ending.

代码:

class Program
{
    // A trivial method that returns a result and takes no arguments.
    static bool MyTask()
    {
        Thread.Sleep(2000);
        return true;
    }

    // This method returns the summation of a positive integer
    // which is passed to it.
    static int SumIt(object v)
    {
        int x = (int)v;
        int sum = 0;
        for (; x …

注意:这篇文章完全改写了2011-06-10; 感谢彼得帮助我.另外,如果我不接受一个答案,请不要被冒犯,因为这个问题似乎相当开放.(但是,如果你解决它,你当然会得到复选标记).

另一位用户发布了有关并行化合并排序的问题.我以为我会写一个简单的解决方案,但唉,它并不比顺序版快得多.

问题陈述

合并排序是一种分而治之的算法,其中计算的叶子可以并行化.

代码的工作原理如下:将列表转换为树,表示计算节点.然后,合并步骤返回每个节点的列表.从理论上讲,我们应该看到一些重要的性能增益,因为我们将从O(n log n)算法转变为具有无限处理器的O(n)算法.

当参数l(水平)大于零时,计算的第一步是并行化的.这是通过[经由可变完成STRAT ]选择RPAR策略,这将使得子计算归并"×发生在具有平行归并" Y.然后,我们合并结果,并使用rdeepseq强制进行评估.

data Tree a = Leaf a | Node (Tree a) (Tree a) deriving (Show)

instance NFData a => NFData (Tree a) where
    rnf (Leaf v) = deepseq v ()
    rnf (Node x y) = deepseq (x, y) ()

listToTree [] = error "listToTree -- empty list"
listToTree [x] = Leaf x
listToTree xs …

我试图使用boost.python将一段C++代码包装到python lib中,但是,我发现多个实例不能同时运行:

代码(C++):

class Foo{
public:
    Foo(){}
    void run(){
        int seconds = 2;
        clock_t endwait;
        endwait = clock () + seconds * CLOCKS_PER_SEC ;
        while (clock() < endwait) {}
    }   

};

BOOST_PYTHON_MODULE(run_test)
{
   using namespace boost::python;

   class_<Foo>("test", init<>())
      .def("run", &Foo::run)
      ;   

}

这是使用CMake(CMake)编译的:

add_library(run_test SHARED run_test.cpp)
target_link_libraries(run_test boost_python python2.7)

并使用以下代码(Python)进行测试:

class Dos(threading.Thread):
    def run(self):
        printl('performing DoS attack')

        proc = test()
        proc.run()

for i in range(5):
    t = Dos()
    t.start()

输出表明代码以非常奇怪的方式并行化.每个线程应该只需要2秒钟,并且我的四核机器上应该同时运行4个线程:

[2011-11-04 13:57:01] performing DoS attack
 [2011-11-04 13:57:01] performing DoS attack …

我已经习惯在.Net的并行扩展中使用Parallel.For(),因为它是一种简单的并行化代码的方法,而无需手动启动和维护线程(这可能是繁琐的).我现在正在看一个无限循环(做一些事情,直到我发出信号停止),我希望并行化,没有一个参数可以自由Parallel.For()重载这样做,所以想知道这里最好的方法是什么是.原则上我可以这样做:

Parallel.For(0, int.Max)

但我怀疑这可能不是工作分区逻辑处理的预期/有效模式(？)

另一种选择是:

for(;;)
{
    Parallel.For(0, 128, delegate()
    {
       // Do stuff.
    }
}

但这似乎不够优雅,也可能导致低效的工作分区.

现在我的直觉是通过创建和维护我自己的线程来手动执行此操作,但我有兴趣获得一些反馈/意见.谢谢.

===更新===

我在接受的答案中使用了文章中的代码的简化版本(我删除了ParallelOptions参数).这是代码......

public class ParallelUtils
{
    public static void While(Func<bool> condition, Action body) 
    { 
        Parallel.ForEach(IterateUntilFalse(condition), ignored => body()); 
    }

    private static IEnumerable<bool> IterateUntilFalse(Func<bool> condition) 
    { 
        while (condition()) yield return true; 
    }
}

一个示例用法是:

Func<bool> whileCondFn = () => !_requestStopFlag;
ParallelUtils.While(whileCondFn, delegate()
{
    // Do stuff.


});

我已经阅读了Mark Harris的文章"优化并行缩减CUDA",我发现它非常有用,但我仍然无法理解1或2个概念.它写在第18页:

//First add during load

// each thread loads one element from global to shared mem

unsigned int tid = threadIdx.x;

unsigned int i = blockIdx.x*blockDim.x + threadIdx.x;

sdata[tid] = g_idata[i];
__syncthreads();

优化代码:有2个负载和第一个减少的添加:

// perform first level of reduction,

// reading from global memory, writing to shared memory
unsigned int tid = threadIdx.x;                                    ...1

unsigned int i = blockIdx.x*(blockDim.x*2) + threadIdx.x;          ...2

sdata[tid] = g_idata[i] + g_idata[i+blockDim.x];                   ...3

__syncthreads();                                                   ...4

我无法理解第2行; 如果我有256个元素,如果我选择128作为我的块大小,那么为什么我将它乘以2？请解释如何确定块大小？

过去几天我一直在清理我对排序算法的记忆,而且我遇到了一个我无法找到最佳解决方案的情况.

我写了一个quicksort的基本实现,我想通过并行执行来提高性能.

我得到的是:

template <typename IteratorType>
void quicksort(IteratorType begin, IteratorType end)
{
  if (distance(begin, end) > 1)
  {
    const IteratorType pivot = partition(begin, end);

    if (distance(begin, end) > 10000)
    {
      thread t1([&begin, &pivot](){ quicksort(begin, pivot); });
      thread t2([&pivot, &end](){ quicksort(pivot + 1, end); });

      t1.join();
      t2.join();
    }
  }
}

虽然这比天真的"无线程"实现更好,但这具有严重的局限性,即:

• 如果要排序的数组太大或者递归太深,系统可能会用尽线程并且执行失败.
• 可能可以避免在每个递归调用中创建线程的成本,特别是考虑到线程不是无限资源.

我想使用线程池来避免后期线程创建,但我面临另一个问题:

• 我创建的大多数线程最初都是完成所有工作,然后在等待完成时不做任何事情.这导致很多线程只是等待子调用完成,这似乎是次优的.

是否有一种技术/实体可以用来避免浪费线程(允许重用)？

我可以使用boost或任何C++ 11工具.

我对r有点新,我想使用一个允许多核处理的软件包,以便更快地运行glm函数.我想知道是否有一种语法我可以用于此事.这是我写的一个示例glm模型,我可以添加一个使用多核的参数吗？

g<-glm(IsChurn~.,data=dat,family='binomial')

谢谢.

当前场景:我在名为directoryA的目录中有900个文件.这些文件通过文件899.txt命名为file0.txt,每个文件大小为15MB.我在python中按顺序遍历每个文件.我将每个文件作为列表加载,执行一些操作,并在directoryB中写出输出文件.当循环结束时,我在目录B中有900个文件.这些文件通过out899.csv命名为out0.csv.

问题:每个文件的处理需要3分钟,使脚本运行超过40小时.我希望以并行方式运行该过程,因为所有文件彼此独立(没有任何相互依赖性).我的机器里有12个核心.

以下脚本按顺序运行.请帮我平行运行.我已经使用相关的stackoverflow问题查看了python中的一些并行处理模块,但是我很难理解,因为我没有太多的python接触.万分感谢.

伪脚本

    from os import listdir 
    import csv

    mypath = "some/path/"

    inputDir = mypath + 'dirA/'
    outputDir = mypath + 'dirB/'

    for files in listdir(inputDir):
        #load the text file as list using csv module 
        #run a bunch of operations
        #regex the int from the filename. for ex file1.txt returns 1, and file42.txt returns 42
        #write out a corresponsding csv file in dirB. For example input file file99.txt is written as out99.csv

如果我没有记错的话，我可以std::transform执行到位 ，通过使用同一范围内的输入和输出迭代器。假设我有一些std::vector对象vec，那么我会写

std::transform(vec.cbegin(),vec.cend(),vec.begin(),unary_op)

使用合适的一元运算unary_op。

使用C ++ 17标准，我想通过std::execution::par在其中插入第一个参数来并行执行转换。这会使该函数从上cppreference文章std::transform中的重载（1）变为（2）。但是，对此超载的注释说：

unary_op[...]不得使任何迭代器（包括最终迭代器）无效，或修改所涉及范围的任何元素。（自C ++ 11起）

“修改任何元素”是否真的意味着我无法就地使用算法，还是在谈论我误解的其他细节？