using System; using System.Security.Principal; using System.Collections; using System.Threading; using log4net; namespace Novell.Zenworks.Zmd { internal class PoolLog { private static readonly ILog log = LogManager.GetLogger (typeof (ZenThread)); public static void Debug (string msg, params object[] args) { if (Environment.GetEnvironmentVariable ("ZMD_THREAD_POOL_DEBUG") == null) return; log.Info (String.Format (msg, args)); } } internal struct ThreadInfo { public ThreadStart ThreadStart; public IPrincipal Principal; public WaitCallback WaitCallback; public object UserData; public static ThreadInfo Zero { get { return new ThreadInfo (); } } } internal class ZenThread { private static readonly ILog log = LogManager.GetLogger (typeof (ZenThread)); private static int nextId = 500; private Thread thread; private ThreadInfo tinfo; private bool running; private DateTime finishTime = DateTime.MinValue; private int id; public event EventHandler Finished; public int Id { get { return id; } } public DateTime FinishTime { get { return finishTime; } } public ZenThread () { id = nextId++; running = true; thread = new Thread (new ThreadStart (WorkerLoop)); thread.IsBackground = true; thread.Start (); } public void RunThread (ThreadInfo tinfo) { lock (this) { PoolLog.Debug ("Thread {0}: got new worker", id); this.tinfo = tinfo; Monitor.Pulse (this); } } public void Stop () { lock (this) { PoolLog.Debug ("Thread {0}: stopping", id); running = false; Monitor.Pulse (this); } } private void WorkerLoop () { while (running) { lock (this) { if (tinfo.Equals (ThreadInfo.Zero)) { PoolLog.Debug ("Thread {0}: waiting on job...", id); Monitor.Wait (this); } if (!running) { PoolLog.Debug ("Thread {0}: terminating", id); break; } // set the principal if (tinfo.Principal != null) { Thread.CurrentPrincipal = tinfo.Principal; } try { PoolLog.Debug ("Thread {0}: running job...", id); if (tinfo.ThreadStart != null) { tinfo.ThreadStart (); } else { tinfo.WaitCallback (tinfo.UserData); } } catch (Exception e) { log.Info ("Unhandled exception: " + e); } finally { PoolLog.Debug ("Thread {0}: job finished", id); tinfo = ThreadInfo.Zero; finishTime = DateTime.Now; if (Finished != null) Finished (this, new EventArgs ()); } } } } } public class ZenThreadPool { private static TimeSpan ThreadTimeout = TimeSpan.FromMinutes (10); private ArrayList readyThreads = new ArrayList (); private ArrayList activeThreads = new ArrayList (); private Queue queue = new Queue (); private int max = 50; private int min = 3; private static ZenThreadPool sharedPool; public static ZenThreadPool SharedPool { get { lock (typeof (ZenThreadPool)) { if (sharedPool == null) sharedPool = new ZenThreadPool (); return sharedPool; } } } public int MaxThreads { get { return max; } set { max = value; } } public int MinThreads { get { return min; } set { min = value; } } private int TotalThreads { get { lock (readyThreads) { return readyThreads.Count + activeThreads.Count; } } } public ZenThreadPool () : this (3, 50) { } public ZenThreadPool (int min, int max) { this.min = min; this.max = max; for (int i = 0; i < min; i++) { AddThread (ThreadInfo.Zero); } } private void AddThread (ThreadInfo tinfo) { lock (readyThreads) { ZenThread thread = new ZenThread (); PoolLog.Debug ("Added a new thread ({0}), total count is {1}", thread.Id, (TotalThreads + 1)); thread.Finished += OnThreadFinished; if (!tinfo.Equals (ThreadInfo.Zero)) { activeThreads.Add (thread); thread.RunThread (tinfo); } else { readyThreads.Add (thread); } } } private void OnThreadFinished (object o, EventArgs args) { lock (readyThreads) { ZenThread thread = (ZenThread) o; if (queue.Count == 0) { activeThreads.Remove (thread); readyThreads.Add (thread); return; } PoolLog.Debug ("Dequeuing job..."); ThreadInfo tinfo = (ThreadInfo) queue.Dequeue (); thread.RunThread (tinfo); } } public void RunThread (WaitCallback cb) { RunThread (cb, null); } public void RunThread (WaitCallback cb, object data) { RunThread (cb, data, null); } public void RunThread (WaitCallback cb, object data, IPrincipal principal) { ThreadInfo tinfo = new ThreadInfo (); tinfo.WaitCallback = cb; tinfo.UserData = data; tinfo.Principal = principal; RunThread (tinfo); } public void RunThread (ThreadStart start) { RunThread (start, Thread.CurrentPrincipal); } public void RunThread (ThreadStart start, IPrincipal principal) { ThreadInfo tinfo = new ThreadInfo (); tinfo.ThreadStart = start; tinfo.Principal = principal; RunThread (tinfo); } private void RunThread (ThreadInfo tinfo) { lock (readyThreads) { CleanupThreads (); // find an unused thread if possible if (readyThreads.Count > 0) { ZenThread thread = (ZenThread) readyThreads[0]; readyThreads.Remove (thread); activeThreads.Add (thread); PoolLog.Debug ("Running job in inactive thread {0}", thread.Id); thread.RunThread (tinfo); return; } // no free threads, see if we can create a new one if (TotalThreads < max) { PoolLog.Debug ("Running job on new thread"); AddThread (tinfo); return; } PoolLog.Debug ("Queueing new job"); // can't create a new one, queue it up queue.Enqueue (tinfo); } } private void CleanupThreads () { lock (readyThreads) { foreach (ZenThread thread in readyThreads.ToArray (typeof (ZenThread))) { if (TotalThreads == min) break; if (DateTime.Now - thread.FinishTime > ThreadTimeout || TotalThreads > max) { PoolLog.Debug ("Stopping thread {0}", thread.Id); thread.Stop (); readyThreads.Remove (thread); } } } } } }