Improve parallel processing with BiocParallel

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: CoGAPS
-Version: 3.33.2
-Date: 2025-03-11
+Version: 3.33.3
+Date: 2026-08-12
 Title: Coordinated Gene Activity in Pattern Sets
 Author: Jeanette Johnson, Ashley Tsang, Jacob Mitchell, Thomas Sherman, Wai-shing Lee, Conor Kelton, Ondrej Maxian, Jacob Carey,
     Genevieve Stein-O'Brien, Michael Considine, Maggie Wodicka, John Stansfield,

R/DistributedCogaps.R

@@ -8,22 +8,35 @@
 #' @param uncertainty uncertainty of data in the same format as data
 #' @param subsetIndices indices of the subset of data to run on
 #' @param workerID worker ID for parallelization
+#' @param skipInternalSubset if TRUE, data is already subsetted and only names
+#' should be subsetted
 #' @return CogapsResult object
 callInternalCoGAPS <- function(data, allParams, uncertainty, subsetIndices,
-workerID)
+workerID, skipInternalSubset=FALSE)
 {
     # identify which mode of parallelization
     genomeWide <- allParams$gaps@distributed == "genome-wide"
     allParams$gaps@distributed <- NULL
 
-    # subset gene/sample names
-    if (genomeWide)
-        allParams$geneNames <- allParams$geneNames[subsetIndices]
-    else
-        allParams$sampleNames <- allParams$sampleNames[subsetIndices]
+    if (!is.null(subsetIndices))
+    {
+        # subset gene/sample names
+        if (genomeWide)
+            allParams$geneNames <- allParams$geneNames[subsetIndices]
+        else
+            allParams$sampleNames <- allParams$sampleNames[subsetIndices]
+    }
 
-    allParams$gaps@subsetIndices <- subsetIndices
-    allParams$gaps@subsetDim <- ifelse(genomeWide, 1, 2)
+    if (skipInternalSubset)
+    {
+        allParams$gaps@subsetIndices <- NULL
+        allParams$gaps@subsetDim <- 0
+    }
+    else
+    {
+        allParams$gaps@subsetIndices <- subsetIndices
+        allParams$gaps@subsetDim <- ifelse(genomeWide, 1, 2)
+    }
     allParams$workerID <- workerID
 
     # Distributed CoGAPS parallelizes across data subsets instead of using the
@@ -51,24 +64,54 @@ workerID)
 #' @importFrom BiocParallel bplapply MulticoreParam
 distributedCogaps <- function(data, allParams, uncertainty)
 {
+    subsetData <- function(dataInput, setIndices, subsetRows)
+    {
+        if (subsetRows)
+            return(dataInput[setIndices,,drop=FALSE])
+        return(dataInput[,setIndices,drop=FALSE])
+    }
+
+    subsetRows <- xor(allParams$transposeData,
+        allParams$gaps@distributed == "genome-wide")
+
     # randomly sample either rows or columns into subsets to break the data up
     set.seed(allParams$gaps@seed)
     sets <- createSets(data, allParams)
     if (min(sapply(sets, length)) < allParams$gaps@nPatterns)
         stop("data subset dimension less than nPatterns")
 
     if (is.null(allParams$BPPARAM))
-        allParams$BPPARAM <- BiocParallel::MulticoreParam(workers=length(sets))
-
+    {
+        cores <- min(length(sets), parallel::detectCores())
+        allParams$BPPARAM <- BiocParallel::MulticoreParam(workers=max(1, cores-2))
+    }
     initialResult <- NULL
     if (is.null(allParams$gaps@fixedPatterns))
     {
         # run Cogaps normally on each subset of the data
         gapsCat(allParams, "Running Across Subsets...\n\n")
-        initialResult <- bplapply(1:length(sets), BPPARAM=allParams$BPPARAM,
+        initialResult <- bplapply(seq_along(sets), BPPARAM=allParams$BPPARAM,
         FUN=function(i)
         {
-            callInternalCoGAPS(data, allParams, uncertainty, sets[[i]], i)
+            if (is(data, "character"))
+            {
+                callInternalCoGAPS(data, allParams, uncertainty, sets[[i]], i)
+            }
+            else
+            {
+                setIndices <- sets[[i]]
+                subsetUncertainty <- uncertainty
+                if (!is.null(uncertainty) && !is(uncertainty, "character"))
+                    subsetUncertainty <- subsetData(uncertainty, setIndices, subsetRows)
+                callInternalCoGAPS(
+                    data=subsetData(data, setIndices, subsetRows),
+                    allParams=allParams,
+                    uncertainty=subsetUncertainty,
+                    subsetIndices=setIndices,
+                    workerID=i,
+                    skipInternalSubset=TRUE
+                )
+            }
         })
 
         # get all unmatched patterns
@@ -94,10 +137,28 @@ distributedCogaps <- function(data, allParams, uncertainty)
 
     # run final phase with fixed matrix
     gapsCat(allParams, "Running Final Stage...\n\n")
-    finalResult <- bplapply(1:length(sets), BPPARAM=allParams$BPPARAM,
+    finalResult <- bplapply(seq_along(sets), BPPARAM=allParams$BPPARAM,
     FUN=function(i)
     {
-        callInternalCoGAPS(data, allParams, uncertainty, sets[[i]], i)
+        if (is(data, "character"))
+        {
+            callInternalCoGAPS(data, allParams, uncertainty, sets[[i]], i)
+        }
+        else
+        {
+            setIndices <- sets[[i]]
+            subsetUncertainty <- uncertainty
+            if (!is.null(uncertainty) && !is(uncertainty, "character"))
+                subsetUncertainty <- subsetData(uncertainty, setIndices, subsetRows)
+            callInternalCoGAPS(
+                data=subsetData(data, setIndices, subsetRows),
+                allParams=allParams,
+                uncertainty=subsetUncertainty,
+                subsetIndices=setIndices,
+                workerID=i,
+                skipInternalSubset=TRUE
+            )
+        }
     })
 
     # concatenate final result

tests/testthat/test_DistributedCogaps.R

@@ -70,4 +70,128 @@ test_that("a one-pattern distributed run keeps its dim names", {
     expect_false(is.null(rownames(cg@featureLoadings)))
     expect_false(is.null(rownames(cg@sampleFactors)))
   }
+})
+
+test_that("distributed workers receive subsetted in-memory data", {
+  set.seed(1)
+  mat <- matrix(rexp(20 * 10), nrow = 20, ncol = 10)
+
+  params <- CogapsParams(seed = 42,
+                         nIterations = 30,
+                         nPatterns = 2,
+                         sparseOptimization = as.logical(0),
+                         distributed = "genome-wide")
+  params <- setDistributedParams(params, nSets = 2)
+  params <- setParam(params, "explicitSets", list(1:10, 11:20))
+
+  observed <- list()
+  ns <- asNamespace("CoGAPS")
+  original <- get("callInternalCoGAPS", envir = ns)
+
+  wrapper <- function(data, allParams, uncertainty, subsetIndices, workerID,
+                     skipInternalSubset=FALSE)
+  {
+    observed[[length(observed) + 1]] <<- list(
+      nrow = nrow(data),
+      ncol = ncol(data),
+      subsetLength = length(subsetIndices),
+      workerID = workerID,
+      skipInternalSubset = skipInternalSubset,
+      payloadBytes = as.numeric(object.size(data))
+    )
+    original(data, allParams, uncertainty, subsetIndices, workerID,
+             skipInternalSubset)
+  }
+
+  assignInNamespace("callInternalCoGAPS", wrapper, ns = "CoGAPS")
+  on.exit(assignInNamespace("callInternalCoGAPS", original, ns = "CoGAPS"),
+          add = TRUE)
+
+  CoGAPS(mat,
+         params = params,
+         BPPARAM = BiocParallel::SerialParam(),
+         messages = FALSE)
+
+  expect_true(length(observed) > 0)
+  expect_true(all(vapply(observed, function(x) x$nrow < nrow(mat), logical(1))))
+  expect_true(all(vapply(observed, function(x) x$ncol == ncol(mat), logical(1))))
+  expect_true(all(vapply(observed, function(x) x$subsetLength == x$nrow, logical(1))))
+  expect_true(all(vapply(observed, function(x) isTRUE(x$skipInternalSubset), logical(1))))
+
+  # benchmark-style check: worker payload should be smaller than full matrix
+  fullBytes <- as.numeric(object.size(mat))
+  expect_true(all(vapply(observed, function(x) x$payloadBytes < fullBytes, logical(1))))
+})
+
+test_that("single-cell distributed workers receive subsetted in-memory data", {
+  set.seed(1)
+  mat <- matrix(rexp(20 * 10), nrow = 20, ncol = 10)
+
+  params <- CogapsParams(seed = 42,
+                         nIterations = 30,
+                         nPatterns = 2,
+                         sparseOptimization = as.logical(0),
+                         distributed = "single-cell")
+  params <- setDistributedParams(params, nSets = 2)
+  params <- setParam(params, "explicitSets", list(1:5, 6:10))
+
+  observed <- list()
+  ns <- asNamespace("CoGAPS")
+  original <- get("callInternalCoGAPS", envir = ns)
+
+  wrapper <- function(data, allParams, uncertainty, subsetIndices, workerID,
+                     skipInternalSubset=FALSE)
+  {
+    observed[[length(observed) + 1]] <<- list(
+      nrow = nrow(data),
+      ncol = ncol(data),
+      subsetLength = length(subsetIndices),
+      workerID = workerID,
+      skipInternalSubset = skipInternalSubset,
+      payloadBytes = as.numeric(object.size(data))
+    )
+    original(data, allParams, uncertainty, subsetIndices, workerID,
+             skipInternalSubset)
+  }
+
+  assignInNamespace("callInternalCoGAPS", wrapper, ns = "CoGAPS")
+  on.exit(assignInNamespace("callInternalCoGAPS", original, ns = "CoGAPS"),
+          add = TRUE)
+
+  CoGAPS(mat,
+         params = params,
+         BPPARAM = BiocParallel::SerialParam(),
+         messages = FALSE)
+
+  expect_true(length(observed) > 0)
+  expect_true(all(vapply(observed, function(x) x$nrow == nrow(mat), logical(1))))
+  expect_true(all(vapply(observed, function(x) x$ncol < ncol(mat), logical(1))))
+  expect_true(all(vapply(observed, function(x) x$subsetLength == x$ncol, logical(1))))
+  expect_true(all(vapply(observed, function(x) isTRUE(x$skipInternalSubset), logical(1))))
+
+  # benchmark-style check: worker payload should be smaller than full matrix
+  fullBytes <- as.numeric(object.size(mat))
+  expect_true(all(vapply(observed, function(x) x$payloadBytes < fullBytes, logical(1))))
+})
+
+
+#test more nSets than workers
+test_that("distributed workers fewer than nSets proceed successfully", {
+  set.seed(1)
+  mat <- matrix(rexp(20 * 10), nrow = 20, ncol = 10)
+
+  params <- CogapsParams(seed = 42,
+                         nIterations = 30,
+                         nPatterns = 2,
+                         sparseOptimization = as.logical(0),
+                         distributed = "genome-wide")
+  params <- setDistributedParams(params, nSets = 4)
+
+
+  cg <- CoGAPS(mat,
+         params = params,
+         BPPARAM = BiocParallel::MulticoreParam(workers=1),
+         messages = FALSE)
+
+  expect_true(is(cg, "CogapsResult"))
 })