From 2030a4e5f79395a7edfb88762a60c37575969f17 Mon Sep 17 00:00:00 2001 From: mgravell Date: Thu, 11 Jul 2019 09:02:07 +0100 Subject: [PATCH 1/7] move core of Async*Call into a new internal AsyncCallState struct; abstracts over Foo() vs Foo(object), avoiding delegate allocations --- src/csharp/Grpc.Core.Api/AsyncCallState.cs | 92 +++++++++++++++++++ .../Grpc.Core.Api/AsyncClientStreamingCall.cs | 41 ++++++--- .../Grpc.Core.Api/AsyncDuplexStreamingCall.cs | 41 ++++++--- .../Grpc.Core.Api/AsyncServerStreamingCall.cs | 38 +++++--- src/csharp/Grpc.Core.Api/AsyncUnaryCall.cs | 38 +++++--- .../Internal/AsyncCallStateTest.cs | 82 +++++++++++++++++ src/csharp/Grpc.Core/Calls.cs | 29 +++++- 7 files changed, 309 insertions(+), 52 deletions(-) create mode 100644 src/csharp/Grpc.Core.Api/AsyncCallState.cs create mode 100644 src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs diff --git a/src/csharp/Grpc.Core.Api/AsyncCallState.cs b/src/csharp/Grpc.Core.Api/AsyncCallState.cs new file mode 100644 index 00000000000..9958dbc47e7 --- /dev/null +++ b/src/csharp/Grpc.Core.Api/AsyncCallState.cs @@ -0,0 +1,92 @@ +#region Copyright notice and license + +// Copyright 2015 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#endregion + + +using System; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// + /// Provides an abstraction over the callback providers + /// used by AsyncUnaryCall, AsyncDuplexStreamingCall, etc + /// + internal /* readonly */ struct AsyncCallState // can be made readonly in C# 7.2 + { + readonly object responseHeadersAsync; // Task or Func> + readonly object getStatusFunc; // Func or Func + readonly object getTrailersFunc; // Func or Func + readonly object disposeAction; // Action or Action + readonly object callbackState; // arg0 for the callbacks above, if needed + + internal AsyncCallState( + Func> responseHeadersAsync, + Func getStatusFunc, + Func getTrailersFunc, + Action disposeAction, + object callbackState) + { + this.responseHeadersAsync = responseHeadersAsync; + this.getStatusFunc = getStatusFunc; + this.getTrailersFunc = getTrailersFunc; + this.disposeAction = disposeAction; + this.callbackState = callbackState; + } + + internal AsyncCallState( + Task responseHeadersAsync, + Func getStatusFunc, + Func getTrailersFunc, + Action disposeAction) + { + this.responseHeadersAsync = responseHeadersAsync; + this.getStatusFunc = getStatusFunc; + this.getTrailersFunc = getTrailersFunc; + this.disposeAction = disposeAction; + this.callbackState = null; + } + + internal Task ResponseHeadersAsync() + { + var withState = responseHeadersAsync as Func>; + return withState != null ? withState(callbackState) + : (Task)responseHeadersAsync; + } + + internal Status GetStatus() + { + var withState = getStatusFunc as Func; + return withState != null ? withState(callbackState) + : ((Func)getStatusFunc)(); + } + + internal Metadata GetTrailers() + { + var withState = getTrailersFunc as Func; + return withState != null ? withState(callbackState) + : ((Func)getTrailersFunc)(); + } + + internal void Dispose() + { + var withState = disposeAction as Action; + if (withState != null) withState(callbackState); + else ((Action)disposeAction)(); + } + } +} diff --git a/src/csharp/Grpc.Core.Api/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core.Api/AsyncClientStreamingCall.cs index f59989655ec..bc63d8a04c3 100644 --- a/src/csharp/Grpc.Core.Api/AsyncClientStreamingCall.cs +++ b/src/csharp/Grpc.Core.Api/AsyncClientStreamingCall.cs @@ -31,10 +31,7 @@ namespace Grpc.Core { readonly IClientStreamWriter requestStream; readonly Task responseAsync; - readonly Task responseHeadersAsync; - readonly Func getStatusFunc; - readonly Func getTrailersFunc; - readonly Action disposeAction; + readonly AsyncCallState callState; /// /// Creates a new AsyncClientStreamingCall object with the specified properties. @@ -54,10 +51,30 @@ namespace Grpc.Core { this.requestStream = requestStream; this.responseAsync = responseAsync; - this.responseHeadersAsync = responseHeadersAsync; - this.getStatusFunc = getStatusFunc; - this.getTrailersFunc = getTrailersFunc; - this.disposeAction = disposeAction; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction); + } + + /// + /// Creates a new AsyncClientStreamingCall object with the specified properties. + /// + /// Stream of request values. + /// The response of the asynchronous call. + /// Response headers of the asynchronous call. + /// Delegate returning the status of the call. + /// Delegate returning the trailing metadata of the call. + /// Delegate to invoke when Dispose is called on the call object. + /// State object for use with the callback parameters. + public AsyncClientStreamingCall(IClientStreamWriter requestStream, + Task responseAsync, + Func> responseHeadersAsync, + Func getStatusFunc, + Func getTrailersFunc, + Action disposeAction, + object state) + { + this.requestStream = requestStream; + this.responseAsync = responseAsync; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction, state); } /// @@ -78,7 +95,7 @@ namespace Grpc.Core { get { - return this.responseHeadersAsync; + return callState.ResponseHeadersAsync(); } } @@ -108,7 +125,7 @@ namespace Grpc.Core /// public Status GetStatus() { - return getStatusFunc(); + return callState.GetStatus(); } /// @@ -117,7 +134,7 @@ namespace Grpc.Core /// public Metadata GetTrailers() { - return getTrailersFunc(); + return callState.GetTrailers(); } /// @@ -132,7 +149,7 @@ namespace Grpc.Core /// public void Dispose() { - disposeAction.Invoke(); + callState.Dispose(); } } } diff --git a/src/csharp/Grpc.Core.Api/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core.Api/AsyncDuplexStreamingCall.cs index 1cb1a918595..99660ab7174 100644 --- a/src/csharp/Grpc.Core.Api/AsyncDuplexStreamingCall.cs +++ b/src/csharp/Grpc.Core.Api/AsyncDuplexStreamingCall.cs @@ -30,10 +30,7 @@ namespace Grpc.Core { readonly IClientStreamWriter requestStream; readonly IAsyncStreamReader responseStream; - readonly Task responseHeadersAsync; - readonly Func getStatusFunc; - readonly Func getTrailersFunc; - readonly Action disposeAction; + readonly AsyncCallState callState; /// /// Creates a new AsyncDuplexStreamingCall object with the specified properties. @@ -53,10 +50,30 @@ namespace Grpc.Core { this.requestStream = requestStream; this.responseStream = responseStream; - this.responseHeadersAsync = responseHeadersAsync; - this.getStatusFunc = getStatusFunc; - this.getTrailersFunc = getTrailersFunc; - this.disposeAction = disposeAction; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction); + } + + /// + /// Creates a new AsyncDuplexStreamingCall object with the specified properties. + /// + /// Stream of request values. + /// Stream of response values. + /// Response headers of the asynchronous call. + /// Delegate returning the status of the call. + /// Delegate returning the trailing metadata of the call. + /// Delegate to invoke when Dispose is called on the call object. + /// State object for use with the callback parameters. + public AsyncDuplexStreamingCall(IClientStreamWriter requestStream, + IAsyncStreamReader responseStream, + Func> responseHeadersAsync, + Func getStatusFunc, + Func getTrailersFunc, + Action disposeAction, + object state) + { + this.requestStream = requestStream; + this.responseStream = responseStream; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction, state); } /// @@ -88,7 +105,7 @@ namespace Grpc.Core { get { - return this.responseHeadersAsync; + return callState.ResponseHeadersAsync(); } } @@ -98,7 +115,7 @@ namespace Grpc.Core /// public Status GetStatus() { - return getStatusFunc(); + return callState.GetStatus(); } /// @@ -107,7 +124,7 @@ namespace Grpc.Core /// public Metadata GetTrailers() { - return getTrailersFunc(); + return callState.GetTrailers(); } /// @@ -122,7 +139,7 @@ namespace Grpc.Core /// public void Dispose() { - disposeAction.Invoke(); + callState.Dispose(); } } } diff --git a/src/csharp/Grpc.Core.Api/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core.Api/AsyncServerStreamingCall.cs index 4303b0b1b02..360482d90b2 100644 --- a/src/csharp/Grpc.Core.Api/AsyncServerStreamingCall.cs +++ b/src/csharp/Grpc.Core.Api/AsyncServerStreamingCall.cs @@ -28,10 +28,7 @@ namespace Grpc.Core public sealed class AsyncServerStreamingCall : IDisposable { readonly IAsyncStreamReader responseStream; - readonly Task responseHeadersAsync; - readonly Func getStatusFunc; - readonly Func getTrailersFunc; - readonly Action disposeAction; + readonly AsyncCallState callState; /// /// Creates a new AsyncDuplexStreamingCall object with the specified properties. @@ -48,10 +45,27 @@ namespace Grpc.Core Action disposeAction) { this.responseStream = responseStream; - this.responseHeadersAsync = responseHeadersAsync; - this.getStatusFunc = getStatusFunc; - this.getTrailersFunc = getTrailersFunc; - this.disposeAction = disposeAction; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction); + } + + /// + /// Creates a new AsyncDuplexStreamingCall object with the specified properties. + /// + /// Stream of response values. + /// Response headers of the asynchronous call. + /// Delegate returning the status of the call. + /// Delegate returning the trailing metadata of the call. + /// Delegate to invoke when Dispose is called on the call object. + /// State object for use with the callback parameters. + public AsyncServerStreamingCall(IAsyncStreamReader responseStream, + Func> responseHeadersAsync, + Func getStatusFunc, + Func getTrailersFunc, + Action disposeAction, + object state) + { + this.responseStream = responseStream; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction, state); } /// @@ -72,7 +86,7 @@ namespace Grpc.Core { get { - return this.responseHeadersAsync; + return callState.ResponseHeadersAsync(); } } @@ -82,7 +96,7 @@ namespace Grpc.Core /// public Status GetStatus() { - return getStatusFunc(); + return callState.GetStatus(); } /// @@ -91,7 +105,7 @@ namespace Grpc.Core /// public Metadata GetTrailers() { - return getTrailersFunc(); + return callState.GetTrailers(); } /// @@ -106,7 +120,7 @@ namespace Grpc.Core /// public void Dispose() { - disposeAction.Invoke(); + callState.Dispose(); } } } diff --git a/src/csharp/Grpc.Core.Api/AsyncUnaryCall.cs b/src/csharp/Grpc.Core.Api/AsyncUnaryCall.cs index 17747f86caa..13e2e3b1136 100644 --- a/src/csharp/Grpc.Core.Api/AsyncUnaryCall.cs +++ b/src/csharp/Grpc.Core.Api/AsyncUnaryCall.cs @@ -29,10 +29,7 @@ namespace Grpc.Core public sealed class AsyncUnaryCall : IDisposable { readonly Task responseAsync; - readonly Task responseHeadersAsync; - readonly Func getStatusFunc; - readonly Func getTrailersFunc; - readonly Action disposeAction; + readonly AsyncCallState callState; /// @@ -50,10 +47,27 @@ namespace Grpc.Core Action disposeAction) { this.responseAsync = responseAsync; - this.responseHeadersAsync = responseHeadersAsync; - this.getStatusFunc = getStatusFunc; - this.getTrailersFunc = getTrailersFunc; - this.disposeAction = disposeAction; + this.callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction); + } + + /// + /// Creates a new AsyncUnaryCall object with the specified properties. + /// + /// The response of the asynchronous call. + /// Response headers of the asynchronous call. + /// Delegate returning the status of the call. + /// Delegate returning the trailing metadata of the call. + /// Delegate to invoke when Dispose is called on the call object. + /// State object for use with the callback parameters. + public AsyncUnaryCall(Task responseAsync, + Func> responseHeadersAsync, + Func getStatusFunc, + Func getTrailersFunc, + Action disposeAction, + object state) + { + this.responseAsync = responseAsync; + callState = new AsyncCallState(responseHeadersAsync, getStatusFunc, getTrailersFunc, disposeAction, state); } /// @@ -74,7 +88,7 @@ namespace Grpc.Core { get { - return this.responseHeadersAsync; + return callState.ResponseHeadersAsync(); } } @@ -92,7 +106,7 @@ namespace Grpc.Core /// public Status GetStatus() { - return getStatusFunc(); + return callState.GetStatus(); } /// @@ -101,7 +115,7 @@ namespace Grpc.Core /// public Metadata GetTrailers() { - return getTrailersFunc(); + return callState.GetTrailers(); } /// @@ -116,7 +130,7 @@ namespace Grpc.Core /// public void Dispose() { - disposeAction.Invoke(); + callState.Dispose(); } } } diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs new file mode 100644 index 00000000000..e5513c3f017 --- /dev/null +++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs @@ -0,0 +1,82 @@ +#region Copyright notice and license + +// Copyright 2015 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#endregion + +using System.Threading.Tasks; +using NUnit.Framework; + +namespace Grpc.Core.Internal.Tests +{ + public class AsyncCallStateTest + { + [Test] + public void Stateless() + { + bool disposed = false; + Task responseHeaders = Task.FromResult(new Metadata()); + Metadata trailers = new Metadata(); + var state = new AsyncCallState(responseHeaders, () => new Status(StatusCode.DataLoss, "oops"), + () => trailers, () => disposed = true); + + Assert.AreSame(responseHeaders, state.ResponseHeadersAsync()); + + var status = state.GetStatus(); + Assert.AreEqual(StatusCode.DataLoss, status.StatusCode); + Assert.AreEqual("oops", status.Detail); + + Assert.AreSame(trailers, state.GetTrailers()); + + Assert.False(disposed); + state.Dispose(); + Assert.True(disposed); + } + + class State + { + public bool disposed = false; + public Task responseHeaders = Task.FromResult(new Metadata()); + public Metadata trailers = new Metadata(); + public Status status = new Status(StatusCode.DataLoss, "oops"); + public void Dispose() { disposed = true; } + } + + [Test] + public void WithState() + { + var callbackState = new State(); + + var state = new AsyncCallState( + obj => ((State)obj).responseHeaders, + obj => ((State)obj).status, + obj => ((State)obj).trailers, + obj => ((State)obj).Dispose(), + callbackState); + + Assert.AreSame(callbackState.responseHeaders, state.ResponseHeadersAsync()); + + var status = state.GetStatus(); + Assert.AreEqual(StatusCode.DataLoss, status.StatusCode); + Assert.AreEqual("oops", status.Detail); + + Assert.AreSame(callbackState.trailers, state.GetTrailers()); + + Assert.False(callbackState.disposed); + state.Dispose(); + Assert.True(callbackState.disposed); + } + } +} diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index 05b9bedfccf..5700e3980fc 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -16,6 +16,7 @@ #endregion +using System; using System.Threading.Tasks; using Grpc.Core.Internal; @@ -59,7 +60,10 @@ namespace Grpc.Core { var asyncCall = new AsyncCall(call); var asyncResult = asyncCall.UnaryCallAsync(req); - return new AsyncUnaryCall(asyncResult, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); + return new AsyncUnaryCall(asyncResult, + Callbacks.GetHeaders, Callbacks.GetStatus, + Callbacks.GetTrailers, Callbacks.Cancel, + asyncCall); } /// @@ -78,7 +82,10 @@ namespace Grpc.Core var asyncCall = new AsyncCall(call); asyncCall.StartServerStreamingCall(req); var responseStream = new ClientResponseStream(asyncCall); - return new AsyncServerStreamingCall(responseStream, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); + return new AsyncServerStreamingCall(responseStream, + Callbacks.GetHeaders, Callbacks.GetStatus, + Callbacks.GetTrailers, Callbacks.Cancel, + asyncCall); } /// @@ -96,7 +103,10 @@ namespace Grpc.Core var asyncCall = new AsyncCall(call); var resultTask = asyncCall.ClientStreamingCallAsync(); var requestStream = new ClientRequestStream(asyncCall); - return new AsyncClientStreamingCall(requestStream, resultTask, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); + return new AsyncClientStreamingCall(requestStream, resultTask, + Callbacks.GetHeaders, Callbacks.GetStatus, + Callbacks.GetTrailers, Callbacks.Cancel, + asyncCall); } /// @@ -116,7 +126,18 @@ namespace Grpc.Core asyncCall.StartDuplexStreamingCall(); var requestStream = new ClientRequestStream(asyncCall); var responseStream = new ClientResponseStream(asyncCall); - return new AsyncDuplexStreamingCall(requestStream, responseStream, asyncCall.ResponseHeadersAsync, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); + return new AsyncDuplexStreamingCall(requestStream, responseStream, + Callbacks.GetHeaders, Callbacks.GetStatus, + Callbacks.GetTrailers, Callbacks.Cancel, + asyncCall); + } + + private static class Callbacks + { + internal static readonly Func> GetHeaders = state => ((AsyncCall)state).ResponseHeadersAsync; + internal static readonly Func GetStatus = state => ((AsyncCall)state).GetStatus(); + internal static readonly Func GetTrailers = state => ((AsyncCall)state).GetTrailers(); + internal static readonly Action Cancel = state => ((AsyncCall)state).Cancel(); } } } From 931ce49cf1633f014978c0ab9adfa2b94b76a2f6 Mon Sep 17 00:00:00 2001 From: Jan Tattermusch Date: Tue, 17 Sep 2019 17:26:01 +0200 Subject: [PATCH 2/7] fix a few nits --- src/csharp/Grpc.Core.Api/AsyncCallState.cs | 4 ++-- src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs | 2 +- src/csharp/tests.json | 1 + 3 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/csharp/Grpc.Core.Api/AsyncCallState.cs b/src/csharp/Grpc.Core.Api/AsyncCallState.cs index 9958dbc47e7..7f97e322c3c 100644 --- a/src/csharp/Grpc.Core.Api/AsyncCallState.cs +++ b/src/csharp/Grpc.Core.Api/AsyncCallState.cs @@ -1,6 +1,6 @@ #region Copyright notice and license -// Copyright 2015 gRPC authors. +// Copyright 2019 The gRPC Authors // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -26,7 +26,7 @@ namespace Grpc.Core /// Provides an abstraction over the callback providers /// used by AsyncUnaryCall, AsyncDuplexStreamingCall, etc /// - internal /* readonly */ struct AsyncCallState // can be made readonly in C# 7.2 + internal readonly struct AsyncCallState { readonly object responseHeadersAsync; // Task or Func> readonly object getStatusFunc; // Func or Func diff --git a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs index e5513c3f017..f85b0f59d60 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/AsyncCallStateTest.cs @@ -1,6 +1,6 @@ #region Copyright notice and license -// Copyright 2015 gRPC authors. +// Copyright 2019 The gRPC Authors // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. diff --git a/src/csharp/tests.json b/src/csharp/tests.json index 5b9ca6dbd14..50eea00c040 100644 --- a/src/csharp/tests.json +++ b/src/csharp/tests.json @@ -3,6 +3,7 @@ "Grpc.Core.Interceptors.Tests.ClientInterceptorTest", "Grpc.Core.Interceptors.Tests.ServerInterceptorTest", "Grpc.Core.Internal.Tests.AsyncCallServerTest", + "Grpc.Core.Internal.Tests.AsyncCallStateTest", "Grpc.Core.Internal.Tests.AsyncCallTest", "Grpc.Core.Internal.Tests.ChannelArgsSafeHandleTest", "Grpc.Core.Internal.Tests.CompletionQueueEventTest", From 74f3291a13fe71df37923fabaf3b11e1845a4de4 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Wed, 4 Dec 2019 15:33:13 +1300 Subject: [PATCH 3/7] Remove readonly struct to fix build --- src/csharp/Grpc.Core.Api/AsyncCallState.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/csharp/Grpc.Core.Api/AsyncCallState.cs b/src/csharp/Grpc.Core.Api/AsyncCallState.cs index 7f97e322c3c..d91e4e3c398 100644 --- a/src/csharp/Grpc.Core.Api/AsyncCallState.cs +++ b/src/csharp/Grpc.Core.Api/AsyncCallState.cs @@ -26,7 +26,7 @@ namespace Grpc.Core /// Provides an abstraction over the callback providers /// used by AsyncUnaryCall, AsyncDuplexStreamingCall, etc /// - internal readonly struct AsyncCallState + internal struct AsyncCallState { readonly object responseHeadersAsync; // Task or Func> readonly object getStatusFunc; // Func or Func From f31e9d5721491c2a4da87cbf545f4d90886793c0 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Wed, 4 Dec 2019 09:35:07 -0800 Subject: [PATCH 4/7] Support multiple xds servers in bootstrap file. --- .../client_channel/xds/xds_bootstrap.cc | 145 +++++++++++------- .../client_channel/xds/xds_bootstrap.h | 22 +-- .../filters/client_channel/xds/xds_channel.cc | 5 +- .../client_channel/xds/xds_channel_secure.cc | 14 +- .../filters/client_channel/xds/xds_client.cc | 2 +- .../core/client_channel/xds_bootstrap_test.cc | 115 ++++++++------ 6 files changed, 184 insertions(+), 119 deletions(-) diff --git a/src/core/ext/filters/client_channel/xds/xds_bootstrap.cc b/src/core/ext/filters/client_channel/xds/xds_bootstrap.cc index 533b62a2dce..5824391cf1e 100644 --- a/src/core/ext/filters/client_channel/xds/xds_bootstrap.cc +++ b/src/core/ext/filters/client_channel/xds/xds_bootstrap.cc @@ -58,23 +58,23 @@ XdsBootstrap::XdsBootstrap(grpc_slice contents, grpc_error** error) return; } InlinedVector error_list; - bool seen_xds_server = false; + bool seen_xds_servers = false; bool seen_node = false; for (grpc_json* child = tree_->child; child != nullptr; child = child->next) { if (child->key == nullptr) { error_list.push_back( GRPC_ERROR_CREATE_FROM_STATIC_STRING("JSON key is null")); - } else if (strcmp(child->key, "xds_server") == 0) { - if (child->type != GRPC_JSON_OBJECT) { + } else if (strcmp(child->key, "xds_servers") == 0) { + if (child->type != GRPC_JSON_ARRAY) { error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "\"xds_server\" field is not an object")); + "\"xds_servers\" field is not an array")); } - if (seen_xds_server) { + if (seen_xds_servers) { error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "duplicate \"xds_server\" field")); + "duplicate \"xds_servers\" field")); } - seen_xds_server = true; - grpc_error* parse_error = ParseXdsServer(child); + seen_xds_servers = true; + grpc_error* parse_error = ParseXdsServerList(child); if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error); } else if (strcmp(child->key, "node") == 0) { if (child->type != GRPC_JSON_OBJECT) { @@ -90,9 +90,9 @@ XdsBootstrap::XdsBootstrap(grpc_slice contents, grpc_error** error) if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error); } } - if (!seen_xds_server) { + if (!seen_xds_servers) { error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "\"xds_server\" field not present")); + "\"xds_servers\" field not present")); } *error = GRPC_ERROR_CREATE_FROM_VECTOR("errors parsing xds bootstrap file", &error_list); @@ -103,47 +103,7 @@ XdsBootstrap::~XdsBootstrap() { grpc_slice_unref_internal(contents_); } -grpc_error* XdsBootstrap::ParseXdsServer(grpc_json* json) { - InlinedVector error_list; - server_uri_ = nullptr; - bool seen_channel_creds = false; - for (grpc_json* child = json->child; child != nullptr; child = child->next) { - if (child->key == nullptr) { - error_list.push_back( - GRPC_ERROR_CREATE_FROM_STATIC_STRING("JSON key is null")); - } else if (strcmp(child->key, "server_uri") == 0) { - if (child->type != GRPC_JSON_STRING) { - error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "\"server_uri\" field is not a string")); - } - if (server_uri_ != nullptr) { - error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "duplicate \"server_uri\" field")); - } - server_uri_ = child->value; - } else if (strcmp(child->key, "channel_creds") == 0) { - if (child->type != GRPC_JSON_ARRAY) { - error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "\"channel_creds\" field is not an array")); - } - if (seen_channel_creds) { - error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "duplicate \"channel_creds\" field")); - } - seen_channel_creds = true; - grpc_error* parse_error = ParseChannelCredsArray(child); - if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error); - } - } - if (server_uri_ == nullptr) { - error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "\"server_uri\" field not present")); - } - return GRPC_ERROR_CREATE_FROM_VECTOR("errors parsing \"xds_server\" object", - &error_list); -} - -grpc_error* XdsBootstrap::ParseChannelCredsArray(grpc_json* json) { +grpc_error* XdsBootstrap::ParseXdsServerList(grpc_json* json) { InlinedVector error_list; size_t idx = 0; for (grpc_json *child = json->child; child != nullptr; @@ -158,7 +118,81 @@ grpc_error* XdsBootstrap::ParseChannelCredsArray(grpc_json* json) { gpr_asprintf(&msg, "array element %" PRIuPTR " is not an object", idx); error_list.push_back(GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg)); } else { - grpc_error* parse_error = ParseChannelCreds(child, idx); + grpc_error* parse_error = ParseXdsServer(child, idx); + if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error); + } + } + return GRPC_ERROR_CREATE_FROM_VECTOR("errors parsing \"xds_servers\" array", + &error_list); +} + +grpc_error* XdsBootstrap::ParseXdsServer(grpc_json* json, size_t idx) { + InlinedVector error_list; + servers_.emplace_back(); + XdsServer& server = servers_[servers_.size() - 1]; + bool seen_channel_creds = false; + for (grpc_json* child = json->child; child != nullptr; child = child->next) { + if (child->key == nullptr) { + error_list.push_back( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("JSON key is null")); + } else if (strcmp(child->key, "server_uri") == 0) { + if (child->type != GRPC_JSON_STRING) { + error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "\"server_uri\" field is not a string")); + } + if (server.server_uri != nullptr) { + error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "duplicate \"server_uri\" field")); + } + server.server_uri = child->value; + } else if (strcmp(child->key, "channel_creds") == 0) { + if (child->type != GRPC_JSON_ARRAY) { + error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "\"channel_creds\" field is not an array")); + } + if (seen_channel_creds) { + error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "duplicate \"channel_creds\" field")); + } + seen_channel_creds = true; + grpc_error* parse_error = ParseChannelCredsArray(child, &server); + if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error); + } + } + if (server.server_uri == nullptr) { + error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "\"server_uri\" field not present")); + } + // Can't use GRPC_ERROR_CREATE_FROM_VECTOR() here, because the error + // string is not static in this case. + if (error_list.empty()) return GRPC_ERROR_NONE; + char* msg; + gpr_asprintf(&msg, "errors parsing index %" PRIuPTR, idx); + grpc_error* error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); + gpr_free(msg); + for (size_t i = 0; i < error_list.size(); ++i) { + error = grpc_error_add_child(error, error_list[i]); + } + return error; +} + +grpc_error* XdsBootstrap::ParseChannelCredsArray(grpc_json* json, + XdsServer* server) { + InlinedVector error_list; + size_t idx = 0; + for (grpc_json *child = json->child; child != nullptr; + child = child->next, ++idx) { + if (child->key != nullptr) { + char* msg; + gpr_asprintf(&msg, "array element %" PRIuPTR " key is not null", idx); + error_list.push_back(GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg)); + } + if (child->type != GRPC_JSON_OBJECT) { + char* msg; + gpr_asprintf(&msg, "array element %" PRIuPTR " is not an object", idx); + error_list.push_back(GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg)); + } else { + grpc_error* parse_error = ParseChannelCreds(child, idx, server); if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error); } } @@ -166,7 +200,8 @@ grpc_error* XdsBootstrap::ParseChannelCredsArray(grpc_json* json) { &error_list); } -grpc_error* XdsBootstrap::ParseChannelCreds(grpc_json* json, size_t idx) { +grpc_error* XdsBootstrap::ParseChannelCreds(grpc_json* json, size_t idx, + XdsServer* server) { InlinedVector error_list; ChannelCreds channel_creds; for (grpc_json* child = json->child; child != nullptr; child = child->next) { @@ -195,7 +230,9 @@ grpc_error* XdsBootstrap::ParseChannelCreds(grpc_json* json, size_t idx) { channel_creds.config = child; } } - if (channel_creds.type != nullptr) channel_creds_.push_back(channel_creds); + if (channel_creds.type != nullptr) { + server->channel_creds.push_back(channel_creds); + } // Can't use GRPC_ERROR_CREATE_FROM_VECTOR() here, because the error // string is not static in this case. if (error_list.empty()) return GRPC_ERROR_NONE; diff --git a/src/core/ext/filters/client_channel/xds/xds_bootstrap.h b/src/core/ext/filters/client_channel/xds/xds_bootstrap.h index d8705fa3ff2..5176ac749ee 100644 --- a/src/core/ext/filters/client_channel/xds/xds_bootstrap.h +++ b/src/core/ext/filters/client_channel/xds/xds_bootstrap.h @@ -58,6 +58,11 @@ class XdsBootstrap { grpc_json* config = nullptr; }; + struct XdsServer { + const char* server_uri = nullptr; + InlinedVector channel_creds; + }; + // If *error is not GRPC_ERROR_NONE after returning, then there was an // error reading the file. static std::unique_ptr ReadFromFile(grpc_error** error); @@ -66,16 +71,16 @@ class XdsBootstrap { XdsBootstrap(grpc_slice contents, grpc_error** error); ~XdsBootstrap(); - const char* server_uri() const { return server_uri_; } - const InlinedVector& channel_creds() const { - return channel_creds_; - } + // TODO(roth): We currently support only one server. Fix this when we + // add support for fallback for the xds channel. + const XdsServer& server() const { return servers_[0]; } const Node* node() const { return node_.get(); } private: - grpc_error* ParseXdsServer(grpc_json* json); - grpc_error* ParseChannelCredsArray(grpc_json* json); - grpc_error* ParseChannelCreds(grpc_json* json, size_t idx); + grpc_error* ParseXdsServerList(grpc_json* json); + grpc_error* ParseXdsServer(grpc_json* json, size_t idx); + grpc_error* ParseChannelCredsArray(grpc_json* json, XdsServer* server); + grpc_error* ParseChannelCreds(grpc_json* json, size_t idx, XdsServer* server); grpc_error* ParseNode(grpc_json* json); grpc_error* ParseLocality(grpc_json* json); @@ -90,8 +95,7 @@ class XdsBootstrap { grpc_slice contents_; grpc_json* tree_ = nullptr; - const char* server_uri_ = nullptr; - InlinedVector channel_creds_; + InlinedVector servers_; std::unique_ptr node_; }; diff --git a/src/core/ext/filters/client_channel/xds/xds_channel.cc b/src/core/ext/filters/client_channel/xds/xds_channel.cc index 31f598556a6..e8ba3706f13 100644 --- a/src/core/ext/filters/client_channel/xds/xds_channel.cc +++ b/src/core/ext/filters/client_channel/xds/xds_channel.cc @@ -30,8 +30,9 @@ grpc_channel_args* ModifyXdsChannelArgs(grpc_channel_args* args) { grpc_channel* CreateXdsChannel(const XdsBootstrap& bootstrap, const grpc_channel_args& args) { - if (!bootstrap.channel_creds().empty()) return nullptr; - return grpc_insecure_channel_create(bootstrap.server_uri(), &args, nullptr); + if (!bootstrap.server().channel_creds.empty()) return nullptr; + return grpc_insecure_channel_create(bootstrap.server().server_uri, &args, + nullptr); } } // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/xds/xds_channel_secure.cc b/src/core/ext/filters/client_channel/xds/xds_channel_secure.cc index 64e5f34607a..9a752fe5826 100644 --- a/src/core/ext/filters/client_channel/xds/xds_channel_secure.cc +++ b/src/core/ext/filters/client_channel/xds/xds_channel_secure.cc @@ -67,12 +67,14 @@ grpc_channel* CreateXdsChannel(const XdsBootstrap& bootstrap, const grpc_channel_args& args) { grpc_channel_credentials* creds = nullptr; RefCountedPtr creds_to_unref; - if (!bootstrap.channel_creds().empty()) { - for (size_t i = 0; i < bootstrap.channel_creds().size(); ++i) { - if (strcmp(bootstrap.channel_creds()[i].type, "google_default") == 0) { + if (!bootstrap.server().channel_creds.empty()) { + for (size_t i = 0; i < bootstrap.server().channel_creds.size(); ++i) { + if (strcmp(bootstrap.server().channel_creds[i].type, "google_default") == + 0) { creds = grpc_google_default_credentials_create(); break; - } else if (strcmp(bootstrap.channel_creds()[i].type, "fake") == 0) { + } else if (strcmp(bootstrap.server().channel_creds[i].type, "fake") == + 0) { creds = grpc_fake_transport_security_credentials_create(); break; } @@ -83,7 +85,7 @@ grpc_channel* CreateXdsChannel(const XdsBootstrap& bootstrap, creds = grpc_channel_credentials_find_in_args(&args); if (creds == nullptr) { // Built with security but parent channel is insecure. - return grpc_insecure_channel_create(bootstrap.server_uri(), &args, + return grpc_insecure_channel_create(bootstrap.server().server_uri, &args, nullptr); } } @@ -91,7 +93,7 @@ grpc_channel* CreateXdsChannel(const XdsBootstrap& bootstrap, grpc_channel_args* new_args = grpc_channel_args_copy_and_remove(&args, &arg_to_remove, 1); grpc_channel* channel = grpc_secure_channel_create( - creds, bootstrap.server_uri(), new_args, nullptr); + creds, bootstrap.server().server_uri, new_args, nullptr); grpc_channel_args_destroy(new_args); return channel; } diff --git a/src/core/ext/filters/client_channel/xds/xds_client.cc b/src/core/ext/filters/client_channel/xds/xds_client.cc index c977844854a..56b8c2a1c16 100644 --- a/src/core/ext/filters/client_channel/xds/xds_client.cc +++ b/src/core/ext/filters/client_channel/xds/xds_client.cc @@ -1268,7 +1268,7 @@ XdsClient::XdsClient(Combiner* combiner, grpc_pollset_set* interested_parties, } if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { gpr_log(GPR_INFO, "[xds_client %p: creating channel to %s", this, - bootstrap_->server_uri()); + bootstrap_->server().server_uri); } chand_ = MakeOrphanable( Ref(DEBUG_LOCATION, "XdsClient+ChannelState"), channel_args); diff --git a/test/core/client_channel/xds_bootstrap_test.cc b/test/core/client_channel/xds_bootstrap_test.cc index e1fb3a9e2df..e5de1afc695 100644 --- a/test/core/client_channel/xds_bootstrap_test.cc +++ b/test/core/client_channel/xds_bootstrap_test.cc @@ -38,16 +38,28 @@ void VerifyRegexMatch(grpc_error* error, const std::regex& e) { TEST(XdsBootstrapTest, Basic) { const char* json = "{" - " \"xds_server\": {" - " \"server_uri\": \"fake:///lb\"," - " \"channel_creds\": [" - " {" - " \"type\": \"fake\"," - " \"ignore\": 0" - " }" - " ]," - " \"ignore\": 0" - " }," + " \"xds_servers\": [" + " {" + " \"server_uri\": \"fake:///lb\"," + " \"channel_creds\": [" + " {" + " \"type\": \"fake\"," + " \"ignore\": 0" + " }" + " ]," + " \"ignore\": 0" + " }," + " {" + " \"server_uri\": \"ignored\"," + " \"channel_creds\": [" + " {" + " \"type\": \"ignored\"," + " \"ignore\": 0" + " }" + " ]," + " \"ignore\": 0" + " }" + " ]," " \"node\": {" " \"id\": \"foo\"," " \"cluster\": \"bar\"," @@ -74,11 +86,11 @@ TEST(XdsBootstrapTest, Basic) { grpc_slice slice = grpc_slice_from_copied_string(json); grpc_error* error = GRPC_ERROR_NONE; grpc_core::XdsBootstrap bootstrap(slice, &error); - EXPECT_EQ(error, GRPC_ERROR_NONE); - EXPECT_STREQ(bootstrap.server_uri(), "fake:///lb"); - ASSERT_EQ(bootstrap.channel_creds().size(), 1); - EXPECT_STREQ(bootstrap.channel_creds()[0].type, "fake"); - EXPECT_EQ(bootstrap.channel_creds()[0].config, nullptr); + EXPECT_EQ(error, GRPC_ERROR_NONE) << grpc_error_string(error); + EXPECT_STREQ(bootstrap.server().server_uri, "fake:///lb"); + ASSERT_EQ(bootstrap.server().channel_creds.size(), 1); + EXPECT_STREQ(bootstrap.server().channel_creds[0].type, "fake"); + EXPECT_EQ(bootstrap.server().channel_creds[0].config, nullptr); ASSERT_NE(bootstrap.node(), nullptr); EXPECT_STREQ(bootstrap.node()->id, "foo"); EXPECT_STREQ(bootstrap.node()->cluster, "bar"); @@ -152,16 +164,18 @@ TEST(XdsBootstrapTest, Basic) { TEST(XdsBootstrapTest, ValidWithoutChannelCredsAndNode) { const char* json = "{" - " \"xds_server\": {" - " \"server_uri\": \"fake:///lb\"" - " }" + " \"xds_servers\": [" + " {" + " \"server_uri\": \"fake:///lb\"" + " }" + " ]" "}"; grpc_slice slice = grpc_slice_from_copied_string(json); grpc_error* error = GRPC_ERROR_NONE; grpc_core::XdsBootstrap bootstrap(slice, &error); EXPECT_EQ(error, GRPC_ERROR_NONE); - EXPECT_STREQ(bootstrap.server_uri(), "fake:///lb"); - EXPECT_EQ(bootstrap.channel_creds().size(), 0); + EXPECT_STREQ(bootstrap.server().server_uri, "fake:///lb"); + EXPECT_EQ(bootstrap.server().channel_creds.size(), 0); EXPECT_EQ(bootstrap.node(), nullptr); } @@ -185,30 +199,31 @@ TEST(XdsBootstrapTest, MalformedJson) { VerifyRegexMatch(error, e); } -TEST(XdsBootstrapTest, MissingXdsServer) { +TEST(XdsBootstrapTest, MissingXdsServers) { grpc_slice slice = grpc_slice_from_copied_string("{}"); grpc_error* error = GRPC_ERROR_NONE; grpc_core::XdsBootstrap bootstrap(slice, &error); gpr_log(GPR_ERROR, "%s", grpc_error_string(error)); ASSERT_TRUE(error != GRPC_ERROR_NONE); - std::regex e(std::string("\"xds_server\" field not present")); + std::regex e(std::string("\"xds_servers\" field not present")); VerifyRegexMatch(error, e); } -TEST(XdsBootstrapTest, BadXdsServer) { +TEST(XdsBootstrapTest, BadXdsServers) { grpc_slice slice = grpc_slice_from_copied_string( "{" - " \"xds_server\":1," - " \"xds_server\":{}" + " \"xds_servers\":1," + " \"xds_servers\":[{}]" "}"); grpc_error* error = GRPC_ERROR_NONE; grpc_core::XdsBootstrap bootstrap(slice, &error); gpr_log(GPR_ERROR, "%s", grpc_error_string(error)); ASSERT_TRUE(error != GRPC_ERROR_NONE); std::regex e( - std::string("\"xds_server\" field is not an object(.*)" - "duplicate \"xds_server\" field(.*)" - "errors parsing \"xds_server\" object(.*)" + std::string("\"xds_servers\" field is not an array(.*)" + "duplicate \"xds_servers\" field(.*)" + "errors parsing \"xds_servers\" array(.*)" + "errors parsing index 0(.*)" "\"server_uri\" field not present")); VerifyRegexMatch(error, e); } @@ -216,19 +231,22 @@ TEST(XdsBootstrapTest, BadXdsServer) { TEST(XdsBootstrapTest, BadXdsServerContents) { grpc_slice slice = grpc_slice_from_copied_string( "{" - " \"xds_server\":{" - " \"server_uri\":1," - " \"server_uri\":\"foo\"," - " \"channel_creds\":1," - " \"channel_creds\":{}" - " }" + " \"xds_servers\":[" + " {" + " \"server_uri\":1," + " \"server_uri\":\"foo\"," + " \"channel_creds\":1," + " \"channel_creds\":{}" + " }" + " ]" "}"); grpc_error* error = GRPC_ERROR_NONE; grpc_core::XdsBootstrap bootstrap(slice, &error); gpr_log(GPR_ERROR, "%s", grpc_error_string(error)); ASSERT_TRUE(error != GRPC_ERROR_NONE); std::regex e( - std::string("errors parsing \"xds_server\" object(.*)" + std::string("errors parsing \"xds_servers\" array(.*)" + "errors parsing index 0(.*)" "\"server_uri\" field is not a string(.*)" "duplicate \"server_uri\" field(.*)" "\"channel_creds\" field is not an array(.*)" @@ -240,24 +258,27 @@ TEST(XdsBootstrapTest, BadXdsServerContents) { TEST(XdsBootstrapTest, BadChannelCredsContents) { grpc_slice slice = grpc_slice_from_copied_string( "{" - " \"xds_server\":{" - " \"server_uri\":\"foo\"," - " \"channel_creds\":[" - " {" - " \"type\":0," - " \"type\":\"fake\"," - " \"config\":1," - " \"config\":{}" - " }" - " ]" - " }" + " \"xds_servers\":[" + " {" + " \"server_uri\":\"foo\"," + " \"channel_creds\":[" + " {" + " \"type\":0," + " \"type\":\"fake\"," + " \"config\":1," + " \"config\":{}" + " }" + " ]" + " }" + " ]" "}"); grpc_error* error = GRPC_ERROR_NONE; grpc_core::XdsBootstrap bootstrap(slice, &error); gpr_log(GPR_ERROR, "%s", grpc_error_string(error)); ASSERT_TRUE(error != GRPC_ERROR_NONE); std::regex e( - std::string("errors parsing \"xds_server\" object(.*)" + std::string("errors parsing \"xds_servers\" array(.*)" + "errors parsing index 0(.*)" "errors parsing \"channel_creds\" array(.*)" "errors parsing index 0(.*)" "\"type\" field is not a string(.*)" From 4f834e73b2dec4fa3c40c5d0b937d06a43573fa2 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Thu, 5 Dec 2019 09:32:37 +1300 Subject: [PATCH 5/7] Clean up --- src/csharp/Grpc.Core.Api/AsyncCallState.cs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/csharp/Grpc.Core.Api/AsyncCallState.cs b/src/csharp/Grpc.Core.Api/AsyncCallState.cs index d91e4e3c398..02e2e2cc615 100644 --- a/src/csharp/Grpc.Core.Api/AsyncCallState.cs +++ b/src/csharp/Grpc.Core.Api/AsyncCallState.cs @@ -85,8 +85,14 @@ namespace Grpc.Core internal void Dispose() { var withState = disposeAction as Action; - if (withState != null) withState(callbackState); - else ((Action)disposeAction)(); + if (withState != null) + { + withState(callbackState); + } + else + { + ((Action)disposeAction)(); + } } } } From 408ec6e867c0c3d942be8824d69b0ec70df41f43 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Wed, 4 Dec 2019 13:20:15 -0800 Subject: [PATCH 6/7] fix xds_end2end_test --- test/cpp/end2end/xds_end2end_test.cc | 36 +++++++++++++++------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/test/cpp/end2end/xds_end2end_test.cc b/test/cpp/end2end/xds_end2end_test.cc index 2cc704d6f95..03b2044fab4 100644 --- a/test/cpp/end2end/xds_end2end_test.cc +++ b/test/cpp/end2end/xds_end2end_test.cc @@ -105,14 +105,16 @@ constexpr int kDefaultLocalityPriority = 0; constexpr char kBootstrapFile[] = "{\n" - " \"xds_server\": {\n" - " \"server_uri\": \"fake:///lb\",\n" - " \"channel_creds\": [\n" - " {\n" - " \"type\": \"fake\"\n" - " }\n" - " ]\n" - " },\n" + " \"xds_servers\": [\n" + " {\n" + " \"server_uri\": \"fake:///lb\",\n" + " \"channel_creds\": [\n" + " {\n" + " \"type\": \"fake\"\n" + " }\n" + " ]\n" + " }\n" + " ],\n" " \"node\": {\n" " \"id\": \"xds_end2end_test\",\n" " \"cluster\": \"test\",\n" @@ -129,14 +131,16 @@ constexpr char kBootstrapFile[] = constexpr char kBootstrapFileBad[] = "{\n" - " \"xds_server\": {\n" - " \"server_uri\": \"fake:///wrong_lb\",\n" - " \"channel_creds\": [\n" - " {\n" - " \"type\": \"fake\"\n" - " }\n" - " ]\n" - " },\n" + " \"xds_servers\": [\n" + " {\n" + " \"server_uri\": \"fake:///wrong_lb\",\n" + " \"channel_creds\": [\n" + " {\n" + " \"type\": \"fake\"\n" + " }\n" + " ]\n" + " }\n" + " ],\n" " \"node\": {\n" " }\n" "}\n"; From 9cedb80c6f01e643c4d16dedc55eef14c203c8b5 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Thu, 5 Dec 2019 12:12:43 +1300 Subject: [PATCH 7/7] Fix flaky health check test --- .../HealthServiceImplTest.cs | 20 ++++++++------ .../TestResponseStreamWriter.cs | 26 ++++++++++++++++--- .../Grpc.HealthCheck/HealthServiceImpl.cs | 7 ++--- 3 files changed, 38 insertions(+), 15 deletions(-) diff --git a/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs b/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs index fbded690c97..08fe84cafd7 100644 --- a/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs +++ b/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs @@ -201,18 +201,22 @@ namespace Grpc.HealthCheck.Tests { var cts = new CancellationTokenSource(); var context = new TestServerCallContext(cts.Token); - var writer = new TestResponseStreamWriter(); + var writer = new TestResponseStreamWriter(started: false); var impl = new HealthServiceImpl(); var callTask = impl.Watch(new HealthCheckRequest { Service = "" }, writer, context); - // Write new 10 statuses. Only last 5 statuses will be returned when we read them from watch writer + // Write new statuses. Only last statuses will be returned when we read them from watch writer for (var i = 0; i < HealthServiceImpl.MaxStatusBufferSize * 2; i++) { // These statuses aren't "valid" but it is useful for testing to have an incrementing number - impl.SetStatus("", (HealthCheckResponse.Types.ServingStatus)i); + impl.SetStatus("", (HealthCheckResponse.Types.ServingStatus)i + 10); } + // Start reading responses now that statuses have been queued up + // This is to keep the test non-flakey + writer.Start(); + // Read messages in a background task var statuses = new List(); var readStatusesTask = Task.Run(async () => { @@ -240,11 +244,11 @@ namespace Grpc.HealthCheck.Tests Assert.AreEqual(HealthCheckResponse.Types.ServingStatus.ServiceUnknown, statuses[0]); // Last 5 queued messages - Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)5, statuses[1]); - Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)6, statuses[2]); - Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)7, statuses[3]); - Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)8, statuses[4]); - Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)9, statuses[5]); + Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)15, statuses[statuses.Count - 5]); + Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)16, statuses[statuses.Count - 4]); + Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)17, statuses[statuses.Count - 3]); + Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)18, statuses[statuses.Count - 2]); + Assert.AreEqual((HealthCheckResponse.Types.ServingStatus)19, statuses[statuses.Count - 1]); } #endif diff --git a/src/csharp/Grpc.HealthCheck.Tests/TestResponseStreamWriter.cs b/src/csharp/Grpc.HealthCheck.Tests/TestResponseStreamWriter.cs index 9c7ad53c0b1..a0b1a51c8aa 100644 --- a/src/csharp/Grpc.HealthCheck.Tests/TestResponseStreamWriter.cs +++ b/src/csharp/Grpc.HealthCheck.Tests/TestResponseStreamWriter.cs @@ -25,24 +25,42 @@ namespace Grpc.HealthCheck.Tests { internal class TestResponseStreamWriter : IServerStreamWriter { - private Channel _channel; + private readonly Channel _channel; + private readonly TaskCompletionSource _startTcs; - public TestResponseStreamWriter(int maxCapacity = 1) + public TestResponseStreamWriter(int maxCapacity = 1, bool started = true) { _channel = System.Threading.Channels.Channel.CreateBounded(new BoundedChannelOptions(maxCapacity) { SingleReader = false, SingleWriter = true, FullMode = BoundedChannelFullMode.Wait }); + if (!started) + { + _startTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } } public ChannelReader WrittenMessagesReader => _channel.Reader; public WriteOptions WriteOptions { get; set; } - public Task WriteAsync(HealthCheckResponse message) + public async Task WriteAsync(HealthCheckResponse message) { - return _channel.Writer.WriteAsync(message).AsTask(); + if (_startTcs != null) + { + await _startTcs.Task; + } + + await _channel.Writer.WriteAsync(message); + } + + public void Start() + { + if (_startTcs != null) + { + _startTcs.TrySetResult(null); + } } public void Complete() diff --git a/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs b/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs index 685a93985fe..8ff95552323 100644 --- a/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs +++ b/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs @@ -157,9 +157,6 @@ namespace Grpc.HealthCheck { string service = request.Service; - HealthCheckResponse response = GetHealthCheckResponse(service, throwOnNotFound: false); - await responseStream.WriteAsync(response); - // Channel is used to to marshall multiple callers updating status into a single queue. // This is required because IServerStreamWriter is not thread safe. // @@ -205,6 +202,10 @@ namespace Grpc.HealthCheck channel.Writer.Complete(); }); + // Send current status immediately + HealthCheckResponse response = GetHealthCheckResponse(service, throwOnNotFound: false); + await responseStream.WriteAsync(response); + // Read messages. WaitToReadAsync will wait until new messages are available. // Loop will exit when the call is canceled and the writer is marked as complete. while (await channel.Reader.WaitToReadAsync())