Merge pull request #2003 from quanteda/fix-threads

Fix threads option

DESCRIPTION

@@ -1,5 +1,5 @@
 Package: quanteda
-Version: 2.1.2
+Version: 2.1.2.9000
 Title: Quantitative Analysis of Textual Data
 Description: A fast, flexible, and comprehensive framework for 
     quantitative text analysis in R.  Provides functionality for corpus management,

NEWS.md

@@ -1,3 +1,11 @@
+# quanteda 2.1.3
+
+## Changes
+
+## Bug fixes and stability enhancements
+
+* Allow use of multi-threading with more than two threads by fixing `quanteda_options()`.
+
 # quanteda 2.1.2
 
 ## Changes

R/quanteda_options.R

@@ -13,7 +13,9 @@
 #' \item{`verbose`}{logical; if `TRUE` then use this as the default
 #' for all functions with a `verbose` argument}
 #' \item{`threads`}{integer; specifies the number of threads to use in
-#' parallelized functions}
+#' parallelized functions; defaults to `RcppParallel::defaultNumThreads()` 
+#' unless `OMP_THREAD_LIMIT` is set; the number of threads can be changed 
+#' only once in a session}
 #' \item{`print_dfm_max_ndoc`}{integer; specifies the number of documents
 #' to display when using the defaults for printing a dfm}
 #' \item{`print_dfm_max_nfeat`}{integer; specifies the number of
@@ -67,7 +69,7 @@ quanteda_options <- function(..., reset = FALSE, initialize = FALSE) {
         args <- args[[1]]
 
     # initialize automatically it not yet done so
-    if (is.null(getOption('quanteda_initialized')) || !"package:quanteda" %in% search())
+    if (is.null(getOption("quanteda_initialized")) || !"package:quanteda" %in% search())
         quanteda_initialize()
 
     if (initialize) {
@@ -98,48 +100,61 @@ quanteda_initialize <- function() {
         if (is.null(getOption(paste0("quanteda_", key))))
             set_option_value(key, opts[[key]])
     }
-    unlist(options('quanteda_initialized' = TRUE), use.names = FALSE)
+    unlist(options("quanteda_initialized" = TRUE), use.names = FALSE)
 }
 
 quanteda_reset <- function() {
     opts <- get_options_default()
     for (key in names(opts)) {
-        set_option_value(key, opts[[key]])
+        if (key != "threads") 
+            set_option_value(key, opts[[key]])
     }
-    unlist(options('quanteda_initialized' = TRUE), use.names = FALSE)
+    unlist(options("quanteda_initialized" = TRUE), use.names = FALSE)
 }
 
 set_option_value <- function(key, value) {
 
     opts <- get_options_default()
     # check for key validity
     if (!key %in% names(opts))
-        stop(key, " is not a valid quanteda option")
+        stop(key, " is not a valid quanteda option", call. = FALSE)
 
     # special setting for threads
     if (key == "threads") {
         value <- as.integer(value)
-        value_default <- RcppParallel::defaultNumThreads()
+        thread <- get_threads()
         if (value < 1)
-            stop("Number of threads must be greater or equal to 1")
-        if (value > value_default) {
-            warning("Setting threads instead to maximum available ", value_default, call. = FALSE)
-            value <- value_default
+            stop("Number of threads must be greater or equal to 1", call. = FALSE)
+        if (value > thread["max"]) {
+            warning("Setting threads instead to maximum available ", thread["max"], call. = FALSE)
+            value <- thread["max"]
+        }
+        if (is.na(thread["tbb"])) {
+            if (value != thread["max"])
+                RcppParallel::setThreadOptions(value)
+        } else {
+            if (value != thread["tbb"])
+                warning("Number of threads can be changed only once in a session", call. = FALSE)   
         }
-        RcppParallel::setThreadOptions(value)
-        Sys.setenv("OMP_THREAD_LIMIT" = value)
     }
 
     # assign the key-value
     opts <- list(value)
     names(opts) <- paste0("quanteda_", key)
     options(opts)
-
+}
+
+# returns thread settings
+get_threads <- function() {
+    c("omp" = as.integer(Sys.getenv("OMP_THREAD_LIMIT")),
+      "tbb" = as.integer(Sys.getenv("RCPP_PARALLEL_NUM_THREADS")),
+      "max" = RcppParallel::defaultNumThreads())
 }
 
 # returns default options
-get_options_default <- function(){
-    opts <- list(threads = min(RcppParallel::defaultNumThreads(), 2),
+get_options_default <- function() {
+
+    opts <- list(threads = unname(min(get_threads(), na.rm = TRUE)),
                  verbose = FALSE,
                  print_dfm_max_ndoc = 6L,
                  print_dfm_max_nfeat = 10L,

README.md

@@ -6,7 +6,7 @@ data](https://cdn.rawgit.com/quanteda/quanteda/master/images/quanteda_logo.svg)]
 
 [![CRAN
 Version](https://www.r-pkg.org/badges/version/quanteda)](https://CRAN.R-project.org/package=quanteda)
-[![](https://img.shields.io/badge/devel%20version-2.1.2-royalblue.svg)](https://github.com/quanteda/quanteda)
+[![](https://img.shields.io/badge/devel%20version-2.1.2.9000-royalblue.svg)](https://github.com/quanteda/quanteda)
 [![Downloads](https://cranlogs.r-pkg.org/badges/quanteda)](https://CRAN.R-project.org/package=quanteda)
 [![Total
 Downloads](https://cranlogs.r-pkg.org/badges/grand-total/quanteda?color=orange)](https://CRAN.R-project.org/package=quanteda)

tests/testthat.R

@@ -1,12 +1,9 @@
 Sys.setenv("R_TESTS" = "")
+Sys.setenv("_R_CHECK_LENGTH_1_CONDITION_" = TRUE)
 
 library(testthat)
 library(quanteda)
-
-Sys.setenv("_R_CHECK_LENGTH_1_CONDITION_" = TRUE)
-
-ops <- quanteda_options()
 quanteda_options(reset = TRUE)
 
 test_check("quanteda")
-quanteda_options(ops)
+

tests/testthat/test-quanteda_options.R

@@ -8,12 +8,14 @@ test_that("quanteda_options initialization works", {
 
 test_that("quanteda_options initialize works correctly", {
 
-    threads_temp <- getOption("quanteda_threads")
-    quanteda_options(verbose = TRUE, threads = 1)
+    quanteda_options(verbose = TRUE, 
+                     print_dfm_max_ndoc = 1L, 
+                     print_dfm_max_nfeat = NULL)
     quanteda_options(initialize = TRUE)
-    expect_equal(quanteda_options("threads"), 1)
     expect_equal(quanteda_options("verbose"), TRUE)
-    quanteda_options(threads = threads_temp)
+    expect_equal(quanteda_options("print_dfm_max_ndoc"), 1L)
+    expect_equal(quanteda_options("print_dfm_max_nfeat"), 10L)
+
 })
 
 test_that("quanteda_options returns an error for non-existing options", {
@@ -35,12 +37,6 @@ test_that("quanteda_options works correctly to set options", {
         getOption("quanteda_verbose")
     )
 
-    quanteda_options(threads = 2)
-    expect_equal(
-        quanteda_options("threads"),
-        getOption("quanteda_threads")
-    )
-
     quanteda_options(print_dfm_max_ndoc = 13L)
     expect_equal(
         quanteda_options("print_dfm_max_ndoc"),
@@ -74,30 +70,18 @@ test_that("quanteda_options reset works correctly", {
 })
 
 test_that("quanteda_options works with threads", {
-    quanteda_options(reset = TRUE)
-    expect_equal(
-        as.numeric(Sys.getenv('RCPP_PARALLEL_NUM_THREADS')),
-        min(2L, RcppParallel::defaultNumThreads())
-    )
-    expect_equal(
-        as.numeric(Sys.getenv('RCPP_PARALLEL_NUM_THREADS')),
-        quanteda_options("threads")
-    )
-    quanteda_options(threads = 2)
+    if (RcppParallel::defaultNumThreads() == 2) {
+        quanteda_options(threads = 1)
+    } else {
+        quanteda_options(threads = 2)
+    }
     expect_equal(
         as.numeric(Sys.getenv('RCPP_PARALLEL_NUM_THREADS')),
         quanteda_options("threads")
     )
-    expect_equal(
-        as.numeric(Sys.getenv('OMP_THREAD_LIMIT')),
-        quanteda_options("threads")
-    )
-    quanteda_options(threads = 1.5)
-    expect_equal(
-        quanteda_options("threads"), 1L
-    )
-    expect_equal(
-        quanteda_options("threads"), 1L
+    expect_warning(
+        quanteda_options(threads = 4),
+        "Number of threads can be changed only once"
     )
     expect_error(quanteda_options(threads = 0),
                  "^Number of threads must be greater or equal to 1")