MultiReadSingleWriteObject.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Waher.Runtime.Collections;
 
namespace Waher.Runtime.Threading
{
    public class MultiReadSingleWriteObject : IMultiReadSingleWriteObject, IDisposable
    {
        private static readonly Stopwatch watch = new Stopwatch();
 
        static MultiReadSingleWriteObject()
        {
            watch.Start();
        }
 
        private readonly bool recordStackTraces;
        private readonly object owner;
        private readonly string creatorStackTrace;
        private string lockStackTrace;
        private LinkedList<WaitRec> noWriters = new LinkedList<WaitRec>();
        private LinkedList<WaitRec> noReadersOrWriters = new LinkedList<WaitRec>();
        private readonly object synchObj = new object();
        private long token = 0;
        private long start = 0;
        private int nrReaders = 0;
        private bool isWriting = false;
        private bool disposed = false;
 
        private class WaitRec
        {
            public TaskCompletionSource<bool> Pending;
            public string StackTrace;
 
            public WaitRec(bool RecordStackTraces)
            {
                this.Pending = new TaskCompletionSource<bool>();
 
                if (RecordStackTraces)
                    this.StackTrace = Environment.StackTrace;
                else
                    this.StackTrace = null;
            }
        }
 
        public MultiReadSingleWriteObject(object Owner)
            : this()
        {
            this.owner = Owner;
        }
 
        public MultiReadSingleWriteObject()
#if DEBUG
            : this(true)
#else
            : this(false)
#endif
        {
        }
 
        public MultiReadSingleWriteObject(object Owner, bool RecordStackTraces)
            : this(RecordStackTraces)
        {
            this.owner = Owner;
        }
 
        public MultiReadSingleWriteObject(bool RecordStackTraces)
        {
            this.recordStackTraces = RecordStackTraces;
 
            if (this.recordStackTraces)
                this.creatorStackTrace = Environment.StackTrace;
            else
                this.creatorStackTrace = null;
 
            this.lockStackTrace = null;
        }
 
        public object Owner => this.owner;
 
        public bool RecordStackTraces => this.recordStackTraces;
 
        public string CreatorStackTrace => this.creatorStackTrace;
 
        public string LockStackTrace => this.lockStackTrace;
 
        public int NrReaders
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                    return this.nrReaders;
                }
            }
        }
 
        public bool IsReading
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                    return this.nrReaders > 0;
                }
            }
        }
 
        public bool IsWriting
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                    return this.isWriting;
                }
            }
        }
 
        public bool IsReadingOrWriting
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                    return this.nrReaders > 0 || this.isWriting;
                }
            }
        }
 
        public int QueueSize
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
                    
                    return this.noReadersOrWriters.Count;
                }
            }
        }
 
        public string[] QueuedStackTraces
        {
            get
            {
                ChunkedList<string> Result = null;
 
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                    foreach (WaitRec Rec in this.noReadersOrWriters)
                    {
                        if (!string.IsNullOrEmpty(Rec.StackTrace))
                        {
                            if (Result is null)
                                Result = new ChunkedList<string>();
 
                            Result.Add(Rec.StackTrace);
                        }
                    }
                }
 
                return Result?.ToArray() ?? Array.Empty<string>();
            }
        }
 
        public long TicksLocked
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        return 0;
                    else if (this.nrReaders > 0 || this.isWriting)
                        return watch.ElapsedTicks - this.start;
                    else
                        return 0;
                }
            }
        }
 
        public double MillisecondsLocked => this.TicksLocked * 1000.0 / Stopwatch.Frequency;
 
        public bool Disposed => this.disposed;
 
        public void AssertReading()
        {
            lock (this.synchObj)
            {
                if (this.disposed)
                    throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                if (this.nrReaders <= 0)
                    throw new InvalidOperationException("Not in a reading state.");
            }
        }
 
        public void AssertWriting()
        {
            lock (this.synchObj)
            {
                if (this.disposed)
                    throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                if (!this.isWriting)
                    throw new InvalidOperationException("Not in a writing state.");
            }
        }
 
        public void AssertReadingOrWriting()
        {
            lock (this.synchObj)
            {
                if (this.disposed)
                    throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                if (this.nrReaders <= 0 && !this.isWriting)
                    throw new InvalidOperationException("Not in a reading or writing state.");
            }
        }
 
        public long Token
        {
            get
            {
                lock (this.synchObj)
                {
                    if (this.disposed)
                        throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
                    return this.token;
                }
            }
        }
 
        public virtual async Task<int> BeginRead()
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            WaitRec Wait = null;
            int Result = 0;
            bool RecordStackTrace = false;
 
            while (true)
            {
                lock (this.synchObj)
                {
                    if (!this.isWriting)
                    {
                        if (this.nrReaders == 0)
                        {
                            this.start = watch.ElapsedTicks;
                            this.token++;
 
                            if (this.recordStackTraces)
                                RecordStackTrace = true;
                        }
 
                        Result = ++this.nrReaders;
                    }
                    else
                    {
                        Wait = new WaitRec(this.recordStackTraces);
                        this.noWriters.AddLast(Wait);
                    }
                }
 
                if (Wait is null)
                {
                    if (RecordStackTrace)
                        this.lockStackTrace = Environment.StackTrace;
 
                    return Result;
                }
                else
                {
                    await Wait.Pending.Task;
                    Wait = null;
                }
            }
        }
 
        public virtual Task<int> EndRead()
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            LinkedList<WaitRec> List = null;
 
            lock (this.synchObj)
            {
                if (this.nrReaders <= 0)
                    throw new InvalidOperationException("Not in a reading state.");
 
                this.nrReaders--;
                if (this.nrReaders == 0)
                {
                    this.token++;
                    this.start = 0;
 
                    if (this.recordStackTraces)
                        this.lockStackTrace = null;
                }
                else
                    return Task.FromResult(this.nrReaders);
 
                if (this.noReadersOrWriters.First is null)
                    return Task.FromResult(0);
 
                List = this.noReadersOrWriters;
                this.noReadersOrWriters = new LinkedList<WaitRec>();
            }
 
            foreach (WaitRec Rec in List)
                Rec.Pending.TrySetResult(true);
 
            return Task.FromResult(0);
        }
 
        public virtual async Task<bool> TryBeginRead(int Timeout)
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            WaitRec Wait = null;
            DateTime Start = DateTime.UtcNow;
            bool RecordStackTrace = false;
 
            while (true)
            {
                lock (this.synchObj)
                {
                    if (!this.isWriting)
                    {
                        if (this.nrReaders == 0)
                        {
                            this.start = watch.ElapsedTicks;
                            this.token++;
 
                            if (this.recordStackTraces)
                                RecordStackTrace = true;
                        }
 
                        this.nrReaders++;
                    }
                    else if (Timeout <= 0)
                        return false;
                    else
                    {
                        Wait = new WaitRec(this.recordStackTraces);
                        this.noWriters.AddLast(Wait);
                    }
                }
 
                if (Wait is null)
                {
                    if (RecordStackTrace)
                        this.lockStackTrace = Environment.StackTrace;
 
                    return true;
                }
                else
                {
                    DateTime Now = DateTime.UtcNow;
                    bool Result;
 
                    using (Timer Timer = new Timer((P) =>
                    {
                        Wait?.Pending.TrySetResult(false);
 
                    }, null, Timeout, System.Threading.Timeout.Infinite))
                    {
                        Result = await Wait.Pending.Task;
                    }
 
                    if (!Result)
                    {
                        lock (this.synchObj)
                        {
                            this.noWriters.Remove(Wait);
                        }
 
                        return false;
                    }
 
                    Timeout -= (int)((Now - Start).TotalMilliseconds + 0.5);
                    Start = Now;
                    Wait = null;
                }
            }
        }
 
        public virtual async Task<bool> TryBeginRead(CancellationToken Cancel)
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            if (Cancel.CanBeCanceled)
            {
                WaitRec Wait = null;
                bool RecordStackTrace = false;
 
                Cancel.Register(() =>
                {
                    Wait?.Pending.TrySetResult(false);
                });
 
                while (true)
                {
                    lock (this.synchObj)
                    {
                        if (!this.isWriting)
                        {
                            if (this.nrReaders == 0)
                            {
                                this.start = watch.ElapsedTicks;
                                this.token++;
 
                                if (this.recordStackTraces)
                                    RecordStackTrace = true;
                            }
 
                            this.nrReaders++;
                        }
                        else if (Cancel.IsCancellationRequested)
                            return false;
                        else
                        {
                            Wait = new WaitRec(this.recordStackTraces);
                            this.noWriters.AddLast(Wait);
                        }
                    }
 
                    if (Wait is null)
                    {
                        if (RecordStackTrace)
                            this.lockStackTrace = Environment.StackTrace;
 
                        return true;
                    }
                    else
                    {
                        bool Result = await Wait.Pending.Task;
 
                        if (!Result)
                        {
                            lock (this.synchObj)
                            {
                                this.noWriters.Remove(Wait);
                            }
 
                            return false;
                        }
 
                        Wait = null;
                    }
                }
            }
            else
            {
                await this.BeginRead();
                return true;
            }
        }
 
        public virtual async Task BeginWrite()
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            WaitRec Prev = null;
            WaitRec Wait = null;
            bool RecordStackTrace = false;
 
            while (true)
            {
                lock (this.synchObj)
                {
                    if (!(Prev is null))
                    {
                        this.noWriters.Remove(Prev);    // In case previously locked for reading
                        this.noReadersOrWriters.Remove(Prev);
                    }
 
                    if (this.nrReaders == 0 && !this.isWriting)
                    {
                        this.start = watch.ElapsedTicks;
                        this.token++;
                        this.isWriting = true;
 
                        if (this.recordStackTraces)
                            RecordStackTrace = true;
                    }
                    else
                    {
                        Wait = new WaitRec(this.recordStackTraces);
                        this.noReadersOrWriters.AddLast(Wait);
                        this.noWriters.AddLast(Wait);
                    }
                }
 
                if (Wait is null)
                {
                    if (RecordStackTrace)
                        this.lockStackTrace = Environment.StackTrace;
 
                    return;
                }
                else
                {
                    await Wait.Pending.Task;
                    Prev = Wait;
                    Wait = null;
                }
            }
        }
 
        public virtual Task EndWrite()
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            LinkedList<WaitRec> List = null;
            LinkedList<WaitRec> List2 = null;
 
            lock (this.synchObj)
            {
                if (!this.isWriting)
                    throw new InvalidOperationException("Not in a writing state.");
 
                this.token++;
                this.isWriting = false;
                this.start = 0;
 
                if (this.recordStackTraces)
                    this.lockStackTrace = null;
 
                if (!(this.noReadersOrWriters.First is null))
                {
                    List = this.noReadersOrWriters;
                    this.noReadersOrWriters = new LinkedList<WaitRec>();
                }
 
                if (!(this.noWriters.First is null))
                {
                    List2 = this.noWriters;
                    this.noWriters = new LinkedList<WaitRec>();
                }
            }
 
            if (!(List is null))
            {
                foreach (WaitRec Rec in List)
                    Rec.Pending.TrySetResult(true);
            }
 
            if (!(List2 is null))
            {
                foreach (WaitRec Rec in List2)
                    Rec.Pending.TrySetResult(true);
            }
 
            return Task.CompletedTask;
        }
 
        public virtual async Task<bool> TryBeginWrite(int Timeout)
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            WaitRec Prev = null;
            WaitRec Wait = null;
            DateTime Start = DateTime.UtcNow;
            bool RecordStackTrace = false;
 
            while (true)
            {
                lock (this.synchObj)
                {
                    if (!(Prev is null))
                    {
                        this.noWriters.Remove(Prev);    // In case previously locked for reading
                        this.noReadersOrWriters.Remove(Prev);
                    }
 
                    if (this.nrReaders == 0 && !this.isWriting)
                    {
                        this.start = watch.ElapsedTicks;
                        this.token++;
                        this.isWriting = true;
 
                        if (this.recordStackTraces)
                            RecordStackTrace = true;
                    }
                    else if (Timeout <= 0)
                        return false;
                    else
                    {
                        Wait = new WaitRec(this.recordStackTraces);
                        this.noWriters.AddLast(Wait);
                        this.noReadersOrWriters.AddLast(Wait);
                    }
                }
 
                if (Wait is null)
                {
                    if (RecordStackTrace)
                        this.lockStackTrace = Environment.StackTrace;
 
                    return true;
                }
                else
                {
                    DateTime Now = DateTime.UtcNow;
                    bool Result;
 
                    using (Timer Timer = new Timer((P) =>
                    {
                        Wait?.Pending.TrySetResult(false);
 
                    }, null, Timeout, System.Threading.Timeout.Infinite))
                    {
                        Result = await Wait.Pending.Task;
                    }
 
                    if (!Result)
                    {
                        lock (this.synchObj)
                        {
                            this.noWriters.Remove(Wait);
                            this.noReadersOrWriters.Remove(Wait);
                        }
 
                        return false;
                    }
 
                    Timeout -= (int)((Now - Start).TotalMilliseconds + 0.5);
                    Start = Now;
                    Prev = Wait;
                    Wait = null;
                }
            }
        }
 
        public virtual async Task<bool> TryBeginWrite(CancellationToken Cancel)
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
 
            if (Cancel.CanBeCanceled)
            {
                WaitRec Prev = null;
                WaitRec Wait = null;
                DateTime Start = DateTime.UtcNow;
                bool RecordStackTrace = false;
 
                Cancel.Register(() =>
                {
                    Prev?.Pending.TrySetResult(false);
                    Wait?.Pending.TrySetResult(false);
                });
 
                while (true)
                {
                    lock (this.synchObj)
                    {
                        if (!(Prev is null))
                        {
                            this.noWriters.Remove(Prev);    // In case previously locked for reading
                            this.noReadersOrWriters.Remove(Prev);
                        }
 
                        if (this.nrReaders == 0 && !this.isWriting)
                        {
                            this.start = watch.ElapsedTicks;
                            this.token++;
                            this.isWriting = true;
 
                            if (this.recordStackTraces)
                                RecordStackTrace = true;
                        }
                        else if (Cancel.IsCancellationRequested)
                            return false;
                        else
                        {
                            Wait = new WaitRec(this.recordStackTraces);
                            this.noWriters.AddLast(Wait);
                            this.noReadersOrWriters.AddLast(Wait);
                        }
                    }
 
                    if (Wait is null)
                    {
                        if (RecordStackTrace)
                            this.lockStackTrace = Environment.StackTrace;
 
                        return true;
                    }
                    else
                    {
                        bool Result = await Wait.Pending.Task;
 
                        if (!Result)
                        {
                            lock (this.synchObj)
                            {
                                this.noWriters.Remove(Wait);
                                this.noReadersOrWriters.Remove(Wait);
                            }
 
                            return false;
                        }
 
                        Prev = Wait;
                        Wait = null;
                    }
                }
            }
            else
            {
                await this.BeginWrite();
                return true;
            }
        }
 
        public virtual Task Unlock()
        {
            LinkedList<WaitRec> List = null;
            LinkedList<WaitRec> List2 = null;
 
            lock (this.synchObj)
            {
                this.nrReaders = 0;
                this.isWriting = false;
                this.token++;
                this.start = 0;
 
                if (this.recordStackTraces)
                    this.lockStackTrace = null;
 
                if (!(this.noReadersOrWriters.First is null))
                {
                    List = this.noReadersOrWriters;
                    this.noReadersOrWriters = new LinkedList<WaitRec>();
                }
 
                if (!(this.noWriters.First is null))
                {
                    List2 = this.noWriters;
                    this.noWriters = new LinkedList<WaitRec>();
                }
            }
 
            if (!(List is null))
            {
                foreach (WaitRec Rec in List)
                    Rec.Pending.TrySetResult(false);
            }
 
            if (!(List2 is null))
            {
                foreach (WaitRec Rec in List2)
                    Rec.Pending.TrySetResult(false);
            }
 
            return Task.CompletedTask;
        }
 
        public virtual void Dispose()
        {
            if (this.disposed)
                throw new ObjectDisposedException(nameof(MultiReadSingleWriteObject));
            else
            {
                this.disposed = true;
                this.Unlock();
            }
        }
 
    }
}