add concurrent connection semaphore to prevent DNS resolve races (#205)

The libcurl spawns a new thread for each DNS resolution that it needs to
do, which can cause problems when too many connections are started at
the same time. This adds a semaphore which is acqured when a connection
starts and gets released once the actual request starts. This limits the
number of concurrent requests that are in the connecting state to the
size of the semaphore (16). Hopefully this prevents the "getaddrinfo()
thread failed" errors that occur sometimes.

Author: StefanKarpinski

src/Curl/Curl.jl

@@ -33,6 +33,8 @@ using LibCURL: curl_off_t, libcurl
 # constants that LibCURL should have but doesn't
 const CURLE_PEER_FAILED_VERIFICATION = 60
 const CURLSSLOPT_REVOKE_BEST_EFFORT = 1 << 3
+const CURLOPT_PREREQFUNCTION = 20000 + 312
+const CURLOPT_PREREQDATA = 10000 + 313
 
 # these are incorrectly defined on Windows by LibCURL:
 if Sys.iswindows()

src/Curl/Easy.jl

@@ -10,6 +10,7 @@ mutable struct Easy
     code     :: CURLcode
     errbuf   :: Vector{UInt8}
     debug    :: Union{Function,Nothing}
+    consem   :: Bool
 end
 
 const EMPTY_BYTE_VECTOR = UInt8[]
@@ -27,6 +28,7 @@ function Easy()
         typemax(CURLcode),
         zeros(UInt8, CURL_ERROR_SIZE),
         nothing,
+        false,
     )
     finalizer(done!, easy)
     add_callbacks(easy)
@@ -35,13 +37,38 @@ function Easy()
 end
 
 function done!(easy::Easy)
+    connect_semaphore_release(easy)
     easy.handle == C_NULL && return
     curl_easy_cleanup(easy.handle)
     curl_slist_free_all(easy.req_hdrs)
     easy.handle = C_NULL
     return
 end
 
+# connect semaphore
+
+# This semaphore limits the number of requests that can be in the connecting
+# state at any given time, globally. Throttling this prevents libcurl from
+# trying to start too many DNS resolver threads concurrently. It also helps
+# ensure that not-yet-started requests get ßa chance to make some progress
+# before adding more events from new requests to the system's workload.
+
+const CONNECT_SEMAPHORE = Base.Semaphore(16) # empirically chosen (ie guessed)
+
+function connect_semaphore_acquire(easy::Easy)
+    @assert !easy.consem
+    Base.acquire(CONNECT_SEMAPHORE)
+    easy.consem = true
+    return
+end
+
+function connect_semaphore_release(easy::Easy)
+    easy.consem || return
+    Base.release(CONNECT_SEMAPHORE)
+    easy.consem = false
+    return
+end
+
 # request options
 
 function set_defaults(easy::Easy)
@@ -309,6 +336,18 @@ end
 
 # callbacks
 
+function prereq_callback(
+    easy_p           :: Ptr{Cvoid},
+    conn_remote_ip   :: Ptr{Cchar},
+    conn_local_ip    :: Ptr{Cchar},
+    conn_remote_port :: Cint,
+    conn_local_port  :: Cint,
+)::Cint
+    easy = unsafe_pointer_to_objref(easy_p)::Easy
+    connect_semaphore_release(easy)
+    return 0
+end
+
 function header_callback(
     data   :: Ptr{Cchar},
     size   :: Csize_t,
@@ -424,6 +463,12 @@ function add_callbacks(easy::Easy)
     errbuf_p = pointer(easy.errbuf)
     setopt(easy, CURLOPT_ERRORBUFFER, errbuf_p)
 
+    # set pre-request callback
+    prereq_cb = @cfunction(prereq_callback,
+        Cint, (Ptr{Cvoid}, Ptr{Cchar}, Ptr{Cchar}, Cint, Cint))
+    setopt(easy, CURLOPT_PREREQFUNCTION, prereq_cb)
+    setopt(easy, CURLOPT_PREREQDATA, easy_p)
+
     # set header callback
     header_cb = @cfunction(header_callback,
         Csize_t, (Ptr{Cchar}, Csize_t, Csize_t, Ptr{Cvoid}))

src/Curl/Multi.jl

@@ -41,6 +41,7 @@ end
 # adding & removing easy handles
 
 function add_handle(multi::Multi, easy::Easy)
+    connect_semaphore_acquire(easy)
     lock(multi.lock) do
         if isempty(multi.easies)
             preserve_handle(multi)
@@ -70,6 +71,7 @@ function remove_handle(multi::Multi, easy::Easy)
         end
         unpreserve_handle(multi)
     end
+    connect_semaphore_release(easy)
 end
 
 # multi-socket options