Feedback

C# - Parallelisierung von Schleifen

Veröffentlicht von am 07.11.2008
(4 Bewertungen)
Microsoft bietet in .net 4.0 die Parallel Extension an. Diese sind bereits als Preview verfügbar.

Hier ist ein selbstgemachte Möglichkeit for-Schleifen zu parallelisieren um den Schleifenkörper auf mehrere CPUs aufzuteilen.

Dies soll allerdings nur verwendet werden wenn der Schleifkörper für jeden Durchlauf unabhängige Ergebnisse liefert. D.h. das Ergebnis bezieht sich nicht auf einen anderen Durchlauf der Schleife.

Ein Beispiel wäre die Matrix-Multiplikation.
public static void MatmulParallel(double[,] a, double[,] b, double[,] c)
{
int s = a.GetLength(0);

gfoidl.Parallel.For(0, s, i => // Lambda-Ausdruck
{
for (int j = 0; j < s; j++)
{
double v = 0;

for (int k = 0; k < s; k++)
v += a[i, k] * b[k, j];

c[i, j] = v;
}
});
}
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
	}
}
Abgelegt unter Thread, Parallel.

6 Kommentare zum Snippet

Sperneder Patrick schrieb am 15.11.2008:
Klingt wirklich interessant..

Hast du ausser der Matrix-Multiplikation eine etwas weltlichere Anwendung dieser Schleife(n) ?
Besteht eine Möglichkeit dies auch unter .NET 3.5 zu verwenden ?

Günther Foidl schrieb am 15.11.2008:
Es funktioniert mit .net 3.5.

Für mich ist die Marix-Multiplikaton durchaus eine weltliche Anwendung und sie ist das Paradebeispiel für Parallelisierung.

Anderes Bsp.:
Du hast eine (große) Liste von Objekten. In einer Schleife soll für jedes Objekt eine Operation durchgeführt werden. Die durchzuführende Operation bezieht sich nur auf das jeweilige Objekte und hat keine Auswirkungen auf die anderen Objekte der Liste.
Die Leistung der Schleife kann verbessert werden wenn dies parallelisiert wird.
Sperneder Patrick schrieb am 15.11.2008:
Hab mir gerade die ParallelExtensions CTP von MS runtergeladen...
->
http://www.microsoft.com/downloads/details.aspx?FamilyId=348F73FD-593D-4B3C-B055-694C50D2B0F3&displaylang=en

wie gesagt, sieht wirklich interessant aus.
Ich muss nur noch einen Anwendungsfall dafür finden und versuch mal wie's mit
der Leistungsverbesserung aussieht.
Günther Foidl schrieb am 15.11.2008:
Bei vielen Schleifendurchläufen ist durchaus eine enorme Leistungsverbesserung vorhanden. Dabei ist meine Variante sogar schneller als die von MS. Der Grund liegt wohl darin dass meine Lösung nur für die Paralleslisierung von Schleifen ist und die MS-Lösung eher von allgemeiner Natur. D.h. bei MS ist mehr Overhead dabei und somit ein wenig langsamer. Bei Riesen-Schleifen geht der Unterschied aber verloren.
Desweiteren ist die MS-Lösung momentan eine Vorschau. Dies ist auch in der Doku dazu vermerkt. Bis zur endgültigen Version wird sich die Leistung der MS-Lösung sicherlich noch verbessern.
Sperneder Patrick schrieb am 16.11.2008:
Hast du Lust ein Forum - Thema daraus zu machen ?
Ich denke mir es gibt bestimmt noch mehrere interessante Fragen und Ansichten zu dem Thema.
Günther Foidl schrieb am 16.11.2008:
Hab unter http://dotnet-forum.de/forums/p/1040/3739.aspx die Diskussion eingerichtet.

Bitte alle weiteren Kommentare, etc. dort posten.
Danke!
 

Logge dich ein, um hier zu kommentieren!