RaspberryIO_26/Swan/Threading/TimerWorkerBase.cs

using System;
using System.Threading;
using System.Threading.Tasks;

namespace Swan.Threading {
  /// <summary>
  /// Provides a base implementation for application workers.
  /// </summary>
  /// <seealso cref="IWorker" />
  public abstract class TimerWorkerBase : WorkerBase {
    private readonly Object _syncLock = new Object();
    private readonly Timer _timer;
    private Boolean _isTimerAlive = true;

    /// <summary>
    /// Initializes a new instance of the <see cref="TimerWorkerBase"/> class.
    /// </summary>
    /// <param name="name">The name.</param>
    /// <param name="period">The execution interval.</param>
    protected TimerWorkerBase(String name, TimeSpan period) : base(name, period) =>
      // Instantiate the timer that will be used to schedule cycles
      this._timer = new Timer(this.ExecuteTimerCallback, this, Timeout.Infinite, Timeout.Infinite);

    /// <inheritdoc />
    public override Task<WorkerState> StartAsync() {
      lock(this._syncLock) {
        if(this.WorkerState == WorkerState.Paused || this.WorkerState == WorkerState.Waiting) {
          return this.ResumeAsync();
        }

        if(this.WorkerState != WorkerState.Created) {
          return Task.FromResult(this.WorkerState);
        }

        if(this.IsStopRequested) {
          return Task.FromResult(this.WorkerState);
        }

        Task<WorkerState> task = this.QueueStateChange(StateChangeRequest.Start);
        this.Interrupt();
        return task;
      }
    }

    /// <inheritdoc />
    public override Task<WorkerState> PauseAsync() {
      lock(this._syncLock) {
        if(this.WorkerState != WorkerState.Running && this.WorkerState != WorkerState.Waiting) {
          return Task.FromResult(this.WorkerState);
        }

        if(this.IsStopRequested) {
          return Task.FromResult(this.WorkerState);
        }

        Task<WorkerState> task = this.QueueStateChange(StateChangeRequest.Pause);
        this.Interrupt();
        return task;
      }
    }

    /// <inheritdoc />
    public override Task<WorkerState> ResumeAsync() {
      lock(this._syncLock) {
        if(this.WorkerState == WorkerState.Created) {
          return this.StartAsync();
        }

        if(this.WorkerState != WorkerState.Paused && this.WorkerState != WorkerState.Waiting) {
          return Task.FromResult(this.WorkerState);
        }

        if(this.IsStopRequested) {
          return Task.FromResult(this.WorkerState);
        }

        Task<WorkerState> task = this.QueueStateChange(StateChangeRequest.Resume);
        this.Interrupt();
        return task;
      }
    }

    /// <inheritdoc />
    public override Task<WorkerState> StopAsync() {
      lock(this._syncLock) {
        if(this.WorkerState == WorkerState.Stopped || this.WorkerState == WorkerState.Created) {
          this.WorkerState = WorkerState.Stopped;
          return Task.FromResult(this.WorkerState);
        }

        Task<WorkerState> task = this.QueueStateChange(StateChangeRequest.Stop);
        this.Interrupt();
        return task;
      }
    }

    /// <summary>
    /// Schedules a new cycle for execution. The delay is given in
    /// milliseconds. Passing a delay of 0 means a new cycle should be executed
    /// immediately.
    /// </summary>
    /// <param name="delay">The delay.</param>
    protected void ScheduleCycle(Int32 delay) {
      lock(this._syncLock) {
        if(!this._isTimerAlive) {
          return;
        }

        _ = this._timer.Change(delay, Timeout.Infinite);
      }
    }

    /// <inheritdoc />
    protected override void Dispose(Boolean disposing) {
      base.Dispose(disposing);

      lock(this._syncLock) {
        if(!this._isTimerAlive) {
          return;
        }

        this._isTimerAlive = false;
        this._timer.Dispose();
      }
    }

    /// <summary>
    /// Cancels the current token and schedules a new cycle immediately.
    /// </summary>
    private void Interrupt() {
      lock(this._syncLock) {
        if(this.WorkerState == WorkerState.Stopped) {
          return;
        }

        this.CycleCancellation.Cancel();
        this.ScheduleCycle(0);
      }
    }

    /// <summary>
    /// Executes the worker cycle control logic.
    /// This includes processing state change requests,
    /// the execution of use cycle code,
    /// and the scheduling of new cycles.
    /// </summary>
    private void ExecuteWorkerCycle() {
      this.CycleStopwatch.Restart();

      lock(this._syncLock) {
        if(this.IsDisposing || this.IsDisposed) {
          this.WorkerState = WorkerState.Stopped;

          // Cancel any awaiters
          try {
            this.StateChangedEvent.Set();
          } catch { /* Ignore */ }

          return;
        }

        // Prevent running another instance of the cycle
        if(this.CycleCompletedEvent.IsSet == false) {
          return;
        }

        // Lock the cycle and capture relevant state valid for this cycle
        this.CycleCompletedEvent.Reset();
      }

      CancellationToken interruptToken = this.CycleCancellation.Token;
      WorkerState initialWorkerState = this.WorkerState;

      // Process the tasks that are awaiting
      if(this.ProcessStateChangeRequests()) {
        return;
      }

      try {
        if(initialWorkerState == WorkerState.Waiting &&
            !interruptToken.IsCancellationRequested) {
          // Mark the state as Running
          this.WorkerState = WorkerState.Running;

          // Call the execution logic
          this.ExecuteCycleLogic(interruptToken);
        }
      } catch(Exception ex) {
        this.OnCycleException(ex);
      } finally {
        // Update the state
        this.WorkerState = initialWorkerState == WorkerState.Paused
                    ? WorkerState.Paused
                    : WorkerState.Waiting;

        lock(this._syncLock) {
          // Signal the cycle has been completed so new cycles can be executed
          this.CycleCompletedEvent.Set();

          // Schedule a new cycle
          this.ScheduleCycle(!interruptToken.IsCancellationRequested
                        ? this.ComputeCycleDelay(initialWorkerState)
                        : 0);
        }
      }
    }

    /// <summary>
    /// Represents the callback that is executed when the <see cref="_timer"/> ticks.
    /// </summary>
    /// <param name="state">The state -- this contains the worker.</param>
    private void ExecuteTimerCallback(Object state) => this.ExecuteWorkerCycle();

    /// <summary>
    /// Queues a transition in worker state for processing. Returns a task that can be awaited
    /// when the operation completes.
    /// </summary>
    /// <param name="request">The request.</param>
    /// <returns>The awaitable task.</returns>
    private Task<WorkerState> QueueStateChange(StateChangeRequest request) {
      lock(this._syncLock) {
        if(this.StateChangeTask != null) {
          return this.StateChangeTask;
        }

        Task<WorkerState> waitingTask = new Task<WorkerState>(() => {
          this.StateChangedEvent.Wait();
          lock(this._syncLock) {
            this.StateChangeTask = null;
            return this.WorkerState;
          }
        });

        this.StateChangeTask = waitingTask;
        this.StateChangedEvent.Reset();
        this.StateChangeRequests[request] = true;
        waitingTask.Start();
        this.CycleCancellation.Cancel();

        return waitingTask;
      }
    }

    /// <summary>
    /// Processes the state change queue by checking pending events and scheduling
    /// cycle execution accordingly. The <see cref="WorkerState"/> is also updated.
    /// </summary>
    /// <returns>Returns <c>true</c> if the execution should be terminated. <c>false</c> otherwise.</returns>
    private Boolean ProcessStateChangeRequests() {
      lock(this._syncLock) {
        WorkerState currentState = this.WorkerState;
        Boolean hasRequest = false;
        Int32 schedule = 0;

        // Update the state according to request priority
        if(this.StateChangeRequests[StateChangeRequest.Stop] || this.IsDisposing || this.IsDisposed) {
          hasRequest = true;
          this.WorkerState = WorkerState.Stopped;
          schedule = this.StateChangeRequests[StateChangeRequest.Stop] ? Timeout.Infinite : 0;
        } else if(this.StateChangeRequests[StateChangeRequest.Pause]) {
          hasRequest = true;
          this.WorkerState = WorkerState.Paused;
          schedule = Timeout.Infinite;
        } else if(this.StateChangeRequests[StateChangeRequest.Start] || this.StateChangeRequests[StateChangeRequest.Resume]) {
          hasRequest = true;
          this.WorkerState = WorkerState.Waiting;
        }

        // Signals all state changes to continue
        // as a command has been handled.
        if(hasRequest) {
          this.ClearStateChangeRequests(schedule, currentState, this.WorkerState);
        }

        return hasRequest;
      }
    }

    /// <summary>
    /// Signals all state change requests to set.
    /// </summary>
    /// <param name="schedule">The cycle schedule.</param>
    /// <param name="oldState">The previous worker state.</param>
    /// <param name="newState">The new worker state.</param>
    private void ClearStateChangeRequests(Int32 schedule, WorkerState oldState, WorkerState newState) {
      lock(this._syncLock) {
        // Mark all events as completed
        this.StateChangeRequests[StateChangeRequest.Start] = false;
        this.StateChangeRequests[StateChangeRequest.Pause] = false;
        this.StateChangeRequests[StateChangeRequest.Resume] = false;
        this.StateChangeRequests[StateChangeRequest.Stop] = false;

        this.StateChangedEvent.Set();
        this.CycleCompletedEvent.Set();
        this.OnStateChangeProcessed(oldState, newState);
        this.ScheduleCycle(schedule);
      }
    }
  }
}