Port RateLimiter fixes from Lucene 4.8.1, #1381

src/Lucene.Net.Tests/Store/TestRateLimiter.cs

@@ -1,5 +1,11 @@
+using J2N;
+using J2N.Threading;
+using J2N.Threading.Atomic;
+using Lucene.Net.Util;
 using NUnit.Framework;
+using System;
 using Assert = Lucene.Net.TestFramework.Assert;
+using ThreadInterruptedException = Lucene.Net.Util.ThreadInterruptedException;
 
 namespace Lucene.Net.Store
 {
@@ -46,5 +52,71 @@ public void TestPause()
             Assert.IsTrue(convert < 2000L, "we should sleep less than 2 seconds but did: " + convert + " millis");
             Assert.IsTrue(convert > 1000L, "we should sleep at least 1 second but did only: " + convert + " millis");
         }
+
+        [Test]
+        public void TestThreads()
+        {
+            double targetMBPerSec = 10.0 + 20 * Random.NextDouble();
+            SimpleRateLimiter limiter = new SimpleRateLimiter(targetMBPerSec);
+
+            CountdownLatch startingGun = new CountdownLatch(1);
+
+            ThreadJob[] threads = new ThreadJob[TestUtil.NextInt32(Random, 3, 6)];
+            AtomicInt64 totBytes = new AtomicInt64();
+            for (int i = 0; i < threads.Length; i++)
+            {
+                threads[i] = new TestThreadsThreadJobAnonymousClass(startingGun, totBytes, limiter);
+                threads[i].Start();
+            }
+
+            long startNS = Time.NanoTime();
+            startingGun.Signal();
+            foreach (ThreadJob thread in threads)
+            {
+                thread.Join();
+            }
+            long endNS = Time.NanoTime();
+            double actualMBPerSec = (totBytes.Value/1024.0/1024.0)/((endNS-startNS)/1000000000.0);
+
+            // TODO: this may false trip .... could be we can only assert that it never exceeds the max, so slow jenkins doesn't trip:
+            double ratio = actualMBPerSec/targetMBPerSec;
+
+            // LUCENENET: backport commits 090b804 (Lucene 6.0.0) and a893aaa (Lucene 7.0.0) with assertion fixes for test reliability
+            // Only enforce that it wasn't too fast; if machine is bogged down (can't schedule threads / sleep properly) then it may falsely be too slow:
+            AssumeTrue("actualMBPerSec=" + actualMBPerSec + " targetMBPerSec=" + targetMBPerSec, 0.9 <= ratio);
+            Assert.IsTrue(ratio <= 1.1, "targetMBPerSec=" + targetMBPerSec + " actualMBPerSec=" + actualMBPerSec);
+        }
+
+        private class TestThreadsThreadJobAnonymousClass(
+            CountdownLatch startingGun,
+            AtomicInt64 totBytes,
+            SimpleRateLimiter limiter)
+            : ThreadJob
+        {
+            public override void Run()
+            {
+                try
+                {
+                    startingGun.Wait();
+                }
+                catch (Exception ie) when (ie.IsInterruptedException())
+                {
+                    throw new ThreadInterruptedException(ie);
+                }
+
+                long bytesSinceLastPause = 0;
+                for (int i = 0; i < 500; i++)
+                {
+                    long numBytes = TestUtil.NextInt32(Random, 1000, 10000);
+                    totBytes.AddAndGet(numBytes);
+                    bytesSinceLastPause += numBytes;
+                    if (bytesSinceLastPause > limiter.MinPauseCheckBytes)
+                    {
+                        limiter.Pause(bytesSinceLastPause);
+                        bytesSinceLastPause = 0;
+                    }
+                }
+            }
+        }
     }
 }

src/Lucene.Net/Store/RateLimitedIndexOutput.cs

@@ -32,6 +32,11 @@ internal sealed class RateLimitedIndexOutput : BufferedIndexOutput
         private readonly RateLimiter rateLimiter;
         private int disposed = 0; // LUCENENET specific - allow double-dispose
 
+        /// <summary>
+        /// How many bytes we've written since we last called <see cref="RateLimiter.Pause(long)"/>.
+        /// </summary>
+        private long bytesSinceLastPause;
+
         internal RateLimitedIndexOutput(RateLimiter rateLimiter, IndexOutput @delegate)
         {
             // TODO should we make buffer size configurable
@@ -53,7 +58,14 @@ internal RateLimitedIndexOutput(RateLimiter rateLimiter, IndexOutput @delegate)
         protected internal override void FlushBuffer(ReadOnlySpan<byte> source)
         {
             int len = source.Length;
-            rateLimiter.Pause(len);
+            bytesSinceLastPause += len;
+
+            if (bytesSinceLastPause > rateLimiter.MinPauseCheckBytes)
+            {
+                rateLimiter.Pause(bytesSinceLastPause);
+                bytesSinceLastPause = 0;
+            }
+
             if (bufferedDelegate != null)
             {
                 bufferedDelegate.FlushBuffer(source);

src/Lucene.Net/Store/RateLimiter.cs

@@ -27,8 +27,8 @@ namespace Lucene.Net.Store
     /// Abstract base class to rate limit IO.  Typically implementations are
     /// shared across multiple <see cref="IndexInput"/>s or <see cref="IndexOutput"/>s (for example
     /// those involved all merging).  Those <see cref="IndexInput"/>s and
-    /// <see cref="IndexOutput"/>s would call <see cref="Pause"/> whenever they
-    /// want to read bytes or write bytes.
+    /// <see cref="IndexOutput"/>s would call <see cref="Pause"/> whenever they have read
+    /// or written more than <see cref="MinPauseCheckBytes"/> bytes.
     /// </summary>
     public abstract class RateLimiter
     {
@@ -52,16 +52,24 @@ public abstract class RateLimiter
         /// <returns> the pause time in nano seconds </returns>
         public abstract long Pause(long bytes);
 
+        /// <summary>
+        /// How many bytes the caller should add up itself before invoking <see cref="Pause"/>.
+        /// </summary>
+        public abstract long MinPauseCheckBytes { get; }
+
         /// <summary>
         /// Simple class to rate limit IO.
         /// </summary>
         public class SimpleRateLimiter : RateLimiter
         {
-            // LUCENENET: these fields are volatile in Lucene, but that is not
-            // valid in .NET. Instead, we use AtomicInt64/AtomicDouble to ensure atomicity.
+            private const int MIN_PAUSE_CHECK_MSEC = 5;
+
+            // LUCENENET: mbPerSec/minPauseCheckBytes are volatile in Lucene; we use
+            // AtomicDouble/AtomicInt64 for atomicity. lastNS is a plain long guarded by
+            // UninterruptableMonitor in Pause() (matches upstream, which is non-volatile).
             private readonly AtomicDouble mbPerSec = new AtomicDouble();
-            private readonly AtomicDouble nsPerByte = new AtomicDouble();
-            private readonly AtomicInt64 lastNS = new AtomicInt64();
+            private readonly AtomicInt64 minPauseCheckBytes = new AtomicInt64();
+            private long lastNS;
 
             // TODO: we could also allow eg a sub class to dynamically
             // determine the allowed rate, eg if an app wants to
@@ -80,43 +88,60 @@ public SimpleRateLimiter(double mbPerSec)
             public override void SetMbPerSec(double mbPerSec)
             {
                 this.mbPerSec.Value = mbPerSec;
-                if (mbPerSec == 0)
-                    nsPerByte.Value = 0;
-                else
-                    nsPerByte.Value = 1000000000.0 / (1024 * 1024 * mbPerSec);
+                minPauseCheckBytes.Value = (long) ((MIN_PAUSE_CHECK_MSEC / 1000.0) * mbPerSec * 1024 * 1024);
             }
 
+            public override long MinPauseCheckBytes => minPauseCheckBytes;
+
             /// <summary>
             /// The current mb per second rate limit.
             /// </summary>
             public override double MbPerSec => this.mbPerSec;
 
             /// <summary>
             /// Pauses, if necessary, to keep the instantaneous IO
-            /// rate at or below the target. NOTE: multiple threads
-            /// may safely use this, however the implementation is
-            /// not perfectly thread safe but likely in practice this
-            /// is harmless (just means in some rare cases the rate
-            /// might exceed the target).  It's best to call this
-            /// with a biggish count, not one byte at a time. </summary>
-            /// <returns> the pause time in nano seconds </returns>
+            /// rate at or below the target. Be sure to only call
+            /// this method when <paramref name="bytes"/> &gt; <see cref="MinPauseCheckBytes"/>,
+            /// otherwise it will pause way too long!
+            /// </summary>
+            /// <returns> the pause time in nanoseconds </returns>
             public override long Pause(long bytes)
             {
-                if (bytes == 1)
+                long startNS = Time.NanoTime();
+
+                double secondsToPause = (bytes/1024.0/1024.0) / mbPerSec;
+
+                long targetNS;
+
+                // Sync'd to read + write lastNS:
+                UninterruptableMonitor.Enter(this);
+                try
                 {
-                    return 0;
-                }
+                    // Time we should sleep until; this is purely instantaneous
+                    // rate (just adds seconds onto the last time we had paused to);
+                    // maybe we should also offer decayed recent history one?
+                    targetNS = lastNS + (long) (1000000000 * secondsToPause);
+
+                    if (startNS >= targetNS)
+                    {
+                        // OK, current time is already beyond the target sleep time,
+                        // no pausing to do.
+
+                        // Set to startNS, not targetNS, to enforce the instant rate, not
+                        // the "averaged over all history" rate:
+                        lastNS = startNS;
+                        return 0;
+                    }
 
-                // TODO: this is purely instantaneous rate; maybe we
-                // should also offer decayed recent history one?
-                var targetNS = /*lastNS =*/ lastNS.AddAndGet((long)(bytes * nsPerByte));
-                long startNS;
-                var curNS = startNS = Time.NanoTime() /* ns */;
-                if (lastNS < curNS)
+                    lastNS = targetNS;
+                }
+                finally
                 {
-                    lastNS.Value = curNS;
+                    UninterruptableMonitor.Exit(this);
                 }
 
+                long curNS = startNS;
+
                 // While loop because Thread.sleep doesn't always sleep
                 // enough:
                 while (true)
@@ -126,6 +151,10 @@ public override long Pause(long bytes)
                     {
                         try
                         {
+                            // LUCENENET NOTE: retaining original comment re: JVMs below
+                            // NOTE: except maybe on real-time JVMs, minimum realistic sleep time
+                            // is 1 msec; if you pass just 1 nsec the default impl rounds
+                            // this up to 1 msec:
                             Thread.Sleep(TimeSpan.FromMilliseconds(pauseNS / 1000000));
                         }
                         catch (Exception ie) when (ie.IsInterruptedException())
@@ -138,6 +167,7 @@ public override long Pause(long bytes)
                     }
                     break;
                 }
+
                 return curNS - startNS;
             }
         }