hodzanassredin · October 9, 2022 12:00 · hodzanassredin · Oct 9, 2022
diff --git a/ActiveObject.cs b/ActiveObject.cs
 namespace ConsoleApp7
 {

    public class AsyncMonitor
    {
        private readonly SemaphoreSlim signal = new SemaphoreSlim(1, 1);

        public void Exit()
        {
            try
            {
                signal.Release();
            }
            catch (SemaphoreFullException) { }
        }

        public async Task Enter(CancellationToken cancellationToken)
        {
            await signal.WaitAsync(cancellationToken);
        }
    }
    public struct ConditionalContinuation
    {
        public ConditionalContinuation(TaskCompletionSource continuation, Func<bool> predicate)
        {
            Continuation = continuation;
            Predicate = predicate;
        }
        public TaskCompletionSource Continuation { get; }
        public Func<bool> Predicate { get; }
    }

    public abstract class ActiveObject
    {
        private AsyncMonitor monitor = new();
        private List<ConditionalContinuation> queue = new();
        public virtual Func<CancellationToken, Task>[] Active() => Array.Empty<Func<CancellationToken, Task>>();
        public ActiveObject(CancellationToken ct)
        {
            foreach (var active in Active())
            {
                Task.Run(() => active(ct));
            }

        }
        public async Task<ExclusiveScope> Exclusive(CancellationToken ct)
        {
            await monitor.Enter(ct);
            return new ExclusiveScope(monitor, queue);
        }

        public class ExclusiveScope : IDisposable
        {
            private readonly AsyncMonitor monitor;
            private readonly List<ConditionalContinuation> queue;
            public ExclusiveScope(AsyncMonitor monitor, List<ConditionalContinuation> queue)
            {
                this.monitor = monitor;
                this.queue = queue;
            }

            public Task Await(Func<bool> predicate)
            {
                if (predicate()) return Task.CompletedTask;
                var tcs = new TaskCompletionSource();

                queue.Add(new ConditionalContinuation(tcs, predicate));
                monitor.Exit();
                return tcs.Task;
            }

            public void Dispose()
            {
                for (int i = 0; i < queue.Count; i++)
                {
                    var condContinuation = queue[i];
                    if (condContinuation.Predicate())
                    {
                        queue.RemoveAt(i);
                        condContinuation.Continuation.SetResult();
                        return;
                    }
                }
                monitor.Exit();
            }
        }
    }

    public class BoundedBuffer<T> : ActiveObject
    {
        int h = 0;
        int n = 0;
        T[] B;
        public BoundedBuffer(int max, CancellationToken ct) : base(ct)
        {
            B = new T[max];
        }
        public async Task<T> Get(CancellationToken ct)
        {
            using (var region = await this.Exclusive(ct))
            {
                await region.Await(() => n != 0);//buffer not empty
                T x = B[h]; h = (h + 1) % B.Length; n--;
                return x;
            }
        }
        public async Task Put(T x, CancellationToken ct)
        {
            using (var region = await this.Exclusive(ct))
            {
                await region.Await(() => n != B.Length); /* buffer not full */
                B[(h + n) % B.Length] = x; n++;
            }
        }

        public override Func<CancellationToken, Task>[] Active()
        {
            return new Func<CancellationToken, Task>[] {
                async (ct)=>{
                    while (!ct.IsCancellationRequested)
                    {
                        using (var region = await this.Exclusive(ct))
                        {
                            Console.WriteLine($"Channel length {n}");
                        }
                        //await Task.Delay(1000, ct);
                    }
                }
            };
        }
    }
 }
diff --git a/Program.cs b/Program.cs
 // See https://aka.ms/new-console-template for more information

 using ConsoleApp7;

 var cts = new CancellationTokenSource();
 var ch = new BoundedBuffer<string>(2, cts.Token);

 var consumers = Enumerable.Range(0,3).Select(i=> Task.Run(async () =>
 {
    while (!cts.Token.IsCancellationRequested)
    {
        try
        {
            var t = await ch.Get(cts.Token);
            Console.WriteLine($"Recieved {t} by {i}");
        }
        catch (Exception)
        {
        }
        
    }
 })).ToArray();
 var producers = Enumerable.Range(0, 3).Select(producer => Task.Run(async () =>
 {
    var rnd = new Random();
    for (int i = 0; i < 100; i++)
    {
        //await Task.Delay(TimeSpan.FromMilliseconds(rnd.Next(3)));
        await ch.Put($"Message {i} from {producer}", cts.Token);
    }
 })).ToArray();
 await Task.WhenAll(producers);
 cts.Cancel();
 await Task.WhenAll(consumers);

 Console.WriteLine("Finished");
	namespace ConsoleApp7
	{

	public class AsyncMonitor
	{
	private readonly SemaphoreSlim signal = new SemaphoreSlim(1, 1);

	public void Exit()
	{
	try
	{
	signal.Release();
	}
	catch (SemaphoreFullException) { }
	}

	public async Task Enter(CancellationToken cancellationToken)
	{
	await signal.WaitAsync(cancellationToken);
	}
	}
	public struct ConditionalContinuation
	{
	public ConditionalContinuation(TaskCompletionSource continuation, Func<bool> predicate)
	{
	Continuation = continuation;
	Predicate = predicate;
	}
	public TaskCompletionSource Continuation { get; }
	public Func<bool> Predicate { get; }
	}

	public abstract class ActiveObject
	{
	private AsyncMonitor monitor = new();
	private List<ConditionalContinuation> queue = new();
	public virtual Func<CancellationToken, Task>[] Active() => Array.Empty<Func<CancellationToken, Task>>();
	public ActiveObject(CancellationToken ct)
	{
	foreach (var active in Active())
	{
	Task.Run(() => active(ct));
	}

	}
	public async Task<ExclusiveScope> Exclusive(CancellationToken ct)
	{
	await monitor.Enter(ct);
	return new ExclusiveScope(monitor, queue);
	}

	public class ExclusiveScope : IDisposable
	{
	private readonly AsyncMonitor monitor;
	private readonly List<ConditionalContinuation> queue;
	public ExclusiveScope(AsyncMonitor monitor, List<ConditionalContinuation> queue)
	{
	this.monitor = monitor;
	this.queue = queue;
	}

	public Task Await(Func<bool> predicate)
	{
	if (predicate()) return Task.CompletedTask;
	var tcs = new TaskCompletionSource();

	queue.Add(new ConditionalContinuation(tcs, predicate));
	monitor.Exit();
	return tcs.Task;
	}

	public void Dispose()
	{
	for (int i = 0; i < queue.Count; i++)
	{
	var condContinuation = queue[i];
	if (condContinuation.Predicate())
	{
	queue.RemoveAt(i);
	condContinuation.Continuation.SetResult();
	return;
	}
	}
	monitor.Exit();
	}
	}
	}

	public class BoundedBuffer<T> : ActiveObject
	{
	int h = 0;
	int n = 0;
	T[] B;
	public BoundedBuffer(int max, CancellationToken ct) : base(ct)
	{
	B = new T[max];
	}
	public async Task<T> Get(CancellationToken ct)
	{
	using (var region = await this.Exclusive(ct))
	{
	await region.Await(() => n != 0);//buffer not empty
	T x = B[h]; h = (h + 1) % B.Length; n--;
	return x;
	}
	}
	public async Task Put(T x, CancellationToken ct)
	{
	using (var region = await this.Exclusive(ct))
	{
	await region.Await(() => n != B.Length); /* buffer not full */
	B[(h + n) % B.Length] = x; n++;
	}
	}

	public override Func<CancellationToken, Task>[] Active()
	{
	return new Func<CancellationToken, Task>[] {
	async (ct)=>{
	while (!ct.IsCancellationRequested)
	{
	using (var region = await this.Exclusive(ct))
	{
	Console.WriteLine($"Channel length {n}");
	}
	//await Task.Delay(1000, ct);
	}
	}
	};
	}
	}
	}
	// See https://aka.ms/new-console-template for more information

	using ConsoleApp7;

	var cts = new CancellationTokenSource();
	var ch = new BoundedBuffer<string>(2, cts.Token);

	var consumers = Enumerable.Range(0,3).Select(i=> Task.Run(async () =>
	{
	while (!cts.Token.IsCancellationRequested)
	{
	try
	{
	var t = await ch.Get(cts.Token);
	Console.WriteLine($"Recieved {t} by {i}");
	}
	catch (Exception)
	{
	}

	}
	})).ToArray();
	var producers = Enumerable.Range(0, 3).Select(producer => Task.Run(async () =>
	{
	var rnd = new Random();
	for (int i = 0; i < 100; i++)
	{
	//await Task.Delay(TimeSpan.FromMilliseconds(rnd.Next(3)));
	await ch.Put($"Message {i} from {producer}", cts.Token);
	}
	})).ToArray();
	await Task.WhenAll(producers);
	cts.Cancel();
	await Task.WhenAll(consumers);

	Console.WriteLine("Finished");