%dofuture% interferes with sink()?

[pglpm]

I'm calling the package Nimble from a foreach() loop with %dofuture%. Nimble outputs some messages, which I'd like to be redirected to log files, together with other messages generated with cat(). Nimble seems to be using the message() function, here is its code

I try to achieve this by opening a connection to the log file and redirecting sink() there. This approach worked with the doParallel backend, but the future backend keeps on sending the messages from Nimble or message() to the main R session.

Is this a bug? Or should I approach my goal differently?

Example with future:

library(`foreach`)
library(`doFuture`)
library('nimble')

future::plan('multisession', workers = 2)

foreach(i = 1:2) %dofuture% {
        outcon <- file(paste0('testout', i, '.log'), open = 'w') # open file for writing
        sink(outcon, type = c("output"))
        sink(outcon, type = c("message"))

        nimble::messageIfVerbose('nimble message') # message from Nimble
        message('\nmessage message\n')
        cat('\ncat message\n')
        
        sink(NULL, type = c("output"))
        sink(NULL, type = c("message"))
        close(outcon)
        i
    }
future::plan('sequential')

in this case the "cat message" correctly appears in each testout....log file, but the "nimble message" and the output from message() are displayed in the main R session.

Compare with doParallel:

library(`foreach`)
library(`doParallel`)
library('nimble')

cl <- parallel::makeCluster(2)
doParallel::registerDoParallel(cl)

foreach(i = 1:2) %dopar% {
        outcon <- file(paste0('testout', i, '.log'), open = 'w')
        sink(outcon, type = c("output"))
        sink(outcon, type = c("message"))

        nimble::messageIfVerbose('nimble message')
        message('\nmessage message\n')
        cat('\ncat message\n')

        sink(NULL, type = c("output"))
        sink(NULL, type = c("message"))
        close(outcon)
        i
    }
foreach::registerDoSEQ()
parallel::stopCluster(cl)

In this case all three messages appear in the log file.

I would expect the code with future to behave in a similar way as the one with doParallel, redirecting both messages to the log file.

Session information

> sessionInfo()
R version 4.4.3 (2025-02-28)
Platform: x86_64-pc-linux-gnu
Running under: Ubuntu 20.04.6 LTS

Matrix products: default
BLAS:   /usr/lib/x86_64-linux-gnu/openblas-pthread/libblas.so.3 
LAPACK: /usr/lib/x86_64-linux-gnu/openblas-pthread/liblapack.so.3;  LAPACK version 3.9.0

locale:
 [1] LC_CTYPE=en_GB.UTF-8    LC_NUMERIC=C            LC_TIME=en_SE.UTF-8    
 [4] LC_COLLATE=C            LC_MONETARY=C           LC_MESSAGES=en_GB.UTF-8
 [7] LC_PAPER=en_GB.UTF-8    LC_NAME=C               LC_ADDRESS=C           
[10] LC_TELEPHONE=C          LC_MEASUREMENT=C        LC_IDENTIFICATION=C    

time zone: Europe/Oslo
tzcode source: system (glibc)

attached base packages:
[1] stats     graphics  grDevices utils     datasets  methods   base     

other attached packages:
[1] inferno_0.2.1  doFuture_1.0.2 future_1.40.0  foreach_1.5.2 

loaded via a namespace (and not attached):
 [1] doParallel_1.0.17   cli_3.6.4           rlang_1.1.6         listenv_0.9.1      
 [5] future.apply_1.11.3 rngtools_1.5.2      pracma_2.4.4        nimble_1.3.0       
 [9] grid_4.4.3          lifecycle_1.0.4     numDeriv_2016.8-1.1 doRNG_1.8.6.2      
[13] compiler_4.4.3      codetools_0.2-20    igraph_2.1.4        coda_0.19-4.1      
[17] pkgconfig_2.0.3     lattice_0.22-7      digest_0.6.37       R6_2.6.1           
[21] parallelly_1.43.0   parallel_4.4.3      magrittr_2.0.3      tools_4.4.3        
[25] iterators_1.0.14    globals_0.16.3     

> future::futureSessionInfo()
*** Package versions
future 1.40.0, parallelly 1.43.0, parallel 4.4.3, globals 0.16.3, listenv 0.9.1

*** Allocations
availableCores():
           system /proc/self/status    cgroups.cpuset             nproc 
               16                16                16                16 
availableWorkers():
$cgroups.cpuset
 [1] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
 [7] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
[13] "localhost" "localhost" "localhost" "localhost"

$nproc
 [1] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
 [7] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
[13] "localhost" "localhost" "localhost" "localhost"

$system
 [1] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
 [7] "localhost" "localhost" "localhost" "localhost" "localhost" "localhost"
[13] "localhost" "localhost" "localhost" "localhost"


*** Settings
- future.plan=<not set>
- future.fork.multithreading.enable=<not set>
- future.globals.maxSize=<not set>
- future.globals.onReference=<not set>
- future.resolve.recursive=<not set>
- future.rng.onMisuse=<not set>
- future.wait.timeout=<not set>
- future.wait.interval=<not set>
- future.wait.alpha=<not set>
- future.startup.script=FALSE

*** Backends
Number of workers: 1
List of future strategies:
1. sequential:
   - args: function (..., gc = FALSE, earlySignal = FALSE, envir = parent.frame(), workers = "<NULL>")
   - tweaked: FALSE
   - call: future::plan("sequential")

*** Basic tests
Main R session details:
     pid     r sysname            release
1 394319 4.4.3   Linux 5.15.0-136-generic
                                               version nodename machine   login
1 #147~20.04.1-Ubuntu SMP Wed Mar 19 16:13:14 UTC 2025  host001  x86_64 user001
     user effective_user
1 user001        user001
Worker R session details:
  worker    pid     r sysname            release
1      1 394319 4.4.3   Linux 5.15.0-136-generic
                                               version nodename machine   login
1 #147~20.04.1-Ubuntu SMP Wed Mar 19 16:13:14 UTC 2025  host001  x86_64 user001
     user effective_user
1 user001        user001
Number of unique worker PIDs: 1 (as expected)

[HenrikBengtsson]

Hello.

TL;DR

The following uses best practices in R to capture output sent to standard output and messages to one file per iteration:

library(doFuture)
library(nimble)

plan(multisession, workers = 2)

y <- foreach(i = 1:2) %dofuture% {
    ## Send all standard output to file
    outcon <- file(paste0('testout', i, '.log'), open = 'w') # open file for writing
    sink(outcon, type = "output")
    on.exit({
      sink(NULL, type = "output")
      close(outcon)
    })
    
    withCallingHandlers({
      nimble::messageIfVerbose('nimble message') # message from Nimble
      message('message message')
      cat('cat message\n')
    }, message = function(msg) {
      ## Output messages to stdout, so they end up in the file
      cat(conditionMessage(msg))
      ## Prevent messages from being relayed to the main R session
      invokeRestart("muffleMessage")
    })
    
    i
}

plan(sequential)

Personally, I would tidy it up by using a helper function that I can reuse, e.g.

library(doFuture)
library(nimble)

redirect_stdout_and_messages_to_file <- function(file, expr, envir = parent.frame()) {
  expr <- substitute(expr)
  
  ## Send all standard output to file
  outcon <- file(file, open = "wt")
  sink(outcon, type = "output")
  on.exit({
    sink(NULL, type = "output")
    close(outcon)
  })
  
  withCallingHandlers({
    eval(expr, envir = envir)
  }, message = function(msg) {
    ## Output message to stdout, so they end up in the file
    cat(conditionMessage(msg))
    ## Prevent messages from being relayed to the main R session
    invokeRestart("muffleMessage")
  })
} ## redirect_stdout_and_messages_to_file()


    
## Configure parallelization
plan(multisession, workers = 2)

y <- foreach(i = 1:2) %dofuture% {
    redirect_stdout_and_messages_to_file(paste0('testout', i, '.log'), {
      nimble::messageIfVerbose('nimble message') # message from Nimble
      message('message message')
      cat('cat message\n')
    })
    i
}

## Disable parallelization
plan(sequential)

Note that this is the approach you should take regardless of parallelization or not, e.g. with foreach(...) %dopar% { ... }, in a regular for-loop, or in an lapply() call. For example,

y <- list()
for(i in 1:2) {
  redirect_stdout_and_messages_to_file(paste0('testout', i, '.log'), {
    nimble::messageIfVerbose('nimble message') # message from Nimble
    message('message message')
    cat('cat message\n')
  })
  y[[i]] <- i
}

Details

There are a few things going on in your original example. I'll try to clarify what's going, that things are working as expected and by design, and how you can get what you want.

For a started, yes, correct nimble::messageIfVerbose() sends a simple message condition;

res <- tryCatch(nimble::messageIfVerbose('nimble message'), condition = identity)
str(res)
#> List of 2
#>  $ message: chr "nimble message\n"
#>  $ call   : language message(...)
#>  - attr(*, "class")= chr [1:3] "simpleMessage" "message" "condition"

So, in that sense, and for the discussion here, nimble::messageIfVerbose() is no different from message(). If they had extended class, e.g. nimbleMessage, then we could treat them separately, but given the facts, we don't have to worry about that.

Second, you can think of message conditions as being SMS/TXT messages. They are delieved to the reciever and there presented in one way or the other, depending on phone, apps, notification settings, smart watches, etc. In R we can "capture" these messages and handle them as we'd like. In Futureverse, the default is that futures captures them on the parallel worker and relays them back to you in the main R session. FWIW, no other parallel framework in R does this - for those frameworks messages (and warnings) often become completely invisible, or sometimes visible depending in what environment the user runs R, e.g. terminal, RStudio Console, ... In Futureverse, the behavior is consistent regardless of environment and parallel backend.

Third, output from cat(), print(), str() goes to standard output (stdout). That is not a condition - that's just a big internal buffer that gets outputted "somewhere". Similarly to conditions, Futureverse captures the standard output on the parallel workers by defaults and relays it to you in your main R session. The correct way to capture standard output is to use utils::capture.output(...), or low-level sink(..., type = "output") as you did.

Fourth, the correct way to capture messages is not analogue to howe we capture standard output. That is, we do not want to use utils::capture.output(..., type = "message") or low-level sink(..., type = "message"). The main reason is that it's better to capture messages via withCallingHandlers(..., message = ...) and do what we want there, i.e. forward them to the file. It is true that messages eventually end up on the "message" stream (= standard error) if we don't handle handle them. It is also true that sink(..., type = "message") would grab that output as a last resort, but it's much better to do it via withCallingHandlers(), despite being a bit of coding. That said, an even more important reason is that sink(..., type = "message") is not safe to use. No one should ever use it! That is because how R is designed. Any attempt to use it is unreliable and you can never be in full control of it. See HenrikBengtsson/Wishlist-for-R#55 for an explanation. That it worked with %dopar% and doParallel is basically mostly luck. So, just wipe utils::capture.output(..., type = "message") and sink(..., type = "message") from your memory and never look back. Please trust me on that one.