using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

using System.Threading;

using System.Threading.Tasks;

​

public class Program

{

    public static void Main()

    {

        CtsTest ctsTest = new CtsTest();

​

        // fill the queue with 200 jobs

        for (int i = 0; i < 200; i++) ctsTest.QueueJob();

​

        // start the worker

        Task.Run(() => ctsTest.ProcessQueueAsync(CancellationToken.None));

​

        // Console.Read();

​

    }

​

    internal class CtsTest

    {

​

        public class ProcessDescriptor

        {

            public Task Task { get; set; }

            public JobDescriptor Descriptor { get; set; }

            public CancellationTokenSource Cts { get; set; }

        }

​

        public class JobDescriptor

        {

            public string ProcessId { get; set; }

        }

​

        private List<ProcessDescriptor> _processQueue = new List<ProcessDescriptor>(); // running task queue

        private ConcurrentDictionary<string, JobDescriptor> _waitQueue = new ConcurrentDictionary<string, JobDescriptor>(); // jobs to be done queue

​

​

        public async Task ProcessQueueAsync(CancellationToken ct)

        {

            await Task.Yield();

​

            while (!ct.IsCancellationRequested)

            {

                try

                {

                    // remove completed jobs from the process queue

                    // in the "real" app there are checks for other states too, but it's not relevant for repro

                    foreach (var process in _processQueue.ToList())

                    {

                        if (process.Task.Status == TaskStatus.RanToCompletion)

                        {

                            // uncomment next line for tests 2 & 3

                            // process.CancellationTokenSource.Dispose();

                            _processQueue.Remove(process);

                        }

                    }

​

​

                    // get the next jobs from wait queue and run them

                    var runningProcesses = _processQueue.Select(f => f.Descriptor.ProcessId).Distinct();

                    var jobsToRun = _waitQueue.Where(f => !runningProcesses.Contains(f.Key))

                        .Take(20 - _processQueue.Count).ToList();

​

                    foreach (var job in jobsToRun)

                    {

                        var runProcess = new ProcessDescriptor()

                        {

                            Descriptor = job.Value,

                            // force cancellation after 5 mins

                            Cts = new CancellationTokenSource(TimeSpan.FromMinutes(5))

                        };

​

                        // link cts with "global" token (ct)

                        var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(ct, runProcess.Cts.Token);

​

                        // The job cannot be awaited. We want to have jobs running in parallel.

​

                        // use the next line for "benchmark" & test 3 - TOKEN is passed and will not be not disposed 

                        runProcess.Task = RunJob1(job.Value, linkedCts.Token);

​

                        // use the next line tests 1 & 2 - CTS is passed and will be disposed (probably not best practice, but OK for testing)

                        //runProcess.Task = RunJob2(job.Value, linkedCts);

​

                        _processQueue.Add(runProcess);

                    }

​

                }

                catch (Exception)

                {

                    // not relevant for repro

                }

                await Task.Delay(1000);

            }

​

            Console.WriteLine("Worker stopped");

        }

​

        private async Task RunJob1(JobDescriptor descriptor, CancellationToken cancellationToken)

        {

            await Task.Delay(3000, cancellationToken);

            Console.WriteLine("Job done: " + descriptor.ProcessId);

​

            // remove job from queue, without disposing

            _waitQueue.TryRemove(descriptor.ProcessId, out descriptor);

            return;

        }

​

        // takes the cts and not the token, being able to dispose cts

        private async Task RunJob2(JobDescriptor descriptor, CancellationTokenSource cancellationTokenSource)

        {

            await Task.Delay(3000, cancellationTokenSource.Token);

            Console.WriteLine("Job done: " + descriptor.ProcessId);

​

            if (_waitQueue.TryRemove(descriptor.ProcessId, out descriptor))

            {

                cancellationTokenSource.Dispose();

            }

            return;

        }

​

​

        public void QueueJob()

        {

            string processId = Guid.NewGuid().ToString();

            JobDescriptor item = new JobDescriptor();

            item.ProcessId = processId;

            _waitQueue.AddOrUpdate(processId, item, (s, updaterDescriptor) => item);

            Console.WriteLine("Job added to queue: " + processId);

        }

    }

}