Sending metadata request once per scriptHost instance (#9248)

Co-authored-by: Brett Samblanet <[email protected]>

Author: soninaren

src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs

@@ -52,6 +52,7 @@ internal class GrpcWorkerChannel : IRpcWorkerChannel, IDisposable
         private readonly IOptions<WorkerConcurrencyOptions> _workerConcurrencyOptions;
         private readonly WaitCallback _processInbound;
         private readonly object _syncLock = new object();
+        private readonly object _metadataLock = new object();
         private readonly Dictionary<MsgType, Queue<PendingItem>> _pendingActions = new();
         private readonly ChannelWriter<OutboundGrpcEvent> _outbound;
         private readonly ChannelReader<InboundGrpcEvent> _inbound;
@@ -85,6 +86,7 @@ internal class GrpcWorkerChannel : IRpcWorkerChannel, IDisposable
         private IHttpProxyService _httpProxyService;
         private Uri _httpProxyEndpoint;
         private System.Timers.Timer _timer;
+        private bool _functionMetadataRequestSent = false;
 
         internal GrpcWorkerChannel(
             string workerId,
@@ -540,6 +542,8 @@ internal FunctionLoadRequestCollection GetFunctionLoadRequestCollection(IEnumera
         public Task SendFunctionEnvironmentReloadRequest()
         {
             _functionsIndexingTask = new TaskCompletionSource<List<RawFunctionMetadata>>(TaskCreationOptions.RunContinuationsAsynchronously);
+            _functionMetadataRequestSent = false;
+
             _workerChannelLogger.LogDebug("Sending FunctionEnvironmentReloadRequest to WorkerProcess with Pid: '{0}'", _rpcWorkerProcess.Id);
             IDisposable latencyEvent = _metricsLogger.LatencyEvent(MetricEventNames.SpecializationEnvironmentReloadRequestResponse);
 
@@ -795,22 +799,32 @@ public Task<List<RawFunctionMetadata>> GetFunctionMetadata()
 
         internal Task<List<RawFunctionMetadata>> SendFunctionMetadataRequest()
         {
-            // reset indexing task when in case we need to send another request
-            _functionsIndexingTask = new TaskCompletionSource<List<RawFunctionMetadata>>(TaskCreationOptions.RunContinuationsAsynchronously);
+            _workerChannelLogger.LogDebug("Fetching worker metadata, FunctionMetadataReceived set to: {functionMetadataReceived}", _functionMetadataRequestSent);
+            if (!_functionMetadataRequestSent)
+            {
+                lock (_metadataLock)
+                {
+                    if (!_functionMetadataRequestSent)
+                    {
+                        RegisterCallbackForNextGrpcMessage(MsgType.FunctionMetadataResponse, _functionLoadTimeout, 1,
+                    msg => ProcessFunctionMetadataResponses(msg.Message.FunctionMetadataResponse), HandleWorkerMetadataRequestError);
 
-            RegisterCallbackForNextGrpcMessage(MsgType.FunctionMetadataResponse, _functionLoadTimeout, 1,
-                msg => ProcessFunctionMetadataResponses(msg.Message.FunctionMetadataResponse), HandleWorkerMetadataRequestError);
+                        _workerChannelLogger.LogDebug("Sending WorkerMetadataRequest to {language} worker with worker ID {workerID}", _runtime, _workerId);
 
-            _workerChannelLogger.LogDebug("Sending WorkerMetadataRequest to {language} worker with worker ID {workerID}", _runtime, _workerId);
+                        // sends the function app directory path to worker for indexing
+                        SendStreamingMessage(new StreamingMessage
+                        {
+                            FunctionsMetadataRequest = new FunctionsMetadataRequest()
+                            {
+                                FunctionAppDirectory = _applicationHostOptions.CurrentValue.ScriptPath
+                            }
+                        });
 
-            // sends the function app directory path to worker for indexing
-            SendStreamingMessage(new StreamingMessage
-            {
-                FunctionsMetadataRequest = new FunctionsMetadataRequest()
-                {
-                    FunctionAppDirectory = _applicationHostOptions.CurrentValue.ScriptPath
+                        _functionMetadataRequestSent = true;
+                    }
                 }
-            });
+            }
+
             return _functionsIndexingTask.Task;
         }
 

test/WebJobs.Script.Tests/Workers/Rpc/GrpcWorkerChannelTests.cs

@@ -1303,6 +1303,54 @@ public async Task SendInvocationRequest_ValidateTraceContext_SessionId()
             }
         }
 
+        [Fact]
+        public async Task GetFunctionMetadata_MultipleCalls_ReturnSameTask()
+        {
+            using var block1 = new SemaphoreSlim(0, 1);
+            using var block2 = new SemaphoreSlim(0, 1);
+            int count = 0;
+
+            await CreateDefaultWorkerChannel();
+
+            _testFunctionRpcService.OnMessage(StreamingMessage.ContentOneofCase.FunctionsMetadataRequest,
+                async _ =>
+                {
+                    if (Interlocked.Increment(ref count) == 1)
+                    {
+                        // notify the second request it can start
+                        block2.Release();
+
+                        // make the first call sit and wait until we know we've issued the second
+                        await block1.WaitAsync();
+                    }
+
+                    _testFunctionRpcService.PublishWorkerMetadataResponse("TestFunctionId1", null, null, false, false, false);
+                });
+
+            var functionsTask1 = _workerChannel.GetFunctionMetadata();
+            await Task.Yield();
+
+            // wait until the first request has made it to the callback before issuing the second
+            await block2.WaitAsync();
+
+            var functionsTask2 = _workerChannel.GetFunctionMetadata();
+            await Task.Yield();
+
+            // now that both requests have been made, let the first return
+            block1.Release();
+
+            //Assert.Same(functionsTask1, functionsTask2);
+
+            var allTask = Task.WhenAll(functionsTask1, functionsTask2);
+            var timeoutTask = Task.Delay(5000);
+
+            // the timeout should never fire
+            var completedTask = await Task.WhenAny(allTask, timeoutTask);
+
+            Assert.True(completedTask == allTask, "Timed out waiting for tasks to complete");
+            Assert.Same(functionsTask1, functionsTask2);
+        }
+
         private IEnumerable<FunctionMetadata> GetTestFunctionsList(string runtime, bool addWorkerProperties = false)
         {
             var metadata1 = new FunctionMetadata()