Task synchronization without UI thread

Question

In the code below, I want to synchronize reporting on the results of a task list. This now works because task.Result blocks until the task completes. However, the id = 3 task takes a long time to complete and blocks all other completed tasks, reporting its status.

I think I can do this by moving the report (Console.Write) to the .ContinueWith statement, but I don't have a user interface thread, so how can I get TaskScheduler to synchronize .ContinueWith tasks?

What I have now:

static void Main(string[] args) { Console.WriteLine("Starting on {0}", Thread.CurrentThread.ManagedThreadId); var tasks = new List<Task<int>>(); for (var i = 0; i < 10; i++) { var num = i; var t = Task<int>.Factory.StartNew(() => { if (num == 3) { Thread.Sleep(20000); } Thread.Sleep(new Random(num).Next(1000, 5000)); Console.WriteLine("Done {0} on {1}", num, Thread.CurrentThread.ManagedThreadId); return num; }); tasks.Add(t); } foreach (var task in tasks) { Console.WriteLine("Completed {0} on {1}", task.Result, Thread.CurrentThread.ManagedThreadId); } Console.WriteLine("End of Main"); Console.ReadKey(); } 

I would like to move on to this or something similar, but I need Console.Write ("Completed ...") so that everything happens in the same thread:

 static void Main(string[] args) { Console.WriteLine("Starting on {0}", Thread.CurrentThread.ManagedThreadId); for (var i = 0; i < 10; i++) { var num = i; Task<int>.Factory.StartNew(() => { if (num == 3) { Thread.Sleep(20000); } Thread.Sleep(new Random(num).Next(1000, 10000)); Console.WriteLine("Done {0} on {1}", num, Thread.CurrentThread.ManagedThreadId); return num; }).ContinueWith(value => { Console.WriteLine("Completed {0} on {1}", value.Result, Thread.CurrentThread.ManagedThreadId); } /* need syncronization context */); } Console.WriteLine("End of Main"); Console.ReadKey(); } 

- SOLUTION - After receiving some comments and reading some solutions, this is a complete solution that does what I want. The goal here is to process long-running tasks as quickly as possible, and then do something with the results of each task one at a time.

 static void Main(string[] args) { Console.WriteLine("Starting on {0}", Thread.CurrentThread.ManagedThreadId); var results = new BlockingCollection<int>(); Task.Factory.StartNew(() => { while (!results.IsCompleted) { try { var x = results.Take(); Console.WriteLine("Completed {0} on {1}", x, Thread.CurrentThread.ManagedThreadId); } catch (InvalidOperationException) { } } Console.WriteLine("\r\nNo more items to take."); }); var tasks = new List<Task>(); for (var i = 0; i < 10; i++) { var num = i; var t = Task.Factory.StartNew(() => { if (num == 3) { Thread.Sleep(20000); } Thread.Sleep(new Random(num).Next(1000, 10000)); Console.WriteLine("Done {0} on {1}", num, Thread.CurrentThread.ManagedThreadId); results.Add(num); }); tasks.Add(t); } Task.Factory.ContinueWhenAll(tasks.ToArray(), _ => results.CompleteAdding()); Console.WriteLine("End of Main"); Console.ReadKey(); } 

Answer 1

You will have to create some kind of task for recording, but remember that even this task can be transferred to another native or managed stream! Using the default scheduler in TPL, you cannot control which managed thread gets the job.

 public class ConcurrentConsole { private static BlockingCollection<string> output = new BlockingCollection<string>(); public static Task CreateWriterTask(CancellationToken token) { return new Task( () => { while (!token.IsCancellationRequested) { string nextLine = output.Take(token); Console.WriteLine(nextLine); } }, token); } public static void WriteLine(Func<string> writeLine) { output.Add(writeLine()); } } 

When I switched my code to use it, I got the following output:

 End of Main Done 1 on 6 Completed 1 on 6 Done 5 on 9 Completed 5 on 9 Done 0 on 4 Completed 0 on 4 Done 2 on 5 Completed 2 on 13 Done 7 on 10 Completed 7 on 10 Done 4 on 8 Completed 4 on 5 Done 9 on 12 Completed 9 on 9 Done 6 on 6 Completed 6 on 5 Done 8 on 11 Completed 8 on 4 Done 3 on 7 Completed 3 on 7 

Even if your code sends () => String.Format("Completed {0} on {1}"... in ConcurrentConsole.WriteLine , make sure ManagedThreadId is found in the ConcurrentConsole Task, it will still change which thread it is in It starts, although with less variability than the tasks being performed.

Answer 2

You can use the OrderedTaskScheduler to ensure that only one job is executed at a time; however, they will run in the thread stream (not necessarily all in one thread).

If you really need everything in the same thread (not just one at a time), you can use ActionThread from the Nito.Async library . It provides a SynchronizationContext for its code, which can be selected from FromCurrentSynchronizationContext .

Answer 3

I would suggest:

1) Creating a lock object
2) Create a list of lines to be written
3) Create a thread that loops, sleeps a bit, then locks the list of lines, and then if it is not empty, write everything down and omit the list
4) Other threads then block the list, add their status, unlock and continue.

 object writeListLocker = new object(); List<string> linesToWrite = new List<string>(); // Main thread loop for (; ; ) { lock (writerListLocker) { foreach (string nextLine in linesToWrite) Console.WriteLine(nextLine); linesToWrite.Clear(); } Thread.Sleep(500); } // Reporting threads lock (writerListLocker) { linesToWrite.Add("Completed (etc.)"); } 

Answer 4

I think you expect a result similar to the following.

 Starting on 8 Done 1 on 11 Completed 1 on 9 Done 5 on 11 Completed 5 on 9 Done 0 on 10 Completed 0 on 9 Done 2 on 12 Completed 2 on 9 Done 7 on 16 Completed 7 on 9 Done 4 on 14 Completed 4 on 9 Done 9 on 18 Completed 9 on 9 Done 6 on 15 Completed 6 on 9 Done 8 on 17 Completed 8 on 9 Done 3 on 13 Completed 3 on 9 

As below, I used the StaSynchronizationContext in my code from SynchronizationContext Overview , where a synchronized call in a single thread is well explained. Please refer to it.

My code snippet:

 static void Main(string[] args) { StaSynchronizationContext context = new StaSynchronizationContext(); StaSynchronizationContext.SetSynchronizationContext(context); Console.WriteLine("Starting on {0}", Thread.CurrentThread.ManagedThreadId); for (var i = 0; i < 10; i++) { var num = i; Task<int>.Factory.StartNew(() => { if (num == 3) { Thread.Sleep(20000); } Thread.Sleep(new Random(num).Next(1000, 10000)); Console.WriteLine("Done {0} on {1}", num, Thread.CurrentThread.ManagedThreadId); return num; }).ContinueWith( value => { Console.WriteLine("Completed {0} on {1}", value.Result, Thread.CurrentThread.ManagedThreadId); } ,TaskScheduler.FromCurrentSynchronizationContext()); } Console.WriteLine("End of Main"); Console.ReadKey(); }