diff --git a/DESCRIPTION b/DESCRIPTION index 41025f67..805c1a45 100644 --- a/DESCRIPTION +++ b/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, diff --git a/R/DistributedCogaps.R b/R/DistributedCogaps.R index 53bf6d50..90d00a63 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) @@ -58,17 +81,37 @@ 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)) { # 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 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 diff --git a/tests/testthat/test_DistributedCogaps.R b/tests/testthat/test_DistributedCogaps.R index 55024b8f..322b76c1 100644 --- a/tests/testthat/test_DistributedCogaps.R +++ b/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")) }) \ No newline at end of file