// // ConcurrentExclusiveSchedulerPair.cs // // Authors: // Jérémie "Garuma" Laval // Marek Safar // // Copyright (c) 2011 Jérémie "Garuma" Laval // Copyright 2012 Xamarin, Inc (http://www.xamarin.com) // // Permission is hereby granted, free of charge, to any person obtaining a copy // of this software and associated documentation files (the "Software"), to deal // in the Software without restriction, including without limitation the rights // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell // copies of the Software, and to permit persons to whom the Software is // furnished to do so, subject to the following conditions: // // The above copyright notice and this permission notice shall be included in // all copies or substantial portions of the Software. // // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN // THE SOFTWARE. // // #if NET_4_5 using System; using System.Threading; using System.Collections.Generic; using System.Collections.Concurrent; using System.Diagnostics; namespace System.Threading.Tasks { [DebuggerDisplay ("Concurrent={ConcurrentTaskCount}, Exclusive={ExclusiveTaskCount}")] [DebuggerTypeProxy (typeof (SchedulerDebuggerView))] public class ConcurrentExclusiveSchedulerPair { sealed class SchedulerDebuggerView { readonly ConcurrentExclusiveSchedulerPair owner; public SchedulerDebuggerView (ConcurrentExclusiveSchedulerPair owner) { this.owner = owner; } public IEnumerable ScheduledConcurrent { get { return owner.concurrentTasks; } } public IEnumerable ScheduledExclusive { get { return owner.exclusiveTasks; } } public TaskScheduler TargetScheduler { get { return owner.target; } } } readonly int maxConcurrencyLevel; readonly int maxItemsPerTask; readonly TaskScheduler target; readonly TaskFactory factory; readonly Action taskHandler; readonly ConcurrentQueue concurrentTasks = new ConcurrentQueue (); readonly ConcurrentQueue exclusiveTasks = new ConcurrentQueue (); //readonly ReaderWriterLockSlim rwl = new ReaderWriterLockSlim (); readonly TaskCompletionSource completion = new TaskCompletionSource (); readonly InnerTaskScheduler concurrent; readonly InnerTaskScheduler exclusive; int numTask; class InnerTaskScheduler : TaskScheduler { readonly ConcurrentExclusiveSchedulerPair scheduler; readonly ConcurrentQueue queue; public InnerTaskScheduler (ConcurrentExclusiveSchedulerPair scheduler, ConcurrentQueue queue) { this.scheduler = scheduler; this.queue = queue; } public int TaskCount { get { return queue.Count; } } public override int MaximumConcurrencyLevel { get { return scheduler.maxConcurrencyLevel; } } protected internal override void QueueTask (Task t) { scheduler.DoQueue (t, queue); } protected override bool TryExecuteTaskInline (Task task, bool taskWasPreviouslyQueued) { if (task.Status != TaskStatus.Created) return false; task.RunSynchronously (scheduler.target); return true; } public void Execute (Task t) { TryExecuteTask (t); } [MonoTODO ("Only useful for debugger support")] protected override IEnumerable GetScheduledTasks () { throw new NotImplementedException (); } } public ConcurrentExclusiveSchedulerPair () : this (TaskScheduler.Current) { } public ConcurrentExclusiveSchedulerPair (TaskScheduler taskScheduler) : this (taskScheduler, taskScheduler.MaximumConcurrencyLevel) { } public ConcurrentExclusiveSchedulerPair (TaskScheduler taskScheduler, int maxConcurrencyLevel) : this (taskScheduler, maxConcurrencyLevel, -1) { } public ConcurrentExclusiveSchedulerPair (TaskScheduler taskScheduler, int maxConcurrencyLevel, int maxItemsPerTask) { this.target = taskScheduler; this.maxConcurrencyLevel = maxConcurrencyLevel; this.maxItemsPerTask = maxItemsPerTask; this.factory = new TaskFactory (taskScheduler); this.taskHandler = InternalTaskProcesser; this.concurrent = new InnerTaskScheduler (this, concurrentTasks); this.exclusive = new InnerTaskScheduler (this, exclusiveTasks); } public void Complete () { completion.SetResult (null); } public TaskScheduler ConcurrentScheduler { get { return concurrent; } } private int ConcurrentTaskCount { get { return concurrent.TaskCount; } } public TaskScheduler ExclusiveScheduler { get { return exclusive; } } private int ExclusiveTaskCount { get { return exclusive.TaskCount; } } public Task Completion { get { return completion.Task; } } void DoQueue (Task task, ConcurrentQueue queue) { queue.Enqueue (task); SpinUpTasks (); } void InternalTaskProcesser () { int times = 0; // const int lockWaitTime = 2; while (!concurrentTasks.IsEmpty || !exclusiveTasks.IsEmpty) { if (maxItemsPerTask != -1 && ++times == maxItemsPerTask) break; throw new NotImplementedException (); /* bool locked = false; try { if (!concurrentTasks.IsEmpty && rwl.TryEnterReadLock (lockWaitTime)) { locked = true; Task task; while (concurrentTasks.TryDequeue (out task)) { RunTask (task); } } } finally { if (locked) { rwl.ExitReadLock (); locked = false; } } try { if (!exclusiveTasks.IsEmpty && rwl.TryEnterWriteLock (lockWaitTime)) { locked = true; Task task; while (exclusiveTasks.TryDequeue (out task)) { RunTask (task); } } } finally { if (locked) { rwl.ExitWriteLock (); } } */ } // TODO: there's a race here, task adding + spinup check may be done while here Interlocked.Decrement (ref numTask); } void SpinUpTasks () { int currentTaskNumber; do { currentTaskNumber = numTask; if (currentTaskNumber >= maxConcurrencyLevel) return; } while (Interlocked.CompareExchange (ref numTask, currentTaskNumber + 1, currentTaskNumber) != currentTaskNumber); factory.StartNew (taskHandler); } void RunTask (Task task) { concurrent.Execute (task); } } } #endif