Merge pull request #9 from nblumhardt/f-captiveworker

Captive thread and BlockingCollection (WIP)

Author: jezzsantos

src/Serilog.Sinks.Async.PerformanceTests/Benchmarks.cs

@@ -10,6 +10,7 @@ public class Benchmarks
         public void Benchmark()
         {
             BenchmarkRunner.Run<ThroughputBenchmark>();
+            BenchmarkRunner.Run<LatencyBenchmark>();
         }
     }
 }

src/Serilog.Sinks.Async.PerformanceTests/LatencyBenchmark.cs

@@ -0,0 +1,65 @@
+using System;
+using BenchmarkDotNet.Attributes;
+using Serilog.Core;
+using Serilog.Events;
+using Serilog.Parsing;
+
+namespace Serilog.Sinks.Async.PerformanceTests
+{
+    public class LatencyBenchmark
+    {
+        private const int Count = 10000;
+
+        private readonly LogEvent _evt = new LogEvent(DateTimeOffset.Now, LogEventLevel.Information, null,
+            new MessageTemplate(new[] {new TextToken("Hello")}), new LogEventProperty[0]);
+
+        private Logger _syncLogger, _asyncLogger, _async2Logger;
+
+        [Setup]
+        public void Reset()
+        {
+            _syncLogger?.Dispose();
+            _asyncLogger?.Dispose();
+            _async2Logger?.Dispose();
+
+            _syncLogger = new LoggerConfiguration()
+                .WriteTo.Sink(new SilentSink())
+                .CreateLogger();
+
+            _asyncLogger = new LoggerConfiguration()
+                .WriteTo.Async(a => a.Sink(new SilentSink()))
+                .CreateLogger();
+
+            _async2Logger = new LoggerConfiguration()
+                .WriteTo.Async2(a => a.Sink(new SilentSink()))
+                .CreateLogger();
+        }
+
+        [Benchmark(Baseline = true)]
+        public void Sync()
+        {
+            for (var i = 0; i < Count; ++i)
+            {
+                _syncLogger.Write(_evt);
+            }
+        }
+
+        [Benchmark]
+        public void Async()
+        {
+            for (var i = 0; i < Count; ++i)
+            {
+                _asyncLogger.Write(_evt);
+            }
+        }
+
+        [Benchmark]
+        public void Async2()
+        {
+            for (var i = 0; i < Count; ++i)
+            {
+                _async2Logger.Write(_evt);
+            }
+        }
+    }
+}

src/Serilog.Sinks.Async.PerformanceTests/Serilog.Sinks.Async.PerformanceTests.csproj

@@ -85,6 +85,8 @@
     <Compile Include="Benchmarks.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="SignallingSink.cs" />
+    <Compile Include="LatencyBenchmark.cs" />
+    <Compile Include="SilentSink.cs" />
     <Compile Include="ThroughputBenchmark.cs" />
   </ItemGroup>
   <ItemGroup>

src/Serilog.Sinks.Async.PerformanceTests/SilentSink.cs

@@ -0,0 +1,12 @@
+using Serilog.Core;
+using Serilog.Events;
+
+namespace Serilog.Sinks.Async.PerformanceTests
+{
+    public class SilentSink : ILogEventSink
+    {
+        public void Emit(LogEvent logEvent)
+        {
+        }
+    }
+}

src/Serilog.Sinks.Async.PerformanceTests/ThroughputBenchmark.cs

@@ -1,5 +1,6 @@
 using System;
 using BenchmarkDotNet.Attributes;
+using Serilog.Core;
 using Serilog.Events;
 using Serilog.Parsing;
 
@@ -13,11 +14,21 @@ public class ThroughputBenchmark
             new MessageTemplate(new[] {new TextToken("Hello")}), new LogEventProperty[0]);
 
         private readonly SignallingSink _signal;
-        private readonly ILogger _syncLogger, _asyncLogger;
+        private Logger _syncLogger, _asyncLogger, _async2Logger;
 
         public ThroughputBenchmark()
         {
             _signal = new SignallingSink(Count);
+        }
+
+        [Setup]
+        public void Reset()
+        {
+            _syncLogger?.Dispose();
+            _asyncLogger?.Dispose();
+            _async2Logger?.Dispose();
+
+            _signal.Reset();
 
             _syncLogger = new LoggerConfiguration()
                 .WriteTo.Sink(_signal)
@@ -26,12 +37,10 @@ public ThroughputBenchmark()
             _asyncLogger = new LoggerConfiguration()
                 .WriteTo.Async(a => a.Sink(_signal))
                 .CreateLogger();
-        }
 
-        [Setup]
-        public void Reset()
-        {
-            _signal.Reset();
+            _async2Logger = new LoggerConfiguration()
+                .WriteTo.Async2(a => a.Sink(_signal))
+                .CreateLogger();
         }
 
         [Benchmark(Baseline = true)]
@@ -56,5 +65,16 @@ public void Async()
 
             _signal.Wait();
         }
+
+        [Benchmark]
+        public void Async2()
+        {
+            for (var i = 0; i < Count; ++i)
+            {
+                _async2Logger.Write(_evt);
+            }
+
+            _signal.Wait();
+        }
     }
 }

src/Serilog.Sinks.Async.UnitTests/AsyncWorkerSinkTests.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Threading;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Serilog.Core;
+using Serilog.Events;
+using Serilog.Parsing;
+
+namespace Serilog.Sinks.Async.UnitTests
+{
+    [TestClass]
+    public class AsyncWorkerSinkTests
+    {
+        [TestMethod]
+        public void EventsArePassedToInnerSink()
+        {
+            var collector = new CollectingSink();
+
+            using (var log = new LoggerConfiguration()
+                .WriteTo.Async2(w => w.Sink(collector))
+                .CreateLogger())
+            {
+                log.Information("Hello, async world!");
+                log.Information("Hello again!");
+            }
+
+            Assert.AreEqual(2, collector.Events.Count);
+        }
+
+        [TestMethod]
+        public void DisposeCompletesWithoutWorkPerformed()
+        {
+            var collector = new CollectingSink();
+
+            using (new LoggerConfiguration()
+                .WriteTo.Async2(w => w.Sink(collector))
+                .CreateLogger())
+            {
+            }
+
+            Assert.AreEqual(0, collector.Events.Count);
+        }
+    }
+}

src/Serilog.Sinks.Async.UnitTests/CollectingSink.cs

@@ -0,0 +1,16 @@
+using System.Collections.Concurrent;
+using Serilog.Core;
+using Serilog.Events;
+
+namespace Serilog.Sinks.Async.UnitTests
+{
+    public class CollectingSink : ILogEventSink
+    {
+        public ConcurrentBag<LogEvent> Events { get; } = new ConcurrentBag<LogEvent>();
+
+        public void Emit(LogEvent logEvent)
+        {
+            Events.Add(logEvent);
+        }
+    }
+}

src/Serilog.Sinks.Async.UnitTests/Serilog.Sinks.Async.UnitTests.csproj

@@ -72,8 +72,10 @@
       <Link>Properties\GlobalAssemblyInfo.cs</Link>
     </Compile>
     <Compile Include="AssertExtensions.cs" />
+    <Compile Include="AsyncWorkerSinkTests.cs" />
     <Compile Include="BufferedQueueSinkSpec.cs" />
     <Compile Include="BufferedQueueSpec.cs" />
+    <Compile Include="CollectingSink.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
   </ItemGroup>
   <ItemGroup>

src/Serilog.Sinks.Async/AsyncWorkerSink.cs

@@ -0,0 +1,72 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using Serilog.Core;
+using Serilog.Debugging;
+using Serilog.Events;
+
+namespace Serilog.Sinks.Async
+{
+    sealed class AsyncWorkerSink : ILogEventSink, IDisposable
+    {
+        readonly Logger _pipeline;
+        readonly int _bufferCapacity;
+        volatile bool _disposed;
+        readonly CancellationTokenSource _cancel = new CancellationTokenSource();
+        readonly BlockingCollection<LogEvent> _queue;
+        readonly Thread _worker;
+
+        public AsyncWorkerSink(Logger pipeline, int bufferCapacity)
+        {
+            if (pipeline == null) throw new ArgumentNullException(nameof(pipeline));
+            if (bufferCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(bufferCapacity));
+            _pipeline = pipeline;
+            _bufferCapacity = bufferCapacity;
+            _queue = new BlockingCollection<LogEvent>(_bufferCapacity);
+            _worker = new Thread(Pump) { IsBackground = true };
+            _worker.Start();
+        }
+
+        public void Emit(LogEvent logEvent)
+        {
+            // The disposed check is racy, but only ensures we don't prevent flush from
+            // completing by pushing more events.
+            if (!_disposed && !_queue.TryAdd(logEvent))
+                SelfLog.WriteLine("{0} unable to enqueue, capacity {1}", typeof(AsyncWorkerSink), _bufferCapacity);
+        }
+
+        public void Dispose()
+        {
+            _disposed = true;
+            _cancel.Cancel();
+            _worker.Join();            
+            _pipeline.Dispose();
+            // _cancel not disposed, because it will make _cancel.Cancel() non-idempotent
+        }
+
+        void Pump()
+        {
+            try
+            {
+                try
+                {
+                    while (true)
+                    {
+                        var next = _queue.Take(_cancel.Token);
+                        _pipeline.Write(next);
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    LogEvent next;
+                    while (_queue.TryTake(out next))
+                        _pipeline.Write(next);
+                }
+            }
+            catch (Exception ex)
+            {
+                SelfLog.WriteLine("{0} fatal error in worker thread: {1}", typeof(AsyncWorkerSink), ex);
+            }
+        }
+    }
+}

src/Serilog.Sinks.Async/LoggerConfigurationExtensions.cs

@@ -1,5 +1,6 @@
 using System;
 using Serilog.Configuration;
+using Serilog.Events;
 
 namespace Serilog.Sinks.Async
 {
@@ -16,5 +17,18 @@ public static LoggerConfiguration Async(this LoggerSinkConfiguration configurati
 
             return configuration.Sink(wrapper);
         }
+
+        public static LoggerConfiguration Async2(this LoggerSinkConfiguration configuration,
+            Action<LoggerSinkConfiguration> sinkConfiguration, int bufferSize = 10000)
+        {
+            var sublogger = new LoggerConfiguration();
+            sublogger.MinimumLevel.Is(LevelAlias.Minimum);
+
+            sinkConfiguration(sublogger.WriteTo);
+
+            var wrapper = new AsyncWorkerSink(sublogger.CreateLogger(), bufferSize);
+
+            return configuration.Sink(wrapper);
+        }
     }
 }