Updates to enable repl_python() to use Positron Python kernel

DESCRIPTION

@@ -60,3 +60,4 @@ LinkingTo: Rcpp
 RoxygenNote: 7.3.2
 Roxygen: list(markdown = TRUE)
 VignetteBuilder: knitr
+Config/build/compilation-database: true

R/array.R

@@ -91,7 +91,7 @@ length.numpy.ndarray <- function(x) {
 array_reshape <- function(x, dim, order = c("C", "F")) {
   np <- import("numpy", convert = FALSE)
   order <- match.arg(order)
-  reshaped <- np$reshape(x, as.integer(dim), order)
+  reshaped <- np$reshape(x, as.integer(dim), order = order)
   if (!is_py_object(x))
     reshaped <- py_to_r(reshaped)
   reshaped

R/repl.R

@@ -89,6 +89,14 @@ repl_python <- function(
     on.exit(teardown(), add = TRUE)
   }
 
+  if(Sys.getenv("POSITRON") == "1" &&
+     numeric_version(Sys.getenv("POSITRON_VERSION")) > "2024.8.0") {
+    if(!is.null(input))
+      warning("`input` to repl_python() ignored.") # TODO: fix
+    eval(call(".ps.reticulate_open"))
+    return()
+  }
+
   # split provided code on newlines
   if (!is.null(input))
     input <- unlist(strsplit(input, "\n", fixed = TRUE))

R/thread.R

@@ -14,6 +14,51 @@
 #'
 #' @export
 py_main_thread_func <- function(f) {
-  tools <- import("rpytools")
-  tools$thread$main_thread_func(f)
+  r_to_py(f, TRUE) # every R func is a main thread func.
 }
+
+
+py_allow_threads <- function(allow = TRUE) {
+  if (allow) {
+    reticulate_ns <- environment(sys.function())
+    for (f in sys.frames()) {
+      if (identical(parent.env(f), reticulate_ns) &&
+          !identical(f, environment()))
+        # Can't release the gil as unlocked while we're holding it
+        # elsewhere on the callstack.
+        stop("Python threads can only be unblocked from a top-level reticulate call")
+    }
+  }
+
+  if (!was_python_initialized_by_reticulate())
+    stop("Can't safely unblock threads when R is running embedded")
+
+  invisible(py_allow_threads_impl(allow))
+}
+
+
+
+## TODO: document how to use sys.unraisablehook() to customize handling of exceptions
+## from background threads. Or, switch to using the threading module, which
+## has more options for customizing exceptions hooks, and document that.
+## TODO: give a meaningful name for the thread that appears in tracebacks.
+## Either use the threading module and pass name=,
+##   or do something like
+##     f = lambda file: run_file(file)
+##     f.__name__ = "run: " + os.path.basename(file)
+py_run_file_on_thread <- function(file, ..., args = NULL) {
+  if (!is.null(args))
+    args <- as.list(as.character(args))
+  import("rpytools.run")$`_launch_lsp_server_on_thread`(file, args)
+}
+
+## used in Positron:
+# reticulate:::py_run_file_on_thread(
+#   file = "${kernelPath}",
+#   args = c(
+#     "-f", "${connnectionFile}",
+#     "--logfile", "${logFile}",
+#     "--loglevel", "${logLevel}",
+#     "--session-mode", "console"
+#   )
+# )

inst/python/rpytools/generator.py

@@ -46,7 +46,7 @@ def _tend_queue(self, min_fetch=0):
         if threading.current_thread() is not threading.main_thread():
             if not self._pending_tend_queue:
                 self._pending_tend_queue = True
-                rpycall.call_python_function_on_main_thread(
+                rpycall.schedule_python_function_on_main_thread(
                     self._tend_queue, min_fetch
                 )
             return

inst/python/rpytools/run.py

@@ -6,45 +6,56 @@ def run_file(path):
     with open(path, "r") as file:
         file_content = file.read()
 
-    d = sys.modules["__main__"].__dict__
+    # ipykernel patches the loader, so that
+    # `import __main__` does not produce the "real" __main__, but rather,
+    # the facade that is the user facing __main__
+    # to get the "real" main, do:
+    # d = sys.modules["__main__"].__dict__
+    from __main__ import __dict__ as d
 
     exec(file_content, d, d)
 
 
 class RunMainScriptContext:
-    def __init__(self, path, args):
+    def __init__(self, path, argv=None):
         self.path = path
-        self.args = tuple(args)
+        self.argv = tuple(argv) if argv is not None else None
 
     def __enter__(self):
         sys.path.insert(0, os.path.dirname(self.path))
 
-        self._orig_sys_argv = sys.argv
-        sys.argv = [self.path] + list(self.args)
+        if self.argv is not None:
+            self._orig_sys_argv = sys.argv
+            sys.argv = [self.path] + list(self.argv)
 
     def __exit__(self, *_):
         # try restore sys.path
         try:
             sys.path.remove(os.path.dirname(self.path))
         except ValueError:
             pass
-        # restore sys.argv if it's been unmodified
-        # otherwise, leave it as-is.
-        set_argv = [self.path] + list(self.args)
-        if sys.argv == set_argv:
-            sys.argv = self._orig_sys_argv
+        # try restore sys.argv if we patched it
+        if self.argv is not None:
+            # restore sys.argv if it's unmodified from what we set it to.
+            # otherwise, leave it as-is.
+            patched_argv = [self.path] + list(self.args)
+            if sys.argv == patched_argv:
+                sys.argv = self._orig_sys_argv
 
 
-def _run_file_on_thread(path, args=None):
+def _launch_lsp_server_on_thread(path, args):
+    # used by Positron reticulate launcher...
+    # TODO: update Positron to replace usage of this with `run_file_on_thread()`
 
-    import _thread
+    return run_file_on_thread(path, args)
 
-    _thread.start_new_thread(run_file, (path, ))
 
 
-def _launch_lsp_server_on_thread(path, args):
+def run_file_on_thread(path, args=None):
     # for now, leave sys.argv and sys.path permanently modified.
     # Later, revisit if it's desirable/safe to restore after the initial
     # lsp event loop startup.
     RunMainScriptContext(path, args).__enter__()
-    _run_file_on_thread(path)
+    import _thread
+
+    _thread.start_new_thread(run_file, (path,))

inst/python/rpytools/thread.py

@@ -1,28 +1,19 @@
-import rpycall
-import threading
-
 import sys
+import queue as queue
+from rpycall import call_r_function, schedule_python_function_on_main_thread
 
-is_py2 = sys.version[0] == "2"
-if is_py2:
-    import Queue as queue
-else:
-    import queue as queue
-
-
-def main_thread_func(f):
-
-    def python_function(*args, **kwargs):
 
-        if isinstance(threading.current_thread(), threading._MainThread):
-            res = f(*args, **kwargs)
-        else:
-            result = queue.Queue()
-            rpycall.call_python_function_on_main_thread(
-                lambda: result.put(f(*args, **kwargs)), None
-            )
-            res = result.get()
+def safe_call_r_function(f, args, kwargs):
+    try:
+        return call_r_function(f, *args, **kwargs), None
+    except Exception as e:  # TODO: should we catch BaseException too? KeyboardInterrupt?
+        return None, e
 
-        return res
 
-    return python_function
+def safe_call_r_function_on_main_thread(f, *args, **kwargs):
+    result = queue.Queue()
+    schedule_python_function_on_main_thread(
+        lambda: result.put(safe_call_r_function(f, args, kwargs)),
+        None,
+    )
+    return result.get()

src/libpython.cpp

@@ -196,6 +196,7 @@ bool LibPython::loadSymbols(bool python3, std::string* pError)
   LOAD_PYTHON_SYMBOL(Py_IsInitialized)
   LOAD_PYTHON_SYMBOL(Py_GetVersion)
   LOAD_PYTHON_SYMBOL(Py_AddPendingCall)
+  LOAD_PYTHON_SYMBOL(Py_MakePendingCalls)
   LOAD_PYTHON_SYMBOL(PyErr_SetInterrupt)
   LOAD_PYTHON_SYMBOL(PyErr_CheckSignals)
   LOAD_PYTHON_SYMBOL(Py_IncRef)

src/libpython.h

@@ -206,6 +206,7 @@ LIBPYTHON_EXTERN wchar_t* (*Py_GetProgramFullPath)();
 
 
 LIBPYTHON_EXTERN int (*Py_AddPendingCall)(int (*func)(void *), void *arg);
+LIBPYTHON_EXTERN int (*Py_MakePendingCalls)();
 LIBPYTHON_EXTERN void (*PyErr_SetInterrupt)();
 LIBPYTHON_EXTERN void (*PyErr_CheckSignals)();
 

src/pending_py_calls_notifier.cpp

@@ -0,0 +1,131 @@
+
+#ifdef _WIN32
+#define WIN32_LEAN_AND_MEAN
+#endif
+
+#include <atomic>
+#include <functional>
+
+// tinythread.h is only used for its includes. It does platform-specific includes
+// nicely like <windows.h>, <unistd.h>, etc.
+#include "tinythread.h"
+
+#define R_NO_REMAP
+#include <Rinternals.h>
+
+#ifndef _WIN32
+#include <R_ext/eventloop.h> // for addInputHandler(), removeInputHandler()
+#endif
+
+#include "pending_py_calls_notifier.h"
+
+namespace pending_py_calls_notifier {
+
+namespace {
+
+std::atomic<bool> notification_pending(false);
+std::function<void()> run_pending_calls;
+
+}
+
+#ifdef _WIN32
+
+namespace {
+
+HWND message_window;
+const UINT WM_PY_PENDING_CALLS = WM_USER + 1;
+
+LRESULT CALLBACK WindowProc(HWND hwnd, UINT uMsg, WPARAM wParam, LPARAM lParam) {
+  if (uMsg == WM_PY_PENDING_CALLS) {
+    if (notification_pending.exchange(false)) {
+      run_pending_calls();
+    }
+    return 0;
+  }
+  return DefWindowProc(hwnd, uMsg, wParam, lParam);
+}
+
+void initialize_windows_message_window() {
+  HINSTANCE hInstance = GetModuleHandle(NULL);
+  WNDCLASS wc = {0};
+  wc.lpfnWndProc = WindowProc;
+  wc.hInstance = hInstance;
+  wc.lpszClassName = TEXT("ReticulatePyPendingCallsNotifier");
+
+  RegisterClass(&wc);
+  message_window = CreateWindow(TEXT("ReticulatePyPendingCallsNotifier"), NULL, 0, 0, 0, 0, 0, HWND_MESSAGE, NULL, hInstance, NULL);
+}
+
+} // end anonymous namespace, windows-specific
+
+
+void initialize(std::function<void()> run_pending_calls_func) {
+  initialize_windows_message_window();
+}
+void notify() {
+    PostMessage(message_window, WM_PY_PENDING_CALLS, 0, 0);
+}
+
+void deinitialize() {
+  if (message_window) {
+    DestroyWindow(message_window);
+    message_window = nullptr;
+  }
+}
+
+#else // end windows, start unix
+
+namespace {
+
+int pipe_fds[2]; // Pipe file descriptors for inter-thread communication
+InputHandler* input_handler = nullptr;
+
+void input_handler_function(void* userData) {
+  char buffer[4];
+  if (read(pipe_fds[0], buffer, sizeof(buffer)) == -1) // Clear the pipe
+    REprintf("Failed to read from pipe for pending Python calls notifier");
+  if (notification_pending.exchange(false)) {
+    run_pending_calls();
+  }
+}
+
+} // end anonymous namespace, unix-specific
+
+
+void initialize(std::function<void()> run_pending_calls_func) {
+  run_pending_calls = run_pending_calls_func;
+  if (pipe(pipe_fds) == -1)
+    Rf_error("Failed to create pipe for pending Python calls notifier");
+
+  input_handler = addInputHandler(R_InputHandlers, pipe_fds[0], input_handler_function, 88);
+}
+
+void notify() {
+  if (!notification_pending.exchange(true)) {
+    if (write(pipe_fds[1], "x", 1) == -1) {
+      // Called from background threads, can't throw R error.
+      REprintf("Failed to write to pipe for pending Python calls notifier\n");
+    }
+  }
+}
+
+void deinitialize() {
+  if (input_handler) {
+    removeInputHandler(&R_InputHandlers, input_handler);
+    input_handler = nullptr;
+  }
+
+  if (pipe_fds[0] != -1) {
+    close(pipe_fds[0]);
+    pipe_fds[0] = -1;
+  }
+
+  if (pipe_fds[1] != -1) {
+    close(pipe_fds[1]);
+    pipe_fds[1] = -1;
+  }
+}
+
+#endif // end unix
+
+} // namespace pending_py_calls_notifier

src/pending_py_calls_notifier.h

@@ -0,0 +1,19 @@
+#ifndef PENDING_PY_CALLS_NOTIFIER_H
+#define PENDING_PY_CALLS_NOTIFIER_H
+
+#include <functional>
+
+namespace pending_py_calls_notifier {
+
+    // Initialize the notifier with a function that runs pending calls.
+    void initialize(std::function<void()> run_pending_calls);
+
+    // Notify the main thread to run pending calls.
+    void notify();
+
+    // Undo initialize
+    void deinitialize();
+
+} // namespace pending_py_calls_notifier
+
+#endif // PENDING_PY_CALLS_NOTIFIER_H