.progress = 'text'の設定が大好きplyr's llplyです。ただし、リスト項目はさまざまなコアに送信され、最後に照合されるため、 mclapply(パッケージから)どこまで進んでいるかがわからないことが私の不安の原因です。multicore
私は次のようなメッセージを出力してきまし*currently in sim_id # ....*たが、リスト項目の何パーセントが完了したかを示す指標が得られないため、あまり役に立ちません (ただし、スクリプトが動かなくなったり動いたりしていないことを知ることは役に立ちます)。
.Rout自分のファイルを見て、進捗状況を把握できるようにする他のアイデアを提案してもらえますか? mclapply手動カウンターを追加することを考えましたが、フィードバックを出す前にすべてのリスト項目の処理を終了する必要があるため、それを実装する方法がわかりません。
mclapplyは複数のプロセスを生成するため、fifo、パイプ、さらにはソケットを使用したい場合があります。次の例を考えてみましょう。
library(multicore)
finalResult <- local({
f <- fifo(tempfile(), open="w+b", blocking=T)
if (inherits(fork(), "masterProcess")) {
# Child
progress <- 0.0
while (progress < 1 && !isIncomplete(f)) {
msg <- readBin(f, "double")
progress <- progress + as.numeric(msg)
cat(sprintf("Progress: %.2f%%\n", progress * 100))
}
exit()
}
numJobs <- 100
result <- mclapply(1:numJobs, function(...) {
# Dome something fancy here
# ...
# Send some progress update
writeBin(1/numJobs, f)
# Some arbitrary result
sample(1000, 1)
})
close(f)
result
})
cat("Done\n")
ここでは、一時ファイルが fifo として使用され、メイン プロセスは、現在の進行状況を報告することだけを義務とする子プロセスを fork します。メイン プロセスは、mclapply評価される式 (より正確には、式ブロック) が を使用して fifo バッファーに部分的な進行状況情報を書き込む を呼び出すことによって続行されwriteBinます。
これは単純な例にすぎないため、おそらく出力全体をニーズに合わせて調整する必要があります。チッ！
基本的に@fotNelsonのソリューションの別のバージョンを追加しますが、いくつかの変更があります:
parallelなく、multicoreCRAN から削除されたものを使用これが誰かに役立つことを願っています...
library(parallel)
#-------------------------------------------------------------------------------
#' Wrapper around mclapply to track progress
#'
#' Based on http://stackoverflow.com/questions/10984556
#'
#' @param X a vector (atomic or list) or an expressions vector. Other
#' objects (including classed objects) will be coerced by
#' ‘as.list’
#' @param FUN the function to be applied to
#' @param ... optional arguments to ‘FUN’
#' @param mc.preschedule see mclapply
#' @param mc.set.seed see mclapply
#' @param mc.silent see mclapply
#' @param mc.cores see mclapply
#' @param mc.cleanup see mclapply
#' @param mc.allow.recursive see mclapply
#' @param mc.progress track progress?
#' @param mc.style style of progress bar (see txtProgressBar)
#'
#' @examples
#' x <- mclapply2(1:1000, function(i, y) Sys.sleep(0.01))
#' x <- mclapply2(1:3, function(i, y) Sys.sleep(1), mc.cores=1)
#'
#' dat <- lapply(1:10, function(x) rnorm(100))
#' func <- function(x, arg1) mean(x)/arg1
#' mclapply2(dat, func, arg1=10, mc.cores=2)
#-------------------------------------------------------------------------------
mclapply2 <- function(X, FUN, ...,
mc.preschedule = TRUE, mc.set.seed = TRUE,
mc.silent = FALSE, mc.cores = getOption("mc.cores", 2L),
mc.cleanup = TRUE, mc.allow.recursive = TRUE,
mc.progress=TRUE, mc.style=3)
{
if (!is.vector(X) || is.object(X)) X <- as.list(X)
if (mc.progress) {
f <- fifo(tempfile(), open="w+b", blocking=T)
p <- parallel:::mcfork()
pb <- txtProgressBar(0, length(X), style=mc.style)
setTxtProgressBar(pb, 0)
progress <- 0
if (inherits(p, "masterProcess")) {
while (progress < length(X)) {
readBin(f, "double")
progress <- progress + 1
setTxtProgressBar(pb, progress)
}
cat("\n")
parallel:::mcexit()
}
}
tryCatch({
result <- mclapply(X, ..., function(...) {
res <- FUN(...)
if (mc.progress) writeBin(1, f)
res
},
mc.preschedule = mc.preschedule, mc.set.seed = mc.set.seed,
mc.silent = mc.silent, mc.cores = mc.cores,
mc.cleanup = mc.cleanup, mc.allow.recursive = mc.allow.recursive
)
}, finally = {
if (mc.progress) close(f)
})
result
}
このpbapplyパッケージは、一般的な場合にこれを実装しています (つまり、Unix ライクおよび Windows 上で、RStudio でも動作します)。と の両方pblapplyに引数pbsapplyがあります。ドキュメントclから：
引数により並列処理を有効にすることができ
clます。parLapplyがcl'<code>cluster' オブジェクトの場合にmclapply呼び出され、clが整数の場合に呼び出されます。プログレス バーを表示すると、メイン プロセスとノード/子プロセスの間の通信オーバーヘッドが、プログレス バーのない同等の関数と比較して増加します。getOption("pboptions")$type == "none"dopb()プログレスバーが無効になっている場合 (つまりである場合)、関数は元の同等の機能にフォールバックしますFALSE。interactive()これは、 ifFALSE(つまり、コマンド ライン R スクリプトから呼び出される)の場合の既定値です。
提供しないcl(または渡すNULL) 場合は、デフォルトの非並列lapplyが使用され、プログレス バーも含まれます。
これは、 @ fotNelton のソリューションに基づく関数で、通常 mcapply を使用する場所ならどこでも適用できます。
mcadply <- function(X, FUN, ...) {
# Runs multicore lapply with progress indicator and transformation to
# data.table output. Arguments mirror those passed to lapply.
#
# Args:
# X: Vector.
# FUN: Function to apply to each value of X. Note this is transformed to
# a data.frame return if necessary.
# ...: Other arguments passed to mclapply.
#
# Returns:
# data.table stack of each mclapply return value
#
# Progress bar code based on https://stackoverflow.com/a/10993589
require(multicore)
require(plyr)
require(data.table)
local({
f <- fifo(tempfile(), open="w+b", blocking=T)
if (inherits(fork(), "masterProcess")) {
# Child
progress <- 0
print.progress <- 0
while (progress < 1 && !isIncomplete(f)) {
msg <- readBin(f, "double")
progress <- progress + as.numeric(msg)
# Print every 1%
if(progress >= print.progress + 0.01) {
cat(sprintf("Progress: %.0f%%\n", progress * 100))
print.progress <- floor(progress * 100) / 100
}
}
exit()
}
newFun <- function(...) {
writeBin(1 / length(X), f)
return(as.data.frame(FUN(...)))
}
result <- as.data.table(rbind.fill(mclapply(X, newFun, ...)))
close(f)
cat("Done\n")
return(result)
})
}
@fotNelson の回答に基づいて、1 行ずつ印刷する代わりにプログレス バーを使用し、mclapply で外部関数を呼び出します。
library('utils')
library('multicore')
prog.indic <- local({ #evaluates in local environment only
f <- fifo(tempfile(), open="w+b", blocking=T) # open fifo connection
assign(x='f',value=f,envir=.GlobalEnv)
pb <- txtProgressBar(min=1, max=MC,style=3)
if (inherits(fork(), "masterProcess")) { #progress tracker
# Child
progress <- 0.0
while (progress < MC && !isIncomplete(f)){
msg <- readBin(f, "double")
progress <- progress + as.numeric(msg)
# Updating the progress bar.
setTxtProgressBar(pb,progress)
}
exit()
}
MC <- 100
result <- mclapply(1:MC, .mcfunc)
cat('\n')
assign(x='result',value=result,envir=.GlobalEnv)
close(f)
})
.mcfunc<-function(i,...){
writeBin(1, f)
return(i)
}
fifo 接続を .GlobalEnv に割り当てることは、mclapply 呼び出しの外部の関数から使用するために必要です。質問と以前の返信をありがとう、私はこれを行う方法をしばらく考えていました。