## ----setup, include = FALSE--------------------------------------------------- knitr::opts_chunk$set(collapse = TRUE, comment = "#>") library(rzig) compiler <- find_zig(required = FALSE) has_zig <- isTRUE(compiler$supported) ## ----create------------------------------------------------------------------- path <- file.path(tempdir(), "parallelpkg") create_rzig_package(path, tests = FALSE) main_zig <- file.path(path, "src", "rzig", "src", "main.zig") ## ----loop, comment = ""------------------------------------------------------- cat( "", "const Squares = struct {", " input: []const f64,", " output: []f64,", "", " fn run(state: *Squares, index: usize) void {", " state.output[index] = state.input[index] * state.input[index];", " }", "};", "", "/// The squares of the values, computed on worker threads.", "/// @param x A numeric vector.", "/// @return A numeric vector.", "/// @export", "pub fn squares(ctx: *rzig.Ctx, x: []const f64) rzig.Error![]f64 {", " const output = try ctx.alloc(f64, x.len);", " var state = Squares{ .input = x, .output = output };", " try rzig.parallelFor(ctx, x.len, &state, Squares.run);", " return output;", "}", sep = "\n", file = main_zig, append = TRUE ) ## ----reduce, comment = ""----------------------------------------------------- cat( "", "const Values = struct {", " values: []const f64,", "", " fn value(state: *const Values, index: usize) f64 {", " return state.values[index];", " }", "", " fn add(left: f64, right: f64) f64 {", " return left + right;", " }", "};", "", "/// The sum of the values, reduced on worker threads.", "/// @param x A numeric vector.", "/// @return A number.", "/// @export", "pub fn parallel_sum(ctx: *rzig.Ctx, x: []const f64) rzig.Error!f64 {", " const state = Values{ .values = x };", " return rzig.parallelReduce(ctx, f64, x.len, &state, 0, Values.value, Values.add);", "}", sep = "\n", file = main_zig, append = TRUE ) ## ----rng, comment = ""-------------------------------------------------------- cat( "", "/// Uniform draws from R's generator.", "/// @param n The number of draws.", "/// @return A numeric vector.", "/// @export", "pub fn r_uniforms(ctx: *rzig.Ctx, n: usize) rzig.Error![]f64 {", " const generator = try ctx.rng();", " const out = try ctx.alloc(f64, n);", " for (out) |*slot| slot.* = generator.uniform();", " return out;", "}", "", "const Draws = struct {", " streams: rzig.RngStreams,", " output: []f64,", "", " fn draw(state: *Draws, index: usize) void {", " var stream = state.streams.stream(index);", " state.output[index] = stream.normal();", " }", "};", "", "/// Normal draws on worker threads, one stream per index.", "/// @param n The number of draws.", "/// @return A numeric vector.", "/// @export", "pub fn stream_normals(ctx: *rzig.Ctx, n: usize) rzig.Error![]f64 {", " const output = try ctx.alloc(f64, n);", " var state = Draws{ .streams = try ctx.rngStreams(), .output = output };", " try rzig.parallelFor(ctx, n, &state, Draws.draw);", " return output;", "}", sep = "\n", file = main_zig, append = TRUE ) document(path) ## ----build, eval = has_zig---------------------------------------------------- build <- build_zig(path) library(build$package, character.only = TRUE, lib.loc = build$library) squares(c(1, 2, 3)) set.seed(1) x <- rnorm(1e5) Sys.setenv(RZIG_NUM_THREADS = "1") one <- parallel_sum(x) Sys.setenv(RZIG_NUM_THREADS = "8") eight <- parallel_sum(x) Sys.unsetenv("RZIG_NUM_THREADS") identical(one, eight) all.equal(one, sum(x)) set.seed(42) ours <- r_uniforms(3) set.seed(42) identical(ours, runif(3)) set.seed(7) first <- stream_normals(5) set.seed(7) identical(first, stream_normals(5)) ## ----cleanup, include = FALSE------------------------------------------------- if (has_zig) unloadNamespace("parallelpkg") unlink(path, recursive = TRUE)