try again

Author: quinnj

src/Connections.jl

@@ -299,7 +299,7 @@ function IOExtras.closeread(c::Connection)
     c.readable = false
     @debugv 3 "✉️  Read done: $c"
     if c.clientconnection
-        t = Threads.@spawn monitor_idle_connection(c)
+        t = @async monitor_idle_connection(c)
         @isdefined(errormonitor) && errormonitor(t)
     end
     return

src/Exceptions.jl

@@ -28,7 +28,7 @@ end # @eval
 function try_with_timeout(f, timeout)
     ch = Channel(0)
     timer = Timer(tm -> close(ch, TimeoutError(timeout)), timeout)
-    Threads.@spawn try
+    @async try
         put!(ch, $f())
     catch e
         if !(e isa HTTPError)

src/clientlayers/ConnectionRequest.jl

@@ -56,7 +56,7 @@ Otherwise leave it open so that it can be reused.
 """
 function connectionlayer(handler)
     return function connections(req; proxy=getproxy(req.url.scheme, req.url.host), socket_type::Type=TCPSocket, socket_type_tls::Type=SOCKET_TYPE_TLS[], readtimeout::Int=0, logerrors::Bool=false, kw...)
-        local io, stream
+        local io, stream, ct, sticky, tid
         if proxy !== nothing
             target_url = req.url
             url = URI(proxy)
@@ -77,6 +77,10 @@ function connectionlayer(handler)
         start_time = time()
         try
             io = newconnection(IOType, url.host, url.port; readtimeout=readtimeout, kw...)
+            ct = current_task()
+            sticky = ct.sticky
+            ct.sticky = true
+            tid = Threads.threadid()
         catch e
             if logerrors
                 @error "HTTP.ConnectError" exception=(e, catch_backtrace()) method=req.method url=req.url context=req.context
@@ -139,10 +143,13 @@ function connectionlayer(handler)
             e isa HTTPError || throw(RequestError(req, e))
             rethrow(e)
         finally
+            # restory stickiness
+            ct.sticky = sticky
             releaseconnection(io, shouldreuse; kw...)
             if !shouldreuse
                 @try Base.IOError close(io)
             end
+            @assert Threads.threadid() == tid
         end
     end
 end

src/clientlayers/StreamRequest.jl

@@ -33,15 +33,15 @@ function streamlayer(stream::Stream; iofunction=nothing, decompress::Union{Nothi
     try
         @sync begin
             if iofunction === nothing
-                Threads.@spawn try
+                @async try
                     writebody(stream, req)
                     @debugv 2 "client closewrite"
                     closewrite(stream)
                 finally
                     req.context[:write_duration_ms] = get(req.context, :write_duration_ms, 0.0) + ((time() - write_start) * 1000)
                 end
                 read_start = time()
-                Threads.@spawn try
+                @async try
                     @debugv 2 "client startread"
                     startread(stream)
                     if isaborted(stream)