As a follow up to this first attempt to create and use my own IObservable<T> implementation, this is version 2.0. The goal has been to correct the code where there were conceptual misunderstandings and to take the issues pointed to by answers into account.
The overall concept is the same: A scheduler that can fire off jobs when they want to be in intervals as small as one second. Instead of using a timer, the Observer pattern is used by implementing an IObservable<DateTime> type, that notifies subscribers each second.
The Scheduler
public class HScheduler : IDisposable
{
public const int UnlimitedJobParallelism = -1;
private const int DefaultWaitTimeout = 60000;
HSecondCounter m_counter;
ConcurrentDictionary<IScheduleJob, int> m_parallelCounters = new ConcurrentDictionary<IScheduleJob, int>();
List<AutoResetEvent> m_waitHandles = new List<AutoResetEvent>();
public HScheduler(int waitTimeout = DefaultWaitTimeout)
{
WaitTimeout = waitTimeout;
}
public bool IsDisposed => m_counter == null;
public bool IsRunning => !IsDisposed;
/// <summary>
/// When the scheduler is stopped the amount of milliseconds to wait for all actions to finish.
/// </summary>
public int WaitTimeout { get; set; }
public void Start()
{
if (IsRunning)
return;
m_counter = HSecondCounter.CountAsync();
Log($"Scheduler started");
}
public ScheduledJobHandle Schedule(IScheduleJob job)
{
if (!RegisterJob(job)) return null;
IDisposable handle = m_counter
.Where(t => CanRun(job) && job.ShouldRun(t))
.Subscribe((time) =>
{
AutoResetEvent waitHandle = null;
try
{
waitHandle = RegisterAction(job);
job.Action(time);
}
catch (Exception ex)
{
Log($"Job Action Error (1): {job.Name}: Error Message: {ex.Message}");
}
finally
{
ReleaseAction(job, waitHandle);
}
},
(ex) =>
{
Log($"Job Action Error (2): {job.Name}: Error Message: {ex.Message}");
},
() =>
{
Log($"Job Action Completed: {job.Name}");
});
return new ScheduledJobHandle(job, Disposable.Create(() =>
{
handle.Dispose();
RemoveJob(job);
}));
}
private bool RegisterJob(IScheduleJob job)
{
if (m_parallelCounters.ContainsKey(job))
{
Log($"Job already registered and running.: {job.Name}");
return false;
}
else
{
m_parallelCounters.AddOrUpdate(job, 0, (sj, c) => c);
Log($"Job Registered: {job.Name}");
return true;
}
}
private void RemoveJob(IScheduleJob job)
{
m_parallelCounters.TryRemove(job, out _);
}
private AutoResetEvent RegisterAction(IScheduleJob job)
{
Log($"Job Action Started: {job.Name}");
m_parallelCounters.AddOrUpdate(job, 1, (scr, c) => c + 1);
lock (m_waitHandles)
{
AutoResetEvent waitHandle = new AutoResetEvent(false);
m_waitHandles.Add(waitHandle);
return waitHandle;
}
}
private void ReleaseAction(IScheduleJob job, AutoResetEvent waitHandle)
{
if (waitHandle != null)
{
lock (m_waitHandles)
{
m_waitHandles.Remove(waitHandle);
if (!waitHandle.SafeWaitHandle.IsClosed && !waitHandle.SafeWaitHandle.IsInvalid)
{
waitHandle.Set();
}
waitHandle.Dispose();
}
}
Log($"Job Action Finished: {job.Name}");
if (job.DegreeOfParallelism != UnlimitedJobParallelism)
m_parallelCounters.AddOrUpdate(job, 0, (scr, c) => c - 1);
}
private bool CanRun(IScheduleJob job)
{
if (job.DegreeOfParallelism == UnlimitedJobParallelism) return true;
int value = m_parallelCounters.GetOrAdd(job, 0);
bool result = value < job.DegreeOfParallelism;
return result;
}
internal void Stop()
{
if (IsRunning)
{
if (m_waitHandles.Count > 0)
{
AutoResetEvent[] waitHandles = m_waitHandles.ToArray(); ;
WaitHandle.WaitAll(waitHandles, WaitTimeout);
Array.ForEach(waitHandles, (wh) => wh.Dispose());
}
Log($"Scheduler STOPPED.");
m_counter.Stop();
m_counter = null;
}
}
private void Log(string format, params object[] parameters)
{
string message = string.Format(format, parameters);
Console.WriteLine($"[{DateTime.Now}]: {message}");
}
public void Dispose()
{
Stop();
}
}
The main changes here are:
1) The naming: Instead of Subscribe, the client now Schedules a job. The use of the term "Subscribe" was as confusion of the two concepts scheduling and subscription to an observable. The client of the Scheduler, should be agnostic about how the Scheduler works internally - and a scheduler schedules jobs.
2) Waithandles: When the Scheduler stops, it will now wait on running jobs to finish before it finally terminates.
3) Loggin: Introduced some basic logging of events.
The "Counter"
This is essentially the same as the previous. It notifies observers for every second and act fundamentally seen as a timer.
public class HSecondCounter : IObservable<DateTime>
{
static public HSecondCounter Count()
{
HSecondCounter counter = new HSecondCounter();
Task.Factory.StartNew(() =>
{
counter.Run((now) =>
{
lock (counter.m_observers)
{
foreach (var observer in counter.m_observers.ToArray())
{
observer.OnNext(now);
}
Console.WriteLine("HSecondCounter: {0}", now);
}
});
}, TaskCreationOptions.LongRunning);
return counter;
}
static public HSecondCounter CountAsync()
{
HSecondCounter counter = new HSecondCounter();
Task.Factory.StartNew(() =>
{
counter.Run((now) =>
{
lock (counter.m_observers)
{
foreach (var observer in counter.m_observers.ToArray())
{
Task.Factory.StartNew(() =>
{
observer.OnNext(now);
});
}
Console.WriteLine("HSecondCounter: {0}", now);
}
});
}, TaskCreationOptions.LongRunning);
return counter;
}
HashSet<IObserver<DateTime>> m_observers = new HashSet<IObserver<DateTime>>();
volatile bool m_doContinue = true;
private HSecondCounter()
{
}
public void Stop()
{
m_doContinue = false;
}
public IDisposable Subscribe(IObserver<DateTime> observer)
{
lock (m_observers)
{
m_observers.Add(observer);
}
return Disposable.Create(() =>
{
lock (m_observers)
{
m_observers.Remove(observer);
}
observer.OnCompleted();
});
}
private void Run(Action<DateTime> notifier)
{
try
{
int lastSecond = 0;
while (m_doContinue)
{
DateTime now = DateTime.Now;
if (now.Second != lastSecond)
{
notifier(now);
}
lastSecond = now.Second;
Thread.Sleep(500);
}
}
catch (Exception ex)
{
lock (m_observers)
{
foreach (var observer in m_observers.ToArray())
{
observer.OnError(ex);
}
}
}
finally
{
lock (m_observers)
{
foreach (var observer in m_observers.ToArray())
{
observer.OnCompleted();
}
}
Console.WriteLine($"HSceondCounter ended at: {DateTime.Now}");
}
}
}
The original Run(...) method had this definition:
async private void Run(Action<DateTime> notifier)
which is a misunderstanding of the async concept in that an async method (unless it is a UI-event handler) should return Task. In the new implementation async is removed. This has the positive effect that when Run(..) is called by Count() and CountAsync() the parameter TaskCreationOptions.LongRunning to Task.Factory.StartNew() has more meaning. In the first version that call returned immediately after starting Run(...) which in turn in fact was the long running thread.
It is in fact possible to await a call to HSceondCounter.CountAsync() like this:
async Task Start()
{
DateTime time = await HSecondCounter.CountAsync();
}
But this will halt the calling thread while HSecondCounter ticking every second infinitely with no way to stop it. So this is not useful.
Other Objects
public interface IScheduleJob
{
string Name { get; }
Action<DateTime> Action { get; }
bool ShouldRun(DateTime time);
int DegreeOfParallelism { get; }
}
public sealed class ScheduledJobHandle : IDisposable
{
IDisposable m_terminateHandle;
public ScheduledJobHandle(IScheduleJob job, IDisposable terminateHandle)
{
Job = job;
m_terminateHandle = terminateHandle;
}
public IScheduleJob Job { get; }
public bool IsDisposed => m_terminateHandle == null;
public void Dispose()
{
if (m_terminateHandle != null)
{
m_terminateHandle.Dispose();
m_terminateHandle = null;
}
}
}
Test Case
void TestSchedule()
{
HScheduler scheduler = new HScheduler();
scheduler.Start();
ScheduleJob job1 = new ScheduleJob
{
Name = "AAAA",
DegreeOfParallelism = HScheduler.UnlimitedJobParallelism,
Action = (value) =>
{
Console.WriteLine($" AAAA Running:{value} - {value.Millisecond} - Thread Id: {Thread.CurrentThread.ManagedThreadId}");
Thread.Sleep(5000);
},
ShouldRunPredicate = (time) =>
{
return true; // time.Second % 2 == 0;
}
};
ScheduleJob job2 = new ScheduleJob
{
Name = "BBBB",
DegreeOfParallelism = 1,
Action = (value) =>
{
Console.WriteLine($" BBBB Running:{value} - {value.Millisecond} - Thread Id: {Thread.CurrentThread.ManagedThreadId}");
Thread.Sleep(4000);
},
ShouldRunPredicate = (time) =>
{
return time.Second % 5 == 0;
}
};
ScheduledJobHandle scheduledJob1 = scheduler.Schedule(job1);
ScheduledJobHandle scheduledJob2 = scheduler.Schedule(job2);
ScheduledJobHandle scheduledJob11 = scheduler.Schedule(job1);
Console.ReadLine();
if (scheduledJob1 != null)
scheduledJob1.Dispose();
if (scheduledJob2 != null)
scheduledJob2.Dispose();
if (scheduledJob11 != null)
scheduledJob11.Dispose();
scheduler.Stop();
Console.WriteLine("After Scheduler Stop");
Console.ReadLine();
}
The previous version worked in general as expected. This one IMO a little more. I'm sure you can find a lot of possible drawbacks, so you're very welcome to do so.
The origin to this work is still t3chb0t's inspiring posts.