Separate socket locks into send & recv locks

Project.toml

@@ -27,4 +27,4 @@ JSON = "0.18,0.19,0.20,0.21,1"
 MbedTLS = "0.5,0.6,0.7,1"
 SoftGlobalScope = "1"
 ZMQ = "1"
-julia = "1.6"
+julia = "1.10"

src/init.jl

@@ -28,7 +28,8 @@ const heartbeat = Ref{Socket}()
 const profile = Dict{String,Any}()
 const read_stdout = Ref{Base.PipeEndpoint}()
 const read_stderr = Ref{Base.PipeEndpoint}()
-const socket_locks = Dict{Socket,ReentrantLock}()
+const socket_locks_recv = Dict{Socket,ReentrantLock}()
+const socket_locks_send = Dict{Socket,ReentrantLock}()
 
 # needed for executing pkg commands on earlier Julia versions
 @static if VERSION < v"1.11"
@@ -97,8 +98,9 @@ function init(args)
 
     # associate a lock with each socket so that multi-part messages
     # on a given socket don't get inter-mingled between tasks.
-    for s in (publish[], raw_input[], requests[], control[], heartbeat[])
-        socket_locks[s] = ReentrantLock()
+    for s in (publish[], raw_input[], requests[], control[])
+        socket_locks_send[s] = ReentrantLock()
+        socket_locks_recv[s] = ReentrantLock()
     end
 
     start_heartbeat(heartbeat[])

src/msg.jl

@@ -41,8 +41,7 @@ function show(io::IO, msg::Msg)
 end
 
 function send_ipython(socket, m::Msg)
-    lock(socket_locks[socket])
-    try
+    @lock socket_locks_send[socket] begin
         @vprintln("SENDING ", m)
         for i in m.idents
             send(socket, i, more=true)
@@ -57,14 +56,11 @@ function send_ipython(socket, m::Msg)
         send(socket, parent_header, more=true)
         send(socket, metadata, more=true)
         send(socket, content)
-    finally
-        unlock(socket_locks[socket])
     end
 end
 
 function recv_ipython(socket)
-    lock(socket_locks[socket])
-    try
+    @lock socket_locks_recv[socket] begin
         idents = String[]
         s = recv(socket, String)
         @vprintln("got msg part $s")
@@ -85,8 +81,6 @@ function recv_ipython(socket)
         m = Msg(idents, JSON.parse(header), JSON.parse(content), JSON.parse(parent_header), JSON.parse(metadata))
         @vprintln("RECEIVED $m")
         return m
-    finally
-        unlock(socket_locks[socket])
     end
 end