Rhpc_intro_ja.R
In Rhpc: Apply-Style Dispatch for High-Performance Computing

### R code from vignette source 'Rhpc_intro_ja.Rnw'

###################################################
### code chunk number 1: Rhpc_intro_ja.Rnw:11-12
###################################################
old_pdf_options <- grDevices::pdf.options(family="Japan1")


###################################################
### code chunk number 2: Rhpc_intro_ja.Rnw:62-80 (eval = FALSE)
###################################################
## library(Rhpc)
## # Rhpc 環境の初期化
## Rhpc_initialize()
## 
## # ハンドルの取得（mpirun 下では引数なし、動的生成時は Rhpc_getHandle(4) など）
## cl <- Rhpc_getHandle()
## 
## # 全ワーカーで関数を実行し、マスターに結果を集約
## pids <- Rhpc_worker_call(cl, Sys.getpid)
## print(pids)
## 
## # 結果を返さずにワーカー上でコードを実行する場合
## Rhpc_worker_noback(cl, function() {
##     cat("worker process:", Sys.getpid(), "\n")
## })
## 
## # 終了処理
## Rhpc_finalize()


###################################################
### code chunk number 3: Rhpc_intro_ja.Rnw:98-112 (eval = FALSE)
###################################################
## library(Rhpc)
## oldoptions <- options(Rhpc.mpiexec = "mpiexec -n 1")
## Rhpc_initialize()
## cl <- Rhpc_getHandle(3)
## 
## worker_env_info <- Rhpc_worker_call(cl, function() {
##     list(
##         computer_name = Sys.getenv("COMPUTERNAME"),
##         username = Sys.getenv("USERNAME")
##     )
## })
## print(worker_env_info)
## Rhpc_finalize()
## options(oldoptions)


###################################################
### code chunk number 4: Rhpc_intro_ja.Rnw:129-205
###################################################
draw_comm_decision_flow <- function() {
    COL <- list(
        start   = "#E8F4FD",
        collect = "#D5F5E3",
        p2p     = "#FDEBD0",
        both    = "#E8DAEF",
        text    = "#2C3E50"
    )
    box <- function(x, y, w, h, label, fill, lwd = 1.5) {
        rect(x - w/2, y - h/2, x + w/2, y + h/2,
             col = fill, border = COL$text, lwd = lwd)
        text(x, y, label, cex = 0.78, col = COL$text)
    }
    arrow <- function(x1, y1, x2, y2, label = NULL) {
        segments(x1, y1, x2, y2, lwd = 1.4, col = COL$text)
        dx <- x2 - x1; dy <- y2 - y1
        len <- sqrt(dx^2 + dy^2)
        if (len > 0) {
            ux <- dx / len; uy <- dy / len
            arr <- 0.18
            arrows(x2 - ux * arr, y2 - uy * arr, x2, y2,
                   length = 0.08, lwd = 1.4, col = COL$text)
        }
        if (!is.null(label)) {
            text((x1 + x2) / 2 + 0.15, (y1 + y2) / 2, label, cex = 0.72, col = "#555555")
        }
    }

    oldpar <- par(mar = c(1, 1, 3, 1))
    plot(c(0, 10), c(0, 14), type = "n", axes = FALSE, xlab = "", ylab = "",
         main = "Rhpc: 集団通信 vs 個別通信の選択")

    box(5, 13, 4.2, 0.9, "Rhpc 関数呼び出し", COL$start)

    box(5, 11.2, 5.0, 0.9, "全ワーカーに同じ命令・関数を送る?", "#FFFFFF")
    arrow(5, 12.55, 5, 11.65)

    box(2.2, 9.2, 3.2, 1.0, "はい\n(MPI_Bcast)", COL$collect)
    box(7.8, 9.2, 3.2, 1.0, "いいえ\n(MPI_Isend 個別配布)", COL$p2p)
    arrow(4.2, 11.2, 2.2, 9.7, "同一")
    arrow(5.8, 11.2, 7.8, 9.7, "個別")

    box(2.2, 7.0, 3.4, 0.9, "結果をマスターに返す?", "#FFFFFF")
    arrow(2.2, 8.7, 2.2, 7.45)

    box(0.9, 5.0, 2.4, 0.9, "いいえ\nBcast のみ", COL$collect)
    box(3.5, 5.0, 2.6, 0.9, "はい\nBcast + P2P", COL$both)
    arrow(1.5, 6.55, 0.9, 5.45, "noback\nExport")
    arrow(2.9, 6.55, 3.5, 5.45, "worker_call\nlapply 結果")

    box(7.8, 7.0, 3.6, 0.9, "入力データを\nワーカー別に分割?", "#FFFFFF")
    arrow(7.8, 8.7, 7.8, 7.45)

    box(6.3, 5.0, 2.6, 0.9, "lapply\n(チャンク分割)", COL$p2p)
    box(9.3, 5.0, 2.6, 0.9, "lapplyLB\n(1要素ずつ)", COL$p2p)
    arrow(7.0, 6.55, 6.3, 5.45, "一括")
    arrow(8.6, 6.55, 9.3, 5.45, "逐次")

    box(5, 2.8, 8.5, 1.6,
        paste(
            "集団通信: 命令・関数・引数 (全員同一)  |  ",
            "個別通信: 入力チャンク・結果・終了信号 (ワーカーごと)",
            sep = "\n"
        ),
        "#F8F9F9", lwd = 1)
    arrow(0.9, 4.55, 2.5, 3.6)
    arrow(3.5, 4.55, 4.0, 3.6)
    arrow(6.3, 4.55, 5.5, 3.6)
    arrow(9.3, 4.55, 7.5, 3.6)

    legend("bottom", inset = 0.02, horiz = TRUE, bty = "n", cex = 0.75,
           legend = c("集団通信 (MPI_Bcast/Gather)", "個別通信 (MPI_Isend/Irecv)", "両方"),
           fill = c(COL$collect, COL$p2p, COL$both), border = COL$text)
    par(oldpar)
}
draw_comm_decision_flow()


###################################################
### code chunk number 5: Rhpc_intro_ja.Rnw:227-297
###################################################
draw_rhpc_seq_diagram <- function(title, events, nworkers = 3,
                                  xlim = c(0, 10), show_legend = TRUE) {
    COL <- list(c = "#27AE60", p = "#E67E22", e = "#3498DB", r = "#C0392B")
    lanes <- c("Master", paste0("W", seq_len(nworkers)))
    nl <- length(lanes)
    y <- rev(seq_len(nl))
    ym <- y[1]

    oldpar <- par(mar = c(3, 2, 3, 2))
    plot(xlim, c(0.3, nl + 0.8), type = "n", xaxt = "n", yaxt = "n",
         xlab = "time ->", ylab = "", main = title)
    for (i in seq_along(lanes)) {
        text(0.25, y[i], lanes[i], adj = 0, font = 2, cex = 0.85)
        segments(0.75, y[i], xlim[2] - 0.2, y[i], lty = 2, col = "#BBBBBB")
    }

    bcast_all <- function(x, label) {
        text(x, nl + 0.55, label, cex = 0.68, col = COL$c, font = 2)
        for (i in 2:nl) arrows(x, ym, x + 0.32, y[i], length = 0.05, col = COL$c, lwd = 1.4)
    }
    gather_all <- function(x, label) {
        text(x, nl + 0.55, label, cex = 0.68, col = COL$c, font = 2)
        for (i in 2:nl) arrows(x + 0.32, y[i], x, ym, length = 0.05, col = COL$c, lwd = 1.4)
    }
    p2p_m2w <- function(x1, x2, w, label) {
        wy <- y[w + 1]
        arrows(x1, ym, x2, wy, length = 0.05, col = COL$p, lwd = 1.4)
        text((x1 + x2) / 2, (ym + wy) / 2 + 0.22, label, cex = 0.6, col = COL$p)
    }
    p2p_w2m <- function(x1, x2, w, label) {
        wy <- y[w + 1]
        arrows(x1, wy, x2, ym, length = 0.05, col = COL$r, lwd = 1.4)
        text((x1 + x2) / 2, (ym + wy) / 2 - 0.22, label, cex = 0.6, col = COL$r)
    }
    exec_w <- function(x, w, label = "run") {
        wy <- y[w + 1]
        rect(x - 0.22, wy - 0.11, x + 0.22, wy + 0.11, col = COL$e, border = NA)
        text(x, wy, label, cex = 0.52, col = "white", font = 2)
    }
    exec_all <- function(x, label = "run") {
        for (i in seq_len(nworkers)) exec_w(x, i, label)
    }
    loop_box <- function(x1, x2, ylo, yhi, label) {
        rect(x1, ylo, x2, yhi, border = "#999999", col = NA, lty = 2)
        text((x1 + x2) / 2, yhi + 0.15, label, cex = 0.65, col = "#666666", font = 3)
    }
    note <- function(x, yy, label) {
        text(x, yy, label, cex = 0.65, col = "#555555")
    }

    for (ev in events) {
        switch(ev$type,
               bcast_all = bcast_all(ev$x, ev$label),
               gather_all = gather_all(ev$x, ev$label),
               p2p_m2w = p2p_m2w(ev$x1, ev$x2, ev$w, ev$label),
               p2p_w2m = p2p_w2m(ev$x1, ev$x2, ev$w, ev$label),
               exec_w = exec_w(ev$x, ev$w, ev$label %||% "run"),
               exec_all = exec_all(ev$x, ev$label %||% "run"),
               loop_box = loop_box(ev$x1, ev$x2, ev$ylo, ev$yhi, ev$label),
               note = note(ev$x, ev$y, ev$label))
    }

    if (show_legend) {
        legend(xlim[2] - 2.8, 0.45,
               legend = c("Collective", "P2P (send)", "Compute", "P2P (result)"),
               fill = c(COL$c, COL$p, COL$e, COL$r), border = NA, bty = "n", cex = 0.7)
    }
    par(oldpar)
}
`%||%` <- function(a, b) if (is.null(a)) b else a


###################################################
### code chunk number 6: Rhpc_intro_ja.Rnw:303-313
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_worker_noback",
    list(
        list(type = "bcast_all", x = 1.0, label = "Bcast: cmd (NORET)"),
        list(type = "bcast_all", x = 2.2, label = "Bcast: FUN+args"),
        list(type = "exec_all",  x = 3.8, label = "FUN()"),
        list(type = "note",      x = 5.5, y = 1.2, label = "(no result returned)")
    ),
    nworkers = 3
)


###################################################
### code chunk number 7: Rhpc_intro_ja.Rnw:320-332
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_Export",
    list(
        list(type = "bcast_all", x = 1.0, label = "Bcast: cmd (EXPORT)"),
        list(type = "bcast_all", x = 2.2, label = "Bcast: name+value"),
        list(type = "exec_all",  x = 3.8, label = "assign()"),
        list(type = "loop_box",  x1 = 0.7, x2 = 4.5, ylo = 0.55, yhi = 4.3,
             label = "repeat for each variable"),
        list(type = "note",      x = 5.8, y = 1.2, label = "(no result returned)")
    ),
    nworkers = 3
)


###################################################
### code chunk number 8: Rhpc_intro_ja.Rnw:339-353
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_worker_call",
    list(
        list(type = "bcast_all", x = 0.8,  label = "Bcast: cmd (RET)"),
        list(type = "bcast_all", x = 1.8,  label = "Bcast: FUN+args"),
        list(type = "exec_all",  x = 3.0,  label = "FUN()"),
        list(type = "gather_all", x = 4.2, label = "Gather: result size"),
        list(type = "p2p_w2m", x1 = 5.2, x2 = 5.8, w = 1, label = "Isend/Irecv"),
        list(type = "p2p_w2m", x1 = 5.5, x2 = 6.1, w = 2, label = "Isend/Irecv"),
        list(type = "p2p_w2m", x1 = 5.8, x2 = 6.4, w = 3, label = "Isend/Irecv"),
        list(type = "note",    x = 7.2, y = 1.2, label = "return list of results")
    ),
    nworkers = 3
)


###################################################
### code chunk number 9: Rhpc_intro_ja.Rnw:360-374
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_EvalQ",
    list(
        list(type = "bcast_all", x = 0.8,  label = "Bcast: cmd (RET)"),
        list(type = "bcast_all", x = 1.8,  label = "Bcast: eval+expr"),
        list(type = "exec_all",  x = 3.0,  label = "eval(expr)"),
        list(type = "gather_all", x = 4.2, label = "Gather: result size"),
        list(type = "p2p_w2m", x1 = 5.2, x2 = 5.8, w = 1, label = "Isend/Irecv"),
        list(type = "p2p_w2m", x1 = 5.5, x2 = 6.1, w = 2, label = "Isend/Irecv"),
        list(type = "p2p_w2m", x1 = 5.8, x2 = 6.4, w = 3, label = "Isend/Irecv"),
        list(type = "note",    x = 7.2, y = 1.2, label = "return list of results")
    ),
    nworkers = 3
)


###################################################
### code chunk number 10: Rhpc_intro_ja.Rnw:381-400
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_lapply",
    list(
        list(type = "bcast_all", x = 0.7, label = "Bcast: cmd+FUN+args"),
        list(type = "p2p_m2w", x1 = 1.6, x2 = 2.1, w = 1, label = "Isend: chunk1"),
        list(type = "p2p_m2w", x1 = 1.75, x2 = 2.25, w = 2, label = "Isend: chunk2"),
        list(type = "p2p_m2w", x1 = 1.9, x2 = 2.4, w = 3, label = "Isend: chunk3"),
        list(type = "exec_all",  x = 3.3, label = "lapply"),
        list(type = "p2p_w2m", x1 = 4.3, x2 = 4.9, w = 2, label = "Probe/Irecv"),
        list(type = "p2p_w2m", x1 = 4.8, x2 = 5.4, w = 1, label = "Probe/Irecv"),
        list(type = "p2p_w2m", x1 = 5.3, x2 = 5.9, w = 3, label = "Probe/Irecv"),
        list(type = "loop_box", x1 = 4.0, x2 = 6.2, ylo = 0.55, yhi = 4.3,
             label = "until all chunks done"),
        list(type = "p2p_m2w", x1 = 7.0, x2 = 7.35, w = 1, label = "exit"),
        list(type = "p2p_m2w", x1 = 7.0, x2 = 7.35, w = 2, label = "exit"),
        list(type = "p2p_m2w", x1 = 7.0, x2 = 7.35, w = 3, label = "exit")
    ),
    nworkers = 3, xlim = c(0, 10)
)


###################################################
### code chunk number 11: Rhpc_intro_ja.Rnw:407-429
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_lapplyLB",
    list(
        list(type = "bcast_all", x = 0.7, label = "Bcast: cmd+FUN+args"),
        list(type = "p2p_m2w", x1 = 1.5, x2 = 1.95, w = 1, label = "Isend: x[1]"),
        list(type = "p2p_m2w", x1 = 1.5, x2 = 1.95, w = 2, label = "Isend: x[2]"),
        list(type = "p2p_m2w", x1 = 1.5, x2 = 1.95, w = 3, label = "Isend: x[3]"),
        list(type = "exec_w",   x = 2.8, w = 1, label = "FUN"),
        list(type = "exec_w",   x = 2.8, w = 2, label = "FUN"),
        list(type = "exec_w",   x = 2.8, w = 3, label = "FUN"),
        list(type = "p2p_w2m", x1 = 3.5, x2 = 4.0, w = 1, label = "Irecv: res"),
        list(type = "p2p_m2w", x1 = 4.2, x2 = 4.65, w = 1, label = "Isend: x[4]"),
        list(type = "p2p_w2m", x1 = 5.0, x2 = 5.5, w = 3, label = "Irecv: res"),
        list(type = "p2p_m2w", x1 = 5.7, x2 = 6.15, w = 3, label = "Isend: x[5]"),
        list(type = "loop_box", x1 = 3.2, x2 = 6.5, ylo = 0.55, yhi = 4.3,
             label = "while tasks remain: send to idle worker, recv result"),
        list(type = "p2p_m2w", x1 = 7.5, x2 = 7.85, w = 1, label = "exit"),
        list(type = "p2p_m2w", x1 = 7.5, x2 = 7.85, w = 2, label = "exit"),
        list(type = "p2p_m2w", x1 = 7.5, x2 = 7.85, w = 3, label = "exit")
    ),
    nworkers = 3, xlim = c(0, 10)
)


###################################################
### code chunk number 12: Rhpc_intro_ja.Rnw:436-451
###################################################
draw_rhpc_seq_diagram(
    "Rhpc_setupRNG (via Rhpc_lapply)",
    list(
        list(type = "bcast_all", x = 0.8, label = "Bcast: cmd+FUN"),
        list(type = "p2p_m2w", x1 = 1.7, x2 = 2.15, w = 1, label = "seed[1]"),
        list(type = "p2p_m2w", x1 = 1.85, x2 = 2.3, w = 2, label = "seed[2]"),
        list(type = "p2p_m2w", x1 = 2.0, x2 = 2.45, w = 3, label = "seed[3]"),
        list(type = "exec_all",  x = 3.2, label = "set seed"),
        list(type = "p2p_w2m", x1 = 4.2, x2 = 4.75, w = 1, label = "done"),
        list(type = "p2p_w2m", x1 = 4.5, x2 = 5.05, w = 2, label = "done"),
        list(type = "p2p_w2m", x1 = 4.8, x2 = 5.35, w = 3, label = "done"),
        list(type = "note", x = 6.5, y = 1.2, label = "L'Ecuyer-CMRG stream per worker")
    ),
    nworkers = 3
)


###################################################
### code chunk number 13: Rhpc_intro_ja.Rnw:457-483
###################################################
draw_comm_comparison <- function() {
    funcs <- c("noback", "Export", "call", "EvalQ", "lapply", "lapplyLB", "setupRNG")
    bcast <- c(1, 1, 1, 1, 1, 1, 1)
    gather <- c(0, 0, 1, 1, 0, 0, 0)
    isend_in <- c(0, 0, 0, 0, 1, 1, 1)
    irecv_out <- c(0, 0, 1, 1, 1, 1, 1)

    mat <- rbind(
        "MPI_Bcast\n(命令・関数)" = bcast,
        "MPI_Gather\n(結果サイズ)" = gather,
        "MPI_Isend\n(入力/タスク)" = isend_in,
        "MPI_Irecv\n(結果回収)" = irecv_out
    )
    colnames(mat) <- funcs

    oldpar <- par(mar = c(6, 8, 3, 2))
    on.exit(par(oldpar))
    barplot(mat, beside = TRUE, col = c("#27AE60", "#8E44AD", "#E67E22", "#C0392B"),
            border = NA, las = 2, cex.names = 0.85,
            main = "関数別: 使用する MPI 通信方式",
            ylab = "使用 (1) / 不使用 (0)", ylim = c(0, 1.35))
    legend("top", inset = 0.02, horiz = TRUE, bty = "n", cex = 0.75,
           legend = rownames(mat),
           fill = c("#27AE60", "#8E44AD", "#E67E22", "#C0392B"))
}
draw_comm_comparison()


###################################################
### code chunk number 14: Rhpc_intro_ja.Rnw:539-549 (eval = FALSE)
###################################################
## Rhpc_initialize()
## cl <- Rhpc_getHandle()
## 
## input_data <- 1:10
## results <- Rhpc_lapply(cl, input_data, function(x) {
##     return(x^2)
## })
## print(unlist(results))
## 
## Rhpc_finalize()


###################################################
### code chunk number 15: Rhpc_intro_ja.Rnw:556-566 (eval = FALSE)
###################################################
## Rhpc_initialize()
## cl <- Rhpc_getHandle()
## input_data <- 1:10
## 
## results_lb <- Rhpc_lapplyLB(cl, input_data, function(x) {
##     Sys.sleep(runif(1, 0, 1))  # 実行時間のばらつきを模擬
##     return(x * 10)
## })
## print(unlist(results_lb))
## Rhpc_finalize()


###################################################
### code chunk number 16: Rhpc_intro_ja.Rnw:572-584 (eval = FALSE)
###################################################
## Rhpc_initialize()
## cl <- Rhpc_getHandle()
## 
## # sapply 相当
## res <- Rhpc_sapply(cl, 1:10000, sqrt)
## 
## # apply 相当
## df <- data.frame(a = 1:4, b = 5:8)
## Rhpc_apply(cl, df, 1, max)  # 行方向
## Rhpc_apply(cl, df, 2, max)  # 列方向
## 
## Rhpc_finalize()


###################################################
### code chunk number 17: Rhpc_intro_ja.Rnw:590-602 (eval = FALSE)
###################################################
## Rhpc_initialize()
## cl <- Rhpc_getHandle()
## 
## multiplier <- 10
## Rhpc_Export(cl, "multiplier")
## 
## res <- Rhpc_worker_call(cl, function() {
##     return(multiplier * 2)
## })
## print(res)
## 
## Rhpc_finalize()


###################################################
### code chunk number 18: Rhpc_intro_ja.Rnw:626-627
###################################################
do.call(grDevices::pdf.options, old_pdf_options)