假设我有一个Stream<Callable<SomeClass>> stream;.该流正在访问超过一百万个不适合内存的对象.
将此转换为a的惯用方法是什么Stream<SomeClass>,以确保在Callable::call传递给非线程安全的消费者(可能通过调用.sequential().forEach()或其他一些瓶颈机制)之前并行执行？
即并行处理流,但顺序传递输出(随机顺序ok,只要它是单线程).
我知道我可以通过在原始流和消费者之间建立一个ExecutionService和一个来做我想做的事Queue.但这似乎是很多代码,是否有一个神奇的单行程？
我想用来rvest抓取网页。它工作正常，但并行执行失败。
例
library(rvest)
library(dplyr)
LINKS <- read_html("https://stackoverflow.com/") %>%
html_nodes(".question-hyperlink") %>%
html_attr(name = "href") %>%
paste("https://stackoverflow.com", ., sep = "")
Get_values <- function(x){
RES <- read_html(x) %>%
html_nodes(".label-key") %>%
html_text()
}
工作正常
DATA <- lapply(LINKS[1:10], Get_values) #works fine
返回NULL
library(parallel)
DATA <- mclapply(LINKS[1:10], Get_values, mc.cores = 2) #returns NULL
我想运行aggregate函数中的dmapply通过所提供的功能ddR包.
期望的结果反映了通过aggregatebase 生成的简单输出:
aggregate(
x = mtcars$mpg,
FUN = function(x) {
mean(x, na.rm = TRUE)
},
by = list(trans = mtcars$am)
)
产生:
trans x
1 0 17.14737
2 1 24.39231
ddmapply我希望在使用时得到相同的结果ddmapply,如下所示:
# ddR
require(ddR)
# ddR object creation
distMtcars <- as.dframe(mtcars)
# Aggregate / ddmapply
dmapply(
FUN = function(x, y) {
aggregate(FUN = mean(x, na.rm = TRUE),
x = x,
by = list(trans …parallel-processing aggregate r distributed-computing dataframe
在柏林的Delphi 10.1中，我想添加一个可能性来停止我的问题中的响应式TParallel。＆For循环。如何使TParallel。＆For循环响应并将值存储在TList <T>中？。
循环计算值并将这些值存储在TList中。它与TTask.Run在单独的线程中运行以使其响应：
type
TCalculationProject=class(TObject)
private
Task: ITask;
...
public
List: TList<Real>;
...
end;
procedure TCalculationProject.CancelButtonClicked;
begin
if Assigned(Task) then
begin
Task.Cancel;
end;
end;
function TCalculationProject.CalculateListItem(const AIndex: Integer): Real;
begin
//a function which takes a lot of calculation time
//however in this example we simulate the calculation time and
//use a simple alogorithm to verify the list afterwards
Sleep(30);
Result:=10*AIndex;
end;
procedure TCalculationProject.CalculateList;
begin
List.Clear;
if Assigned(Task) then
begin
Task.Cancel;
end;
Task:=TTask.Run(
procedure
var
LoopResult: TParallel.TLoopResult;
Lock: …delphi parallel-processing multithreading thread-safety wait
C++ 17为标准库添加了并行扩展(例如std::sort(std::execution::par_unseq, arr, arr + 1000),允许使用多个线程和向量指令进行排序).
我注意到微软的实验性实现提到VC++编译器缺乏对这里做矢量化的支持,这让我感到惊讶 - 我认为现代C++编译器能够推断出循环的可矢量化,但显然VC++编译器/优化器无法生成SIMD代码即使明确告知这样做.看似缺乏自动矢量化支持与2011年关于Quora的问题的答案相矛盾,这表明编译器将在可能的情况下进行矢量化.
也许,编译器只会对非常明显的情况进行矢量化,例如a std::array<int, 4>,而且只不过是这样,因此C++ 17的显式并行化会很有用.
因此我的问题是:当没有明确告知这样做时,当前的编译器会自动向量化我的代码吗？(为了使这个问题更具体,让我们将其缩小到支持SIMD的Intel x86 CPU,以及最新版本的GCC,Clang,MSVC和ICC.)
作为扩展:其他语言的编译器是否可以做更好的自动矢量化(可能是由于语言设计)(因此C++标准委员会认为它对于显式(C++ 17风格)矢量化是必要的)？
parallel-processing simd vectorization auto-vectorization c++17
我尝试创建将两个不同的空间数据集组合在一起的摘要统计信息：一个大的栅格文件和一个多边形文件。这个想法是获得每个面内栅格值的摘要统计信息。
由于栅格太大而无法一次处理，因此我尝试创建子任务并并行处理它们，即一次处理每个多边形SpatialPolgyonsDataframe。
该代码工作正常，但是经过大约100次交互后，我遇到了内存问题。这是我的代码以及我打算做的事情：
# session setup
library("raster")
library("rgdal")
# multicore processing.
library("foreach")
library("doSNOW")
# assign three clusters to be used for current R session
cluster = makeCluster(3, type = "SOCK",outfile="")
registerDoSNOW(cluster)
getDoParWorkers()# check if it worked
# load base data
r.terra.2008<-raster("~/terra.tif")
spodf.malha.2007<-readOGR("~/,"composed")
# bring both data-sets to a common CRS
proj4string(r.terra.2008)
proj4string(spodf.malha.2007)
spodf.malha.2007<-spTransform(spodf.malha.2007,CRSobj = CRS(projargs = proj4string(r.terra.2008)))
proj4string(r.terra.2008)==proj4string(spodf.malha.2007) # should be TRUE
# create a function to extract areas
function.landcover.sum<-function(r.landuse,spodf.pol){
return(table(extract(r.landuse,spodf.pol)))}
# apply it one one subset to …我正在运行一堆shell脚本,
parallel -a my_scripts bash
并且在某些时候我决定我已经运行了足够多的东西,并希望停止产生新的工作,并简单地让所有现有的工作完成.换句话说,我想杀死父进程而不杀死孩子.
在第一次启动GNU并行时似乎有控制终止的方法(例如,如果我事先知道我只想运行x作业,那么我可以使用--halt now,success=x参数),但是我已经找不到如何控制GNU并行运行.
当然我可以只是CTRL+C杀死并行,并重新运行中止的工作,但我认为可能有一个更聪明的方法.
ECMA-2017(ES8)刚刚在一个月前完成,它引入了SharedArrayBuffer和Atomics.此处的链接表明它们已在某些浏览器中得到支持.
我们知道,它们旨在允许跨线程共享数据.我想知道在浏览器和Node中如何实现这种并行性？我们应该分别使用Web Workers和'cluster'包吗？
javascript parallel-processing multithreading shared-memory ecmascript-2017
我在这个主题上看过其他一些帖子,似乎没有一个与我遇到的问题完全相同.但是这里:
我正在使用并行运行一个函数
cores <- detectCores()
cl <- makeCluster(8L,outfile="output.txt")
registerDoParallel(cl)
x <- foreach(i = 1:length(y), .combine='list',.packages=c('httr','jsonlite'),
.multicombine=TRUE,.verbose=F,.inorder=F) %dopar% {function(y[i])}
这通常工作正常,但现在抛出错误:
序列化错误(数据,节点$ con):写入连接时出错
检查output.txt文件后,我看到:
starting worker pid=11112 on localhost:11828 at 12:38:32.867
starting worker pid=10468 on localhost:11828 at 12:38:33.389
starting worker pid=4996 on localhost:11828 at 12:38:33.912
starting worker pid=3300 on localhost:11828 at 12:38:34.422
starting worker pid=10808 on localhost:11828 at 12:38:34.937
starting worker pid=5840 on localhost:11828 at 12:38:35.435
starting worker pid=8764 on localhost:11828 at 12:38:35.940
starting worker pid=7384 on localhost:11828 at 12:38:36.448
Error in …在上市波纹管,我希望当我打电话t.detach()时创建线程行之后,该线程t将在后台运行,而printf("quit the main function now \n")会叫,然后main将退出.
#include <thread>
#include <iostream>
void hello3(int* i)
{
for (int j = 0; j < 100; j++)
{
*i = *i + 1;
printf("From new thread %d \n", *i);
fflush(stdout);
}
char c = getchar();
}
int main()
{
int i;
i = 0;
std::thread t(hello3, &i);
t.detach();
printf("quit the main function now \n");
fflush(stdout);
return 0;
}
然而,从它在屏幕上打印的内容来看并非如此.它打印
From new thread 1
From new thread …