2
\$\begingroup\$

This is a 2nd follow-up to my previous one about a Scheduler built with observables.

Although the last one was working correctly, it was only possible to see this in LINQPad which I didn't like very much. I prefer to have proper tests so I've redesigned it a little bit to make testing possible.


Core

The Scheduler itself is now pretty simple. It requires an observable that produces ticks which it turns into a hot-observable. On each tick a each job's triggers are evaluated and jobs which match the trigger are executed.

Unscheduling jobs is a blocking operation when a timeout is pecified.

public class Scheduler : IDisposable
{
    private readonly IConnectableObservable<DateTime> _scheduler;

    private readonly IDisposable _disconnect;

    public Scheduler(IObservable<DateTime> ticks)
    {
        // Not using .RefCount here because it should be ticking regardless of subscriptions.
        _scheduler = ticks.Publish();
        _disconnect = _scheduler.Connect();
    }

    public IDisposable Schedule(Job job, CancellationToken cancellationToken = default)
    {
        var unschedule =
            _scheduler
                // .ToList the results so that all triggers have the chance to evaluate the tick.
                .Where(tick => job.Triggers.Select(t => t.Matches(tick)).ToList().Any(x => x))
                .Subscribe(timestamp => job.Execute(cancellationToken));

        return Disposable.Create(() =>
        {
            job.Continuation.Wait(job.UnscheduleTimeout);
            unschedule.Dispose();
        });
    }

    public void Dispose()
    {
        // Stop ticking.
        _disconnect.Dispose();
    }
}

The Job class is handling job's triggers and tasks and keeping them within the specified max-degree-of-parallelism.

public class Job
{
    private readonly List<Task> _tasks = new List<Task>();

    public Job(string name, IEnumerable<Trigger> trigger, Func<CancellationToken, Task> action)
    {
        Name = name;
        Triggers = trigger.ToList();
        Action = action;
    }

    public string Name { get; }

    public IEnumerable<Trigger> Triggers { get; }

    public Func<CancellationToken, Task> Action { get; }

    public Action<Job> OnMisfire { get; set; }

    public DegreeOfParallelism MaxDegreeOfParallelism { get; set; } = 1;

    public TimeSpan UnscheduleTimeout { get; set; }

    public Task Continuation => Task.WhenAll(_tasks).ContinueWith(_ => _tasks.Clear());

    public int Count => _tasks.Count;

    public void Execute(CancellationToken cancellationToken)
    {
        if (CanExecute())
        {
            var jobTask = Action(cancellationToken);
            _tasks.Add(jobTask);
            jobTask.ContinueWith(_ => _tasks.Remove(jobTask), cancellationToken);
        }
        else
        {
            OnMisfire?.Invoke(this);
        }
    }

    private bool CanExecute()
    {
        return
            MaxDegreeOfParallelism.Equals(DegreeOfParallelism.Unlimited) ||
            Count < MaxDegreeOfParallelism.Value;
    }
}

The DegreeOfParallelism is an anti-primitive-obsession wrapper for int.

public class DegreeOfParallelism : Primitive<int>
{
    private const int UnlimitedValue = -1;

    public DegreeOfParallelism(int value) : base(value) { }

    public static readonly DegreeOfParallelism Unlimited = new DegreeOfParallelism(UnlimitedValue);

    protected override void Validate(int value)
    {
        if (value == UnlimitedValue)
        {
            return;
        }

        if (value < 1)
        {
            throw new ArgumentException("Value must be positive.");
        }
    }

    public static implicit operator DegreeOfParallelism(int value) => new DegreeOfParallelism(value);
}

It is supported by the base class Primitive<T> which implements basic operators, equality and comparer.

[PublicAPI]
[CannotApplyEqualityOperator]
public abstract class Primitive<T> : IEquatable<Primitive<T>>, IComparable<Primitive<T>>
{
    private static readonly IComparer<Primitive<T>> Comparable = ComparerFactory<Primitive<T>>.Create(p => p.Value);

    protected Primitive(T value)
    {
        // ReSharper disable once VirtualMemberCallInConstructor - it's ok to do this here because Validate is stateless.
        Validate(Value = value);
    }

    protected abstract void Validate(T value);

    [AutoEqualityProperty]
    public T Value { get; }

    #region IEquatable

    public bool Equals(Primitive<T> other) => AutoEquality<Primitive<T>>.Comparer.Equals(this, other);

    public override bool Equals(object obj) => obj is T other && Equals(other);

    public override int GetHashCode() => AutoEquality<Primitive<T>>.Comparer.GetHashCode(this);

    #endregion

    #region IComparable

    public int CompareTo(Primitive<T> other) => Comparable.Compare(this, other);

    #endregion

    public static implicit operator T(Primitive<T> primitive) => primitive.Value;
}

Triggers that I use are currently very simple too. It's just a base class that provides a single method for checking whether the trigger Matches. I have two of them.

public abstract class Trigger
{
    public abstract bool Matches(DateTime tick);
}

public class CronTrigger : Trigger
{
    private readonly CronExpression _cronExpression;

    public CronTrigger(string cronExpression)
    {
        _cronExpression = CronExpression.Parse(cronExpression);
    }

    public string Schedule => _cronExpression.ToString();

    public override bool Matches(DateTime tick)
    {
        return _cronExpression.Contains(tick);
    }
}

public class CountTrigger : Trigger
{
    public CountTrigger(int count)
    {
        Counter = new InfiniteCounter(count);
    }

    public IInfiniteCounter Counter { get; }

    public override bool Matches(DateTime tick)
    {
        Counter.MoveNext();
        return Counter.Position == InfiniteCounterPosition.Last;
    }
}

Testing

So far I've created two tests for it (with XUnit). One testing the Job and a bigger one testing the Scheduler.

The first test checks whether the max-number of tasks is not exceeded and whether the Continuation tasks works correctly.

public class JobTest
{
    [Fact]
    public async Task Job_executes_no_more_than_specified_number_of_times()
    {
        var misfireCount = 0;
        var job = new Job("test", Enumerable.Empty<Trigger>(), async token => await Task.Delay(TimeSpan.FromSeconds(3), token))
        {
            OnMisfire = j => misfireCount++,
            MaxDegreeOfParallelism = 2
        };
        job.Execute(CancellationToken.None);
        job.Execute(CancellationToken.None);
        job.Execute(CancellationToken.None);
        Assert.Equal(2, job.Count);
        Assert.Equal(1, misfireCount);

        // Wait until all jobs are completed.
        await job.Continuation;

        Assert.Equal(0, job.Count);
    }
}

Testing Scheduler is now possible by using an observable that is ticking as I say:

public class SchedulerTest
{
    [Fact]
    public void Executes_job_according_to_triggers()
    {
        var job1ExecuteCount = 0;
        var job2ExecuteCount = 0;
        var misfireCount = 0;
        var subject = new Subject<DateTime>();
        var scheduler = new Scheduler(subject);

        var unschedule1 = scheduler.Schedule(new Job("test-1", new[] { new CountTrigger(2) }, async token =>
        {
            Interlocked.Increment(ref job1ExecuteCount);
            await Task.Delay(TimeSpan.FromSeconds(3), token);
        })
        {
            MaxDegreeOfParallelism = 2,
            OnMisfire = _ => Interlocked.Increment(ref misfireCount),
            UnscheduleTimeout = TimeSpan.FromSeconds(4)
        });

        var unschedule2 = scheduler.Schedule(new Job("test-2", new[] { new CountTrigger(3) }, async token =>
        {
            Interlocked.Increment(ref job2ExecuteCount);
            await Task.Delay(TimeSpan.FromSeconds(3), token);
        })
        {
            MaxDegreeOfParallelism = 2,
            OnMisfire = _ => Interlocked.Increment(ref misfireCount),
            UnscheduleTimeout = TimeSpan.FromSeconds(4)
        });

        // Scheduler was just initialized and should not have executed anything yet.
        Assert.Equal(0, job1ExecuteCount);
        Assert.Equal(0, job2ExecuteCount);

        // Tick once.
        subject.OnNext(DateTime.Now);

        // Still nothing should be executed.
        Assert.Equal(0, job1ExecuteCount);
        Assert.Equal(0, job2ExecuteCount);

        // Now tick twice...
        subject.OnNext(DateTime.Now);
        subject.OnNext(DateTime.Now);

        // Unschedule the job. This blocking call waits until all tasks are completed.
        unschedule1.Dispose();
        unschedule2.Dispose();

        // Tick once again. Nothing should be executed anymore.
        subject.OnNext(DateTime.Now);

        // ...this should have matched the two triggers.
        Assert.Equal(1, job1ExecuteCount);
        Assert.Equal(1, job2ExecuteCount);

        Assert.Equal(0, misfireCount);
    }
}

The extension to fix missing seconds...

public static class ObservableExtensions
{
    public static IObservable<DateTime> TruncateMilliseconds(this IObservable<DateTime> ticks)
    {
        return ticks.Select(DateTimeExtensions.TruncateMilliseconds);
    }

    public static IObservable<DateTime> FixMissingSeconds(this IObservable<DateTime> ticks)
    {
        var last = DateTime.MinValue;

        return ticks.SelectMany(tick =>
        {
            if (tick.Millisecond > 0) throw new InvalidOperationException($"{nameof(FixMissingSeconds)} requires ticks without the millisecond part.");

            // We have to start somewhere so let it be one second before tick if we are currently nowhere.
            last = last == DateTime.MinValue ? tick.AddSeconds(-1) : last;

            // Calculates the gap between tick and last. In normal case it's 1.
            var gap = tick.DiffInSeconds(last);

            // If we missed one second due to time inaccuracy, 
            // this makes sure to publish the missing second too
            // so that all jobs at that second can also be triggered.
            return
                Enumerable
                    .Range(0, gap)
                    .Select(_ => last = last.AddSeconds(1));
        });
    }
}

can now also be tested.

public class ObservableExtensionsTest
{
    [Fact]
    public void Returns_ticks_unchanged_when_no_gap()
    {
        var ticks = new[] { 0, 1, 2 }.Select(s => DateTime.Parse($"2019-01-01 10:00:0{s}")).ToList();
        Assert.Equal(ticks, ticks.ToObservable().TruncateMilliseconds().FixMissingSeconds().ToEnumerable().ToList());
    }

    [Fact]
    public void Fixes_tick_gap()
    {
        var expected = new[] { 0, 1, 2 }.Select(s => DateTime.Parse($"2019-01-01 10:00:0{s}")).ToList();
        var missing = new[] { 0, 2 }.Select(s => DateTime.Parse($"2019-01-01 10:00:0{s}")).ToList();
        Assert.Equal(expected.ToList(), missing.ToObservable().TruncateMilliseconds().FixMissingSeconds().ToEnumerable().ToList());
    }
}

Utilities

There are two more classes that drive the automatic scheduler.

One creates an observable that is ticking every-second:

public static class Tick
{
    public static IObservable<DateTime> EverySecond(IDateTime dateTime)
    {
        return
            Observable
                .Interval(TimeSpan.FromSeconds(1))
                .Select(_ => dateTime.Now());
    }
}

the other provides and extension that is fixing missing seconds due to the occasional glitches in the ticking clock:

public static class ObservableExtensions
{
    public static IObservable<DateTime> FixMissingSeconds(this IObservable<DateTime> ticks)
    {
        var last = DateTime.MinValue;

        return ticks.SelectMany(tick =>
        {
            tick = tick.TruncateMilliseconds();

            // We have to start somewhere so let it be one second before tick if we are currently nowhere.
            last = last == DateTime.MinValue ? tick.AddSeconds(-1) : last;

            // Calculates the gap between tick and last. In normal case it's 1.
            var gap = (int)((tick - last).Ticks / TimeSpan.TicksPerSecond);

            // If we missed one second due to time inaccuracy, 
            // this makes sure to publish the missing second too
            // so that all jobs at that second can also be triggered.
            return
                Enumerable
                    .Range(0, gap)
                    .Select(_ => last = last.AddSeconds(1));
        });
    }
}

public interface IDateTime
{
    DateTime Now();
}

public class DateTimeUtc : IDateTime
{
    public DateTime Now() => DateTime.UtcNow;
}

For working with DateTime I also have these:

public static class DateTimeExtensions
{
    public static DateTime TruncateMilliseconds(this DateTime dateTime)
    {
        return new DateTime(dateTime.Ticks - (dateTime.Ticks % TimeSpan.TicksPerSecond), dateTime.Kind);
    }

    public static int Diff(this DateTime later, DateTime earlier, long ticksPerX)
    {
        if (later < earlier) throw new ArgumentException($"'{nameof(later)}' must be greater or equal '{nameof(earlier)}'.");

        return (int)((later - earlier).Ticks / ticksPerX);
    }

    public static int DiffInSeconds(this DateTime later, DateTime earlier)
    {
        return later.Diff(earlier, TimeSpan.TicksPerSecond);
    }
}

This time my main focus is on testability and thread-safety. Do you think I need any locking or synchronisation anywhere? I'm not entirely sure I have thought of everythig. How about testing it? Can you see anything that cannot be tested and do you think the two tests are sane?

\$\endgroup\$
  • \$\begingroup\$ The code (still a little bit messy and with some other less relevant experimetal stuff) can be found in my repository here. \$\endgroup\$ – t3chb0t Mar 9 at 12:07
  • \$\begingroup\$ O-ha! Downvoted by the competition :-) \$\endgroup\$ – t3chb0t Mar 9 at 15:53

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.