optimize: send trace segments to collector in batch queue. (#62)

* optimize: send trace segments to collector in batch queue.

* bugfix: batch send trace segments process should continue until the number of segments left in the queue lessthan the _batchsize.

* bugfix: improve the batch sending segments process, and limit the max memory usage of the pending queue.

src/SkyWalking.Abstractions/Config/AgentConfig.cs

@@ -46,7 +46,14 @@ namespace SkyWalking.Config
         /// The max number of spans in a single segment. Through this config item, skywalking keep your application memory cost estimated.
         /// </summary>
         public static int SpanLimitPerSegment = 300;
-        
+
+        /// <summary>
+        /// The max number of segments in the memory queue waiting to be sent to collector.
+        /// It means that when the number of queued segments reachs this limit,
+        /// any more segments enqueued into the sending queue, will leads the same number of oldest queued segments dequeued and droped.
+        /// Zero or minus value means no limit.
+        /// </summary>
+        public static int PendingSegmentsLimit = 300000;
         
     }
 }
\ No newline at end of file

src/SkyWalking.AspNet/SkyWalkingOptions.cs

@@ -47,5 +47,13 @@ namespace SkyWalking.AspNet
         /// Namespace isolates headers in cross propagation. The HEADER name will be 'HeaderName:Namespace'.
         /// </summary>
         public string Namespace { get; set; }
+
+        /// <summary>
+        /// The max number of segments in the memory queue waiting to be sent to collector.
+        /// It means that when the number of queued segments reachs this limit,
+        /// any more segments enqueued into the sending queue, will leads the same number of oldest queued segments dequeued and droped.
+        /// Zero or minus value means no limit.
+        /// </summary>
+        public int PendingSegmentsLimit { get; set; } = 300000;
     }
 }
\ No newline at end of file

src/SkyWalking.AspNet/SkyWalkingStartup.cs

@@ -53,6 +53,11 @@ namespace SkyWalking.AspNet
             {
                 AgentConfig.SamplePer3Secs = v;
             }
+            var pendingSegmentsLimit = GetAppSetting(nameof(AgentConfig.PendingSegmentsLimit), false);
+            if(int.TryParse(pendingSegmentsLimit, out v))
+            {
+                AgentConfig.PendingSegmentsLimit = v;
+            }
         }
 
         private string GetAppSetting(string key, bool @throw)

src/SkyWalking.AspNetCore/SkyWalkingHostedService.cs

@@ -54,6 +54,7 @@ namespace SkyWalking.AspNetCore
             AgentConfig.ApplicationCode = options.Value.ApplicationCode;
             CollectorConfig.DirectServers = options.Value.DirectServers;
             AgentConfig.SamplePer3Secs = options.Value.SamplePer3Secs;
+            AgentConfig.PendingSegmentsLimit = options.Value.PendingSegmentsLimit;
             _logger = LogManager.GetLogger<SkyWalkingHostedService>();
             _diagnosticObserver = diagnosticObserver;
         }
@@ -79,8 +80,8 @@ namespace SkyWalking.AspNetCore
             _logger.Info("SkyWalking Agent stopping...");
             try
             {
-                await GrpcConnectionManager.Instance.ShutdownAsync();
                 ServiceManager.Instance.Dispose();
+                await GrpcConnectionManager.Instance.ShutdownAsync();
                 _logger.Info("SkyWalking Agent stopped.");
             }
             catch (Exception e)

src/SkyWalking.AspNetCore/SkyWalkingOptions.cs

@@ -58,5 +58,17 @@ namespace SkyWalking.AspNetCore
             get; 
             set;
         } = -1;
+
+        /// <summary>
+        /// The max number of segments in the memory queue waiting to be sent to collector.
+        /// It means that when the number of queued segments reachs this limit,
+        /// any more segments enqueued into the sending queue, will leads the same number of oldest queued segments dequeued and droped.
+        /// Zero or minus value means no limit.
+        /// </summary>
+        public int PendingSegmentsLimit
+        {
+            get;
+            set;
+        } = 300000;
     }
 }

src/SkyWalking.Core/Remote/GrpcTraceSegmentService.cs

@@ -17,10 +17,13 @@
  */
 
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using SkyWalking.Boot;
+using SkyWalking.Config;
 using SkyWalking.Context;
 using SkyWalking.Context.Trace;
 using SkyWalking.Logging;
@@ -29,21 +32,30 @@ using SkyWalking.Utils;
 
 namespace SkyWalking.Remote
 {
-    public class GrpcTraceSegmentService : IBootService, ITracingContextListener
+    public class GrpcTraceSegmentService : TimerService, ITracingContextListener
     {
         private static readonly ILogger _logger = LogManager.GetLogger<GrpcTraceSegmentService>();
+        private static readonly ConcurrentQueue<ITraceSegment> _traceSegments
+            = new ConcurrentQueue<ITraceSegment>();
 
-        public void Dispose()
+        public override void Dispose()
         {
             TracingContext.ListenerManager.Remove(this);
+            if(_traceSegments.Count > 0)
+            {
+                BatchSendTraceSegments().GetAwaiter().GetResult();
+            }
+            base.Dispose();
         }
 
-        public int Order { get; } = 1;
+        public override int Order { get; } = 1;
 
-        public Task Initialize(CancellationToken token)
+        protected override TimeSpan Interval => TimeSpan.FromSeconds(1);
+
+        protected override Task Initializing(CancellationToken token)
         {
             TracingContext.ListenerManager.Add(this);
-            return TaskUtils.CompletedTask;
+            return base.Initializing(token);
         }
 
         public async void AfterFinished(ITraceSegment traceSegment)
@@ -53,6 +65,23 @@ namespace SkyWalking.Remote
                 return;
             }
 
+            if (_traceSegments.Count >= AgentConfig.PendingSegmentsLimit && AgentConfig.PendingSegmentsLimit > 0)
+            {
+                _traceSegments.TryDequeue(out var v);
+            }
+            _traceSegments.Enqueue(traceSegment);
+        }
+
+        protected async override Task Execute(CancellationToken token)
+        {
+            await BatchSendTraceSegments();
+        }
+
+        private async Task BatchSendTraceSegments()
+        {
+            if (_traceSegments.Count == 0)
+                return;
+
             var availableConnection = GrpcConnectionManager.Instance.GetAvailableConnection();
             if (availableConnection == null)
             {
@@ -63,23 +92,25 @@ namespace SkyWalking.Remote
 
             try
             {
-                var segment = traceSegment.Transform();
                 var traceSegmentService =
                     new TraceSegmentService.TraceSegmentServiceClient(availableConnection.GrpcChannel);
                 using (var asyncClientStreamingCall = traceSegmentService.collect())
                 {
-                    await asyncClientStreamingCall.RequestStream.WriteAsync(segment);
+                    while (_traceSegments.TryDequeue(out var segment))
+                    {
+                        await asyncClientStreamingCall.RequestStream.WriteAsync(segment.Transform());
+                        _logger.Debug(
+                            $"Transform and send UpstreamSegment to collector. [TraceSegmentId] = {segment.TraceSegmentId} [GlobalTraceId] = {segment.RelatedGlobalTraces.FirstOrDefault()}");
+                    }
                     await asyncClientStreamingCall.RequestStream.CompleteAsync();
                     await asyncClientStreamingCall.ResponseAsync;
                 }
-
-                _logger.Debug(
-                    $"Transform and send UpstreamSegment to collector. [TraceSegmentId] = {traceSegment.TraceSegmentId} [GlobalTraceId] = {traceSegment.RelatedGlobalTraces.FirstOrDefault()}");
             }
             catch (Exception e)
             {
                 _logger.Warning($"Transform and send UpstreamSegment to collector fail. {e.Message}");
                 availableConnection?.Failure();
+                return;
             }
         }
     }