Down the Concurrency Rabbit Hole

module ActorStatus =

    [<Literal>]
    let Idle = 0L

    [<Literal>]
    let Occupied = 1L

    [<Literal>]
    let Stopped = 2L

/// <summary> An actor pulls a message off of a (thread-safe) queue and executes it in the isolated context of the actor. </summary>
/// <typeparam name = "tmsg"> the type of messages that this actor will manage </typeparam>
/// <param name = "bodyFn"> the function to execute when the thread is available </param>
type ThreadedActor<'tmsg>(bodyFn: 'tmsg * ThreadedActor<'tmsg> -> unit) as this =

    let mutable status: int64 = ActorStatus.Idle

    //Do not expose these
    let signal = new ManualResetEventSlim(false)
    let queue = ConcurrentQueue<'tmsg>()
    let mutable count = 0
    // This is the main thread function (delegate). It gets executed when the thread is started/signaled
    let threadFn () =
        // While the actor is not stopped, check the queue and process any awaiting messages
        while Interlocked.Read(&status) <> ActorStatus.Stopped do
            while not queue.IsEmpty do
                //assert(queue.Count = Threading.Volatile.Read(&count)) race condition
                // If the thread is idle, update it to 'occupied'
                Interlocked.CompareExchange(&status, ActorStatus.Occupied, ActorStatus.Idle)
                |> ignore
                // Try to get the next message in the queue
                let isSuccessful, message = queue.TryDequeue()
                // If we successfully retrieved the next message, execute it in the context of this thread
                if isSuccessful then 
                    bodyFn (message, this)
                    if Threading.Interlocked.Decrement(&count) = 0 then 
                        signal.Reset()
            // If the thread is 'occupied', mark it as idle
            Interlocked.CompareExchange(&status, ActorStatus.Idle, ActorStatus.Occupied) |> ignore
            if Interlocked.Read(&status) <> ActorStatus.Stopped  then signal.Wait()
        // If the thread is stopped, dispose of it
        signal.Dispose()

    // The thread associated with this actor (one-to-one relationship)
    // Pass the threadFn delegate to the constructor
    let thread =
        Thread(ThreadStart(threadFn), IsBackground = true, Name = "ActorThread")

    // Start the thread
    do thread.Start()

    /// Enqueue a new messages for the thread to pick up and execute
    member this.Enqueue(msg: 'tmsg) =
        if Interlocked.Read(&status) <> ActorStatus.Stopped then
            queue.Enqueue(msg)
            if Threading.Interlocked.Increment(&count) = 1 then 
                signal.Set()
        else
            failwith "Cannot queue to stopped actor."

    // Get the length of the actor's message queue
    member this.QueueCount = queue.Count

    // Stops the actor
    member this.Stop() =
        Interlocked.Exchange(&status, ActorStatus.Stopped)
        |> ignore

        signal.Set()

    interface IDisposable with
        member __.Dispose() =
            this.Stop()
            thread.Join()
            

 type ThreadedActor3<'tmsg>(bodyFn: 'tmsg -> unit,  ct : CancellationToken ) as this =
 
     let queue =  ConcurrentQueue<'tmsg>()
     let signal = new ManualResetEventSlim(false,1000) //avoids spin

 
     let threadFn () =
         while not <| ct.IsCancellationRequested do
             let isSuccessful, message = queue.TryDequeue()
             // If we successfully retrieved the next message, execute it in the context of this thread
             if isSuccessful then 
                 bodyFn message
             else signal.Reset()
             if queue.IsEmpty && (not <| ct.IsCancellationRequested) then 
                 signal.Wait()
                 

         signal.Dispose()
 
 
     let thread =
         Thread(ThreadStart(threadFn), IsBackground = true, Name = "ActorThread")
 
     do thread.Start()
 
 
     member this.Enqueue(msg: 'tmsg) =
         queue.Enqueue(msg)
         if not <| signal.IsSet then signal.Set()

 
 
     member this.QueueCount = queue.Count
 
     interface IDisposable with
         member __.Dispose() =
             thread.Join() 

type ThreadedActor4<'tmsg>(bodyFn: 'tmsg -> unit, qDepth : uint64, ct : CancellationToken ) as this =

     let queue =  ConcurrentQueue<'tmsg>()
     let signal = new ManualResetEventSlim(false,1000) //avoids spin
     let rsignal = new ManualResetEventSlim(false,1000) //throttles
 
     let threadFn () =
         while not <| ct.IsCancellationRequested do
             let isSuccessful, message = queue.TryDequeue()
             if isSuccessful then 
                 bodyFn message
             else signal.Reset()
             if uint64 queue.Count <= qDepth && (not rsignal.IsSet) then rsignal.Set()
             if queue.IsEmpty && (not <| ct.IsCancellationRequested) then 
                 signal.Wait()
                 

         signal.Dispose()
 
 
     let thread =
         Thread(ThreadStart(threadFn), IsBackground = true, Name = "ActorThread")
 
     do thread.Start()
 
 
     member this.Enqueue(msg: 'tmsg) =
        if uint64 queue.Count < qDepth then
            queue.Enqueue(msg)
            if not <| signal.IsSet then signal.Set()

        else
            rsignal.Reset()
            if uint64 queue.Count >= qDepth then rsignal.Wait() //blocking throttle
            queue.Enqueue(msg)


     member this.QueueCount = queue.Count
 
     interface IDisposable with
         member __.Dispose() =
             thread.Join() 

type ResettableValueTaskSource(cancellationToken : CancellationToken) as this =

    let mutable _waitSource = ManualResetValueTaskSourceCore<bool>()
    do _waitSource.RunContinuationsAsynchronously <- false
    let mutable _waitSourceCancellation = cancellationToken.UnsafeRegister((fun x -> (x :?> ResettableValueTaskSource).CancelWaiter(cancellationToken)), this)
    let mutable _hasWaiter = 0
   

    member this.SignalWaiter() = 
        if Interlocked.Exchange(&_hasWaiter, 0) = 1
        then _waitSource.SetResult(true)
        ()
    member private this.CancelWaiter(cancellationToken : CancellationToken) = 
        if Interlocked.Exchange(&_hasWaiter, 0) = 1
        then _waitSource.SetException (ExceptionDispatchInfo.SetCurrentStackTrace (new OperationCanceledException(cancellationToken)))
        ()

    member this.WaitAsync() =
        if _hasWaiter <> 0
        then raise (new InvalidOperationException("Concurrent use is not supported") :> System.Exception)
        _waitSource.Reset ()
        Volatile.Write(&_hasWaiter, 1)
        new ValueTask(this :> IValueTaskSource, _waitSource.Version)
        
    interface IValueTaskSource with
        member this.GetResult(token : int16) = 
            _waitSource.GetResult (token) |> ignore

        member this.GetStatus(token : int16) = 
               (_waitSource.GetStatus (token)) :> ValueTaskSourceStatus

        member this.OnCompleted(continuation : Action<obj>, state : obj, token : System.Int16, flags : ValueTaskSourceOnCompletedFlags) = 
            _waitSource.OnCompleted (continuation, state, token, flags)

    interface IValueTaskSource<bool> with
        member this.GetResult(token : int16) = 
            _waitSource.GetResult (token)

        member this.GetStatus(token : int16) = 
               (_waitSource.GetStatus (token)) :> ValueTaskSourceStatus

        member this.OnCompleted(continuation : Action<obj>, state : obj, token : System.Int16, flags : ValueTaskSourceOnCompletedFlags) = 
            _waitSource.OnCompleted (continuation, state, token, flags)

            
type VACore<'tmsg> = 
    {
        mutable count: uint64
        queueDepth: uint64
        queue: ConcurrentQueue<'tmsg>
        vts: ResettableValueTaskSource
        vtsR: ResettableValueTaskSource
        ct: CancellationToken
    }

[<RequireQualifiedAccess>]    
module AsyncChannels =
    


    let init<'t> qDepth ct =
        let queue = ConcurrentQueue<'t>()
        let vts = ResettableValueTaskSource(ct) //slow down the enqueue (when queue reaches depth)
        let vtsR = ResettableValueTaskSource(ct) //prevents cpu cycling (when queue is empty)
        let mutable count = queue.Count |> uint64
        {
            count = count
            queueDepth = qDepth
            queue = queue
            vts = vts
            vtsR = vtsR
            ct = ct 
        }

    let enqueue (core : VACore<'t>) (item : 't) =
            core.queue.Enqueue(item)
            let cnt = Threading.Interlocked.Increment(&core.count)
            core.vtsR.SignalWaiter()

            if Threading.Interlocked.Read(&core.count) > core.queueDepth then
                core.vts.WaitAsync()
            else
                ValueTask.CompletedTask

    let dequeue (core : VACore<'t>) =
        let isSuccessful, message = core.queue.TryDequeue()
        if isSuccessful then 
            Threading.Interlocked.Decrement(&core.count)  
            core.vts.SignalWaiter()
            ValueSome(message)  
        else ValueNone 


    let iter1 (core : VACore<'t>) (fn : 't -> unit) =
        task {
            while not <| core.ct.IsCancellationRequested do
                let dequeue = dequeue core
                if dequeue.IsSome then
                    fn dequeue.Value
                if core.queue.IsEmpty && Interlocked.Read(&core.count) = 0UL then 
                    do! core.vtsR.WaitAsync()
                   
        }