[REEF-1554] Fix .NET Core compatibility issues in REEF.Wake
This change addressed the issue by
* Replacing System.Timers.Timer in TimerStage class with System.Threading.Timer
* Adding timer period and initialDelay parameter verification since interface
is 64 bit and .NET Core class only supports 32 bit.
* Adding unit tests to verify TimerStage behavior.
JIRA:
[REEF-1554](https://issues.apache.org/jira/browse/REEF-1554)
Pull request:
This closes #1134
diff --git a/lang/cs/Org.Apache.REEF.Wake.Tests/Org.Apache.REEF.Wake.Tests.csproj b/lang/cs/Org.Apache.REEF.Wake.Tests/Org.Apache.REEF.Wake.Tests.csproj
index 283536f..91ed56f 100644
--- a/lang/cs/Org.Apache.REEF.Wake.Tests/Org.Apache.REEF.Wake.Tests.csproj
+++ b/lang/cs/Org.Apache.REEF.Wake.Tests/Org.Apache.REEF.Wake.Tests.csproj
@@ -51,6 +51,7 @@
<Compile Include="RemoteManagerTest.cs" />
<Compile Include="StreamingRemoteManagerTest.cs" />
<Compile Include="StreamingTransportTest.cs" />
+ <Compile Include="TimerStageTest.cs" />
<Compile Include="TimeTest.cs" />
<Compile Include="TcpConnectionRetryTest.cs" />
<Compile Include="TransportTest.cs" />
diff --git a/lang/cs/Org.Apache.REEF.Wake.Tests/TimerStageTest.cs b/lang/cs/Org.Apache.REEF.Wake.Tests/TimerStageTest.cs
new file mode 100644
index 0000000..75e5b60
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Wake.Tests/TimerStageTest.cs
@@ -0,0 +1,121 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License
+
+using System;
+using System.Threading;
+using Xunit;
+using Org.Apache.REEF.Wake.Impl;
+
+namespace Org.Apache.REEF.Wake.Tests
+{
+ // Timer stage tests.
+ public class TimerStageTest
+ {
+ private const long _delay = 100;
+ private const long _period = 1000;
+ private const long _bigValue = long.MaxValue;
+
+ [Fact]
+ public void TestValidTimerPeriod()
+ {
+ RunTest(_delay, _period);
+ }
+
+ [Fact]
+ public void TestInvalidTimerPeriod()
+ {
+ Assert.Throws<ArgumentException>(() => RunTest(_delay, _bigValue));
+ }
+
+ [Fact]
+ public void TestInvalidTimerDelay()
+ {
+ Assert.Throws<ArgumentException>(() => RunTest(_bigValue, _period));
+ }
+
+ void RunTest(long delay, long period)
+ {
+ TimerMonitor monitor = new TimerMonitor();
+ int expected = 10;
+
+ TestEventHandler handler = new TestEventHandler(monitor, expected);
+ IStage stage = new TimerStage(handler, delay, period);
+
+ monitor.Mwait();
+ Assert.Equal(expected, handler.GetCount());
+ }
+
+ private class TestEventHandler : IEventHandler<PeriodicEvent>
+ {
+ private TimerMonitor _monitor;
+ private long _expected;
+ private long _count;
+
+ public TestEventHandler(TimerMonitor monitor, long expected)
+ {
+ _count = 0;
+ _monitor = monitor;
+ _expected = expected;
+ }
+
+ public void OnNext(PeriodicEvent e)
+ {
+ long count = Interlocked.Increment(ref _count);
+ if (Interlocked.Read(ref _count) == _expected)
+ {
+ _monitor.Mnotify();
+ }
+ }
+
+ public long GetCount()
+ {
+ return Interlocked.Read(ref _count);
+ }
+ }
+
+ private class TimerMonitor
+ {
+ private long finished;
+
+ public TimerMonitor()
+ {
+ finished = 0;
+ }
+
+ public void Mwait()
+ {
+ lock (this)
+ {
+ while (Interlocked.Read(ref this.finished) < 1)
+ {
+ Monitor.Wait(this);
+ }
+ Interlocked.CompareExchange(ref finished, 0, 1);
+ }
+ }
+
+ public void Mnotify()
+ {
+ lock (this)
+ {
+ Interlocked.CompareExchange(ref finished, 1, 0);
+ Monitor.Pulse(this);
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.Wake/Impl/TimerStage.cs b/lang/cs/Org.Apache.REEF.Wake/Impl/TimerStage.cs
index ef21ebf..2e93c10 100644
--- a/lang/cs/Org.Apache.REEF.Wake/Impl/TimerStage.cs
+++ b/lang/cs/Org.Apache.REEF.Wake/Impl/TimerStage.cs
@@ -15,14 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-using System.Timers;
+using System;
+using System.Threading;
namespace Org.Apache.REEF.Wake.Impl
{
/// <summary>Stage that triggers an event handler periodically</summary>
public sealed class TimerStage : IStage
{
- // private readonly ScheduledExecutorService executor;
+ // Maximum time period or initial delay supported by the timer.
+ public const long MaxTimeValue = int.MaxValue;
+
private readonly Timer _timer;
private readonly PeriodicEvent _value = new PeriodicEvent();
private readonly IEventHandler<PeriodicEvent> _handler;
@@ -36,14 +39,17 @@
/// <summary>Constructs a timer stage</summary>
/// <param name="handler">an event handler</param>
- /// <param name="initialDelay">an initial delay</param>
- /// <param name="period">a period in milli-seconds</param>
+ /// <param name="initialDelay">an initial delay in the interval [0,MaxTimeValue]</param>
+ /// <param name="period">a period in milli-seconds in the interval [0,MaxTimeValue]</param>
public TimerStage(IEventHandler<PeriodicEvent> handler, long initialDelay, long period)
{
+ // Core .NET only supports 32 bit timers.
+ Validate("initialDelay", initialDelay);
+ Validate("period", period);
+
_handler = handler;
- _timer = new Timer(period);
- _timer.Elapsed += (sender, e) => OnTimedEvent(sender, e, _handler, _value);
- _timer.Enabled = true;
+ _timer = new Timer(
+ (object state) => { OnTimedEvent(_handler, _value); }, this, (int)initialDelay, (int)period);
}
/// <summary>
@@ -51,12 +57,27 @@
/// </summary>
public void Dispose()
{
- _timer.Stop();
+ _timer.Dispose();
}
- private static void OnTimedEvent(object source, ElapsedEventArgs e, IEventHandler<PeriodicEvent> handler, PeriodicEvent value)
+ private static void OnTimedEvent(IEventHandler<PeriodicEvent> handler, PeriodicEvent value)
{
handler.OnNext(value);
}
+
+ /// <summary>
+ /// Validates the input is less than TimerStage.MaxTimeValue.
+ /// </summary>
+ /// <param name="name">Parameter name</param>
+ /// <param name="value">Parameter value</param>
+ /// <exception cref="ArgumentException">Input value exceeds TimerStage.MaxTimeValue</exception>
+ private static void Validate(string name, long value)
+ {
+ if (value > MaxTimeValue)
+ {
+ throw new ArgumentException(string.Format(
+ "Parameter: " + name + " {0} is larger than supported value {1}", value, MaxTimeValue));
+ }
+ }
}
-}
+}
\ No newline at end of file
