R Academy · レッスン

future フレームワーク

Q: 「future フレームワーク」レッスンは無料ですか？

はい。「future フレームワーク」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、R Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 R Academyコースには全4レッスンが含まれています。

Q: 「future フレームワーク」で何を学びますか？

plan(multisession) と future() を使って、非同期かつ移植性の高い並列処理を行います。 ブラウザで直接実行するハンズオンコードでR Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Q: R Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのR Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

Q: 「future フレームワーク」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

Q: このR Academyレッスンでコードを書いて実行できますか？

はい。すべてのR Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

plan(multisession) と future() を使って、非同期かつ移植性の高い並列処理を行います。

「future フレームワーク」はCoddyKit上の無料R Academyレッスンです。これはレッスン2/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはR Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 R Academyコースには全4レッスンが含まれています。

futureパッケージとは

futureパッケージは、Rで非同期プログラミングと並列プログラミングを行うための、統一された高レベルAPIを提供します。futureとは、後で利用可能になる値のプレースホルダーです。その値は、別のコアやマシンで計算される場合もあります。

library(future)

# A simple future: computation happens asynchronously
f <- future({
  Sys.sleep(0.5)  # simulate slow work
  42
})

cat('Future created, doing other work...\n')

# Retrieve the result (blocks until done)
result <- value(f)
cat('Result:', result, '\n')

plan(): バックエンドの選択

plan()は、それ以降に作成されるすべてのfutureの実行戦略を設定します。一般的な戦略には、sequential（デフォルト、シングルスレッド）、multisession（複数のRセッション、あらゆる環境で動作）、multicore（フォーク、Unix/macOSのみ）があります。

library(future)

# Default: sequential (no parallelism)
plan(sequential)
cat('Strategy:', class(plan())[1], '\n')

# Parallel with separate R sessions (works on Windows too)
plan(multisession, workers = 4)
cat('Strategy:', class(plan())[1], '\n')

# Forking (Unix/macOS only, lower overhead)
if (.Platform$OS.type != 'windows') {
  plan(multicore, workers = 4)
  cat('Strategy:', class(plan())[1], '\n')
}

# Reset to sequential
plan(sequential)

value(): 結果の待機

value(f)は、futureの計算が完了するまで現在のプロセスをブロックし、その後結果を返します。futureでエラーが発生した場合、value()は呼び出し元のセッションでそのエラーを再スローします。

library(future)
plan(multisession, workers = 2)

# Launch two slow tasks concurrently
f1 <- future({ Sys.sleep(0.3); sum(1:1000) })
f2 <- future({ Sys.sleep(0.3); prod(1:10) })

# Both run in parallel; total time ~ 0.3s not 0.6s
start <- proc.time()[['elapsed']]
v1 <- value(f1)
v2 <- value(f2)
elapsed <- proc.time()[['elapsed']] - start

cat('v1:', v1, ' v2:', v2, '\n')
cat('Elapsed:', round(elapsed, 2), 's\n')

plan(sequential)

%<-%演算子

%<-%演算子は、future()と遅延評価されるvalue()を組み合わせたシンタックスシュガーです。通常の代入のように見えますが、右辺は非同期で実行されます。値は、変数に初めてアクセスした時点で取得されます。

library(future)
plan(multisession, workers = 2)

# %<-% starts the computation immediately in the background
x %<-% {
  Sys.sleep(0.3)
  rnorm(5, mean = 10)
}

y %<-% {
  Sys.sleep(0.3)
  runif(5, min = 1, max = 5)
}

cat('Both running in background...\n')

# Accessing x or y here blocks until ready
cat('x:', round(x, 2), '\n')
cat('y:', round(y, 2), '\n')

plan(sequential)

グローバル変数の自動検出

futureの最も強力な機能の一つが、グローバル変数の検出です。futureのコードが呼び出し元の環境で参照している変数を自動的に見つけてワーカーに送信するため、手動でclusterExport()を呼び出す必要はありません。

library(future)
plan(multisession, workers = 2)

# These globals are auto-detected and sent to the worker
threshold <- 50
multiplier <- 3

f <- future({
  x <- threshold * multiplier  # uses both globals
  x + 1
})

cat('Result:', value(f), '\n')  # 151

# Inspect which globals were identified
future_obj <- future({
  threshold + multiplier
}, lazy = TRUE)
cat('Globals found:', paste(names(future::getGlobalsAndPackages(future_obj)$globals), collapse = ', '), '\n')

plan(sequential)

future_options: 動作の調整

future.options（options()またはfuture::plan()の引数で設定）は、futureごとに許可される最大メモリ量、グローバル変数の検出方法、タイムアウトの動作など、グローバル設定を制御します。

library(future)

# Cap memory per future at 500 MB
options(future.globals.maxSize = 500 * 1024^2)  # bytes

# Disable automatic global detection (manual control)
options(future.globals.onReference = 'error')

# Inspect current settings
cat('Max globals size (MB):',
    getOption('future.globals.maxSize') / 1024^2, '\n')

# Restore defaults
options(future.globals.maxSize = 500 * 1024^2)
options(future.globals.onReference = NULL)

plan(multisession, workers = 2)

# Large object: you'd get an error if it exceeds the cap
big_vec <- 1:1e5  # small enough
f <- future(sum(big_vec))
cat('Sum:', value(f), '\n')

plan(sequential)

ネストしたFuture

Futureはネストできます。つまり、future自体が内部futureを作成できます。デフォルトでは、内部futureはplan(list(...))でネストしたプランを設定しない限り、逐次実行されます。これは2段階の並列処理に便利です。

library(future)

# Two-level parallelism: outer uses multisession, inner sequential
plan(list(multisession, sequential))

outer_futures <- lapply(1:3, function(i) {
  future({
    # Each outer worker runs its own sequential inner work
    inner <- lapply(1:4, function(j) i * j)
    unlist(inner)
  })
})

results <- lapply(outer_futures, value)
for (i in seq_along(results)) {
  cat('Outer', i, ':', results[[i]], '\n')
}

plan(sequential)

Futureでのエラー処理

futureの計算でエラーが発生すると、value()がそのエラーを再スローします。value()をtryCatch()でラップすると、パイプライン全体を停止せずに、個々のfutureのエラーを処理できます。

library(future)
plan(multisession, workers = 2)

# This future will error
bad_future <- future(log(-1, base = 'oops'))  # invalid arg
good_future <- future(sqrt(144))

# Handle errors gracefully
bad_result <- tryCatch(
  value(bad_future),
  error = function(e) {
    cat('Caught error:', conditionMessage(e), '\n')
    NA
  }
)

good_result <- value(good_future)
cat('bad_result:', bad_result, '\n')
cat('good_result:', good_result, '\n')

plan(sequential)

Futureの状態確認

resolved(f)は、futureがブロックせずに完了していればTRUEを返します。これにより、完了するまでポーリングしながら別の有用な処理を行う、ノンブロッキングのポーリングループを作成できます。

library(future)
plan(multisession, workers = 2)

f <- future({
  Sys.sleep(0.5)
  'done'
})

# Poll without blocking
counter <- 0
while (!resolved(f)) {
  counter <- counter + 1
  Sys.sleep(0.1)
  cat('Polling... (', counter, ')\n')
}

cat('Future resolved! Result:', value(f), '\n')
cat('Polled', counter, 'times\n')

plan(sequential)

futureとlapplyの組み合わせ

lapply()とfuture()を組み合わせるとN個のタスクを分散実行でき、lapply()とvalue()を組み合わせると結果を収集できます。このパターンでは、各futureを開始するタイミングを明示的に制御できます。

library(future)
plan(multisession, workers = 4)

# Fan out: create all futures
params <- list(
  list(n = 1000, mean = 0),
  list(n = 1000, mean = 5),
  list(n = 1000, mean = 10),
  list(n = 1000, mean = -3)
)

futures <- lapply(params, function(p) {
  future({
    x <- rnorm(p$n, mean = p$mean)
    list(mean = mean(x), sd = sd(x))
  })
})

# Fan in: collect results
results <- lapply(futures, value)
for (i in seq_along(results)) {
  cat('Param mean', params[[i]]$mean,
      '-> observed mean:', round(results[[i]]$mean, 3), '\n')
}

plan(sequential)

実用例: CV foldの評価

交差検証は、各foldが独立しているため、非常に並列化しやすい問題です。futureを使うとK個のfoldを同時に評価でき、計算コストの高いモデルにかかる経過時間を大幅に短縮できます。

library(future)
plan(multisession, workers = 4)

set.seed(42)
data <- data.frame(x = rnorm(200), y = rnorm(200))
fold_ids <- sample(rep(1:4, 50))

# Launch all folds in parallel
fold_futures <- lapply(1:4, function(k) {
  future({
    train <- data[fold_ids != k, ]
    test  <- data[fold_ids == k, ]
    fit   <- lm(y ~ x, data = train)
    preds <- predict(fit, newdata = test)
    rmse  <- sqrt(mean((test$y - preds)^2))
    rmse
  })
})

rmses <- unlist(lapply(fold_futures, value))
cat('Fold RMSEs:', round(rmses, 4), '\n')
cat('Mean RMSE:', round(mean(rmses), 4), '\n')

plan(sequential)

確認問題

WindowsとmacOSの両方で動作する並列コードを実行したいとします。どのplan()戦略を選ぶべきでしょうか。

まとめ: futureフレームワーク

重要なポイント:

future({expr})は式を非同期で実行し、value(f)は結果を取得します
%<-%はシンタックスシュガーであり、遅延future評価を伴う代入です
クロスプラットフォームではplan(multisession)、Unix/macOSでのフォークにはplan(multicore)を使用します
グローバル変数は自動検出されてワーカーに送信されるため、clusterExport()は必要ありません
resolved(f)はブロックせずに完了を確認します
エラーに強いパイプラインには、value()をtryCatch()でラップします
処理が終わったら、必ずplan(sequential)またはplan()を呼び出して設定をリセットします

library(future)
plan(multisession, workers = 2)

# Concise pattern: fan-out then fan-in
task <- function(i) future({ i^2 + i })
results <- lapply(1:6, task)
cat(unlist(lapply(results, value)), '\n')  # 2 6 12 20 30 42

plan(sequential)

よくある質問