Mercurial > hg > ucis.core

comparison Util/WorkQueue.cs @ 66:b7bc27c6734e

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Added WorkQueue class
author Ivo Smits <Ivo@UCIS.nl>
date Tue, 15 Oct 2013 16:22:54 +0200
parents
children 5d1b72ba44dc
comparison
equal deleted inserted replaced
65:abe0d55a2201 66:b7bc27c6734e
1 using System;
2 using System.Collections.Generic;
3 using System.Threading;
4 using System.Windows.Forms;
5 using SysThreadPool = System.Threading.ThreadPool;
6
7 namespace UCIS.Util {
8 public class WorkQueue : WorkQueue<MethodInvoker> {
9 public WorkQueue() : base(Handler) { }
10 private static void Handler(MethodInvoker item) { item(); }
11 }
12 public class WorkQueue<TWork> : IDisposable {
13 Queue<TWork> queue = new Queue<TWork>();
14 Action<TWork> callback = null;
15 int maxIdleWorkers = 0;
16 int maxWorkers = 1;
17 int idleWorkers = 0;
18 int workers = 0;
19
20 public Boolean UseFrameworkThreadpool { get; set; }
21
22 public WorkQueue(Action<TWork> callback) {
23 this.callback = callback;
24 UseFrameworkThreadpool = true;
25 }
26 public void Dispose() {
27 maxWorkers = 0;
28 lock (queue) Monitor.PulseAll(queue);
29 }
30
31 public int MaxIdleWorkers {
32 get { return maxIdleWorkers; }
33 set {
34 maxIdleWorkers = value;
35 lock (queue) Monitor.PulseAll(queue);
36 }
37 }
38 public int MaxWorkers {
39 get { return maxWorkers; }
40 set {
41 maxWorkers = value;
42 lock (queue) Monitor.PulseAll(queue);
43 }
44 }
45 public int TotalWorkers { get { return workers; } }
46 public int IdleWorkers { get { return idleWorkers; } }
47
48 public void Enqueue(TWork item) {
49 lock (queue) {
50 queue.Enqueue(item);
51 Monitor.Pulse(queue);
52 if (workers < maxWorkers && idleWorkers == 0) StartWorker();
53 }
54 }
55 public void Clear() { lock (queue) queue.Clear(); }
56 public int Count { get { lock (queue) return queue.Count; } }
57
58 private void StartWorker() {
59 lock (queue) {
60 if (workers >= maxWorkers) return;
61 if (UseFrameworkThreadpool) {
62 SysThreadPool.QueueUserWorkItem(Worker);
63 } else {
64 (new Thread(Worker)).Start();
65 }
66 workers++;
67 }
68 }
69 private void RaiseEvent(Action<WorkQueue<TWork>> callback) {
70 if (callback != null) callback(this);
71 }
72 private void Worker(Object state) {
73 while (true) {
74 TWork item;
75 lock (queue) {
76 if (workers >= maxWorkers) {
77 workers--;
78 break;
79 }
80 if (queue.Count == 0) {
81 if (idleWorkers >= maxIdleWorkers) {
82 workers--;
83 queue.TrimExcess();
84 break;
85 }
86 idleWorkers++;
87 Monitor.Wait(queue);
88 idleWorkers--;
89 if (queue.Count == 0) continue;
90 }
91 item = queue.Dequeue();
92 }
93 callback(item);
94 }
95 }
96 }
97 }