/*
* Copyright (c) Contributors, http://opensimulator.org/
* See CONTRIBUTORS.TXT for a full list of copyright holders.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of the OpenSimulator Project nor the
* names of its contributors may be used to endorse or promote products
* derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE DEVELOPERS ``AS IS'' AND ANY
* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE CONTRIBUTORS BE LIABLE FOR ANY
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
using System;
using System.Collections.Generic;
using System.Threading;
namespace OpenSim.Framework
{
///
/// Provides helper methods for parallelizing loops
///
public static class Parallel
{
private static readonly int processorCount = System.Environment.ProcessorCount;
///
/// Executes a for loop in which iterations may run in parallel
///
/// The loop will be started at this index
/// The loop will be terminated before this index is reached
/// Method body to run for each iteration of the loop
public static void For(int fromInclusive, int toExclusive, Action body)
{
For(processorCount, fromInclusive, toExclusive, body);
}
///
/// Executes a for loop in which iterations may run in parallel
///
/// The number of concurrent execution threads to run
/// The loop will be started at this index
/// The loop will be terminated before this index is reached
/// Method body to run for each iteration of the loop
public static void For(int threadCount, int fromInclusive, int toExclusive, Action body)
{
int counter = threadCount;
AutoResetEvent threadFinishEvent = new AutoResetEvent(false);
Exception exception = null;
--fromInclusive;
for (int i = 0; i < threadCount; i++)
{
ThreadPool.QueueUserWorkItem(
delegate(object o)
{
int threadIndex = (int)o;
while (exception == null)
{
int currentIndex = Interlocked.Increment(ref fromInclusive);
if (currentIndex >= toExclusive)
break;
try { body(currentIndex); }
catch (Exception ex) { exception = ex; break; }
}
if (Interlocked.Decrement(ref counter) == 0)
threadFinishEvent.Set();
}, i
);
}
threadFinishEvent.WaitOne();
threadFinishEvent.Close();
if (exception != null)
throw new Exception(exception.Message, exception);
}
///
/// Executes a foreach loop in which iterations may run in parallel
///
/// Object type that the collection wraps
/// An enumerable collection to iterate over
/// Method body to run for each object in the collection
public static void ForEach(IEnumerable enumerable, Action body)
{
ForEach(processorCount, enumerable, body);
}
///
/// Executes a foreach loop in which iterations may run in parallel
///
/// Object type that the collection wraps
/// The number of concurrent execution threads to run
/// An enumerable collection to iterate over
/// Method body to run for each object in the collection
public static void ForEach(int threadCount, IEnumerable enumerable, Action body)
{
int counter = threadCount;
AutoResetEvent threadFinishEvent = new AutoResetEvent(false);
IEnumerator enumerator = enumerable.GetEnumerator();
Exception exception = null;
for (int i = 0; i < threadCount; i++)
{
ThreadPool.QueueUserWorkItem(
delegate(object o)
{
int threadIndex = (int)o;
while (exception == null)
{
T entry;
lock (enumerator)
{
if (!enumerator.MoveNext())
break;
entry = (T)enumerator.Current; // Explicit typecast for Mono's sake
}
try { body(entry); }
catch (Exception ex) { exception = ex; break; }
}
if (Interlocked.Decrement(ref counter) == 0)
threadFinishEvent.Set();
}, i
);
}
threadFinishEvent.WaitOne();
threadFinishEvent.Close();
if (exception != null)
throw new Exception(exception.Message, exception);
}
///
/// Executes a series of tasks in parallel
///
/// A series of method bodies to execute
public static void Invoke(params Action[] actions)
{
Invoke(processorCount, actions);
}
///
/// Executes a series of tasks in parallel
///
/// The number of concurrent execution threads to run
/// A series of method bodies to execute
public static void Invoke(int threadCount, params Action[] actions)
{
int counter = threadCount;
AutoResetEvent threadFinishEvent = new AutoResetEvent(false);
int index = -1;
Exception exception = null;
for (int i = 0; i < threadCount; i++)
{
ThreadPool.QueueUserWorkItem(
delegate(object o)
{
int threadIndex = (int)o;
while (exception == null)
{
int currentIndex = Interlocked.Increment(ref index);
if (currentIndex >= actions.Length)
break;
try { actions[currentIndex](); }
catch (Exception ex) { exception = ex; break; }
}
if (Interlocked.Decrement(ref counter) == 0)
threadFinishEvent.Set();
}, i
);
}
threadFinishEvent.WaitOne();
threadFinishEvent.Close();
if (exception != null)
throw new Exception(exception.Message, exception);
}
}
}