Problem with interrupting 'parallelFor()' execution on Windows machines

[AndreySamokhin]

Hello, and thank you for creating this package for R. I encountered a potential issue with the parallelFor() function on Windows machines when a C++ function is interrupted by the user from R. Below is a minimal example to reproduce the issue. I tried using the sourceCpp() function and building an R package with the same C++ code. The results were identical. The number of selected threads did not influence the result either.

I ran this code on three machines:

• Intel Core i7-2720QM, Linux Mint 22, R version 4.3.3
• Intel Core i7-4790K, Windows 10 x64 (build 19045), R version 4.1.3
• AMD Ryzen 5 5625U, Windows 10 x64 (build 19045), R version 4.4.1

library(Rcpp)

sourceCpp(code = '
#include <Rcpp.h>
#include <RcppThread.h>
#include <thread>
#include <chrono>

using namespace Rcpp;

// [[Rcpp::depends(RcppThread)]]

// [[Rcpp::export]]
bool TestUserInterrupt(int n_threads = 1L) {
  RcppThread::Rcout << "Main thread ID: " << std::this_thread::get_id()
                    << std::endl;
  RcppThread::parallelFor(0, 5 * 10 * n_threads, [](int i) {
    RcppThread::Rcout << std::this_thread::get_id() << "|";
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
  }, n_threads);
  RcppThread::Rcout << std::endl << "Processing complete." << std::endl;
  return true;
}
')

TestUserInterrupt(1)

On the Linux machine, everything worked as expected (the "Error: C++ call interrupted by the user" message was printed to the console, and the function could be rerun).

On the Windows machines, however, when the function was interrupted, the output to the console stopped without displaying the "Error: C++ call interrupted by the user" message. A subsequent call to the function resulted in no output: R was busy, and the interruption was not possible without restarting the R session. I am not sure if it is relevant, but when using multiple threads (e.g., 4), the output was sometimes shuffled (e.g., "... 4|||53|4||2|352|||4 ...").

One more remark. Initially I encountered a problem when I tried to integrate RcppThread into another R package. In that case, R crashed when the execution was interrupted by the user from R.

[aitap]

To save some time debugging: this doesn't happen on Rterm.exe, but it does happen on Rgui.exe. The second call to TestUserInterrupt(1) hangs in:

#5  std::mutex::lock (
    this=0x7ff859773040 <RcppThread::RMonitor::instance()::instance_>)
    at c:/rtools44/x86_64-w64-mingw32.static.posix/lib/gcc/x86_64-w64-mingw32.static.posix/13.3.0/include/c++/bits/std_mutex.h:113
#6  std::lock_guard<std::mutex>::lock_guard (__m=...,
    this=<synthetic pointer>)
    at c:/rtools44/x86_64-w64-mingw32.static.posix/lib/gcc/x86_64-w64-mingw32.static.posix/13.3.0/include/c++/bits/std_mutex.h:249
#7  RcppThread::RMonitor::safelyPrint<char [17]> (
    this=0x7ff859773040 <RcppThread::RMonitor::instance()::instance_>,
    object=...)
    at C:/R/R-svn/library/RcppThread/include/RcppThread/RMonitor.hpp:95
#8  0x00007ff8596b2196 in RcppThread::RPrinter::operator<< <char [17]> (
    object=..., this=<optimized out>)
    at C:/R/R-svn/library/RcppThread/include/RcppThread/Rcout.hpp:36
#9  TestUserInterrupt (n_threads=1) at file1460219f4f1d.cpp:13

...which suggests that the mutex in the RMonitor instance is left locked after R is interrupted. Why would that happen when RcppThread carefully wraps its own checks in R_ToplevelExec?

RcppThread/inst/include/RcppThread/RMonitor.hpp

Lines 127 to 138 in 3d47c8a

	//! calls R's API to check for user interrupt.
	//!
	//! Since the R API is not thread safe, it must be called in top-level
	//! context to avoid long-jumping out of the current context.
	static void callRCheck(void *dummy) {
	R_CheckUserInterrupt();
	}
	//! checks for R user interruptions in top level context.
	inline bool isInterrupted() {
	return (R_ToplevelExec(callRCheck, NULL) == FALSE);
	}

That's because printing in Rgui requires processing GUI events, which performs its own interrupt checks:

#0  0x00007ff878cd8a1a in ucrtbase!longjmp ()
   from /c/Windows/System32/ucrtbase.dll
#1  0x00007ff84c28ef50 in R_jumpctxt (targetcptr=<optimized out>, mask=1,
    mask@entry=0, val=val@entry=0x0) at context.c:244
#2  0x00007ff84c2c04dc in jump_to_top_ex (traceback=traceback@entry=TRUE,
    tryUserHandler=tryUserHandler@entry=TRUE,
    processWarnings=processWarnings@entry=TRUE,
    resetConsole=resetConsole@entry=TRUE,
    ignoreRestartContexts=ignoreRestartContexts@entry=FALSE)
    at errors.c:1080
#3  0x00007ff84c2c67b6 in onintrEx (resumeOK=resumeOK@entry=TRUE)
    at errors.c:201
#4  0x00007ff84c2c67ea in Rf_onintr () at errors.c:204 # <-- R interrupts itself...
#5  0x00007ff84c20f67f in R_ProcessEvents () at system.c:154 # <-- ...because it saw the interrupt button pressed...
#6  0x00007ff84c20f8b0 in R_WriteConsole (buf=0xa123faf70 "2|", len=2)
    at system.c:266 # <-- ...while trying to print
#7  0x00007ff84c361b38 in Rcons_vprintf (format=<optimized out>,
    arg=<optimized out>) at printutils.c:963
#8  0x00007ff84c361bf4 in Rvprintf (format=0x7ff8559f45c3 "%s",
    arg=0xa123fd038 "\220\320?\022\n") at printutils.c:985
#9  0x00007ff84c361c62 in Rprintf (
    format=0x7ff84e875170 <R_Toplevel+16> "P\367?\022\n")
    at printutils.c:898
#10 0x00007ff85594ebc1 in RcppThread::RMonitor::safelyPrint<char [1]> (
    this=0x7ff8559f3040 <RcppThread::RMonitor::instance()::instance_>,
    object=...)
    at c:/rtools44/x86_64-w64-mingw32.static.posix/lib/gcc/x86_64-w64-mingw32.static.posix/13.3.0/include/c++/bits/basic_string.h:222
#11 0x00007ff85594e468 in RcppThread::RPrinter::operator<< <char [1]> (
    object=..., this=<optimized out>) # <-- parallelFor() flushes the print buffer
    at C:/R/R-svn/library/RcppThread/include/RcppThread/Rcout.hpp:34
#12 RcppThread::ThreadPool::wait (this=0x11bf23d5230)
    at C:/R/R-svn/library/RcppThread/include/RcppThread/ThreadPool.hpp:249
#13 0x00007ff8559324ac in RcppThread::parallelFor<TestUserInterrupt(int)::<lambda(int)> > (begin=0, nBatches=0, f=..., nThreads=<optimized out>,
    end=<optimized out>)
    at C:/R/R-svn/library/RcppThread/include/RcppThread/parallelFor.hpp:51
#14 TestUserInterrupt (n_threads=<optimized out>)
    at filea101ee8296b.cpp:15

Moving the interrupt check before printing will make the problem less likely to happen, but it will leave a window of opportunity for the user to press interrupt after the check but before printing is complete:

RcppThread/inst/include/RcppThread/ThreadPool.hpp

Lines 249 to 251 in 3d47c8a

	Rcout << "";
	Rcerr << "";
	checkUserInterrupt();

I'm not exactly sure, but a complete solution may involve printing from R_ToplevelExec or using Rcpp's unwindProtect to make sure that the destructors will fire. A good question for R-package-devel.

[AndreySamokhin]

I stand corrected, my initial example is not perfect as it works correctly when the RcppThread::Rcout ... line is commented out. I have prepared another example that demonstrates the original issue more clearly.

On both Windows and Linux, R crashes when the function is manually interrupted from R. The only difference is that on Linux, the "Error: C++ call interrupted by the user" message is displayed just before the crash, whereas on Windows, this message does not appear.

library(Rcpp)

sourceCpp(code = '
#include <Rcpp.h>
#include <RcppThread.h>
#include <thread>
#include <chrono>
#include <vector>
#include <cmath>

using namespace Rcpp;

void LoadCpu(std::vector<double>& long_vec,
             std::vector<double>& score,
             int n_threads) {
  
  const int iterations = 1000;
  const int subvec_size = 100;
  size_t n_vectors = long_vec.size() / subvec_size;
  double out = 0;
  RcppThread::parallelFor(0, n_vectors, [&] (unsigned int i) {
    for (int j = 0; j < iterations; j++) {
      int k2 = j % subvec_size;
      for (size_t k = 0; k < subvec_size; k++) {
        score[i] += std::sin(long_vec[subvec_size * i + k]) *
          std::cos(long_vec[subvec_size * i + k]) *
          std::sqrt(std::abs(long_vec[subvec_size * i + k2] * 3.14159));
      }
    }
  }, n_threads);
}

// [[Rcpp::depends(RcppThread)]]

// [[Rcpp::export]]
double TestUserInterrupt3(NumericVector long_vec,
                          int n_threads) {
  
  const int subvec_size = 100;
  std::vector<double> score(long_vec.size() / subvec_size);
  std::vector<double> cpp_long_vec(long_vec.begin(), long_vec.end());
  LoadCpu(cpp_long_vec, score, n_threads);
  return std::accumulate(score.begin(), score.end(), 0.0);
}
')

x <- rnorm(100L * 5000L)
TestUserInterrupt3(x, n_threads = 1L)

[aitap]

That's another problem. In RcppThread::parallelFor, when the wait() method throws a UserInterruptException, the thread pool keeps working, accessing the local variables belonging to the main thread, while the stack of the main thread is being unwound to the exception handler, destroying the local variables in the process.

Even without the crash due to local variable access, you can try to interrupt a parallelFor and see that R remains busy consuming CPU time for a long time after the main thread returns to the prompt. Moreover, submitting a new task to the pool won't work:

library(Rcpp)

sourceCpp(code = '
#include <Rcpp.h>
#include <RcppThread.h>
using namespace Rcpp;

// [[Rcpp::depends(RcppThread)]]
// [[Rcpp::export]]
double TestUserInterrupt4(NumericVector x, int n_threads, int sqrt_iter, double xnew) {
  double * px = &x[0];
  RcppThread::parallelFor(0, x.length(), [sqrt_iter, xnew, px] (unsigned int k) {
    if (k == 0) RcppThread::Rcout << "interrupt me:";
    for (volatile int i = 0; i < sqrt_iter; ++i)
    for (volatile int j = 0; j < sqrt_iter; ++j)
      px[k] = xnew;
  }, n_threads);
}
')

x <- rnorm(100L * 5000L)
TestUserInterrupt4(x, n_threads = 10L, 1000000, 42)
# interrupt me:^C
# Error: C++ call interrupted by the user.
TestUserInterrupt4(x, n_threads = 10L, 1, 1)
# <still busy with the previous task>
# ^C
# Error: C++ call interrupted by the user.
q()
# Save workspace image? [y/n/c]: n
# <now consumes all CPU cores instead of 10, doesn't quit>

I think one possible solution here is to catch the exception in ThreadPool::wait, stop the pool, wait until the threads actually stop executing their jobs, and then rethrow the exception.