标签: parallel-processing

Perl中并行处理的最佳模块是什么？我从未在Perl中完成并行处理.

什么是用于数据库访问和邮件的并行进程的优秀Perl模块？

我查看了模块Parallel :: ForkManager.任何想法都赞赏.

我正在尝试并行过滤元素.对于每个元素,我需要执行距离计算以查看它是否足够接近目标点.不要因为这样做已经存在数据结构,我现在只做初步实验.

无论如何,我想运行一些非常基本的实验,我生成随机向量并过滤它们.这是我的实现,完成所有这些

(defn pfilter [pred coll]
  (map second
    (filter first
      (pmap (fn [item] [(pred item) item]) coll))))

(defn random-n-vector [n]
  (take n (repeatedly rand)))

(defn distance [u v]
  (Math/sqrt (reduce + (map #(Math/pow (- %1 %2) 2) u v))))

(defn -main [& args]
  (let [[n-str vectors-str threshold-str] args
        n (Integer/parseInt n-str)
        vectors (Integer/parseInt vectors-str)
        threshold (Double/parseDouble threshold-str)
        random-vector (partial random-n-vector n)
        u (random-vector)]
    (time (println n vectors 
      (count 
        (pfilter 
          (fn [v] (< (distance u v) threshold))
          (take vectors (repeatedly random-vector)))))))) …

我有一个需要多次读取的xml文件.我正在尝试使用Parallel.ForEach来加速这个过程,因为没有读入的数据与它正在读入的顺序有关.数据只是用于填充对象.我的问题是即使我每次在线程中打开文件只读它抱怨它是由另一个程序打开.(我没有在文本编辑器中打开它或任何东西:))

如何从同一个文件中完成多次读取？

编辑:文件大约是18KB.它读取大约1800次.

谢谢

关于等待任务完成和线程同步的主题.

我目前有一个迭代,我已经包含在Parallel.ForEach中.在下面的例子中,我在评论中提出了一些关于如何最好地处理循环的优雅终止的问题(.NET 4.0);

private void myFunction()
    {

        IList<string> iListOfItems = new List<string>();
        // populate iListOfItems

        CancellationTokenSource cts = new CancellationTokenSource();

        ParallelOptions po = new ParallelOptions();
        po.MaxDegreeOfParallelism = 20; // max threads
        po.CancellationToken = cts.Token;

        try
        {
            var myWcfProxy = new myWcfClientSoapClient();

            if (Parallel.ForEach(iListOfItems, po, (item, loopsate) =>
            {
                try
                {
                    if (_requestedToStop)
                        loopsate.Stop();
                    // long running blocking WS call, check before and after
                    var response = myWcfProxy.ProcessIntervalConfiguration(item);
                    if (_requestedToStop)
                        loopsate.Stop();

                    // perform some local processing of the response object
                } …

我们每天最多有30 GB的GZip压缩日志文件.每个文件保存100.000行,压缩后为6到8 MB.解析逻辑已被剥离的简化代码利用Parallel.ForEach循环.

在两个NUMA节点上,MaxDegreeOfParallelism为8处理的最大行数达到峰值,32个逻辑CPU盒(Intel Xeon E7-2820 @ 2 GHz):

using System;

using System.Collections.Concurrent;

using System.Linq;
using System.IO;
using System.IO.Compression;

using System.Threading.Tasks;

namespace ParallelLineCount
{
    public class ScriptMain
    {
        static void Main(String[] args)
        {
            int    maxMaxDOP      = (args.Length > 0) ? Convert.ToInt16(args[0]) : 2;
            string fileLocation   = (args.Length > 1) ? args[1] : "C:\\Temp\\SomeFiles" ;
            string filePattern    = (args.Length > 1) ? args[2] : "*2012-10-30.*.gz";
            string fileNamePrefix = (args.Length > 1) ? args[3] : "LineCounts";

            Console.WriteLine("Start:                 {0}", DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ"));
            Console.WriteLine("Processing file(s):    {0}", …

我无法从R初始化与AWS EC2实例的连接,因为我似乎一直在收到错误:Permission denied (publickey)我目前正在使用Mac OS X 10.6.8作为我的操作系统

我尝试在终端($)和R(>)中运行的代码如下:

$ R --vanilla
> require(snowfall)
> sfInit(parallel=TRUE,socketHosts =list("ec2-xx-xxx-xx-xx.zone.compute.amazonaws.com"))
Permission denied (publickey)

但奇怪的是,当我尝试ssh到实例时,我不需要密码,因为我已经在初始化时将公钥导入实例,(我认为)

所以从我的正常终端...跑步时

$ ssh ubuntu@ec2-xx-xxx-xx-xx.zone.compute.amazonaws.com

它会自动连接...(所以我不是100%确定它是否是无密码问题,例如在AWS中使用雪(和降雪)在R中进行并行处理)

我试过在键等上查看相当数量的材料,但似乎没有任何差别.我~/.ssh/authorized_keys也是一个文件夹而不是文件由于某种原因,我甚至在尝试时都无法访问它sudo cd .ssh/authorized_keys...在权限方面drw-------

最终的目标是连接到很多ec2实例并foreach用来执行一些并行处理...但是现在连接到一个也很好...我也想使用我自己的ami所以starcluster不是我真正想要的......(除非我能够使用私有amis并私下运行所有​​命令......)

另外如果doRedis比有人能告诉我如何从本地机器连接到ec2实例那样好,那也是好的...

编辑

我已经设法使用parallel包来处理ssh无密码登录makePSOCKcluster,如R和makePSOCKcluter EC2 socketConnection所示 ......但现在遇到的socketConnection问题如链接中的问题所示...

任何想法如何连接到它？

还证明一切正常,我想这意味着以下命令/函数可以用于获取所有不同的IP地址

d <- parLapply(cl1, 1:length(cl1),function(x)system("ifconfig",intern=T)[2])

函数cl1的输出在哪里make*cluster

注意,因为赏金真的是针对链接中的问题....我不介意你提出哪个问题的答案...但只要在这个问题上写了一些东西就把它与正确的答案联系起来关于相关问题,那么我将相应地奖励积分......

您好我在使用flatmap时有关于并行性的问题请考虑以下示例

    IntStream.of(-1, 1).parallel().flatMap(i->IntStream.range(0,1000).parallel()).forEach(System.out::println);

我是否将内部标志设置为并行是否重要？如果我离开或不离开,结果看起来非常相似.为什么代码(ReferencePipeline)会对映射进行顺序化？我对这条线感到困惑:

result.sequential().forEach(downstream);

http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/stream/ReferencePipeline.java#ReferencePipeline.flatMap%28java.util.function.Function% 29

假设我有两个int[]数组input1和input2.我想从第一个中取正数,从第二个中取出不同的数字,将它们合并在一起,排序并存储到结果数组中.这可以使用流执行:

int[] result = IntStream.concat(Arrays.stream(input1).filter(x -> x > 0), 
                   Arrays.stream(input2).distinct()).sorted().toArray();

我想加快任务,所以我考虑使流并行.通常这只意味着我可以.parallel()在流构造和终端操作之间插入任何地方,结果将是相同的.IntDream.concat的JavaDoc 表示如果任何输入流是并行的,则生成的流将是并行的.所以,我认为做parallel()任何input1流或input2流或级联的流将产生相同的结果.

实际上我错了:如果我添加.parallel()到结果流,似乎输入流保持顺序.此外,我可以将输入流(它们中的任何一个或两者)标记为.parallel(),然后将结果流转换为.sequential(),但输入保持平行.所以实际上有8种可能性:input1,input2和连接流都可以是并行的:

int[] sss = IntStream.concat(Arrays.stream(input1).filter(x -> x > 0),
                Arrays.stream(input2).distinct()).sorted().toArray();
int[] ssp = IntStream.concat(Arrays.stream(input1).filter(x -> x > 0),
                Arrays.stream(input2).distinct()).parallel().sorted().toArray();
int[] sps = IntStream.concat(Arrays.stream(input1).filter(x -> x > 0), 
                Arrays.stream(input2).parallel().distinct()).sequential().sorted().toArray();
int[] spp = IntStream.concat(Arrays.stream(input1).filter(x -> x > 0), 
                Arrays.stream(input2).parallel().distinct()).sorted().toArray();
int[] pss = IntStream.concat(Arrays.stream(input1).parallel().filter(x -> x > 0),
                Arrays.stream(input2).distinct()).sequential().sorted().toArray(); …

我最近一直致力于一个使用神经网络进行虚拟机器人控制的项目.我使用tensorflow对其进行编码并且运行顺畅.到目前为止,我使用顺序模拟来评估神经网络有多好,但是,我想并行运行多个模拟以减少获取数据所需的时间.

要做到这一点,我正在导入python的multiprocessing包.最初我将sess变量(sess=tf.Session())传递给一个运行模拟的函数.但是,一旦我找到使用此sess变量的任何语句,该进程将退出而不会发出警告.摸索了一会儿后,我发现这两个职位: Tensorflow:传递会话的蟒蛇多进程 ,并同时运行多个tensorflow会议

虽然它们高度相关,但我还是无法弄清楚如何使其发挥作用.我尝试为每个单独的过程创建一个会话,并将神经网络的权重分配给它的可训练参数,但没有成功.我也尝试将会话保存到文件中,然后在一个进程中加载​​它,但也没有运气.

有人能够将会话(或会话克隆)传递给多个进程吗？

谢谢.

阅读文档后,我仍然对使用它感到困惑par_unseq.我知道由于线程和矢量化,我无法说出执行的顺序,但是我仍然可以依赖输出的顺序吗？

transform([x0, x1, x2], f) == [f(x0), f(x1), f(x2)]]

换句话说,这个测试是否会失败？

std::vector<int> xs = {1, 2, 3, 4};
std::vector<int> ys(xs.size());

std::transform(
    std::execution::par_unseq,
    cbegin(xs), cend(xs),
    begin(ys),
    [](int x) { return x*x; });

std::vector<int> expected = {1, 4, 9, 16};
ASSERT_EQ(expected , ys);