我想使用以下来生成唯一的作业,其中{1}和{2}是唯一的元组:
parallel echo {1} {2} ::: A B C D ::: A B C D
例如在python(itertools)中提供了这样一个组合生成器:
permutations('ABCD', 2)
AB AC AD BA BC BD CA CB CD DA DB DC
有没有办法直接通过bash实现它？还是GNU并行本身？也许以某种方式跳过冗余工作？但是,如何检查已使用的参数组合.
parallel echo {= 'if($_==3) { skip() }' =} ::: {1..5}
我想实现以下函数,将数组的某些元素标记为1.
void mark(std::vector<signed char>& marker)
{
#pragma omp parallel for schedule(dynamic, M)
for (int i = 0; i < marker.size; i++)
marker[i] = 0;
#pragma omp parallel for schedule(dynamic, M)
for (int i = 0; i < marker.size; i++)
marker[getIndex(i)] = 1; // is it ok ?
}
如果我们尝试同时在不同的线程中将同一元素的值设置为1,会发生什么？它通常会设置为1还是这个循环可能导致意外行为？
目前,我在具有4个核心的边缘节点上运行Nifi.假设我有20个传入的流文件,并且我为ExecuteStreamCommand处理器提供10个并发任务,这是否意味着我只获得并发执行或并发和并行执行？
一个简单的程序，用于计算数字平方并存储结果：
import time
from joblib import Parallel, delayed
import multiprocessing
array1 = [ 0 for i in range(100000) ]
def myfun(i):
return i**2
#### Simple loop ####
start_time = time.time()
for i in range(100000):
array1[i]=i**2
print( "Time for simple loop --- %s seconds ---" % ( time.time()
- start_time
)
)
#### Parallelized loop ####
start_time = time.time()
results = Parallel( n_jobs = -1,
verbose = 0,
backend = "threading"
)(
map( delayed( myfun ),
range( 100000 )
)
) …我正在通过Parallel.ForEach处理各种大小的PDF(简单的2MB到几百MB的高DPI扫描)并且偶尔会遇到OutOfMemoryException - 可以理解的是由于进程是32位并且Parallel产生了线程. ForEach占用了大量未知的内存消耗工作.
限制MaxDegreeOfParallelism确实有效,尽管由于所述线程的内存占用量较小而导致有大量(10k +)批量的小型PDF需要处理时的吞吐量不足.这是一个CPU繁重的过程,在遇到偶尔的大型PDF组并获得OutOfMemoryException之前,Parallel.ForEach很容易达到100%的CPU.运行Performance Profiler会将其备份.
根据我的理解,为Parallel.ForEach设置分区器不会提高我的性能.
这导致我使用TaskScheduler传递给我的Parallel.ForEach 的自定义MemoryFailPoint检查.在它周围搜索似乎有关于创建自定义TaskScheduler对象的稀缺信息.
之间寻找在.NET专业任务调度4并行扩展附加功能,在C#中的自定义的TaskScheduler这里#2各种各样的回答,我已经建立了我自己的TaskScheduler,并有我的QueueTask方法,例如:
protected override void QueueTask(Task task)
{
lock (tasks) tasks.AddLast(task);
try
{
using (MemoryFailPoint memFailPoint = new MemoryFailPoint(600))
{
if (runningOrQueuedCount < maxDegreeOfParallelism)
{
runningOrQueuedCount++;
RunTasks();
}
}
}
catch (InsufficientMemoryException e)
{
// somehow return thread to pool?
Console.WriteLine("InsufficientMemoryException");
}
}
虽然try/catch有点贵,但我的目标是捕获600MB的可能最大大小PDF(+一点额外内存开销)将抛出OutOfMemoryException.当我捕获InsufficientMemoryException时,这个解决方案似乎杀掉了试图完成工作的线程.有了足够大的PDF,我的代码最终成为一个单一的线程Parallel.ForEach.
在Parallel.ForEach和OutOfMemoryExceptions上的Stackoverflow上发现的其他问题似乎不适合我在线程上使用动态内存的最大吞吐量的用例,并且通常只是MaxDegreeOfParallelism作为静态解决方案使用,例如:
因此,为了获得可变工作内存大小的最大吞吐量,可以:
MemoryFailPoint支票工作,我如何将线程返回到线程池中？我有一个mpi4py程序间歇性挂起。如何跟踪各个流程在做什么？
我可以在不同的终端上运行该程序，例如使用 pdb
mpiexec -n 4 xterm -e "python -m pdb my_program.py"
但是，如果问题仅通过大量进程（在我的情况下为〜80）表现出来，则将变得很麻烦。另外，很容易捕获异常，pdb但是我需要查看跟踪以找出发生挂起的位置。
我有疑问future(),doFuture()用法.
我想N并行运行计算(使用foreach ... %dopar%) - N我的机器上有多少核心.为此,我使用future:
library(doFuture)
registerDoFuture()
plan(multiprocess)
foreach(i = seq_len(N)) %dopar% {
foo <- rnorm(1e6)
}
这就像一个魅力,因为我N并行运行计算.但是我需要实现另一个使用大量内核的分析步骤(例如,N).这是代码的样子:
foreach(i = seq_len(N)) %dopar% {
foo <- rnorm(1e6)
write.table(foo, paste0("file_", i, ".txt"))
# This step uses high number of cores
system(paste0("head ", "file_", i, ".txt", " > ", "file_head_", i, ".txt")
}
我在运行多个rnorm与head并行,但由于head使用了大量的内核(让我们假设这个)我的分析卡住.
题:
如何使用并行运行部分代码future？(如何仅rnorm并行运行然后head …
我很感激PLYNQ专家的帮助!我会花时间回顾一下答案,我对math.SE有一个更为确定的概况.
我有一个类型的对象ParallelQuery<List<string>>,它有44个列表,我想并行处理(一次五个,比如说).我的流程有一个签名
private ProcessResult Process(List<string> input)
处理将返回一个结果,这是一对布尔值,如下所示.
private struct ProcessResult
{
public ProcessResult(bool initialised, bool successful)
{
ProcessInitialised = initialised;
ProcessSuccessful = successful;
}
public bool ProcessInitialised { get; }
public bool ProcessSuccessful { get; }
}
问题.给定一个IEnumerable<List<string>> processMe,我的PLYNQ查询尝试实现此方法:https://msdn.microsoft.com/en-us/library/dd384151(v = vs.110).aspx .它写成
processMe.AsParallel()
.Aggregate<List<string>, ConcurrentStack<ProcessResult>, ProcessResult>
(
new ConcurrentStack<ProcessResult>, //aggregator seed
(agg, input) =>
{ //updating the aggregate result
var res = Process(input);
agg.Push(res);
return agg;
},
agg =>
{ //obtain the result …我有以下代码用purr :: pmap进行串行处理
library(tidyverse)
set.seed(1)
params <- tribble(
~mean, ~sd, ~n,
5, 1, 1,
10, 5, 3,
-3, 10, 5
)
params %>%
pmap(rnorm)
#> [[1]]
#> [1] 4.373546
#>
#> [[2]]
#> [1] 10.918217 5.821857 17.976404
#>
#> [[3]]
#> [1] 0.2950777 -11.2046838 1.8742905 4.3832471 2.7578135
如何并行化(fork)上面的过程,以便它运行得更快并产生相同的结果？
在这里,我rnorm用于说明目的,实际上我有一个功能,做重型工作.它需要并行化.
我对非purrr(非tidyverse)解决方案持开放态度,只要它在给定rnorm函数和params输入时产生相同的结果.
我正在研究java 8并行流并且想要在并行流中打印元素是一些顺序(比如插入顺序,逆序或顺序顺序).
为此,我尝试了以下代码:
System.out.println("With forEachOrdered:");
listOfIntegers
.parallelStream()
.forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");
System.out.println("With Sequential:");
listOfIntegers.parallelStream()
.sequential()
.forEach(e -> System.out.print(e + " "));
对于这两个,我得到相同的输出如下:
With forEachOrdered:
1 2 3 4 5 6 7 8
With Sequential:
1 2 3 4 5 6 7 8
从api文档中,我可以看到:
forEachOrdered - >这是一个终端操作.
和
顺序 - >这是一个中间操作.
所以我的问题是哪一个更好用？在哪种情况下,一个应该优先于其他情况？