Scheduler built with observables

Question

I sometimes have to schedule some tasks and wanted to have my own reusable scheduler. Since I like the Rx, I build it on top of it.

---

Follow-up

There is a newer version of the Scheduler.

---

The Scheduler class is really simple. It's backed by a newer implementation of my old CronExpression.

There are currently only these two methods. The factory method Create creates a new scheduler that ticks at the specified intervals and provides schedules as DateTimes to the observers.

Jobs are scheduled with the Schedule extension. This one requires a cron-expression and the action to execute.

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

    public static IDisposable Schedule(this IObservable<DateTime> schedules, string cronExpressionString, Action<DateTime> action)
    {
        var cronExpression = CronExpression.Parse(cronExpressionString);
        return
            schedules
                .Where(cronExpression.Contains)
                .Subscribe(action);
    }
}

---

The DateTime abstraction is supported by the IDateTime interface:

public interface IDateTime
{
    DateTime Now();
}

which is implemented as

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

or

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

---

Example

In order to use it I just create a scheduler, specify what kind of timestamp it should generate and schedule some actions:

var scheduler = Scheduler.Create(TimeSpan.FromSeconds(1), new LocalDateTime());

scheduler.Schedule("0/1 * * * * * *", schedule =>
{
    Console.WriteLine($"DEBUG: {schedule} [{Thread.CurrentThread.ManagedThreadId}]");
});

scheduler.Schedule("0/5 * * * * * *", schedule =>
{
    Console.WriteLine($"ACTION: {schedule}");
});

The output is:

DEBUG: 13/04/2018 22:32:09 [10]
DEBUG: 13/04/2018 22:32:10 [12]
ACTION: 13/04/2018 22:32:10
DEBUG: 13/04/2018 22:32:11 [10]
DEBUG: 13/04/2018 22:32:12 [14]
DEBUG: 13/04/2018 22:32:13 [8]
DEBUG: 13/04/2018 22:32:14 [12]
ACTION: 13/04/2018 22:32:15
DEBUG: 13/04/2018 22:32:15 [8]
DEBUG: 13/04/2018 22:32:16 [12]
DEBUG: 13/04/2018 22:32:17 [8]

---

There seems to be no rocket science here but this might elusive. Can/should this scheduler by improved in any way?

Answer 1

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

You forgot to plug interval parameter. And, I would consider renaming it resolution.

I wouldn't trust the values from selecting DatetTime::Now or DateTime::UtcNow:

// try running this for a while
var scheduler = Scheduler.Create(TimeSpan.FromSeconds(1), new LocalDateTime());
scheduler.Schedule("0/1 * * * * * *", schedule =>
{
    Console.WriteLine($"ACTION: {schedule}: {schedule:ss.fff}");
});

ACTION: 2018-04-13 6:28:29 PM: 29.987
ACTION: 2018-04-13 6:28:30 PM: 30.987
ACTION: 2018-04-13 6:28:32 PM: 32.001
ACTION: 2018-04-13 6:28:33 PM: 33.001

Depending on CronExpression::Contains implementation, the scheduler could be skipping task when DateTime.Now.Millisecond gets near 0 or 999.

I've attempted to solve that by adding index * interval to a snapshot of time, however this solution suffered a problem: The delay is cumulative, so the schedule's time will slowly deviate(fall behind) from the real time:

// dont use this
public static IObservable<DateTime> Create2(TimeSpan interval, IDateTime dateTime)
{
    var snapshot = dateTime.Now();
    var offset = interval.Ticks - snapshot.Ticks % interval.Ticks;

    return Observable.Interval(interval)
        .Delay(TimeSpan.FromTicks(offset))
        .Select(x => snapshot.AddTicks(interval.Ticks * (x + 1) + offset));
}

An better solution would be to provide a range of time, and check if expression fall between the range:

public static IDisposable Schedule2(this IObservable<DateTime> schedules, string cronExpressionString, Action<DateTime> action)
{
    var cronExpression = Regex.Match(cronExpressionString, @"^0/(?<s>\d+)");
    return
        schedules
            .Scan(default(TimeRange), (previous, x) => new TimeRange(previous?.End, x))
            .Where(cronExpression.Contains)
            .Subscribe(action);
}

public class TimeRange
{
    public DateTime? Start { get; }
    public DateTime End { get; }

    public TimeRange(DateTime? start, DateTime end)
    {
        this.Start = start;
        this.End = end;
    }
}

Answer 2

Xiaoy312 has a point about the risk of missing seconds when using Interval. It seems that each "tick" of Observable.Interval() waits until the previous returns. So if the OnNext halts the thread a while, the next "tick" is fired too late and you may miss a second or two.

Trying to start each scheduled task on a new thread - for instance via a System.Timers.Timer seems to be a bad idea, since the same Cron job probably can not run concurrently(?).

One way not to lose any seconds using Observable.Interval is as follows:

  Random rand = new Random(5);
  IObservable<Timestamped<long>> source = Observable.Interval(TimeSpan.FromSeconds(1)).Timestamp().ObserveOn(NewThreadScheduler.Default);

  IDisposable subscription = source.Subscribe(
      x =>
      {
        Console.WriteLine("OnNext: {0}", x);
        Thread.Sleep(rand.Next(0, 4001));
        Console.WriteLine("After Sleep OnNext: {0}", x);
      },
      ex => Console.WriteLine("OnError: {0}", ex.Message),
      () => Console.WriteLine("OnCompleted"));


  Console.WriteLine("Press ENTER to unsubscribe...");
  Console.ReadLine();
  subscription.Dispose();

This seems to ensure no slip in the sequence of timestamps (Second-wise at least), but each elapsed interval may be (cumulatively) delayed according to previous intervals "laziness". In this way you are not guarantied that the job is done on schedule, but you won't miss a job because of "missing" seconds. It seems though that the time pump catches up the delay when no action delay the current elapsed interval.