Merge pull request #28 from JuliaComputing/tan/misc

concurrent use is flaky, add to readme

Author: tanmaykm

README.md

@@ -143,6 +143,7 @@ gRPCController(;
     [ max_message_length = DEFAULT_MAX_MESSAGE_LENGTH, ]
     [ max_recv_message_length = 0, ]
     [ max_send_message_length = 0, ]
+    [ enable_shared_locks = false, ]
     [ verbose::Bool = false, ]
 )
 ```
@@ -163,6 +164,8 @@ gRPCController(;
    `max_message_length`, same as setting this to 0)
 - `max_send_message_length`: maximum message length to send (default is
    `max_message_length`, same as setting this to 0)
+- `enable_shared_locks`: whether to enable locks for using gRPCClient across
+    tasks/threads concurrently (experimental, default is false)
 - `verbose`: whether to print out verbose communication logs (default false)
 
 ### `gRPCChannel`
@@ -231,6 +234,10 @@ It has the following members:
 - `grpc_status`: grpc status code for this request
 - `message`: any error message if request was not successful
 
+## TODO
+
+- Concurrent use of gRPCClient is still somewhat flaky.
+
 ## Credits
 
 This package was originally developed at [Julia Computing](https://juliacomputing.com)

src/curl.jl

@@ -42,6 +42,57 @@ function buffer_send_data(input::Channel{T}) where T <: ProtoType
 end
 =#
 
+function share_lock(easy_p::Ptr{Cvoid}, data::curl_lock_data, access::curl_lock_access, userptr::Ptr{Cvoid})
+    share = unsafe_pointer_to_objref(Ptr{CurlShare}(userptr))::CurlShare
+    lock(share.locks[data])
+    nothing
+end
+
+function share_unlock(easy_p::Ptr{Cvoid}, data::curl_lock_data, userptr::Ptr{Cvoid})
+    share = unsafe_pointer_to_objref(Ptr{CurlShare}(userptr))::CurlShare
+    unlock(share.locks[data])
+    nothing
+end
+
+mutable struct CurlShare
+    shptr::Ptr{CURLSH}
+    locks::Vector{ReentrantLock}
+    closed::Bool
+
+    function CurlShare()
+        shptr = curl_share_init()
+        curl_share_setopt(shptr, CURLSHOPT_SHARE, CURL_LOCK_DATA_SHARE)
+        curl_share_setopt(shptr, CURLSHOPT_SHARE, CURL_LOCK_DATA_COOKIE)
+        curl_share_setopt(shptr, CURLSHOPT_SHARE, CURL_LOCK_DATA_DNS)
+        curl_share_setopt(shptr, CURLSHOPT_SHARE, CURL_LOCK_DATA_PSL)
+        curl_share_setopt(shptr, CURLSHOPT_SHARE, CURL_LOCK_DATA_CONNECT)
+
+        share_lock_cb = @cfunction(share_lock, Cvoid, (Ptr{Cvoid}, Cuint, Cuint, Ptr{Cvoid}))
+        share_unlock_cb = @cfunction(share_unlock, Cvoid, (Ptr{Cvoid}, Cuint, Ptr{Cvoid}))
+
+        @ccall LibCURL.LibCURL_jll.libcurl.curl_share_setopt(shptr::Ptr{CURLSH}, CURLSHOPT_LOCKFUNC::CURLSHoption; share_lock_cb::Ptr{Cvoid})::CURLSHcode
+        @ccall LibCURL.LibCURL_jll.libcurl.curl_share_setopt(shptr::Ptr{CURLSH}, CURLSHOPT_UNLOCKFUNC::CURLSHoption; share_unlock_cb::Ptr{Cvoid})::CURLSHcode
+
+        locks = Vector(undef, CURL_LOCK_DATA_LAST)
+        for idx in 1:CURL_LOCK_DATA_LAST
+            locks[idx] = ReentrantLock()
+        end
+
+        obj = new(shptr, locks, false)
+        userptr = pointer_from_objref(obj)
+        @ccall LibCURL.LibCURL_jll.libcurl.curl_share_setopt(shptr::Ptr{CURLSH}, CURLSHOPT_USERDATA::CURLSHoption; userptr::Ptr{Cvoid})::CURLSHcode
+        obj
+    end
+end
+
+function close(share::CurlShare)
+    if share.closed
+        curl_share_cleanup(share.shptr)
+        share.closed = true
+    end
+    nothing
+end
+
 function send_data(easy::Curl.Easy, input::Channel{T}, max_send_message_length::Int) where T <: ProtoType
     while true
         yield()
@@ -95,7 +146,7 @@ function grpc_request_header(request_timeout::Real)
     end
 end
 
-function easy_handle(maxage::Clong, keepalive::Clong, negotiation::Symbol, revocation::Bool, request_timeout::Real)
+function easy_handle(curlshare::Union{Nothing,Ptr{CURLSH}}, maxage::Clong, keepalive::Clong, negotiation::Symbol, revocation::Bool, request_timeout::Real)
     easy = Curl.Easy()
     http_version = (negotiation === :http2) ? CURL_HTTP_VERSION_2_0 :
                    (negotiation === :http2_tls) ? CURL_HTTP_VERSION_2TLS :
@@ -105,6 +156,9 @@ function easy_handle(maxage::Clong, keepalive::Clong, negotiation::Symbol, revoc
     Curl.setopt(easy, CURLOPT_PIPEWAIT, Clong(1))
     Curl.setopt(easy, CURLOPT_POST, Clong(1))
     Curl.setopt(easy, CURLOPT_HTTPHEADER, grpc_request_header(request_timeout))
+    if curlshare !== nothing
+        Curl.setopt(easy, CURLOPT_SHARE, curlshare)
+    end
     if !revocation
         Curl.setopt(easy, CURLOPT_SSL_OPTIONS, CURLSSLOPT_NO_REVOKE)
     end
@@ -200,7 +254,7 @@ function get_grpc_status(easy::Curl.Easy)
     return grpc_status, grpc_message
 end
 
-function grpc_request(downloader::Downloader, url::String, input::Channel{T1}, output::Channel{T2};
+function grpc_request(curlshare::Union{Nothing,Ptr{CURLSH}}, downloader::Downloader, url::String, input::Channel{T1}, output::Channel{T2};
         maxage::Clong = typemax(Clong),
         keepalive::Clong = 60,
         negotiation::Symbol = :http2_prior_knowledge,
@@ -210,7 +264,7 @@ function grpc_request(downloader::Downloader, url::String, input::Channel{T1}, o
         max_recv_message_length::Int = DEFAULT_MAX_RECV_MESSAGE_LENGTH,
         max_send_message_length::Int = DEFAULT_MAX_SEND_MESSAGE_LENGTH,
         verbose::Bool = false)::gRPCStatus where {T1 <: ProtoType, T2 <: ProtoType}
-    Curl.with_handle(easy_handle(maxage, keepalive, negotiation, revocation, request_timeout)) do easy
+    Curl.with_handle(easy_handle(curlshare, maxage, keepalive, negotiation, revocation, request_timeout)) do easy
         # setup the request
         Curl.set_url(easy, url)
         Curl.set_timeout(easy, request_timeout)

src/gRPCClient.jl

@@ -6,6 +6,7 @@ using ProtoBuf
 
 import Downloads: Curl
 import ProtoBuf: call_method
+import Base: close
 
 export gRPCController, gRPCChannel, gRPCException, gRPCServiceCallException, gRPCMessageTooLargeException, gRPCStatus, gRPCCheck, StatusCode
 

src/grpc.jl

@@ -97,6 +97,8 @@ Contains settings to control the behavior of gRPC requests.
    `max_message_length`, same as setting this to 0)
 - `max_send_message_length`: maximum message length to send (default is
    `max_message_length`, same as setting this to 0)
+- `enable_shared_locks`: whether to enable locks for using gRPCClient across
+    tasks/threads concurrently (experimental, default is false)
 - `verbose`: whether to print out verbose communication logs (default false)
 """
 struct gRPCController <: ProtoRpcController
@@ -108,6 +110,7 @@ struct gRPCController <: ProtoRpcController
     connect_timeout::Real
     max_recv_message_length::Int
     max_send_message_length::Int
+    enable_shared_locks::Bool
     verbose::Bool
 
     function gRPCController(;
@@ -120,6 +123,7 @@ struct gRPCController <: ProtoRpcController
             max_message_length::Integer = DEFAULT_MAX_MESSAGE_LENGTH,
             max_recv_message_length::Integer = 0,
             max_send_message_length::Integer = 0,
+            enable_shared_locks::Bool = false,
             verbose::Bool = false
         )
         if maxage < 0 || keepalive < 0 || request_timeout < 0 || connect_timeout < 0 || 
@@ -128,7 +132,17 @@ struct gRPCController <: ProtoRpcController
         end
         (max_recv_message_length == 0) && (max_recv_message_length = max_message_length)
         (max_send_message_length == 0) && (max_send_message_length = max_message_length)
-        new(maxage, keepalive, negotiation, revocation, request_timeout, connect_timeout, max_recv_message_length, max_send_message_length, verbose)
+        new(maxage,
+            keepalive,
+            negotiation,
+            revocation,
+            request_timeout,
+            connect_timeout,
+            max_recv_message_length,
+            max_send_message_length,
+            enable_shared_locks,
+            verbose,
+        )
     end
 end
 
@@ -146,16 +160,22 @@ the server.
 struct gRPCChannel <: ProtoRpcChannel
     downloader::Downloader
     baseurl::String
+    curlshare::CurlShare
 
     function gRPCChannel(baseurl::String)
         downloader = Downloader(; grace=Inf)
         Curl.init!(downloader.multi)
         Curl.setopt(downloader.multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX)
         endswith(baseurl, '/') && (baseurl = baseurl[1:end-1])
-        new(downloader, baseurl)
+        new(downloader, baseurl, CurlShare())
     end
 end
 
+function close(channel::gRPCChannel)
+    close(channel.curlshare)
+    nothing
+end
+
 function to_delimited_message_bytes(msg, max_message_length::Int)
     iob = IOBuffer()
     limitiob = LimitIO(iob, max_message_length)
@@ -193,7 +213,8 @@ function call_method(channel::gRPCChannel, service::ServiceDescriptor, method::M
 end
 function call_method(channel::gRPCChannel, service::ServiceDescriptor, method::MethodDescriptor, controller::gRPCController, input::Channel{T1}, outchannel::Channel{T2}) where {T1 <: ProtoType, T2 <: ProtoType}
     url = string(channel.baseurl, "/", service.name, "/", method.name)
-    status_future = @async grpc_request(channel.downloader, url, input, outchannel;
+    shptr = controller.enable_shared_locks ? channel.curlshare.shptr : nothing
+    status_future = @async grpc_request(shptr, channel.downloader, url, input, outchannel;
         maxage = controller.maxage,
         keepalive = controller.keepalive,
         negotiation = controller.negotiation,

test/runtests_routeguide.jl

@@ -88,8 +88,10 @@ server_endpoint = isempty(ARGS) ? "http://localhost:10000/" : ARGS[1]
     @debug("testing routeclinet...")
     test_clients(server_endpoint)
 
-    @debug("testing async safety...")
-    test_task_safety(server_endpoint)
+    if get(ENV, "TEST_ASYNC", "false") == "true"
+        @debug("testing async safety...")
+        test_task_safety(server_endpoint)
+    end
 
     kill(serverproc)
     @info("stopped test server")