test-parallel.R
In TrialSimulator: Clinical Trial Simulator

# Parallelization
#
# Covers:
#   - Reproducibility: running n_workers > 1 with seeds extracted from a
#     multi-replicate run reproduces identical results under n_workers = 1.
#     This verifies that the Mersenne-Twister stream reset (instead of
#     L'Ecuyer-CMRG) keeps single- and multi-process outputs in sync.

test_that('package behaves the same under single- and multi-process modes with same seed', {

  # covr cannot reliably collect coverage traces from the worker R processes
  # spawned by n_workers > 1; under coverage that yields a truncated trace which
  # breaks covr's merge step (readRDS: error reading from connection). The
  # parallel path is still exercised in ordinary (non-coverage) test runs.
  skip_if(Sys.getenv("R_COVR") == "true",
          "n_workers > 1 spawns R processes that covr cannot trace")

  foo <- function(n, n_workers, seed = NULL){
    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/10)
    os <- endpoint(name = 'os', type = 'tte', generator = rexp, rate = log(2)/17)
    or <- endpoint(name = 'or', type = 'non-tte', readout = c(or = 1), generator = rnorm)

    pbo <- arm(name = 'pbo')
    pbo$add_endpoints(pfs, os, or)

    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/12)
    os <- endpoint(name = 'os', type = 'tte', generator = rexp, rate = log(2)/18)
    or <- endpoint(name = 'or', type = 'non-tte', readout = c(or = 1), generator = rnorm)

    trt1 <- arm(name = 'trt1')
    trt1$add_endpoints(pfs, os, or)


    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/13)
    os <- endpoint(name = 'os', type = 'tte', generator = rexp, rate = log(2)/18.5)
    or <- endpoint(name = 'or', type = 'non-tte', readout = c(or = 1), generator = rnorm)

    trt2 <- arm(name = 'trt2')
    trt2$add_endpoints(pfs, os, or)


    accrual_rate <- data.frame(end_time = c(1, 2, 6, 12, Inf),
                               piecewise_rate = c(2, 8, 20, 25, 50))

    trial <- trial(
      name = 'test', n_patients = 1000, duration = 40,
      enroller = StaggeredRecruiter, accrual_rate = accrual_rate,
      dropout = rweibull, shape = 1.32, scale = 114.4,
      seed = seed,
      silent = TRUE
    )

    trial$add_arms(sample_ratio = c(1, 1, 2), pbo, trt1, trt2)

    interim1 <- milestone(name = 'interim1',
                          when = eventNumber(endpoint = 'or', n = 200),
                          action = function(trial){trial$remove_arms('trt1')})

    action2 <- function(trial){

      locked_data <- trial$get_locked_data('interim2')
      fit1 <- fitLogrank(Surv(pfs, pfs_event) ~ arm,
                         placebo = 'pbo',
                         data = locked_data,
                         alternative = 'less')
      fit2 <- fitCoxph(Surv(os, os_event) ~ arm,
                       placebo = 'pbo',
                       data = locked_data,
                       alternative = 'less',
                       scale = 'hazard ratio',
                       arm != 'trt1')
      fit3 <- fitLinear(or ~ arm,
                        placebo = 'pbo',
                        data = locked_data,
                        alternative = 'less', arm != 'trt2')
      trial$save(fit1$p[1], 'IA_pval11')
      trial$save(fit1$p[2], 'IA_pval12')
      trial$save(fit2$p[1], 'IA_pval2')
      trial$save(fit3$p[1], 'IA_pval3')
    }

    interim2 <- milestone(name = 'interim2',
                          when = eventNumber(endpoint = 'pfs', n = 240) &
                            eventNumber(endpoint = 'os', n = 170),
                          action = action2)

    action3 <- function(trial){

      locked_data <- trial$get_locked_data('final')
      fit1 <- fitLogrank(Surv(pfs, pfs_event) ~ arm,
                         placebo = 'pbo',
                         data = locked_data,
                         alternative = 'less')
      fit2 <- fitCoxph(Surv(os, os_event) ~ arm,
                       placebo = 'pbo',
                       data = locked_data,
                       alternative = 'less',
                       scale = 'hazard ratio',
                       arm != 'trt1')
      fit3 <- fitLinear(or ~ arm,
                        placebo = 'pbo',
                        data = locked_data,
                        alternative = 'less', arm != 'trt2')
      trial$save(fit1$p[1], 'FA_pval11')
      trial$save(fit1$p[2], 'FA_pval12')
      trial$save(fit2$p[1], 'FA_pval2')
      trial$save(fit3$p[1], 'FA_pval3')
    }

    final <- milestone(name = 'final',
                       when = calendarTime(time = 40),
                       action = action3)

    listener <- listener(silent = TRUE)
    listener$add_milestones(interim1, interim2, final)

    controller <- controller(trial, listener)
    controller$run(n = n, n_workers = n_workers, plot_event = FALSE, silent = TRUE)
    controller$get_output(c('seed', 'IA_pval11', 'IA_pval12', 'IA_pval2', 'IA_pval3',
                            'FA_pval11', 'FA_pval12', 'FA_pval2', 'FA_pval3'))
  }

  op2 <- foo(n = 10, n_workers = 2, seed = NULL)

  op1 <- NULL
  for(i in 1:10){
    op1 <- rbind(op1, foo(n = 1, n_workers = 1, seed = op2$seed[i]))
  }

  expect_equal(op1, op2)

})


test_that('stop_followup and update_accrual_rate reproduce across worker modes', {

  skip_if(Sys.getenv("R_COVR") == "true",
          "n_workers > 1 spawns R processes that covr cannot trace")

  bar <- function(n, n_workers, seed = NULL){
    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/10)
    pbo <- arm(name = 'pbo'); pbo$add_endpoints(pfs)
    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/12)
    trt <- arm(name = 'trt'); trt$add_endpoints(pfs)

    trial <- trial(
      name = 'test', n_patients = 300, duration = 30,
      enroller = StaggeredRecruiter,
      accrual_rate = data.frame(end_time = Inf, piecewise_rate = 30),
      dropout = rweibull, shape = 1.32, scale = 114.4,
      seed = seed, silent = TRUE
    )
    trial$add_arms(sample_ratio = c(1, 1), pbo, trt)

    adapt <- milestone(name = 'adapt',
                       when = calendarTime(time = 8),
                       action = function(trial){
                         stop_followup(trial, arm == 'pbo')
                         update_accrual_rate(
                           trial,
                           data.frame(end_time = Inf, piecewise_rate = 15))
                       })
    final <- milestone(name = 'final',
                       when = calendarTime(time = 25),
                       action = function(trial){
                         d <- trial$get_locked_data('final')
                         fit <- fitLogrank(Surv(pfs, pfs_event) ~ arm,
                                           placebo = 'pbo', data = d,
                                           alternative = 'less')
                         trial$save(fit$p[1], 'FA_pval')
                         trial$save(nrow(d), 'FA_n')
                       })

    listener <- listener(silent = TRUE)
    listener$add_milestones(adapt, final)
    controller <- controller(trial, listener)
    controller$run(n = n, n_workers = n_workers, plot_event = FALSE, silent = TRUE)
    controller$get_output(c('seed', 'FA_pval', 'FA_n'))
  }

  op2 <- bar(n = 6, n_workers = 2, seed = NULL)

  op1 <- NULL
  for(i in 1:6){
    op1 <- rbind(op1, bar(n = 1, n_workers = 1, seed = op2$seed[i]))
  }

  expect_equal(op1, op2)
})


test_that('update_milestone reproduces across worker modes', {

  skip_if(Sys.getenv("R_COVR") == "true",
          "n_workers > 1 spawns R processes that covr cannot trace")

  baz <- function(n, n_workers, seed = NULL){
    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/10)
    pbo <- arm(name = 'pbo'); pbo$add_endpoints(pfs)
    pfs <- endpoint(name = 'pfs', type = 'tte', generator = rexp, rate = log(2)/12)
    trt <- arm(name = 'trt'); trt$add_endpoints(pfs)

    trial <- trial(
      name = 'test', n_patients = 300, duration = 40,
      enroller = StaggeredRecruiter,
      accrual_rate = data.frame(end_time = Inf, piecewise_rate = 30),
      dropout = rweibull, shape = 1.32, scale = 114.4,
      seed = seed, silent = TRUE
    )
    trial$add_arms(sample_ratio = c(1, 1), pbo, trt)

    interim <- milestone(name = 'interim',
                         when = eventNumber(endpoint = 'pfs', n = 60),
                         action = function(trial){
                           d <- trial$get_locked_data('interim')
                           fit <- fitLogrank(Surv(pfs, pfs_event) ~ arm,
                                             placebo = 'pbo', data = d,
                                             alternative = 'less')
                           ## data-dependent update: raise the final target
                           ## when the interim z is unimpressive
                           if(fit$z[1] > -1.5){
                             update_milestone(trial, 'final',
                                              when = eventNumber(endpoint = 'pfs',
                                                                 n = 160))
                             trial$save('raised', 'im_decision')
                           }else{
                             trial$save('kept', 'im_decision')
                           }
                         })
    final <- milestone(name = 'final',
                       when = eventNumber(endpoint = 'pfs', n = 120),
                       action = function(trial){
                         d <- trial$get_locked_data('final')
                         fit <- fitLogrank(Surv(pfs, pfs_event) ~ arm,
                                           placebo = 'pbo', data = d,
                                           alternative = 'less')
                         trial$save(fit$p[1], 'FA_pval')
                         trial$save(sum(d$pfs_event), 'FA_events')
                       })

    listener <- listener(silent = TRUE)
    listener$add_milestones(interim, final)
    controller <- controller(trial, listener)
    controller$run(n = n, n_workers = n_workers, plot_event = FALSE, silent = TRUE)
    controller$get_output(c('seed', 'im_decision', 'FA_pval', 'FA_events'))
  }

  op2 <- baz(n = 6, n_workers = 2, seed = NULL)

  op1 <- NULL
  for(i in 1:6){
    op1 <- rbind(op1, baz(n = 1, n_workers = 1, seed = op2$seed[i]))
  }

  expect_equal(op1, op2)
})