AsyncCallBase.cs 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388
  1. #region Copyright notice and license
  2. // Copyright 2015, Google Inc.
  3. // All rights reserved.
  4. //
  5. // Redistribution and use in source and binary forms, with or without
  6. // modification, are permitted provided that the following conditions are
  7. // met:
  8. //
  9. // * Redistributions of source code must retain the above copyright
  10. // notice, this list of conditions and the following disclaimer.
  11. // * Redistributions in binary form must reproduce the above
  12. // copyright notice, this list of conditions and the following disclaimer
  13. // in the documentation and/or other materials provided with the
  14. // distribution.
  15. // * Neither the name of Google Inc. nor the names of its
  16. // contributors may be used to endorse or promote products derived from
  17. // this software without specific prior written permission.
  18. //
  19. // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  20. // "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  21. // LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  22. // A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  23. // OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  24. // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  25. // LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  26. // DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  27. // THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  28. // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  29. // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  30. #endregion
  31. using System;
  32. using System.Diagnostics;
  33. using System.IO;
  34. using System.Runtime.CompilerServices;
  35. using System.Runtime.InteropServices;
  36. using System.Threading;
  37. using System.Threading.Tasks;
  38. using Grpc.Core.Internal;
  39. using Grpc.Core.Logging;
  40. using Grpc.Core.Profiling;
  41. using Grpc.Core.Utils;
  42. namespace Grpc.Core.Internal
  43. {
  44. /// <summary>
  45. /// Base for handling both client side and server side calls.
  46. /// Manages native call lifecycle and provides convenience methods.
  47. /// </summary>
  48. internal abstract class AsyncCallBase<TWrite, TRead>
  49. {
  50. static readonly ILogger Logger = GrpcEnvironment.Logger.ForType<AsyncCallBase<TWrite, TRead>>();
  51. protected static readonly Status DeserializeResponseFailureStatus = new Status(StatusCode.Internal, "Failed to deserialize response message.");
  52. readonly Func<TWrite, byte[]> serializer;
  53. readonly Func<byte[], TRead> deserializer;
  54. protected readonly GrpcEnvironment environment;
  55. protected readonly object myLock = new object();
  56. protected INativeCall call;
  57. protected bool disposed;
  58. protected bool started;
  59. protected bool cancelRequested;
  60. protected AsyncCompletionDelegate<object> sendCompletionDelegate; // Completion of a pending send or sendclose if not null.
  61. protected TaskCompletionSource<TRead> streamingReadTcs; // Completion of a pending streaming read if not null.
  62. protected TaskCompletionSource<object> sendStatusFromServerTcs;
  63. protected bool readingDone; // True if last read (i.e. read with null payload) was already received.
  64. protected bool halfcloseRequested; // True if send close have been initiated.
  65. protected bool finished; // True if close has been received from the peer.
  66. protected bool initialMetadataSent;
  67. protected long streamingWritesCounter; // Number of streaming send operations started so far.
  68. public AsyncCallBase(Func<TWrite, byte[]> serializer, Func<byte[], TRead> deserializer, GrpcEnvironment environment)
  69. {
  70. this.serializer = GrpcPreconditions.CheckNotNull(serializer);
  71. this.deserializer = GrpcPreconditions.CheckNotNull(deserializer);
  72. this.environment = GrpcPreconditions.CheckNotNull(environment);
  73. }
  74. /// <summary>
  75. /// Requests cancelling the call.
  76. /// </summary>
  77. public void Cancel()
  78. {
  79. lock (myLock)
  80. {
  81. GrpcPreconditions.CheckState(started);
  82. cancelRequested = true;
  83. if (!disposed)
  84. {
  85. call.Cancel();
  86. }
  87. }
  88. }
  89. /// <summary>
  90. /// Requests cancelling the call with given status.
  91. /// </summary>
  92. protected void CancelWithStatus(Status status)
  93. {
  94. lock (myLock)
  95. {
  96. cancelRequested = true;
  97. if (!disposed)
  98. {
  99. call.CancelWithStatus(status);
  100. }
  101. }
  102. }
  103. protected void InitializeInternal(INativeCall call)
  104. {
  105. lock (myLock)
  106. {
  107. this.call = call;
  108. }
  109. }
  110. /// <summary>
  111. /// Initiates sending a message. Only one send operation can be active at a time.
  112. /// completionDelegate is invoked upon completion.
  113. /// </summary>
  114. protected void StartSendMessageInternal(TWrite msg, WriteFlags writeFlags, AsyncCompletionDelegate<object> completionDelegate)
  115. {
  116. byte[] payload = UnsafeSerialize(msg);
  117. lock (myLock)
  118. {
  119. GrpcPreconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null");
  120. CheckSendingAllowed(allowFinished: false);
  121. call.StartSendMessage(HandleSendFinished, payload, writeFlags, !initialMetadataSent);
  122. sendCompletionDelegate = completionDelegate;
  123. initialMetadataSent = true;
  124. streamingWritesCounter++;
  125. }
  126. }
  127. /// <summary>
  128. /// Initiates reading a message. Only one read operation can be active at a time.
  129. /// completionDelegate is invoked upon completion.
  130. /// </summary>
  131. protected Task<TRead> ReadMessageInternalAsync()
  132. {
  133. lock (myLock)
  134. {
  135. GrpcPreconditions.CheckState(started);
  136. if (readingDone)
  137. {
  138. // the last read that returns null or throws an exception is idempotent
  139. // and maintain its state.
  140. GrpcPreconditions.CheckState(streamingReadTcs != null, "Call does not support streaming reads.");
  141. return streamingReadTcs.Task;
  142. }
  143. GrpcPreconditions.CheckState(streamingReadTcs == null, "Only one read can be pending at a time");
  144. GrpcPreconditions.CheckState(!disposed);
  145. call.StartReceiveMessage(HandleReadFinished);
  146. streamingReadTcs = new TaskCompletionSource<TRead>();
  147. return streamingReadTcs.Task;
  148. }
  149. }
  150. /// <summary>
  151. /// If there are no more pending actions and no new actions can be started, releases
  152. /// the underlying native resources.
  153. /// </summary>
  154. protected bool ReleaseResourcesIfPossible()
  155. {
  156. using (Profilers.ForCurrentThread().NewScope("AsyncCallBase.ReleaseResourcesIfPossible"))
  157. {
  158. if (!disposed && call != null)
  159. {
  160. bool noMoreSendCompletions = sendCompletionDelegate == null && (halfcloseRequested || cancelRequested || finished);
  161. if (noMoreSendCompletions && readingDone && finished)
  162. {
  163. ReleaseResources();
  164. return true;
  165. }
  166. }
  167. return false;
  168. }
  169. }
  170. protected abstract bool IsClient
  171. {
  172. get;
  173. }
  174. private void ReleaseResources()
  175. {
  176. if (call != null)
  177. {
  178. call.Dispose();
  179. }
  180. disposed = true;
  181. OnAfterReleaseResources();
  182. }
  183. protected virtual void OnAfterReleaseResources()
  184. {
  185. }
  186. protected virtual void CheckSendingAllowed(bool allowFinished)
  187. {
  188. GrpcPreconditions.CheckState(started);
  189. CheckNotCancelled();
  190. GrpcPreconditions.CheckState(!disposed || allowFinished);
  191. GrpcPreconditions.CheckState(!halfcloseRequested, "Already halfclosed.");
  192. GrpcPreconditions.CheckState(!finished || allowFinished, "Already finished.");
  193. GrpcPreconditions.CheckState(sendCompletionDelegate == null, "Only one write can be pending at a time");
  194. }
  195. protected void CheckNotCancelled()
  196. {
  197. if (cancelRequested)
  198. {
  199. throw new OperationCanceledException("Remote call has been cancelled.");
  200. }
  201. }
  202. protected byte[] UnsafeSerialize(TWrite msg)
  203. {
  204. using (Profilers.ForCurrentThread().NewScope("AsyncCallBase.UnsafeSerialize"))
  205. {
  206. return serializer(msg);
  207. }
  208. }
  209. protected Exception TryDeserialize(byte[] payload, out TRead msg)
  210. {
  211. using (Profilers.ForCurrentThread().NewScope("AsyncCallBase.TryDeserialize"))
  212. {
  213. try
  214. {
  215. msg = deserializer(payload);
  216. return null;
  217. }
  218. catch (Exception e)
  219. {
  220. msg = default(TRead);
  221. return e;
  222. }
  223. }
  224. }
  225. protected void FireCompletion<T>(AsyncCompletionDelegate<T> completionDelegate, T value, Exception error)
  226. {
  227. try
  228. {
  229. completionDelegate(value, error);
  230. }
  231. catch (Exception e)
  232. {
  233. Logger.Error(e, "Exception occured while invoking completion delegate.");
  234. }
  235. }
  236. /// <summary>
  237. /// Handles send completion.
  238. /// </summary>
  239. protected void HandleSendFinished(bool success)
  240. {
  241. AsyncCompletionDelegate<object> origCompletionDelegate = null;
  242. lock (myLock)
  243. {
  244. origCompletionDelegate = sendCompletionDelegate;
  245. sendCompletionDelegate = null;
  246. ReleaseResourcesIfPossible();
  247. }
  248. if (!success)
  249. {
  250. FireCompletion(origCompletionDelegate, null, new InvalidOperationException("Send failed"));
  251. }
  252. else
  253. {
  254. FireCompletion(origCompletionDelegate, null, null);
  255. }
  256. }
  257. /// <summary>
  258. /// Handles halfclose (send close from client) completion.
  259. /// </summary>
  260. protected void HandleSendCloseFromClientFinished(bool success)
  261. {
  262. AsyncCompletionDelegate<object> origCompletionDelegate = null;
  263. lock (myLock)
  264. {
  265. origCompletionDelegate = sendCompletionDelegate;
  266. sendCompletionDelegate = null;
  267. ReleaseResourcesIfPossible();
  268. }
  269. if (!success)
  270. {
  271. FireCompletion(origCompletionDelegate, null, new InvalidOperationException("Sending close from client has failed."));
  272. }
  273. else
  274. {
  275. FireCompletion(origCompletionDelegate, null, null);
  276. }
  277. }
  278. /// <summary>
  279. /// Handles send status from server completion.
  280. /// </summary>
  281. protected void HandleSendStatusFromServerFinished(bool success)
  282. {
  283. lock (myLock)
  284. {
  285. ReleaseResourcesIfPossible();
  286. }
  287. if (!success)
  288. {
  289. sendStatusFromServerTcs.SetException(new InvalidOperationException("Error sending status from server."));
  290. }
  291. else
  292. {
  293. sendStatusFromServerTcs.SetResult(null);
  294. }
  295. }
  296. /// <summary>
  297. /// Handles streaming read completion.
  298. /// </summary>
  299. protected void HandleReadFinished(bool success, byte[] receivedMessage)
  300. {
  301. // if success == false, received message will be null. It that case we will
  302. // treat this completion as the last read an rely on C core to handle the failed
  303. // read (e.g. deliver approriate statusCode on the clientside).
  304. TRead msg = default(TRead);
  305. var deserializeException = (success && receivedMessage != null) ? TryDeserialize(receivedMessage, out msg) : null;
  306. TaskCompletionSource<TRead> origTcs = null;
  307. lock (myLock)
  308. {
  309. origTcs = streamingReadTcs;
  310. if (receivedMessage == null)
  311. {
  312. // This was the last read.
  313. readingDone = true;
  314. }
  315. if (deserializeException != null && IsClient)
  316. {
  317. readingDone = true;
  318. // TODO(jtattermusch): it might be too late to set the status
  319. CancelWithStatus(DeserializeResponseFailureStatus);
  320. }
  321. if (!readingDone)
  322. {
  323. streamingReadTcs = null;
  324. }
  325. ReleaseResourcesIfPossible();
  326. }
  327. if (deserializeException != null && !IsClient)
  328. {
  329. origTcs.SetException(new IOException("Failed to deserialize request message.", deserializeException));
  330. return;
  331. }
  332. origTcs.SetResult(msg);
  333. }
  334. }
  335. }