From 10b3382286080951f89f2213234fc7042cd74cf2 Mon Sep 17 00:00:00 2001 From: dimalvovs Date: Wed, 12 Aug 2026 10:02:13 -0400 Subject: [PATCH 1/6] init subset per worker fix --- R/DistributedCogaps.R | 83 ++++++++++++++++--- tests/testthat/test_DistributedCogaps.R | 104 +++++++++++++++++++++++- 2 files changed, 174 insertions(+), 13 deletions(-) diff --git a/R/DistributedCogaps.R b/R/DistributedCogaps.R index 53bf6d50..b7db7cee 100755 --- a/R/DistributedCogaps.R +++ b/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,6 +64,16 @@ 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) @@ -65,10 +88,28 @@ distributedCogaps <- function(data, allParams, uncertainty) { # 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 +135,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 diff --git a/tests/testthat/test_DistributedCogaps.R b/tests/testthat/test_DistributedCogaps.R index 55024b8f..a898dd11 100644 --- a/tests/testthat/test_DistributedCogaps.R +++ b/tests/testthat/test_DistributedCogaps.R @@ -70,4 +70,106 @@ 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))) } -}) \ No newline at end of file +}) + +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)))) +}) From ff2876447eb206415af7d3e118324105132ac6fd Mon Sep 17 00:00:00 2001 From: dimalvovs Date: Wed, 12 Aug 2026 10:29:28 -0400 Subject: [PATCH 2/6] update docs and bump version --- DESCRIPTION | 2 +- R/Package.R | 2 +- man/CoGAPS-package.Rd | 2 +- man/callInternalCoGAPS.Rd | 12 +++++++++++- 4 files changed, 14 insertions(+), 4 deletions(-) diff --git a/DESCRIPTION b/DESCRIPTION index 41025f67..0230034f 100644 --- a/DESCRIPTION +++ b/DESCRIPTION @@ -1,5 +1,5 @@ Package: CoGAPS -Version: 3.33.2 +Version: 3.33.3 Date: 2025-03-11 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, diff --git a/R/Package.R b/R/Package.R index 7584ec08..3964bd16 100755 --- a/R/Package.R +++ b/R/Package.R @@ -19,7 +19,7 @@ #' CoGAPS: an R/C++ package to identify patterns and biological #' process activity in transcriptomic data. #' Bioinformatics. 2010 Nov 1;26(21):2792-3 -#' @docType package +#' @docType _PACKAGE #' @name CoGAPS-package #' @importFrom Rcpp evalCpp #' @useDynLib CoGAPS diff --git a/man/CoGAPS-package.Rd b/man/CoGAPS-package.Rd index c10c6fdb..1e8fd359 100755 --- a/man/CoGAPS-package.Rd +++ b/man/CoGAPS-package.Rd @@ -1,6 +1,6 @@ % Generated by roxygen2: do not edit by hand % Please edit documentation in R/Package.R -\docType{package} +\docType{_PACKAGE} \name{CoGAPS-package} \alias{CoGAPS-package} \title{CoGAPS: Coordinated Gene Activity in Pattern Sets} diff --git a/man/callInternalCoGAPS.Rd b/man/callInternalCoGAPS.Rd index 316dccc7..3eaa13eb 100755 --- a/man/callInternalCoGAPS.Rd +++ b/man/callInternalCoGAPS.Rd @@ -7,7 +7,14 @@ called directly, but to avoid any re-entrant behavior this function is called instead. It is a light wrapper around cogaps_cpp that handles setting the distributed parameters} \usage{ -callInternalCoGAPS(data, allParams, uncertainty, subsetIndices, workerID) +callInternalCoGAPS( + data, + allParams, + uncertainty, + subsetIndices, + workerID, + skipInternalSubset = FALSE +) } \arguments{ \item{data}{data in a supported format} @@ -19,6 +26,9 @@ callInternalCoGAPS(data, allParams, uncertainty, subsetIndices, workerID) \item{subsetIndices}{indices of the subset of data to run on} \item{workerID}{worker ID for parallelization} + +\item{skipInternalSubset}{if TRUE, data is already subsetted and only names +should be subsetted} } \value{ CogapsResult object From 63910f885f4cade8dc6a73a194cc37a1f675b55e Mon Sep 17 00:00:00 2001 From: dimalvovs Date: Wed, 12 Aug 2026 10:35:37 -0400 Subject: [PATCH 3/6] update package date --- DESCRIPTION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/DESCRIPTION b/DESCRIPTION index 0230034f..805c1a45 100644 --- a/DESCRIPTION +++ b/DESCRIPTION @@ -1,6 +1,6 @@ Package: CoGAPS Version: 3.33.3 -Date: 2025-03-11 +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, From bda8873d37cad9c7fd917d032d2c0e1a90427384 Mon Sep 17 00:00:00 2001 From: dimalvovs Date: Wed, 12 Aug 2026 11:45:40 -0400 Subject: [PATCH 4/6] revert _PACKAGE to package --- R/Package.R | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/R/Package.R b/R/Package.R index 3964bd16..7584ec08 100755 --- a/R/Package.R +++ b/R/Package.R @@ -19,7 +19,7 @@ #' CoGAPS: an R/C++ package to identify patterns and biological #' process activity in transcriptomic data. #' Bioinformatics. 2010 Nov 1;26(21):2792-3 -#' @docType _PACKAGE +#' @docType package #' @name CoGAPS-package #' @importFrom Rcpp evalCpp #' @useDynLib CoGAPS From f80ec1db160eedc84da9f39a5e1afa3dc081860d Mon Sep 17 00:00:00 2001 From: dimalvovs Date: Wed, 12 Aug 2026 12:07:08 -0400 Subject: [PATCH 5/6] update doc --- man/CoGAPS-package.Rd | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/man/CoGAPS-package.Rd b/man/CoGAPS-package.Rd index 1e8fd359..c10c6fdb 100755 --- a/man/CoGAPS-package.Rd +++ b/man/CoGAPS-package.Rd @@ -1,6 +1,6 @@ % Generated by roxygen2: do not edit by hand % Please edit documentation in R/Package.R -\docType{_PACKAGE} +\docType{package} \name{CoGAPS-package} \alias{CoGAPS-package} \title{CoGAPS: Coordinated Gene Activity in Pattern Sets} From 3ae5dbff7e939b65a64aad6ffcf6c22a4affe71e Mon Sep 17 00:00:00 2001 From: dimalvovs Date: Wed, 12 Aug 2026 19:50:30 -0400 Subject: [PATCH 6/6] untie workers from nSets --- R/DistributedCogaps.R | 6 ++++-- tests/testthat/test_DistributedCogaps.R | 22 ++++++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/R/DistributedCogaps.R b/R/DistributedCogaps.R index b7db7cee..90d00a63 100755 --- a/R/DistributedCogaps.R +++ b/R/DistributedCogaps.R @@ -81,8 +81,10 @@ distributedCogaps <- function(data, allParams, uncertainty) 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)) { diff --git a/tests/testthat/test_DistributedCogaps.R b/tests/testthat/test_DistributedCogaps.R index a898dd11..322b76c1 100644 --- a/tests/testthat/test_DistributedCogaps.R +++ b/tests/testthat/test_DistributedCogaps.R @@ -173,3 +173,25 @@ test_that("single-cell distributed workers receive subsetted in-memory data", { 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")) +}) \ No newline at end of file