tests/testthat/test-bqs-parallel.R

In qcluster: Clustering via Quadratic Scoring

skip_if_no_psock_cluster <- function(workers = 2L) {
  probe <- tryCatch(
    parallel::makeCluster(workers, type = "PSOCK"),
    error = function(e) e
  )

  if (inherits(probe, "error")) {
    skip(paste("PSOCK cluster unavailable in this environment:", conditionMessage(probe)))
  }

  parallel::stopCluster(probe)
}

env_method_for_bqs <- function(data) {
  p <- ncol(data)
  center <- colMeans(data)

  list(params = list(
    proportion = 1,
    mean = matrix(center, ncol = 1),
    cov = array(diag(p), dim = c(p, p, 1)),
    openblas_threads = Sys.getenv("OPENBLAS_NUM_THREADS", unset = ""),
    omp_threads = Sys.getenv("OMP_NUM_THREADS", unset = ""),
    mkl_threads = Sys.getenv("MKL_NUM_THREADS", unset = "")
  ))
}

test_that("bqs serializes BLAS threading inside PSOCK workers by default", {
  skip_if(parallel::detectCores() < 2)
  skip_if_no_psock_cluster()

  methodset <- mbind(env_method_for_bqs)
  res <- bqs(iris[1:20, -5], methodset, B = 2, ncores = 2, saveparams = TRUE)

  worker_params <- res$raw$params[[1]]
  expect_length(worker_params, 2)
  expect_true(all(vapply(worker_params, function(x) identical(x$openblas_threads, "1"), logical(1))))
  expect_true(all(vapply(worker_params, function(x) identical(x$omp_threads, "1"), logical(1))))
  expect_true(all(vapply(worker_params, function(x) identical(x$mkl_threads, "1"), logical(1))))
})