C# - Parallelisierung von Schleifen

using System;
using System.Threading;

namespace gfoidl
{
	/// <summary>
	/// Unterstützung für parallele Ausführung
	/// </summary>
	/// <remarks>
	/// Diese Klasse erlaubt die Parallelisierung von Schleifen.
	/// Diese werden in separaten Threads ausgeführt was 
	/// wiederum die Möglichkeit gibt diese simultan auf
	/// verschiedene CPUs/Kerne auszuführen
	/// </remarks>
	public sealed class Parallel
	{
		/// <summary>
		/// Delegate für den Schleifenkörper
		/// </summary>
		/// <param name="index">
		/// Schleifenindex
		/// </param>
		public delegate void SchleifenKörper(int index);
		//---------------------------------------------------------------------
		#region Felder
		private static int _anzahlThreads = System.Environment.ProcessorCount;

		// Wird für Synchronisation der Threads benötigt:
		private static object _sync = new object();

		// Wird für Singleton-Pattern benötigt:
		private static volatile Parallel _instanz = null;

		// Hintergrundthreads für parallele Ausführung:
		private Thread[] _threads = null;

		// Ereignisse um Job-Verfügbarkeit und Thread-Verfügbarkeit zu
		// signalisieren:
		private AutoResetEvent[] _jobVerfügbar = null;
		private ManualResetEvent[] _threadLeerlauf = null;

		// Schleife:
		private int _aktuellerIndex;
		private int _stopIndex;
		private SchleifenKörper _schleifenKörper;
		#endregion
		//---------------------------------------------------------------------
		#region Eigenschaften
		/// <summary>
		/// <para>
		/// Anzahl der Threads die für die parallele Ausführung verwendet
		/// werden.
		/// </para>
		/// <para>
		/// Standardmäßig wird die Anzahl auf die Anzahl der CPUs im
		/// System gesetzt (siehe auch
		/// <see cref="System.Environment.ProcessorCount"/>).
		/// </para>
		/// </summary>
		public static int ThreadAnzahl
		{
			get { return _anzahlThreads; }
			set
			{
				lock (_sync)
				{
					_anzahlThreads = Math.Max(1, value);
				}
			}
		}
		//---------------------------------------------------------------------
		private static Parallel Instanz
		{
			get
			{
				if (_instanz == null)
				{
					_instanz = new Parallel();
					_instanz.Initiieren();
				}
				else
				{
					if (_instanz._threads.Length != _anzahlThreads)
					{
						// Alte Threads terminieren:
						_instanz.Terminieren();

						// Reinitiieren:
						_instanz.Initiieren();
					}
				}

				return _instanz;
			}
		}
		#endregion
		//---------------------------------------------------------------------
		#region Konstruktor
		/// <summary>
		/// Privater Konstruktor um Instanziierung der Klasse zu vermeiden
		/// (Singleton).
		/// </summary>
		private Parallel() { }
		#endregion
		//---------------------------------------------------------------------
		#region Methoden
		/// <summary>
		/// For-Schleife in der die Ausführung parallel stattfindet
		/// </summary>
		/// <param name="start">
		/// Startindex
		/// </param>
		/// <param name="ende">
		/// Obergrenze des Index (exklusiv)
		/// </param>
		/// <param name="schleifenKörper">
		/// Schleifenkörper
		/// </param>
		/// <remarks>
		/// <para>
		/// Diese Methode wird benötigt um die For-Schleife parallet 
		/// in verschiedenen Threads auszuführen.
		/// </para>
		/// <para>
		/// Die Anzahl der Iterationen entspricht <paramref name="ende"/>
		/// - <paramref name="start"/> - 1.
		/// </para>
		/// </remarks>
		/// <example>
		/// <code>
		/// Parallel.For(0, 20, delegate(int i)
		/// // Äquivalent zu:
		/// // for (int i = 0; i &lt; 20; i++)
		/// {
		///		Console.WriteLine(i);
		/// });
		/// </code>
		/// </example>
		public static void For(
			int start,
			int ende,
			SchleifenKörper schleifenKörper)
		{
			lock (_sync)
			{
				// Instanz des Managers holen:
				Parallel instanz = Instanz;

				instanz._aktuellerIndex = start - 1;
				instanz._stopIndex = ende;
				instanz._schleifenKörper = schleifenKörper;

				// Allen Threads signalisieren dass ein Job verfügbar ist
				// und sie als beschäftigt markieren:
				for (int i = 0; i < _anzahlThreads; i++)
				{
					instanz._threadLeerlauf[i].Reset();
					instanz._jobVerfügbar[i].Set();
				}

				// Warten bis alle Threads im Leerlauf sind:
				for (int i = 0; i < _anzahlThreads; i++)
				{
					instanz._threadLeerlauf[i].WaitOne();
				}
			}
		}
		//---------------------------------------------------------------------
		/// <summary>
		/// Initiiert die Instanz der Klasse (Singleton) die benötigt wird
		/// um die Anzahl der Threads zu erstellen und diese zu 
		/// synchronisieren
		/// </summary>
		private void Initiieren()
		{
			// Ereignis-Array der über verfügbare Jobs signalisiert:
			_jobVerfügbar = new AutoResetEvent[_anzahlThreads];

			// Ereignis-Array der über verfügbaren Thread signalisiert:
			_threadLeerlauf = new ManualResetEvent[_anzahlThreads];

			// Thread-Array:
			_threads = new Thread[_anzahlThreads];

			for (int i = 0; i < _anzahlThreads; i++)
			{
				_jobVerfügbar[i] = new AutoResetEvent(false);
				_threadLeerlauf[i] = new ManualResetEvent(true);

				_threads[i] = new Thread(
					new ParameterizedThreadStart(ArbeiterThread));
				_threads[i].IsBackground = true;
				_threads[i].Start(i);
			}
		}
		//---------------------------------------------------------------------
		/// <summary>
		/// Terminiert alle Threads und schließt die Synchronisations-Objekte
		/// </summary>
		private void Terminieren()
		{
			// Thread beenden indem der Schleifenkörper auf null 
			// gesetzt wird und verfügbare Arbeit signalisiert wird:
			_schleifenKörper = null;

			for (int i = 0, anzahlThreads = _threads.Length; i < anzahlThreads; i++)
			{
				_jobVerfügbar[i].Set();

				// Warten bis Thread terminiert ist:
				_threads[i].Join();

				// Ereignisse schließen:
				_jobVerfügbar[i].Close();
				_threadLeerlauf[i].Close();
			}

			// Aufräumen:
			_jobVerfügbar = null;
			_threadLeerlauf = null;
			_threads = null;
		}
		//---------------------------------------------------------------------
		/// <summary>
		/// Führt die parallele Ausführung durch
		/// </summary>
		/// <param name="index">
		/// Zählvariable
		/// </param>
		/// <remarks>
		/// Die Zählvariable ist deshalb als object deklariert da diese
		/// Methode über ParameterizedThreadStart aufgerufen wird. Hier sind
		/// nur object-Argumente erlaubt
		/// </remarks>
		private void ArbeiterThread(object index)
		{
			int threadIndex = (int)index;
			int lokalerIndex = 0;

			while (true)
			{
				// Warten bis ein Job zu tun ist:
				_jobVerfügbar[threadIndex].WaitOne();

				// Ist der Schleifenkörper null -> Schleife verlassen:
				if (_schleifenKörper == null) break;

				while (true)
				{
					// Lokalen Index ermitteln indem der globale 
					// Schleifenzähler inkrementiert wird:
					lokalerIndex = Interlocked.Increment(ref _aktuellerIndex);

					if (lokalerIndex >= _stopIndex) break;

					// Schleifenkörper ausführen:
					_schleifenKörper(lokalerIndex);
				}

				// Verfügbare Threads signalisieren:
				_threadLeerlauf[threadIndex].Set();
			}
		}
		#endregion
	}
}