Support for new status and forceclose server commands

CHANGELOG.md

@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- New socket server command `status` that returns a string describing the current server status, including information for each active worker, as well as `forceclose` which is the forced version of `close` that can shut down a worker even if it's currently running [#229](https://github.com/PumasAI/QuartoNotebookRunner.jl/pull/229).
+
+### Changed
+
+- Socket server commands `run`, `close` and `stop` will error if a file lock is currently held (a file worker is running) instead of waiting potentially a long time for that lock to open. Those locks were always intended only as mutation protection and not a queueing mechanism, they only behaved that way by accident and in case some worker hangs, it's impractical that further `quarto render` commands just add on top of the pile without a message [#229](https://github.com/PumasAI/QuartoNotebookRunner.jl/pull/229).
+
 ## [v0.14.0] - 2025-02-26
 
 ### Added

src/QuartoNotebookRunner.jl

@@ -43,6 +43,10 @@ export Server, render, run!, close!
 
 # Includes.
 
+const QNR_VERSION =
+    VersionNumber(TOML.parsefile(joinpath(@__DIR__, "..", "Project.toml"))["version"])
+include_dependency(joinpath(@__DIR__, "..", "Project.toml"))
+
 include("UserError.jl")
 include("Malt.jl")
 include("WorkerSetup.jl")

src/server.jl

@@ -11,6 +11,9 @@
     lock::ReentrantLock
     timeout::Float64
     timeout_timer::Union{Nothing,Timer}
+    run_started::Union{Nothing,Dates.DateTime}
+    run_finished::Union{Nothing,Dates.DateTime}
+    run_decision_channel::Channel{Symbol}
 
     function File(path::String, options::Union{String,Dict{String,Any}})
         if isfile(path)
@@ -24,6 +27,7 @@
                 exeflags, env = _exeflags_and_env(merged_options)
                 timeout = _extract_timeout(merged_options)
 
+
                 exe, _exeflags = _julia_exe(exeflags)
                 worker =
                     cd(() -> Malt.Worker(; exe, exeflags = _exeflags, env), dirname(path))
@@ -38,6 +42,9 @@
                     ReentrantLock(),
                     timeout,
                     nothing,
+                    nothing,
+                    nothing,
+                    Channel{Symbol}(32), # we don't want an unbuffered channel because we might want to `put!` to it without blocking
                 )
                 init!(file, merged_options)
                 return file
@@ -1427,41 +1434,96 @@
     options::Union{String,Dict{String,Any}} = Dict{String,Any}(),
     chunk_callback = (i, n, c) -> nothing,
 )
-    borrow_file!(server, path; optionally_create = true) do file
-        if file.timeout_timer !== nothing
-            close(file.timeout_timer)
-            file.timeout_timer = nothing
-        end
-        result = evaluate!(file, output; showprogress, options, markdown, chunk_callback)
-        if file.timeout > 0
-            file.timeout_timer = Timer(file.timeout) do _
+    try
+        borrow_file!(server, path; optionally_create = true) do file
+            if file.timeout_timer !== nothing
+                close(file.timeout_timer)
+                file.timeout_timer = nothing
+            end
+            file.run_started = Dates.now()
+            file.run_finished = nothing
+
+            # we want to be able to force close the worker while `evaluate!` is running,
+            # so we run `evaluate!` in a task and wait for the `run_decision_channel`
+            # further down. Depending on the value fetched from that channel, we either
+            # know that evaluation has finished, or that force closing was requested.
+            while !isempty(file.run_decision_channel)
+                take!(file.run_decision_channel) # empty! not defined on channels in earlier julia versions
+            end
+
+            result_task = Threads.@spawn begin
+                try
+                    evaluate!(file, output; showprogress, options, markdown, chunk_callback)
+                finally
+                    put!(file.run_decision_channel, :evaluate_finished)
+                end
+            end
+
+            # block until a decision is reached
+            decision = take!(file.run_decision_channel)
+
+            # :forceclose might have been set from another task
+            if decision === :forceclose
+                close!(server, file.path) # this is in the same task, so reentrant lock allows access
+                error("File was force-closed during run")
+            elseif decision === :evaluate_finished
+                result = try
+                    fetch(result_task)
+                catch err
+                    # throw the original exception, not the wrapping TaskFailedException
+                    rethrow(err.task.exception)
+                end
+            else
+                error("Invalid decision $decision")
+            end
+
+            file.run_finished = Dates.now()
+            if file.timeout > 0
+                file.timeout_timer = Timer(file.timeout) do _
+                    close!(server, file.path)
+                    @debug "File at $(file.path) timed out after $(file.timeout) seconds of inactivity."
+                end
+            else
                 close!(server, file.path)
-                @debug "File at $(file.path) timed out after $(file.timeout) seconds of inactivity."
             end
+            return result
+        end
+    catch err
+        if err isa FileBusyError
+            throw(
+                UserError(
+                    "Tried to run file \"$path\" but the corresponding worker is busy.",
+                ),
+            )
         else
-            close!(server, file.path)
+            rethrow(err)
         end
-        return result
     end
 end
 
 struct NoFileEntryError <: Exception
     path::String
 end
 
+struct FileBusyError <: Exception
+    path::String
+end
+
 """
-    borrow_file!(f, server, path; optionally_create = false, options = Dict{String,Any}())
+    borrow_file!(f, server, path; wait = false, optionally_create = false, options = Dict{String,Any}())
 
 Executes `f(file)` while the `file`'s `ReentrantLock` is locked.
 All actions on a `Server`'s `File` should be wrapped in this
 so that no two tasks can mutate the `File` at the same time.
 When `optionally_create` is `true`, the `File` will be created on the server
 if it doesn't exist, in which case it is passed `options`.
+If `wait = false`, `borrow_file!` will throw a `FileBusyError` if the lock cannot be attained immediately.
 """
 function borrow_file!(
     f,
     server,
     path;
+    wait = false,
     optionally_create = false,
     options = Dict{String,Any}(),
 )
@@ -1497,7 +1559,18 @@
         # no file exists or it doesn't match the one we have, we recurse into `borrow_file!`.
         # This could in principle go on forever but is very unlikely to with a small number of
         # concurrent users.
-        lock(file.lock) do
+
+        if wait
+            lock(file.lock)
+            lock_attained = true
+        else
+            lock_attained = trylock(file.lock)
+        end
+
+        try
+            if !lock_attained
+                throw(FileBusyError(apath))
+            end
             current_file = lock(server.lock) do
                 get(server.workers, apath, nothing)
             end
@@ -1506,6 +1579,8 @@
             else
                 return f(file)
             end
+        finally
+            lock_attained && unlock(file.lock)
         end
     end
 end
@@ -1560,14 +1635,55 @@
         end
         return true
     catch err
-        if !(err isa NoFileEntryError)
+        if err isa FileBusyError
+            throw(
+                UserError(
+                    "Tried to close file \"$path\" but the corresponding worker is busy.",
+                ),
+            )
+        elseif !(err isa NoFileEntryError)
             rethrow(err)
         else
             false
         end
     end
 end
 
+function forceclose!(server::Server, path::String)
+    apath = abspath(path)
+    file = lock(server.lock) do
+        if haskey(server.workers, apath)
+            return server.workers[apath]
+        else
+            throw(NoFileEntryError(apath))
+        end
+    end
+    # if the worker is not actually running we need to fall back to normal closing,
+    # for that we try to get the file lock now
+    lock_attained = trylock(file.lock)
+    try
+        # if we've attained the lock, we can close normally
+        if lock_attained
+            close!(server, path)
+            # but if not, we request a forced close via the run decision channel that
+            # is being waited for in `run!` function
+        else
+            put!(file.run_decision_channel, :forceclose)
+            t = time()
+            while Malt.isrunning(file.worker)
+                timeout = 10
+                (time() - t) > timeout && error(
+                    "Force close was requested but worker was still running after $timeout seconds.",
+                )
+                sleep(0.1)
+            end
+        end
+    finally
+        lock_attained && unlock(file.lock)
+    end
+    return
+end
+
 json_reader(str) = JSON3.read(str, Any)
 yaml_reader(str) = YAML.load(str)