cores can be specified

Author: tanlabcode

R/compute_mutual_information.R

@@ -86,8 +86,9 @@ compute_mutual_information <- function(dir_out, cores=NULL) {
     # register cores for parallel computation
     if (is.null(cores)) {
         cores <- max(1, parallel::detectCores() - 2)
+        message(sprintf("Number of cores not specified, using %.0f.", cores))
     } else {
-        cores <- max(1, cores)
+        message(sprintf("Using %.0f specified cores.", cores))
     }
     doParallel::registerDoParallel(cores = cores)
 

R/compute_non_self_talk.R

@@ -43,10 +43,6 @@ compute_non_self_talk_type <- function(ligands, type, letter, dir_in, dir_out) {
     index_valid <- index[valid,]
     ligands_valid <- ligands[valid,]
 
-    # detect and register cores
-    cores <- max(1, parallel::detectCores() - 2)
-    doParallel::registerDoParallel(cores = cores)
-
     # parallel loop for MI distances
     i <- NULL
     score <- foreach::`%dopar%`(
@@ -72,9 +68,6 @@ compute_non_self_talk_type <- function(ligands, type, letter, dir_in, dir_out) {
 
     # write out
     vroom::vroom_write(score, fpath_out, progress = FALSE)
-
-    # unregister cores
-    doParallel::stopImplicitCluster()
     NULL
 }
 
@@ -100,8 +93,22 @@ compute_non_self_talk_type <- function(ligands, type, letter, dir_in, dir_out) {
 #' @param dir_out Output directory
 #' @return None
 #' @export
-compute_non_self_talk <- function(ligands, type_a, type_b, dir_in, dir_out) {
+compute_non_self_talk <- function(
+    ligands, type_a, type_b, dir_in, dir_out, cores=NULL) {
+
+    # detect and register cores
+    if (is.null(cores)) {
+        cores <- max(1, parallel::detectCores() - 2)
+        message(sprintf("Number of cores not specified, using %.0f.", cores))
+    } else {
+        message(sprintf("Using %.0f specified cores.", cores))
+    }
+    doParallel::registerDoParallel(cores = cores)
+
     compute_non_self_talk_type(ligands, type_a, "A", dir_in, dir_out)
     compute_non_self_talk_type(ligands, type_b, "B", dir_in, dir_out)
+
+    # unregister cores
+    doParallel::stopImplicitCluster()
     NULL
 }

R/cytotalk.R

@@ -116,7 +116,7 @@ run_cytotalk <- function(
     ligands=CytoTalk::ligands_human,
     cutoff_a=0.1, cutoff_b=0.1,
     beta_max=100, omega_min=0.5, omega_max=0.5,
-    depth=3, ntrial=10000) {
+    depth=3, ntrial=10000, cores=NULL) {
 
     # must have valid data directory
     type_names <- check_valid_names(dir_in)
@@ -143,7 +143,8 @@ run_cytotalk <- function(
     # compute preferential expression measure
     tick(1, "Preprocessing...")
     preprocess(proteins, type_a, type_b, cutoff_a, cutoff_b, dir_in, dir_out)
-    compute_non_self_talk(ligands, type_a, type_b, dir_in, dir_out)
+    compute_non_self_talk(
+        ligands, type_a, type_b, dir_in, dir_out, cores)
     status <- compute_pem(dir_in, dir_out)
 
     # exit if bad PEM status
@@ -153,7 +154,7 @@ run_cytotalk <- function(
 
     # compute mutual information (within types)
     tick(2, "Mutual information matrix...")
-    compute_mutual_information(dir_out)
+    compute_mutual_information(dir_out, cores)
 
     # use ARACNE.m to filter out indirect edges
     tick(3, "Indirect edge-filtered network...")
@@ -170,7 +171,7 @@ run_cytotalk <- function(
 
     # run Kolmogorov-Smirnov tests
     tick(6, "Determine best signaling network...")
-    generate_signaling_network(dir_out)
+    generate_signaling_network(dir_out, cores)
 
     # generate SIF and SVG files
     tick(7, "Generate network output...")

R/generate_signaling_network.R

@@ -36,7 +36,7 @@ generate_summary <- function(dir_out) {
 }
 
 #' @noRd
-compute_kolmogorov_smirnov <- function(dir_out) {
+compute_kolmogorov_smirnov <- function(dir_out, cores=NULL) {
     # format filepaths
     fpath_edge <- file.path(dir_out, "PCSF_EdgeOccurance.txt")
     fpath_pval <- file.path(dir_out, "PCSF_EdgeTestValues.txt")
@@ -64,7 +64,12 @@ compute_kolmogorov_smirnov <- function(dir_out) {
     lst_counts <- tapply(vec_counts, vec_param, c)
 
     # detect and register cores
-    cores <- max(1, parallel::detectCores() - 2)
+    if (is.null(cores)) {
+        cores <- max(1, parallel::detectCores() - 2)
+        message(sprintf("Number of cores not specified, using %.0f.", cores))
+    } else {
+        message(sprintf("Using %.0f specified cores.", cores))
+    }
     doParallel::registerDoParallel(cores = cores)
 
     # parallel loop for Kolmogorov-Smirnov test
@@ -105,8 +110,8 @@ compute_kolmogorov_smirnov <- function(dir_out) {
 #' @param dir_out Output directory
 #' @return None
 #' @export
-generate_signaling_network <- function(dir_out) {
+generate_signaling_network <- function(dir_out, cores=NULL) {
     generate_summary(dir_out)
-    compute_kolmogorov_smirnov(dir_out)
+    compute_kolmogorov_smirnov(dir_out, cores)
     NULL
 }