我想知道在使用foreach %dopar%循环后是否可以输出两个不同的对象.

我会尝试解释我在寻找什么.假设我在循环中有几个操作,我有两个data.frame:

library(doMC)
library(parallel)
registerDoMC(cores=4)

result <- foreach(i=1:100) %dopar% {
#### some code here
#### some code here
vec1 <- result_from_previous code # It would be the 1st object I'd like to ouput
vec2 <- result_from_previous code # It would be the 2nd object I'd like to output
}

我想要的输出是长度为2的data.frames列表,例如:

dim(result[[1]]) # equals to nrow=length(vec1) and ncol=100
dim(result[[2]]) # equals to nrow=length(vec2) and ncol=100

我从前一篇文章中尝试了这个保存foreach dopar loop的多个输出:

comb <- function(x, ...) {
  lapply(seq_along(x), function(i) c(x[[i]], …

我有96个txt文件需要处理.现在我正在使用for循环并一次执行一个,这个过程非常慢.生成的96个文件,不需要合并.有没有办法使它们并行运行,ala Parallel.foreach在C#中？当前代码:

for src_name in glob.glob(source_dir+'/*.txt'):
   outfile = open (...)
   with open(...) as infile:
      for line in infile:
         --PROCESS--
   for --condition--:
      outfile.write(...)
   infile.close()
   outfile.close()

希望此进程并行运行source_dir中的所有文件.

有没有办法让J使用多个核心？我认为APL/J的部分好处是语言结构很好地适用于并行解决方案.

看看我的CPU使用情况(我在OSX上),显然只有一个处理器在使用中.

我有一个重要的功能f作用于列表,我不明白为什么它不能将列表分成4个部分,并重新组合结果？

对于从XE2到XE8的Delphi编译器,对于非Windows目标平台,是一个整数数据成员的读操作,用[Volatile]注释,原子？

我知道对于Windows平台的情况,当且仅当数据成员对齐到4个字节时才是原子的,但是非windows(Android等)呢？

请注意,我不是在询问线程安全性.线程安全性和原子性是两回事.

想象一下,我们正在做一些过程,我想在程序的开头设置一个整体种子:例如

mylist <- list( as.list(rep(NA,3)), as.list(rep(NA,3)) )
foo <- function(x){  for(i in 1:length(x)){ 
                       x[[i]] <- sample(100,1)
                         }
                      return(x) 
                     } 

# start block
set.seed(1)
l1 <- lapply(mylist, foo)
l2 <- lapply(mylist, foo)
# end

当然在一个区块内l1并且l2是不同的,但是如果我再次运行上面的区块l1将与之前相同并且l2将与之前相同.

想象一下,foo是可怕的耗时,所以我想用mclapply没有lapply,所以我这样做:

library(parallel)

# start block
set.seed(1)
mclapply(mylist , foo,  mc.cores = 3)
mclapply(mylist , foo,  mc.cores = 3)
# end

如果我再次运行此块,我将在下次获得不同的结果.我如何产生与使用lapply但设置一个整体种子相同的行为mclappy.我查看了mclapplydoc,但我不确定因为使用:

set.seed(1)
l1 <-  mclapply(mylist , …

我在Oozie中使用fork/join,以便并行一些子工作流操作.我的workflow.xml如下所示:

<workflow-app name="myName" xmlns="uri:oozie:workflow:0.5"
<start to="fork1"/>
<kill name="Kill">
    <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>

<fork name="fork1">
    <path start="subworkflow1"/>
    <path start="subworkflow2"/>
</fork>
<join name="Completed" to="End"

<action name="subworkflow1">
    <sub-workflow>
        <app-path>....</app-path>
        <propagate-configuration/>
        <configuration>
            <property>
                <name>....</name>
                <value>....</value>
            </property>
        </configuration>
    </sub-workflow>
    <ok to="Completed"/>
    <error to="Completed"/>
</action>

<action name="subworkflow2">
    <sub-workflow>
        <app-path>....</app-path>
        <propagate-configuration/>
        <configuration>
            <property>
                <name>....</name>
                <value>....</value>
            </property>
        </configuration>
    </sub-workflow>
    <ok to="Completed"/>
    <error to="Completed"/>
</action>

<end name="End"></workflow-app>

当subworkflow1被杀死时(由于某种原因失败),它也会杀死subworkflow2.我希望这两个动作是平行的,但不依赖.

在我的工作流程中,当workflow1被杀死时,我看到workflow2也被杀死了,但我的应用程序成功了(我在Oozie仪表板上检查它 - > HUE中的工作流程).

在这种情况下,我希望subworkflow1将被杀死,subworkflow2将成功,我真的不关心我的整个应用程序会说什么.

• 在我的情况下,subworkflow1需要比subworkflow2更长的时间,所以当我检查我的应用程序结束时,我看到虽然它说subworkflow1 + 2被杀了,我的应用程序成功了,但真正发生的是subworkflow2完成了它的部分,即使,它后来被杀死了(它继续"运行",直到叉子的所有路径完成它们的运行).所以workflow2完成了它的部分而不是被杀死因为workflow1被杀了...

我应该怎么做才能使每个路径获得它自己的状态并继续运行,即使同一个fork中的其他路径被杀死了？

这可能是我写过的最糟糕的StackOverflow标题.我实际上要做的是执行一个异步方法,该方法在同步方法中多次并行使用async/await约定(并且本身包含额外的await调用),同时在整个执行每个分支的过程中保持相同的线程.并行执行,包括所有等待延续.换句话说,我想同步执行一些异步代码,但我想多次并行执行.现在你可以看到为什么标题太糟糕了.也许最好用一些代码来说明......

假设我有以下内容:

public class MyAsyncCode
{
    async Task MethodA()
    {
        // Do some stuff...
        await MethodB();
        // Some other stuff
    }

    async Task MethodB()
    {
        // Do some stuff...
        await MethodC();
        // Some other stuff
    }

    async Task MethodC()
    {
        // Do some stuff...
    }
}

调用者是同步的(来自控制台应用程序).让我尝试说明我尝试使用Task.WaitAll(...)和包装任务的尝试:

public void MyCallingMethod()
{
    List<Task> tasks = new List<Task>();
    for(int c = 0 ; c < 4 ; c++)
    {
        MyAsyncCode asyncCode = new MyAsyncCode();
        tasks.Add(Task.Run(() => asyncCode.MethodA()));
    }
    Task.WaitAll(tasks.ToArray());
} …

长篇文章请不要气馁.我尽量提供尽可能多的数据,我真的需要帮助解决这个问题:S.如果有新的提示或想法,我会每天更新

问题:

我尝试在并行进程的帮助下并行运行两个核心机器上的Python代码(以避免GIL),但是存在代码明显变慢的问题.例如,在一个核心机器上运行每个工作负载需要600秒,但在两个核心机器上运行需要1600秒(每个工作负载800秒).

我已经尝试过的:

• 我测量了内存,似乎没有内存问题.[只在高点使用20%].

• 我使用"htop"检查我是否真的在不同的核心上运行程序,或者我的核心亲和力是否搞砸了.但也没有运气,我的程序在我的所有内核上运行.

• 问题是CPU限制的问题,因此我检查并确认我的代码在所有内核上以100%CPU运行,大多数情况下.

• 我检查了进程ID,我确实产生了两个不同的进程.

• 我将我提交给执行程序[e.submit(function,[...])]的函数更改为计算派函数并观察到了巨大的加速.所以问题很可能发生在我的process_function(...)中,我将其提交到执行程序而不是之前的代码中.

• 目前我正在使用"并发"中的"期货"来平行任务.但我也试过"多处理"中的"池"类.但是,结果保持不变.

码:

• 产卵过程:

  result = [None]*psutil.cpu_count()
  
  e = futures.ProcessPoolExecutor( max_workers=psutil.cpu_count() )
  
  for i in range(psutil.cpu_count()):
      result[i] = e.submit(process_function, ...)
  

  Run Code Online (Sandbox Code Playgroud)

• process_function:

  from math import floor
  from math import ceil
  import numpy
  import MySQLdb
  import time
  
  db = MySQLdb.connect(...)
  cursor  = db.cursor()
  query = "SELECT ...."
  cursor.execute(query)
  
  [...]  #save db results into the variable db_matrix (30 columns, 5.000 rows)
  [...]  #save db results into the variable bp_vector (3 columns, 500 rows)
  [...]  #save …

  Run Code Online (Sandbox Code Playgroud)

以下(简化)脚本在unix集群的主节点(4个虚拟核心)上正常工作.

library(foreach)
library(doParallel)

nc = detectCores()
cl = makeCluster(nc)
registerDoParallel(cl)

foreach(i = 1:nrow(data_frame_1), .packages = c("package_1","package_2"), .export = c("variable_1","variable_2"))  %dopar% {     

    row_temp = data_frame_1[i,]
    function(argument_1 = row_temp, argument_2 = variable_1, argument_3 = variable_2)

}

stopCluster(cl)

我想利用集群中的16个节点(16 * 4总共虚拟核心).

我想我需要做的就是更改指定的并行后端makeCluster.但是我应该怎么做呢？文档不是很清楚.

基于这个相当古老的(2013年)帖子http://www.r-bloggers.com/the-wonders-of-foreach/,似乎我应该更改默认类型(sock或者MPI- 哪个可以在unix上工作？ )

编辑

来自foreach作者的这个小插图:

默认情况下,doParallel在类Unix系统上使用多核功能,在Windows上使用snow功能.请注意,多核功能仅在一台计算机上运行任务,而不是在一组计算机上运行.但是,您可以使用snow功能在群集上执行,使用类Unix操作系统,Windows甚至组合.

什么you can use the snow functionality意思？我该怎么办？

请考虑以下示例:

class Example
{
    private readonly List<string> _list = new List<string>();
    private readonly object _lock = new object();

    public IReadOnlyList<string> Contents
    {
        get
        {
            lock (_lock)
            {
                return new List<string>(_list);
            }
        }
    }

    public void ModifyOperation(string example)
    {
        lock (_lock)
        {
            // ...
            _list.Add(example);
            // ...
        }
    }
}

如何Contents在不复制整个List的情况下实现对List的并行读访问？在C#中有并发Collections,但没有线程安全List.在Java中有类似的东西CopyOnWriteArrayList.