Merge pull request #133 from PrometheusClientNet/fix-race-condition-in-summary

phnx47 · web-flow · commit cc8572013994 · 2021-01-29T13:35:02.000+07:00
Fix race condition in summary
diff --git a/src/Prometheus.Client/SummaryImpl/QuantileStream.cs b/src/Prometheus.Client/SummaryImpl/QuantileStream.cs
@@ -47,12 +47,20 @@ public QuantileStream(int bufferSize, TimeSpan streamDuration, int ageBuckets, I
 
         public void Append(double value)
         {
-            if (ShouldFlushBuffer())
-                FlushBuffer();
-
-            lock (_bufferLock)
+            while (true)
             {
-                _buffer[_bufferPosition++] = value;
+                if(ShouldFlushBuffer())
+                    FlushBuffer();
+
+                lock (_bufferLock)
+                {
+                    // use this trick to make FlushBuffer outside of the lock
+                    if (ShouldFlushBuffer())
+                        continue;
+
+                    _buffer[_bufferPosition++] = value;
+                    return;
+                }
             }
         }
 
@@ -62,10 +70,18 @@ public void Reset()
             {
                 _bufferPosition = 0;
                 _headStreamIndex = 0;
+            }
 
+            _sampleStreamsLock.EnterWriteLock();
+            try
+            {
                 for(var i = 0; i < _sampleStreams.Length; i++)
                     _sampleStreams[i].Reset();
             }
+            finally
+            {
+                _sampleStreamsLock.ExitWriteLock();
+            }
         }
 
         public void FlushBuffer()
diff --git a/tests/Prometheus.Client.Tests/CounterInt64Tests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/CounterInt64Tests/ThreadingTests.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.CounterInt64Tests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateCounter();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    long vl = 0;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Inc(rnd.Next());
+                        if(i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private CounterInt64 CreateCounter()
+        {
+            var config = new MetricConfiguration("test", string.Empty, Array.Empty<string>(), false);
+            return new CounterInt64(config, Array.Empty<string>());
+        }
+    }
+}
diff --git a/tests/Prometheus.Client.Tests/CounterTests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/CounterTests/ThreadingTests.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.CounterTests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateCounter();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    double vl = 0;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Inc(rnd.NextDouble());
+                        if(i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private Counter CreateCounter()
+        {
+            var config = new MetricConfiguration("test", string.Empty, Array.Empty<string>(), false);
+            return new Counter(config, Array.Empty<string>());
+        }
+    }
+}
diff --git a/tests/Prometheus.Client.Tests/GaugeInt64Tests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/GaugeInt64Tests/ThreadingTests.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.GaugeInt64Tests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateGauge();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    long vl;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Inc(rnd.Next());
+                        if (i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private IGauge<long> CreateGauge()
+        {
+            var config = new MetricConfiguration("test", string.Empty, Array.Empty<string>(), false);
+            return new GaugeInt64(config, Array.Empty<string>());
+        }
+    }
+}
diff --git a/tests/Prometheus.Client.Tests/GaugeTests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/GaugeTests/ThreadingTests.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.GaugeTests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateGauge();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    double vl;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Inc(rnd.NextDouble());
+                        if (i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private IGauge CreateGauge()
+        {
+            var config = new MetricConfiguration("test", string.Empty, Array.Empty<string>(), false);
+            return new Gauge(config, Array.Empty<string>());
+        }
+    }
+}
diff --git a/tests/Prometheus.Client.Tests/HistogramTests/HistogramConfigurationTests.cs b/tests/Prometheus.Client.Tests/HistogramTests/HistogramConfigurationTests.cs
@@ -1,5 +1,4 @@
 using System;
-using System.Linq;
 using Xunit;
 
 namespace Prometheus.Client.Tests.HistogramTests
diff --git a/tests/Prometheus.Client.Tests/HistogramTests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/HistogramTests/ThreadingTests.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.HistogramTests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateHistogram();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    HistogramState vl;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Observe(rnd.NextDouble());
+                        if (i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private IHistogram CreateHistogram(double[] buckets = null)
+        {
+            var config = new HistogramConfiguration("test", string.Empty, Array.Empty<string>(), buckets, false);
+            return new Histogram(config, Array.Empty<string>());
+        }
+    }
+}
diff --git a/tests/Prometheus.Client.Tests/MetricFactoryLegacyExtensionsTests.cs b/tests/Prometheus.Client.Tests/MetricFactoryLegacyExtensionsTests.cs
@@ -1,5 +1,4 @@
 using NSubstitute;
-using Prometheus.Client;
 using Xunit;
 
 namespace Prometheus.Client.Tests
diff --git a/tests/Prometheus.Client.Tests/SummaryTests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/SummaryTests/ThreadingTests.cs
@@ -0,0 +1,44 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.SummaryTests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateSummary();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    SummaryState vl;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Observe(rnd.NextDouble());
+
+                        if (i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private Summary CreateSummary()
+        {
+            var config = new SummaryConfiguration("test", string.Empty, Array.Empty<string>(), false);
+            return new Summary(config, Array.Empty<string>());
+        }
+    }
+}
diff --git a/tests/Prometheus.Client.Tests/UntypedTests/ThreadingTests.cs b/tests/Prometheus.Client.Tests/UntypedTests/ThreadingTests.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Prometheus.Client.Tests.UntypedTests
+{
+    public class ThreadingTests
+    {
+        [Theory]
+        [InlineData(10000, 1)]
+        [InlineData(10000, 10)]
+        [InlineData(10000, 100)]
+        public async Task ObserveInParallel(int observations, int threads)
+        {
+            var metric = CreateUntyped();
+
+            var tasks = Enumerable.Range(0, threads)
+                .Select(n => Task.Run(() =>
+                {
+                    double vl;
+                    var rnd = new Random();
+                    for (var i = 0; i < observations; i++)
+                    {
+                        metric.Set(rnd.NextDouble());
+                        if (i % 100 == 0)
+                            vl = metric.Value;
+                    }
+
+                    metric.Reset();
+                }))
+                .ToArray();
+
+            await Task.WhenAll(tasks);
+        }
+
+        private IUntyped CreateUntyped()
+        {
+            var config = new MetricConfiguration("test", string.Empty, Array.Empty<string>(), false);
+            return new Untyped(config, Array.Empty<string>());
+        }
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`using System;`
`2`		`-using System.Linq;`
`3`	`2`	`using Xunit;`
`4`	`3`
`5`	`4`	`namespace Prometheus.Client.Tests.HistogramTests`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`using NSubstitute;`
`2`		`-using Prometheus.Client;`
`3`	`2`	`using Xunit;`
`4`	`3`
`5`	`4`	`namespace Prometheus.Client.Tests`