每当我在R中看到关于并行处理的问题时,它就会使用该foreach函数.由于for循环不是很像R,是否有并行版本apply,如果是这样,为什么它不是更受欢迎？
背景:
我的工作需要能够捕捉程序stdout,stderr并返回程序的值.理想情况下,我想在我存储在我的对象中的字符串中捕获这些字符串,该字符串包含进程的详细信息.我目前有一些代码,通过使用一些(在我看来)古老的C文件句柄魔术将输出保存到文件中.任何时候我想输出结果,我打开该文件,然后打印内容.
有时(当我生成的进程继续运行时)我的可执行文件的下一次执行将会中断,因为它无法打开文件进行写入.
问题陈述:
我正在寻找一种方法来将stdout窗口中创建的进程的输出保存为一个字符串,并stderr以更安全,更现代的方式保存到另一个字符串.这样我就可以在每次输出每个创建过程的结果时打印这些内容.
我丑陋的代码:
主要块 -
int stdoutold = _dup(_fileno(stdout)); //make a copy of stdout
int stderrold = _dup(_fileno(stdout)); //make a copy of stderr
FILE *f;
if(!fopen_s(&f, "name_of_my_file", "w")){ //make sure I can write to the file
_dup2(_fileno(f), _fileno(stdout)); //make stdout point to f
_dup2(_fileno(f), _fileno(stderr)); //make stderr point to f
fork("command_I_want_to_run", &pi); //run my fake fork (see below)
}
else{
...//error handling
}
_close(_fileno(stdout)); //close tainted stdout
_close(_fileno(stderr)); //close …好吧,如果这感觉像是重复旧问题,我很抱歉,我已经通过tanenbaum对Stack Overflow,现代操作系统手册进行了几个问题,并且仍然要清除我对此的怀疑.
首先,我将非常感谢我应该更详细地阅读的任何书籍/资源,以便更好地理解这种结构.我不明白这些是OS书籍或编程语言或架构书籍中通常解释的概念.
在我提出问题之前,我会根据有关堆栈/堆的读数列出我的发现
堆
堆
现在,关于我的一些问题.
我知道这很多,而且我似乎总是非常困惑,如果你能指出我正确的方向让这些事情得到澄清,我将不胜感激!
c++ parallel-processing memory-management heap-memory stack-memory
如何将以下内容转换为Parallel.ForEach？
public async void getThreadContents(String[] threads)
{
HttpClient client = new HttpClient();
List<String> usernames = new List<String>();
int i = 0;
foreach (String url in threads)
{
i++;
progressLabel.Text = "Scanning thread " + i.ToString() + "/" + threads.Count<String>();
HttpResponseMessage response = await client.GetAsync(url);
String content = await response.Content.ReadAsStringAsync();
String user;
Predicate<String> userPredicate;
foreach (Match match in regex.Matches(content))
{
user = match.Groups[1].ToString();
userPredicate = (String x) => x == user;
if (usernames.Find(userPredicate) != user)
{
usernames.Add(match.Groups[1].ToString());
}
}
progressBar1.PerformStep();
} …我已经看到了很多关于写入文件的问题,但我想知道打开文本文件最有效的方法是什么,附加一些数据然后在你要从多个连接写入时再次关闭它(即并行计算情况),并不能保证每个连接何时都要写入文件.
例如,在下面的玩具示例中,它只使用我桌面上的核心,它似乎工作正常,但我想知道如果写入时间越长并且写入文件的进程数量增加,此方法是否容易失败(特别是在可能存在延迟的网络共享中).
任何人都可以建议一种强大的,明确的方式,当可能有其他想要同时写入文件的从属进程时,应该打开,写入然后关闭连接吗？
require(doParallel)
require(doRNG)
ncores <- 7
cl <- makeCluster( ncores , outfile = "" )
registerDoParallel( cl )
res <- foreach( j = 1:100 , .verbose = TRUE , .inorder= FALSE ) %dorng%{
d <- matrix( rnorm( 1e3 , j ) , nrow = 1 )
conn <- file( "~/output.txt" , open = "a" )
write.table( d , conn , append = TRUE , col.names = FALSE )
close( conn )
}
我正在寻找最好的 …
我有一个非常庞大的字符串向量,并希望使用foreach和dosnow包进行并行计算.我注意到foreach会为每个进程复制矢量,从而快速耗尽系统内存.我试图将矢量分解为列表对象中的较小部分,但仍然没有看到任何内存使用量减少.有没有人有这个想法？以下是一些演示代码:
library(foreach)
library(doSNOW)
library(snow)
x<-rep('some string', 200000000)
# split x into smaller pieces in a list object
splits<-getsplits(x, mode='bysize', size=1000000)
tt<-vector('list', length(splits$start))
for (i in 1:length(tt)) tt[[i]]<-x[splits$start[i]: splits$end[i]]
ret<-foreach(i = 1:length(splits$start), .export=c('somefun'), .combine=c) %dopar% somefun(tt[[i]])
在多CPU机器中,不同的CPU是否竞争相同的内存带宽,还是独立访问DRAM？
换句话说,如果一个程序的内存带宽有限,例如1-CPU 8核系统,那么转向4-CPU 4*8核机器就有机会加速它(假设CPU和DRAM)是可比的)？
我正在编写一个执行一些长计算的程序,我可以根据需要分成任意数量的任务.为了便于讨论,让我们假设我正在编写一种算法,通过尝试将它除以2和p-1之间的所有数来找出数p是否为素数.显然,这个任务可以分解为许多线程.
我实际上写了一个样本应用程序就是这样做的.作为一个参数,我给出了我想要检查的数字,以及要使用的线程数(每个线程都有一个相同大小的数字范围,试图将p除以 - 它们一起覆盖整个范围).
我的机器有8个核心.我开始使用大量的程序运行程序,我知道它是素数(2971215073),并且有1,2,3个线程等,直到达到8个线程 - 每次程序运行速度比前一个快,这是我的预期.但是,当我尝试大于8的数字时,计算时间实际上变得越来越小(即使是一点点)!
在我的线程中没有I/O或类似的东西,只是纯粹的cpu计算.当我通过8个线程时,我预计运行时间会变得更糟,因为会有更多的上下文切换,并行运行线程的数量保持在8个.很难说峰值在哪里因为差异很小而且变化很大从一次运行到另一次运行,但很明显,即50个线程以某种方式运行速度超过8(约300毫秒)......
我的猜测是因为我有这么多线程,所以我得到更多的运行时间,因为我在系统的线程池中有更大的部分,所以我的线程被选中更多.但是,我创建的线程越多,程序运行得越快就越有意义(否则为什么不是每个人都创建1000个线程？).
任何人都可以提供一个解释,也许是最佳实践,关于创建相对于机器上的核心数量的线程数量？
谢谢.
我感兴趣的人的代码(在Windows上编译,VS2012):
#include <Windows.h>
#include <conio.h>
#include <iostream>
#include <thread>
#include <vector>
using namespace std;
typedef struct
{
unsigned int primeCandidate;
unsigned int rangeStart;
unsigned int rangeEnd;
} param_t;
DWORD WINAPI isDivisible(LPVOID p)
{
param_t* param = reinterpret_cast<param_t*>(p);
for (unsigned int d = param->rangeStart; d < param->rangeEnd; ++d)
{
if (param->primeCandidate % d == 0)
{
cout << param->primeCandidate << " is divisible by " << d << endl;
return …c++ parallel-processing multithreading distributed-computing threadpool
我一直在学习Cuda,我仍然在处理并行问题.我目前遇到的问题是对一组值实现最大减少.这是我的内核
__global__ void max_reduce(const float* const d_array,
float* d_max,
const size_t elements)
{
extern __shared__ float shared[];
int tid = threadIdx.x;
int gid = (blockDim.x * blockIdx.x) + tid;
if (gid < elements)
shared[tid] = d_array[gid];
__syncthreads();
for (unsigned int s=blockDim.x/2; s>0; s>>=1)
{
if (tid < s && gid < elements)
shared[tid] = max(shared[tid], shared[tid + s]);
__syncthreads();
}
if (gid == 0)
*d_max = shared[tid];
}
我已经使用相同的方法(用min替换max函数)实现了min reduce,这很好.
为了测试内核,我使用串行for循环找到了最小值和最大值.最小值和最大值在内核中总是相同,但只有min reduce匹配.
有什么明显的东西我错过了/做错了吗？
我试图parLapply在全局环境中未定义的另一个函数内部使用.worker函数使用我想要的其他函数列表,这些函数clusterExport也未在全局环境中定义.我的问题是两个函数都将它们的评估环境导出到集群,这些集群很庞大而且不需要.
让我们调用worker函数workerFunction和函数列表functionList.
workerFunction <- function(i) {
intermediateOutput <- functionList[[i]](y)
result <- otherCalculations(intermediateOutput)
return(result)
}
library(parallel)
cl <- makeCluster(detectCores())
environment(workerFunction) <- .GlobalEnv
environment(functionList) <- .GlobalEnv
clusterExport(cl, varlist=c("functionList", "y"), envir=.GlobalEnv)
output <- parLapply(cl, inputVector, workerFunction)
我明白了:
Error in get(name, envir = envir) (from <text>#53) : object 'functionList' not found
如果我没有设置environment(functionList) <- .GlobalEnv,则将巨大的封闭环境functionList导出到集群.为什么R不能functionList在全球环境中找到？