From e7c877407f2a72a9519eb53debca5aeef20cded9 Mon Sep 17 00:00:00 2001 From: John Hurliman Date: Tue, 6 Oct 2009 02:38:00 -0700 Subject: * Continued work on the new LLUDP implementation. Appears to be functioning, although not everything is reimplemented yet * Replaced logic in ThreadTracker with a call to System.Diagnostics that does the same thing * Added Util.StringToBytes256() and Util.StringToBytes1024() to clamp output at byte[256] and byte[1024], respectively * Fixed formatting for a MySQLAssetData error logging line --- OpenSim/Framework/Parallel.cs | 207 +++++++++++++++++++++++++++++++++++++ OpenSim/Framework/ThreadTracker.cs | 127 +---------------------- OpenSim/Framework/Util.cs | 36 +++++++ 3 files changed, 248 insertions(+), 122 deletions(-) create mode 100644 OpenSim/Framework/Parallel.cs (limited to 'OpenSim/Framework') diff --git a/OpenSim/Framework/Parallel.cs b/OpenSim/Framework/Parallel.cs new file mode 100644 index 0000000..74537ba --- /dev/null +++ b/OpenSim/Framework/Parallel.cs @@ -0,0 +1,207 @@ +/* + * 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(); + + 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(); + + 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(); + + if (exception != null) + throw new Exception(exception.Message, exception); + } + } +} diff --git a/OpenSim/Framework/ThreadTracker.cs b/OpenSim/Framework/ThreadTracker.cs index d3a239d..b68d9b3 100644 --- a/OpenSim/Framework/ThreadTracker.cs +++ b/OpenSim/Framework/ThreadTracker.cs @@ -26,138 +26,21 @@ */ using System; -using System.Collections; using System.Collections.Generic; using System.Reflection; -using System.Threading; +using System.Diagnostics; using log4net; namespace OpenSim.Framework { public static class ThreadTracker { - private static readonly ILog m_log - = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); - - private static readonly long ThreadTimeout = 30 * 10000000; - public static List m_Threads; - public static Thread ThreadTrackerThread; + private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); - static ThreadTracker() + public static ProcessThreadCollection GetThreads() { -#if DEBUG - m_Threads = new List(); - ThreadTrackerThread = new Thread(ThreadTrackerThreadLoop); - ThreadTrackerThread.Name = "ThreadTrackerThread"; - ThreadTrackerThread.IsBackground = true; - ThreadTrackerThread.Priority = ThreadPriority.BelowNormal; - ThreadTrackerThread.Start(); - Add(ThreadTrackerThread); -#endif + Process thisProc = Process.GetCurrentProcess(); + return thisProc.Threads; } - - private static void ThreadTrackerThreadLoop() - { - try - { - while (true) - { - Thread.Sleep(5000); - CleanUp(); - } - } - catch (Exception e) - { - m_log.ErrorFormat( - "[THREAD TRACKER]: Thread tracker cleanup thread terminating with exception. Please report this error. Exception is {0}", - e); - } - } - - public static void Add(Thread thread) - { -#if DEBUG - if (thread != null) - { - lock (m_Threads) - { - ThreadTrackerItem tti = new ThreadTrackerItem(); - tti.Thread = thread; - tti.LastSeenActive = DateTime.Now.Ticks; - m_Threads.Add(tti); - } - } -#endif - } - - public static void Remove(Thread thread) - { -#if DEBUG - lock (m_Threads) - { - foreach (ThreadTrackerItem tti in new ArrayList(m_Threads)) - { - if (tti.Thread == thread) - m_Threads.Remove(tti); - } - } -#endif - } - - public static void CleanUp() - { - lock (m_Threads) - { - foreach (ThreadTrackerItem tti in new ArrayList(m_Threads)) - { - try - { - - - if (tti.Thread.IsAlive) - { - // Its active - tti.LastSeenActive = DateTime.Now.Ticks; - } - else - { - // Its not active -- if its expired then remove it - if (tti.LastSeenActive + ThreadTimeout < DateTime.Now.Ticks) - m_Threads.Remove(tti); - } - } - catch (NullReferenceException) - { - m_Threads.Remove(tti); - } - } - } - } - - public static List GetThreads() - { - if (m_Threads == null) - return null; - - List threads = new List(); - lock (m_Threads) - { - foreach (ThreadTrackerItem tti in new ArrayList(m_Threads)) - { - threads.Add(tti.Thread); - } - } - return threads; - } - - #region Nested type: ThreadTrackerItem - - public class ThreadTrackerItem - { - public long LastSeenActive; - public Thread Thread; - } - - #endregion } } diff --git a/OpenSim/Framework/Util.cs b/OpenSim/Framework/Util.cs index 0851d26..189fa38 100644 --- a/OpenSim/Framework/Util.cs +++ b/OpenSim/Framework/Util.cs @@ -1231,6 +1231,42 @@ namespace OpenSim.Framework return (ipaddr1 != null) ? "http://" + ipaddr1.ToString() + ":" + port1 : uri; } + public static byte[] StringToBytes256(string str) + { + if (String.IsNullOrEmpty(str)) { return Utils.EmptyBytes; } + if (str.Length > 254) str = str.Remove(254); + if (!str.EndsWith("\0")) { str += "\0"; } + + // Because this is UTF-8 encoding and not ASCII, it's possible we + // might have gotten an oversized array even after the string trim + byte[] data = UTF8.GetBytes(str); + if (data.Length > 256) + { + Array.Resize(ref data, 256); + data[255] = 0; + } + + return data; + } + + public static byte[] StringToBytes1024(string str) + { + if (String.IsNullOrEmpty(str)) { return Utils.EmptyBytes; } + if (str.Length > 1023) str = str.Remove(1023); + if (!str.EndsWith("\0")) { str += "\0"; } + + // Because this is UTF-8 encoding and not ASCII, it's possible we + // might have gotten an oversized array even after the string trim + byte[] data = UTF8.GetBytes(str); + if (data.Length > 1024) + { + Array.Resize(ref data, 1024); + data[1023] = 0; + } + + return data; + } + #region FireAndForget Threading Pattern public static void FireAndForget(System.Threading.WaitCallback callback) -- cgit v1.1