Merge pull request #15 from JuliaWeb/concurrency

Fix some concurrency bugs

docs/src/changelog.md

@@ -21,6 +21,8 @@ Changelog](https://keepachangelog.com).
 
 - Fixed segfaults that would occur in [`SshChannel`](@ref) when its
   [`Session`](@ref) is disconnected by the remote end ([#13]).
+- Fixed some concurrency bugs in the [`Demo.DemoServer`](@ref) and
+  [`SessionEvent`](@ref) ([#15]).
 
 ## [v0.5.0] - 2024-08-10
 

src/server.jl

@@ -6,11 +6,12 @@ $(TYPEDEF)
 $(TYPEDFIELDS)
 
 This object wraps a `lib.ssh_event`, but it's designed to only allow adding a
-single session to it. Use this in a server to poll the session.
+single session to it. Use this in a server to poll the session. It is threadsafe.
 """
 mutable struct SessionEvent
     ptr::Union{lib.ssh_event, Nothing}
     session::Session
+    lock::ReentrantLock
 
     @doc """
     $(TYPEDSIGNATURES)
@@ -29,18 +30,12 @@ mutable struct SessionEvent
             throw(LibSSHException("Could not add Session to SessionEvent: $(ret)"))
         end
 
-        self = new(ptr, session)
-        finalizer(close, self)
+        self = new(ptr, session, ReentrantLock())
+        finalizer(_finalizer, self)
     end
 end
 
-function _finalizer(event::SessionEvent)
-    try
-        close(event)
-    catch ex
-        Threads.@spawn @error "Exception when finalizing SessionEvent" exception=(ex, catch_backtrace())
-    end
-end
+_finalizer(event::SessionEvent) = close(event; unsafe=true)
 
 function Base.unsafe_convert(::Type{lib.ssh_event}, event::SessionEvent)
     if !isassigned(event)
@@ -50,6 +45,13 @@ function Base.unsafe_convert(::Type{lib.ssh_event}, event::SessionEvent)
     return event.ptr
 end
 
+Base.lock(event::SessionEvent) = lock(event.lock)
+Base.unlock(event::SessionEvent) = unlock(event.lock)
+
+function Base.show(io::IO, event::SessionEvent)
+    print(io, SessionEvent, "(ptr=$(event.ptr), session=$(event.session))")
+end
+
 """
 $(TYPEDSIGNATURES)
 
@@ -74,7 +76,7 @@ function event_dopoll(event::SessionEvent)
 
         # Always check if the event is still assigned within the loop since it
         # may be closed at any time.
-        ret = if isassigned(event)
+        ret = @lock event if isassigned(event)
             lib.ssh_event_dopoll(event, 0)
         else
             SSH_ERROR
@@ -97,11 +99,11 @@ Removes the [`Session`](@ref) from the underlying `ssh_event` and frees the
 event memory. This function may be safely called multiple times, and the event
 will be unusable afterwards.
 """
-function Base.close(event::SessionEvent)
+function Base.close(event::SessionEvent; unsafe=false)
     # Developer note: this function is called by the finalizer so we can't do
-    # any task switches, including print statements.
+    # any task switches if `unsafe=true`, including print statements.
 
-    if isassigned(event)
+    _close =  () -> if isassigned(event)
         # Attempt to remove the session. Note that we don't bother checking the
         # return code because some other callback may have already removed the
         # session, which will cause this to return SSH_ERROR.
@@ -113,6 +115,12 @@ function Base.close(event::SessionEvent)
         lib.ssh_event_free(event)
         event.ptr = nothing
     end
+
+    if unsafe
+        _close()
+    else
+        @lock event _close()
+    end
 end
 
 """
@@ -712,8 +720,8 @@ function Base.close(client::Client)
     end
 
     close(client.session_event)
-    close(client.session)
     wait(client.task)
+    close(client.session)
 end
 
 #=
@@ -825,8 +833,13 @@ function DemoServer(f::Function, args...; timeout=10, kill_timeout=3, kwargs...)
     start(demo_server)
 
     timer = Timer(timeout)
+    parent_testsets = get(task_local_storage(), :__BASETESTNEXT__, [])
     still_running = true
     t = Threads.@spawn try
+        # Copy the testsets from the parent task so all @test uses in f() get
+        # reported properly.
+        task_local_storage(:__BASETESTNEXT__, parent_testsets)
+
         f()
     finally
         still_running = false

test/LibSSHTests.jl

@@ -56,6 +56,30 @@ function http_server(f::Function, port)
     end
 end
 
+# Helper function to start a DemoServer and create a session connected to it
+function demo_server_with_session(f::Function, port, args...;
+                                  timeout=10,
+                                  kill_timeout=3,
+                                  password="foo",
+                                  log_verbosity=lib.SSH_LOG_NOLOG,
+                                  kwargs...)
+    demo_server = DemoServer(port, args...; password, timeout, kill_timeout, kwargs...) do
+        # Create a session
+        session = ssh.Session("127.0.0.1", port; log_verbosity)
+        @test ssh.isconnected(session)
+        @test ssh.userauth_password(session, password) == ssh.AuthStatus_Success
+
+        try
+            f(session)
+        finally
+            ssh.disconnect(session)
+            close(session)
+        end
+    end
+
+    return demo_server
+end
+
 @testset "Server" begin
     hostkey = joinpath(@__DIR__, "ed25519_test_key")
 
@@ -86,6 +110,24 @@ end
         @test server.ptr == nothing
     end
 
+    @testset "SessionEvent" begin
+        demo_server_with_session(2222; timeout=10) do session
+            event = ssh.SessionEvent(session)
+
+            # Smoke test
+            show(IOBuffer(), event)
+
+            @test event.ptr isa lib.ssh_event
+
+            close(event)
+            @test isnothing(event.ptr)
+
+            event = ssh.SessionEvent(session)
+            finalize(event)
+            @test isnothing(event.ptr)
+        end
+    end
+
     # Helper function to set up an `ssh` command. Slightly ugly workaround to
     # set the necessary environment variables comes from:
     # https://github.com/JuliaLang/julia/issues/39282
@@ -390,30 +432,6 @@ end
     end
 end
 
-# Helper function to start a DemoServer and create a session connected to it
-function demo_server_with_session(f::Function, port, args...;
-                                  timeout=10,
-                                  kill_timeout=3,
-                                  password="foo",
-                                  log_verbosity=lib.SSH_LOG_NOLOG,
-                                  kwargs...)
-    demo_server = DemoServer(port, args...; password, timeout, kill_timeout, kwargs...) do
-        # Create a session
-        session = ssh.Session("127.0.0.1", port; log_verbosity)
-        @test ssh.isconnected(session)
-        @test ssh.userauth_password(session, password) == ssh.AuthStatus_Success
-
-        try
-            f(session)
-        finally
-            ssh.disconnect(session)
-            close(session)
-        end
-    end
-
-    return demo_server
-end
-
 @testset "SshChannel" begin
     session = ssh.Session("localhost"; auto_connect=false)