Rework threading

Author: kgiszewski

Threading/IManageThreads.cs

@@ -1,10 +1,13 @@
-namespace Threading
+using System;
+
+namespace Threading
 {
     public interface IManageThreads
     {
         void Queue(object o);
-        void ProcessQueue();
+        void ProcessQueue(Action<object> action, int maxThreads);
         int NumberThreadsRunning { get; }
         int MaxConcurrentThreads { get; }
+        void RemoveThread();
     }
 }

Threading/Program.cs

@@ -6,50 +6,63 @@ namespace Threading
 {
     class Program
     {
+        private static readonly IManageThreads _threadManager = new ThreadManager();
+        private static int _mainThreadId;
+
         static void Main(string[] args)
         {
-            Console.WriteLine($"Main thread Id: {Thread.CurrentThread.ManagedThreadId}");
+            var rand = new Random();
 
-            var mySomethingDoer = new DoSomething
-            {
-                Name = "Thing 1",
-                SleepFactor = 100
-            };
+            _mainThreadId = Thread.CurrentThread.ManagedThreadId;
 
-            var mySomethingDoer2 = new DoSomething
+            Console.WriteLine($"Main thread Id: {_mainThreadId}");
+            
+            //queue up 10 things that need done
+            for (var i = 1; i < 10; i++)
             {
-                Name = "Thing 2",
-                SleepFactor = 1000
-            };
+                _threadManager.Queue(new DoSomething
+                {
+                   Id = i,
+                   SleepFactor = rand.Next(1, 5) * 1000
+                });
+            }
+            
+            //work thru the queue, use 2 max threads
+            _threadManager.ProcessQueue(_workerBee, 5);
 
-            var mySomethingDoer3 = new DoSomething
+            //since work is outside of this thread, we have to wait for the other threads to finish
+            while (_threadManager.NumberThreadsRunning > 0)
             {
-                Name = "Thing 3",
-                SleepFactor = 1000
-            };
+                
+            }
 
-            var mySomethingDoer4 = new DoSomething
-            {
-                Name = "Thing 4",
-                SleepFactor = 100
-            };
+            Console.WriteLine("All threads complete.");
+            Console.WriteLine($"Current number of threads: {_threadManager.NumberThreadsRunning}, Max Threads: {_threadManager.MaxConcurrentThreads}");
+            Console.ReadKey();
+        }
 
-            var threadManager = new ThreadManager();
+        //this is the method that does the work
+        //many threads will be in this method at the same time, so don't assume you get to do whatever you want with out locking
+        //work inside the DoSomething class should be thread-safe already
+        private static void _workerBee(object o)
+        {
+            //illustrating that the main thread and this method should always be different threads
+            if (_mainThreadId == Thread.CurrentThread.ManagedThreadId)
+            {
+                throw new Exception("This shouldn't be the case!");
+            }
 
-            threadManager.Queue(mySomethingDoer);
-            threadManager.Queue(mySomethingDoer2);
-            threadManager.Queue(mySomethingDoer3);
-            threadManager.Queue(mySomethingDoer4);
+            //the thread manager is using objects, so we need to cast
+            var myDoerClass = o as DoSomething;
 
-            threadManager.ProcessQueue();
+            myDoerClass?.Process();
 
-            while (threadManager.NumberThreadsRunning > 0)
-            {
-                
-            }
+            //this could be done differently if we choose to want to remove it
+            //perhaps raise an event that the thread manager subscribes to
+            //otherwise I don't see an easy way for the thread to be removed from the manager
+            _threadManager.RemoveThread();
 
-            Console.WriteLine("All threads complete.");
-            Console.ReadKey();
+            Thread.Sleep(5000);
         }
     }
 }

Threading/ThreadManager.cs

@@ -1,53 +1,80 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
-using Threading.Work;
 
 namespace Threading
 {
     public class ThreadManager : IManageThreads
     {
-        private readonly List<int> _currentRunningThreads = new List<int>();
+        //let's keep track of our threads
+        private readonly ConcurrentDictionary<int, Thread> _currentRunningThreads = new ConcurrentDictionary<int, Thread>();
+
+        //let's keep track of our work to do
         private readonly Queue<object> _workQueue = new Queue<object>();
 
+        //helper to add a new worker
         public void Queue(object o)
         {
             _workQueue.Enqueue(o);
         }
 
-        public void ProcessQueue()
+        //process the queue
+        //essentially this is the brains that spawns a new thread
+        //this runs on the main thread
+        public void ProcessQueue(Action<object> action, int maxThreads)
         {
+            //throttle the work
+            MaxConcurrentThreads = maxThreads;
+
+            //if anything to do...
             while (_workQueue.Any())
             {
-                Console.WriteLine($"Work queue items: {_workQueue.Count}");
-
+                //using a count on a concurrent dictionary should get us the right count (I hope)
+                //limit to the max threads
                 if (_currentRunningThreads.Count < MaxConcurrentThreads)
                 {
+                    Console.WriteLine($"Work queue items: {_workQueue.Count}");
+
+                    //grab some work off the top of the queue
                     var work = _workQueue.Dequeue();
-                    var thread = new Thread(WorkerBee);
 
-                    _currentRunningThreads.Add(thread.ManagedThreadId);
+                    //define a new thread
+                    var thread = new Thread(new ParameterizedThreadStart(action));
+
+                    //try to add it to the dictionary
+                    //maybe we should add some logic here in case the dictionary doesn't let us
+                    _currentRunningThreads.TryAdd(thread.ManagedThreadId, thread);
+
+                    Console.WriteLine($"Spawning new thread: {thread.ManagedThreadId}...");
+                    Console.WriteLine($"Current number of threads: {_currentRunningThreads.Count}, Max Threads: {MaxConcurrentThreads}");
+
+                    //add some delay so that the console messages slowly crawl
+                    Thread.Sleep(5000);
 
+                    //begin the work
                     thread.Start(work);
                 }
                 else
                 {
-                    Console.WriteLine($"Waiting for a thread to come free...");
+                    //guess we should wait until we get a thread to use
+                    //Console.WriteLine($"Waiting for a thread to come free...");
                     Thread.Sleep(1000);
                 }
             }
         }
 
         public int NumberThreadsRunning => _currentRunningThreads.Count;
-        public int MaxConcurrentThreads => 2;
-        private void WorkerBee(object o)
+        public int MaxConcurrentThreads { get; set; }
+        public void RemoveThread()
         {
-            var myDoerClass = o as DoSomething;
+            Thread t;
 
-            myDoerClass?.Process();
+            //take the thread out of the mix, this might need better error handling
+            _currentRunningThreads.TryRemove(Thread.CurrentThread.ManagedThreadId, out t);
 
-            _currentRunningThreads.Remove(Thread.CurrentThread.ManagedThreadId);
+            Console.WriteLine($"Ending thread Id: {t.ManagedThreadId}...");
         }
     }
 }

Threading/Work/DoSomething.cs

@@ -5,20 +5,20 @@ namespace Threading.Work
 {
     public class DoSomething : IDoSomething
     {
-        private readonly Random _rand = new Random();
-        public string Name { get; set; }
+        //just a placeholder class that is unaware that it's in a multithreaded env
+        public int Id { get; set; }
         public int SleepFactor { get; set; }
 
         public void Process()
         {
             for (var i = 0; i < 10; i++)
             {
-                Console.WriteLine($"Thread Id: {Thread.CurrentThread.ManagedThreadId} Processing: {Name} - {i + 1}");
+                Console.WriteLine($"Thread Id: {Thread.CurrentThread.ManagedThreadId} Processing Object: {Id} Task: {i + 1}/10");
 
-                Thread.Sleep(_rand.Next(1, 5) * SleepFactor);
+                Thread.Sleep(SleepFactor);
             }
 
-            Console.WriteLine($"Thread Id: {Thread.CurrentThread.ManagedThreadId} {Name} is complete.");
+            Console.WriteLine($"Thread Id: {Thread.CurrentThread.ManagedThreadId} Object {Id} is complete.");
         }
     }
 }

Threading/Work/IDoSomething.cs

@@ -2,7 +2,7 @@
 {
     public interface IDoSomething
     {
-        string Name { get; set; }
+        int Id { get; set; }
         int SleepFactor { get; set; }
         void Process();
     }