Replace Spinlock with parking_lot::Mutex

Cargo.toml

@@ -24,6 +24,7 @@ protobuf = { version = "2.0", optional = true }
 prost = { version = "0.6", optional = true }
 bytes = { version = "0.5", optional = true }
 log = "0.4"
+parking_lot = "0.10"
 
 [workspace]
 members = ["proto", "benchmark", "compiler", "interop", "tests-and-examples"]

src/call/client.rs

@@ -10,6 +10,7 @@ use futures::ready;
 use futures::sink::Sink;
 use futures::stream::Stream;
 use futures::task::{Context, Poll};
+use parking_lot::Mutex;
 use std::future::Future;
 
 use super::{ShareCall, ShareCallHolder, SinkBase, WriteFlags};
@@ -18,7 +19,7 @@ use crate::channel::Channel;
 use crate::codec::{DeserializeFn, SerializeFn};
 use crate::error::{Error, Result};
 use crate::metadata::Metadata;
-use crate::task::{BatchFuture, BatchType, SpinLock};
+use crate::task::{BatchFuture, BatchType};
 
 /// Update the flag bit in res.
 #[inline]
@@ -144,7 +145,7 @@ impl Call {
             )
         });
 
-        let share_call = Arc::new(SpinLock::new(ShareCall::new(call, cq_f)));
+        let share_call = Arc::new(Mutex::new(ShareCall::new(call, cq_f)));
         let sink = ClientCStreamSender::new(share_call.clone(), method.req_ser());
         let recv = ClientCStreamReceiver {
             call: share_call,
@@ -209,7 +210,7 @@ impl Call {
             grpc_sys::grpcwrap_call_recv_initial_metadata(call.call, ctx, tag)
         });
 
-        let share_call = Arc::new(SpinLock::new(ShareCall::new(call, cq_f)));
+        let share_call = Arc::new(Mutex::new(ShareCall::new(call, cq_f)));
         let sink = ClientDuplexSender::new(share_call.clone(), method.req_ser());
         let recv = ClientDuplexReceiver::new(share_call, method.resp_de());
         Ok((sink, recv))
@@ -266,7 +267,7 @@ impl<T> Future for ClientUnaryReceiver<T> {
 /// [`Cancelled`]: ./enum.RpcStatusCode.html#variant.Cancelled
 #[must_use = "if unused the ClientCStreamReceiver may immediately cancel the RPC"]
 pub struct ClientCStreamReceiver<T> {
-    call: Arc<SpinLock<ShareCall>>,
+    call: Arc<Mutex<ShareCall>>,
     resp_de: DeserializeFn<T>,
     finished: bool,
 }
@@ -314,14 +315,14 @@ impl<T> Future for ClientCStreamReceiver<T> {
 /// [`close`]: #method.close
 #[must_use = "if unused the StreamingCallSink may immediately cancel the RPC"]
 pub struct StreamingCallSink<Req> {
-    call: Arc<SpinLock<ShareCall>>,
+    call: Arc<Mutex<ShareCall>>,
     sink_base: SinkBase,
     close_f: Option<BatchFuture>,
     req_ser: SerializeFn<Req>,
 }
 
 impl<Req> StreamingCallSink<Req> {
-    fn new(call: Arc<SpinLock<ShareCall>>, req_ser: SerializeFn<Req>) -> StreamingCallSink<Req> {
+    fn new(call: Arc<Mutex<ShareCall>>, req_ser: SerializeFn<Req>) -> StreamingCallSink<Req> {
         StreamingCallSink {
             call,
             sink_base: SinkBase::new(false),
@@ -520,11 +521,11 @@ impl<Resp> Stream for ClientSStreamReceiver<Resp> {
 /// [`Cancelled`]: ./enum.RpcStatusCode.html#variant.Cancelled
 #[must_use = "if unused the ClientDuplexReceiver may immediately cancel the RPC"]
 pub struct ClientDuplexReceiver<Resp> {
-    imp: ResponseStreamImpl<Arc<SpinLock<ShareCall>>, Resp>,
+    imp: ResponseStreamImpl<Arc<Mutex<ShareCall>>, Resp>,
 }
 
 impl<Resp> ClientDuplexReceiver<Resp> {
-    fn new(call: Arc<SpinLock<ShareCall>>, de: DeserializeFn<Resp>) -> ClientDuplexReceiver<Resp> {
+    fn new(call: Arc<Mutex<ShareCall>>, de: DeserializeFn<Resp>) -> ClientDuplexReceiver<Resp> {
         ClientDuplexReceiver {
             imp: ResponseStreamImpl::new(call, de),
         }

src/call/mod.rs

@@ -14,12 +14,13 @@ use futures::future::Future;
 use futures::ready;
 use futures::task::{Context, Poll};
 use libc::c_void;
+use parking_lot::Mutex;
 
 use crate::buf::{GrpcByteBuffer, GrpcByteBufferReader};
 use crate::codec::{DeserializeFn, Marshaller, SerializeFn};
 use crate::error::{Error, Result};
 use crate::grpc_sys::grpc_status_code::*;
-use crate::task::{self, BatchFuture, BatchType, CallTag, SpinLock};
+use crate::task::{self, BatchFuture, BatchType, CallTag};
 
 // By default buffers in `SinkBase` will be shrink to 4K size.
 const BUF_SHRINK_SIZE: usize = 4 * 1024;
@@ -505,7 +506,7 @@ impl ShareCallHolder for ShareCall {
     }
 }
 
-impl ShareCallHolder for Arc<SpinLock<ShareCall>> {
+impl ShareCallHolder for Arc<Mutex<ShareCall>> {
     fn call<R, F: FnOnce(&mut ShareCall) -> R>(&mut self, f: F) -> R {
         let mut call = self.lock();
         f(&mut call)

src/call/server.rs

@@ -13,6 +13,7 @@ use futures::ready;
 use futures::sink::Sink;
 use futures::stream::Stream;
 use futures::task::{Context, Poll};
+use parking_lot::Mutex;
 
 use super::{RpcStatus, ShareCall, ShareCallHolder, WriteFlags};
 use crate::auth_context::AuthContext;
@@ -24,7 +25,7 @@ use crate::cq::CompletionQueue;
 use crate::error::{Error, Result};
 use crate::metadata::Metadata;
 use crate::server::{BoxHandler, RequestCallContext};
-use crate::task::{BatchFuture, CallTag, Executor, Kicker, SpinLock};
+use crate::task::{BatchFuture, CallTag, Executor, Kicker};
 
 pub struct Deadline {
     spec: gpr_timespec,
@@ -239,13 +240,13 @@ impl UnaryRequestContext {
 /// finish before dropping.
 #[must_use = "if unused the RequestStream may immediately cancel the RPC"]
 pub struct RequestStream<T> {
-    call: Arc<SpinLock<ShareCall>>,
+    call: Arc<Mutex<ShareCall>>,
     base: StreamingBase,
     de: DeserializeFn<T>,
 }
 
 impl<T> RequestStream<T> {
-    fn new(call: Arc<SpinLock<ShareCall>>, de: DeserializeFn<T>) -> RequestStream<T> {
+    fn new(call: Arc<Mutex<ShareCall>>, de: DeserializeFn<T>) -> RequestStream<T> {
         RequestStream {
             call,
             base: StreamingBase::new(None),
@@ -393,7 +394,7 @@ impl_unary_sink!(
     #[must_use = "if unused the sink may immediately cancel the RPC"]
     ClientStreamingSink,
     ClientStreamingSinkResult,
-    Arc<SpinLock<ShareCall>>
+    Arc<Mutex<ShareCall>>
 );
 
 // A macro helper to implement server side streaming sink.
@@ -570,7 +571,7 @@ impl_stream_sink!(
     #[must_use = "if unused the sink may immediately cancel the RPC"]
     DuplexSink,
     DuplexSinkFailure,
-    Arc<SpinLock<ShareCall>>
+    Arc<Mutex<ShareCall>>
 );
 
 /// A context for rpc handling.
@@ -688,7 +689,7 @@ pub fn execute_client_streaming<P, Q, F>(
 {
     let mut call = ctx.call();
     let close_f = accept_call!(call);
-    let call = Arc::new(SpinLock::new(ShareCall::new(call, close_f)));
+    let call = Arc::new(Mutex::new(ShareCall::new(call, close_f)));
 
     let req_s = RequestStream::new(call.clone(), de);
     let sink = ClientStreamingSink::new(call, ser);
@@ -735,7 +736,7 @@ pub fn execute_duplex_streaming<P, Q, F>(
 {
     let mut call = ctx.call();
     let close_f = accept_call!(call);
-    let call = Arc::new(SpinLock::new(ShareCall::new(call, close_f)));
+    let call = Arc::new(Mutex::new(ShareCall::new(call, close_f)));
 
     let req_s = RequestStream::new(call.clone(), de);
     let sink = DuplexSink::new(call, ser);

src/task/mod.rs

@@ -2,7 +2,6 @@
 
 mod callback;
 mod executor;
-mod lock;
 mod promise;
 
 use std::fmt::{self, Debug, Formatter};
@@ -11,6 +10,7 @@ use std::sync::Arc;
 
 use futures::future::Future;
 use futures::task::{Context, Poll, Waker};
+use parking_lot::Mutex;
 
 use self::callback::{Abort, Request as RequestCallback, UnaryRequest as UnaryRequestCallback};
 use self::executor::SpawnTask;
@@ -22,7 +22,6 @@ use crate::error::{Error, Result};
 use crate::server::RequestCallContext;
 
 pub(crate) use self::executor::{Executor, Kicker, UnfinishedWork};
-pub use self::lock::SpinLock;
 pub use self::promise::BatchType;
 
 /// A handle that is used to notify future that the task finishes.
@@ -49,10 +48,10 @@ impl<T> NotifyHandle<T> {
     }
 }
 
-type Inner<T> = SpinLock<NotifyHandle<T>>;
+type Inner<T> = Mutex<NotifyHandle<T>>;
 
 fn new_inner<T>() -> Arc<Inner<T>> {
-    Arc::new(SpinLock::new(NotifyHandle::new()))
+    Arc::new(Mutex::new(NotifyHandle::new()))
 }
 
 /// Get the future status without the need to poll.