OpenSim
/
OpenSimMirror
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'htb-throttle'

prioritization
Melanie 2009-10-08 08:07:38 +01:00
parent d9f15fbf40 68961ec092
commit fe679be9e7
46 changed files with 3387 additions and 4873 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
OpenSim/Client/MXP/PacketHandler/MXPPacketServer.cs
View File

@ -89,7 +89,6 @@ namespace OpenSim.Client.MXP.PacketHandler
m_clientThread.Name = "MXPThread";
m_clientThread.IsBackground = true;
m_clientThread.Start();
ThreadTracker.Add(m_clientThread);
}
public void StartListener()

6
OpenSim/Data/MySQL/MySQLAssetData.cs
View File

@ -239,10 +239,8 @@ namespace OpenSim.Data.MySQL
}
catch (Exception e)
{
m_log.ErrorFormat(
"[ASSETS DB]: " +
"MySql failure creating asset {0} with name {1}" + Environment.NewLine + e.ToString()
+ Environment.NewLine + "Attempting reconnection", asset.FullID, asset.Name);
m_log.ErrorFormat("[ASSET DB]: MySQL failure creating asset {0} with name \"{1}\". Attempting reconnect. Error: {2}",
asset.FullID, asset.Name, e.Message);
_dbConnection.Reconnect();
}
}

4
OpenSim/Framework/ConfigSettings.cs
View File

@ -168,7 +168,7 @@ namespace OpenSim.Framework
public const bool DefaultUserServerHttpSSL = false;
public const uint DefaultMessageServerHttpPort = 8006;
public const bool DefaultMessageServerHttpSSL = false;
public const uint DefaultGridServerHttpPort = 8003;
public const uint DefaultInventoryServerHttpPort = 8003;
public const uint DefaultGridServerHttpPort = 8001;
public const uint DefaultInventoryServerHttpPort = 8004;
}
}

6
OpenSim/Framework/InventoryConfig.cs
View File

@ -56,15 +56,15 @@ namespace OpenSim.Framework
m_configMember.addConfigurationOption("default_inventory_server",
ConfigurationOption.ConfigurationTypes.TYPE_STRING_NOT_EMPTY,
"Default Inventory Server URI (this server's external name)",
"http://127.0.0.1:" + ConfigSettings.DefaultInventoryServerHttpPort, false);
"http://127.0.0.1:8004", false);
m_configMember.addConfigurationOption("default_user_server",
ConfigurationOption.ConfigurationTypes.TYPE_STRING_NOT_EMPTY,
"Default User Server URI",
"http://127.0.0.1:" + ConfigSettings.DefaultUserServerHttpPort, false);
"http://127.0.0.1:8002", false);
m_configMember.addConfigurationOption("default_asset_server",
ConfigurationOption.ConfigurationTypes.TYPE_STRING_NOT_EMPTY,
"Default Asset Server URI",
"http://127.0.0.1:" + ConfigSettings.DefaultAssetServerHttpPort, false);
"http://127.0.0.1:8003", false);
m_configMember.addConfigurationOption("database_provider", ConfigurationOption.ConfigurationTypes.TYPE_STRING,
"DLL for database provider", "OpenSim.Data.MySQL.dll", false);
m_configMember.addConfigurationOption("database_connect", ConfigurationOption.ConfigurationTypes.TYPE_STRING,

207
OpenSim/Framework/Parallel.cs Normal file
View File

@ -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
{
/// <summary>
/// Provides helper methods for parallelizing loops
/// </summary>
public static class Parallel
{
private static readonly int processorCount = System.Environment.ProcessorCount;
/// <summary>
/// Executes a for loop in which iterations may run in parallel
/// </summary>
/// <param name="fromInclusive">The loop will be started at this index</param>
/// <param name="toExclusive">The loop will be terminated before this index is reached</param>
/// <param name="body">Method body to run for each iteration of the loop</param>
public static void For(int fromInclusive, int toExclusive, Action<int> body)
{
For(processorCount, fromInclusive, toExclusive, body);
}
/// <summary>
/// Executes a for loop in which iterations may run in parallel
/// </summary>
/// <param name="threadCount">The number of concurrent execution threads to run</param>
/// <param name="fromInclusive">The loop will be started at this index</param>
/// <param name="toExclusive">The loop will be terminated before this index is reached</param>
/// <param name="body">Method body to run for each iteration of the loop</param>
public static void For(int threadCount, int fromInclusive, int toExclusive, Action<int> 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);
}
/// <summary>
/// Executes a foreach loop in which iterations may run in parallel
/// </summary>
/// <typeparam name="T">Object type that the collection wraps</typeparam>
/// <param name="enumerable">An enumerable collection to iterate over</param>
/// <param name="body">Method body to run for each object in the collection</param>
public static void ForEach<T>(IEnumerable<T> enumerable, Action<T> body)
{
ForEach<T>(processorCount, enumerable, body);
}
/// <summary>
/// Executes a foreach loop in which iterations may run in parallel
/// </summary>
/// <typeparam name="T">Object type that the collection wraps</typeparam>
/// <param name="threadCount">The number of concurrent execution threads to run</param>
/// <param name="enumerable">An enumerable collection to iterate over</param>
/// <param name="body">Method body to run for each object in the collection</param>
public static void ForEach<T>(int threadCount, IEnumerable<T> enumerable, Action<T> body)
{
int counter = threadCount;
AutoResetEvent threadFinishEvent = new AutoResetEvent(false);
IEnumerator<T> 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);
}
/// <summary>
/// Executes a series of tasks in parallel
/// </summary>
/// <param name="actions">A series of method bodies to execute</param>
public static void Invoke(params Action[] actions)
{
Invoke(processorCount, actions);
}
/// <summary>
/// Executes a series of tasks in parallel
/// </summary>
/// <param name="threadCount">The number of concurrent execution threads to run</param>
/// <param name="actions">A series of method bodies to execute</param>
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);
}
}
}

30
OpenSim/Framework/Servers/BaseOpenSimServer.cs
View File

@ -27,6 +27,7 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Reflection;
using System.Text;
@ -109,9 +110,8 @@ namespace OpenSim.Framework.Servers
m_periodicDiagnosticsTimer.Elapsed += new ElapsedEventHandler(LogDiagnostics);
m_periodicDiagnosticsTimer.Enabled = true;
// Add ourselves to thread monitoring. This thread will go on to become the console listening thread
// This thread will go on to become the console listening thread
Thread.CurrentThread.Name = "ConsoleThread";
ThreadTracker.Add(Thread.CurrentThread);
ILoggerRepository repository = LogManager.GetRepository();
IAppender[] appenders = repository.GetAppenders();
@ -235,7 +235,7 @@ namespace OpenSim.Framework.Servers
{
StringBuilder sb = new StringBuilder();
List<Thread> threads = ThreadTracker.GetThreads();
ProcessThreadCollection threads = ThreadTracker.GetThreads();
if (threads == null)
{
sb.Append("OpenSim thread tracking is only enabled in DEBUG mode.");
@ -243,25 +243,15 @@ namespace OpenSim.Framework.Servers
else
{
sb.Append(threads.Count + " threads are being tracked:" + Environment.NewLine);
foreach (Thread t in threads)
foreach (ProcessThread t in threads)
{
if (t.IsAlive)
{
sb.Append(
"ID: " + t.ManagedThreadId + ", Name: " + t.Name + ", Alive: " + t.IsAlive
+ ", Pri: " + t.Priority + ", State: " + t.ThreadState + Environment.NewLine);
}
sb.Append("ID: " + t.Id + ", TotalProcessorTime: " + t.TotalProcessorTime + ", TimeRunning: " +
(DateTime.Now - t.StartTime) + ", Pri: " + t.CurrentPriority + ", State: " + t.ThreadState );
if (t.ThreadState == System.Diagnostics.ThreadState.Wait)
sb.Append(", Reason: " + t.WaitReason + Environment.NewLine);
else
{
try
{
sb.Append("ID: " + t.ManagedThreadId + ", Name: " + t.Name + ", DEAD" + Environment.NewLine);
}
catch
{
sb.Append("THREAD ERROR" + Environment.NewLine);
}
}
sb.Append(Environment.NewLine);
}
}
int workers = 0, ports = 0, maxWorkers = 0, maxPorts = 0;

140
OpenSim/Framework/Tests/ThreadTrackerTests.cs
View File

@ -41,7 +41,7 @@ namespace OpenSim.Framework.Tests
[Test]
public void DefaultThreadTrackerTest()
{
List<Thread> lThread = ThreadTracker.GetThreads();
System.Diagnostics.ProcessThreadCollection lThread = ThreadTracker.GetThreads();
/*
foreach (Thread t in lThread)
@ -50,143 +50,7 @@ namespace OpenSim.Framework.Tests
}
*/
Assert.That(lThread.Count == 1);
Assert.That(lThread[0].Name == "ThreadTrackerThread");
}
/// <summary>
/// Validate that adding a thread to the thread tracker works
/// Validate that removing a thread from the thread tracker also works.
/// </summary>
[Test]
public void AddThreadToThreadTrackerTestAndRemoveTest()
{
Thread t = new Thread(run);
t.Name = "TestThread";
t.Priority = ThreadPriority.BelowNormal;
t.IsBackground = true;
t.SetApartmentState(ApartmentState.MTA);
t.Start();
ThreadTracker.Add(t);
List<Thread> lThread = ThreadTracker.GetThreads();
Assert.That(lThread.Count == 2);
foreach (Thread tr in lThread)
{
Assert.That((tr.Name == "ThreadTrackerThread" || tr.Name == "TestThread"));
}
running = false;
ThreadTracker.Remove(t);
lThread = ThreadTracker.GetThreads();
Assert.That(lThread.Count == 1);
foreach (Thread tr in lThread)
{
Assert.That((tr.Name == "ThreadTrackerThread"));
}
}
/// <summary>
/// Test a dead thread removal by aborting it and setting it's last seen active date to 50 seconds
/// </summary>
[Test]
public void DeadThreadTest()
{
Thread t = new Thread(run2);
t.Name = "TestThread";
t.Priority = ThreadPriority.BelowNormal;
t.IsBackground = true;
t.SetApartmentState(ApartmentState.MTA);
t.Start();
ThreadTracker.Add(t);
t.Abort();
Thread.Sleep(5000);
ThreadTracker.m_Threads[1].LastSeenActive = DateTime.Now.Ticks - (50*10000000);
ThreadTracker.CleanUp();
List<Thread> lThread = ThreadTracker.GetThreads();
Assert.That(lThread.Count == 1);
foreach (Thread tr in lThread)
{
Assert.That((tr.Name == "ThreadTrackerThread"));
Assert.That(lThread.Count > 0);
}
}
[Test]
public void UnstartedThreadTest()
{
Thread t = new Thread(run2);
t.Name = "TestThread";
t.Priority = ThreadPriority.BelowNormal;
t.IsBackground = true;
t.SetApartmentState(ApartmentState.MTA);
ThreadTracker.Add(t);
ThreadTracker.m_Threads[1].LastSeenActive = DateTime.Now.Ticks - (50 * 10000000);
ThreadTracker.CleanUp();
List<Thread> lThread = ThreadTracker.GetThreads();
Assert.That(lThread.Count == 1);
foreach (Thread tr in lThread)
{
Assert.That((tr.Name == "ThreadTrackerThread"));
}
}
[Test]
public void NullThreadTest()
{
Thread t = null;
ThreadTracker.Add(t);
List<Thread> lThread = ThreadTracker.GetThreads();
Assert.That(lThread.Count == 1);
foreach (Thread tr in lThread)
{
Assert.That((tr.Name == "ThreadTrackerThread"));
}
}
/// <summary>
/// Worker thread 0
/// </summary>
/// <param name="o"></param>
public void run(object o)
{
while (running)
{
Thread.Sleep(5000);
}
}
/// <summary>
/// Worker thread 1
/// </summary>
/// <param name="o"></param>
public void run2(object o)
{
try
{
while (running2)
{
Thread.Sleep(5000);
}
}
catch (ThreadAbortException)
{
}
}
}
}

127
OpenSim/Framework/ThreadTracker.cs
View File

@ -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 ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
private static readonly long ThreadTimeout = 30 * 10000000;
public static List<ThreadTrackerItem> m_Threads;
public static Thread ThreadTrackerThread;
static ThreadTracker()
public static ProcessThreadCollection GetThreads()
{
#if DEBUG
m_Threads = new List<ThreadTrackerItem>();
ThreadTrackerThread = new Thread(ThreadTrackerThreadLoop);
ThreadTrackerThread.Name = "ThreadTrackerThread";
ThreadTrackerThread.IsBackground = true;
ThreadTrackerThread.Priority = ThreadPriority.BelowNormal;
ThreadTrackerThread.Start();
Add(ThreadTrackerThread);
#endif
}
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);
Process thisProc = Process.GetCurrentProcess();
return thisProc.Threads;
}
}
}
}
public static List<Thread> GetThreads()
{
if (m_Threads == null)
return null;
List<Thread> threads = new List<Thread>();
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
}
}

8
OpenSim/Framework/ThrottleOutPacketType.cs
View File

@ -29,9 +29,9 @@ using System;
namespace OpenSim.Framework
{
[Flags]
public enum ThrottleOutPacketType : int
{
Unknown = -1, // Also doubles as 'do not throttle'
Resend = 0,
Land = 1,
Wind = 2,
@ -39,11 +39,5 @@ namespace OpenSim.Framework
Task = 4,
Texture = 5,
Asset = 6,
Unknown = 7, // Also doubles as 'do not throttle'
Back = 8,
TypeMask = 15, // The mask to mask off the flags
LowPriority = 128 // Additional flags
}
}

40
OpenSim/Framework/Util.cs
View File

@ -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<byte>(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<byte>(ref data, 1024);
data[1023] = 0;
}
return data;
}
#region FireAndForget Threading Pattern
public static void FireAndForget(System.Threading.WaitCallback callback)
@ -1247,7 +1283,9 @@ namespace OpenSim.Framework
{
System.Threading.WaitCallback callback = (System.Threading.WaitCallback)ar.AsyncState;
callback.EndInvoke(ar);
try { callback.EndInvoke(ar); }
catch (Exception ex) { m_log.Error("[UTIL]: Asynchronous method threw an exception: " + ex.Message, ex); }
ar.AsyncWaitHandle.Close();
}

12
OpenSim/Region/Application/Application.cs
View File

@ -91,6 +91,18 @@ namespace OpenSim
m_log.Info("[OPENSIM MAIN]: configured log4net using default OpenSim.exe.config");
}
// Increase the number of IOCP threads available. Mono defaults to a tragically low number
int workerThreads, iocpThreads;
System.Threading.ThreadPool.GetMaxThreads(out workerThreads, out iocpThreads);
m_log.InfoFormat("[OPENSIM MAIN]: Runtime gave us {0} worker threads and {1} IOCP threads", workerThreads, iocpThreads);
if (workerThreads < 500 || iocpThreads < 1000)
{
workerThreads = 500;
iocpThreads = 1000;
m_log.Info("[OPENSIM MAIN]: Bumping up to 500 worker threads and 1000 IOCP threads");
System.Threading.ThreadPool.SetMaxThreads(workerThreads, iocpThreads);
}
// Check if the system is compatible with OpenSimulator.
// Ensures that the minimum system requirements are met
m_log.Info("Performing compatibility checks... ");

2
OpenSim/Region/Application/OpenSimBase.cs
View File

@ -675,7 +675,7 @@ namespace OpenSim
if (foundClientServer)
{
m_clientServers[clientServerElement].Server.Close();
m_clientServers[clientServerElement].NetworkStop();
m_clientServers.RemoveAt(clientServerElement);
}
IScene scene;

2
OpenSim/Region/ClientStack/IClientNetworkServer.cs
View File

@ -38,7 +38,7 @@ namespace OpenSim.Region.ClientStack
IPAddress _listenIP, ref uint port, int proxyPortOffsetParm, bool allow_alternate_port, IConfigSource configSource,
AgentCircuitManager authenticateClass);
Socket Server { get; }
void NetworkStop();
bool HandlesRegion(Location x);
void AddScene(IScene x);

38
OpenSim/Region/ClientStack/LindenUDP/ILLClientStackNetworkHandler.cs
View File

@ -1,38 +0,0 @@
/*
* 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.Net.Sockets;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public interface ILLClientStackNetworkHandler
{
void SendPacketTo(byte[] buffer, int size, SocketFlags flags, uint circuitcode); // EndPoint packetSender);
void RemoveClientCircuit(uint circuitcode);
void RegisterPacketServer(LLPacketServer server);
}
}

83
OpenSim/Region/ClientStack/LindenUDP/ILLPacketHandler.cs
View File

@ -1,83 +0,0 @@
/*
* 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 OpenMetaverse;
using OpenMetaverse.Packets;
using OpenSim.Framework;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public delegate void PacketStats(int inPackets, int outPackets, int unAckedBytes);
public delegate void PacketDrop(Packet pack, Object id);
public delegate void QueueEmpty(ThrottleOutPacketType queue);
public delegate bool SynchronizeClientHandler(IScene scene, Packet packet, UUID agentID, ThrottleOutPacketType throttlePacketType);
/// <summary>
/// Interface to a class that handles all the activity involved with maintaining the client circuit (handling acks,
/// resends, pings, etc.)
/// </summary>
public interface ILLPacketHandler : IDisposable
{
event PacketStats OnPacketStats;
event PacketDrop OnPacketDrop;
event QueueEmpty OnQueueEmpty;
SynchronizeClientHandler SynchronizeClient { set; }
int PacketsReceived { get; }
int PacketsReceivedReported { get; }
uint ResendTimeout { get; set; }
bool ReliableIsImportant { get; set; }
int MaxReliableResends { get; set; }
/// <summary>
/// Initial handling of a received packet. It will be processed later in ProcessInPacket()
/// </summary>
/// <param name="packet"></param>
void InPacket(Packet packet);
/// <summary>
/// Take action depending on the type and contents of an received packet.
/// </summary>
/// <param name="item"></param>
void ProcessInPacket(LLQueItem item);
void ProcessOutPacket(LLQueItem item);
void OutPacket(Packet NewPack,
ThrottleOutPacketType throttlePacketType);
void OutPacket(Packet NewPack,
ThrottleOutPacketType throttlePacketType, Object id);
LLPacketQueue PacketQueue { get; }
void Flush();
void Clear();
ClientInfo GetClientInfo();
void SetClientInfo(ClientInfo info);
void AddImportantPacket(PacketType type);
void RemoveImportantPacket(PacketType type);
int GetQueueCount(ThrottleOutPacketType queue);
}
}

36
OpenSim/Region/ClientStack/LindenUDP/LLQueItem.cs → OpenSim/Region/ClientStack/LindenUDP/IncomingPacket.cs
View File

@ -1,4 +1,4 @@
/*
/*
* Copyright (c) Contributors, http://opensimulator.org/
* See CONTRIBUTORS.TXT for a full list of copyright holders.
*
@ -26,24 +26,32 @@
*/
using System;
using OpenMetaverse.Packets;
using OpenSim.Framework;
using OpenMetaverse;
using OpenMetaverse.Packets;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public class LLQueItem
/// <summary>
/// Holds a reference to a <seealso cref="LLUDPClient"/> and a <seealso cref="Packet"/>
/// for incoming packets
/// </summary>
public sealed class IncomingPacket
{
public LLQueItem()
{
}
/// <summary>Client this packet came from</summary>
public LLUDPClient Client;
/// <summary>Packet data that has been received</summary>
public Packet Packet;
public bool Incoming;
public ThrottleOutPacketType throttleType;
public int TickCount;
public Object Identifier;
public int Resends;
public int Length;
public uint Sequence;
/// <summary>
/// Default constructor
/// </summary>
/// <param name="client">Reference to the client this packet came from</param>
/// <param name="packet">Packet data</param>
public IncomingPacket(LLUDPClient client, Packet packet)
{
Client = client;
Packet = packet;
}
}
}

58
OpenSim/Region/ClientStack/LindenUDP/KillPacket.cs → OpenSim/Region/ClientStack/LindenUDP/IncomingPacketHistoryCollection.cs
View File

@ -1,4 +1,4 @@
/*
/*
* Copyright (c) Contributors, http://opensimulator.org/
* See CONTRIBUTORS.TXT for a full list of copyright holders.
*
@ -25,47 +25,49 @@
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
using OpenMetaverse.Packets;
using System;
using System.Collections.Generic;
namespace OpenSim.Region.ClientStack.LindenUDP
{
/// <summary>
/// When packetqueue dequeues this packet in the outgoing stream, it thread aborts
/// Ensures that the thread abort happens from within the client thread
/// regardless of where the close method is called
/// A circular buffer and hashset for tracking incoming packet sequence
/// numbers
/// </summary>
class KillPacket : Packet
public sealed class IncomingPacketHistoryCollection
{
public override int Length
private readonly uint[] m_items;
private HashSet<uint> m_hashSet;
private int m_first;
private int m_next;
private int m_capacity;
public IncomingPacketHistoryCollection(int capacity)
{
get { return 0; }
this.m_capacity = capacity;
m_items = new uint[capacity];
m_hashSet = new HashSet<uint>();
}
public override void FromBytes(Header header, byte[] bytes, ref int i, ref int packetEnd)
public bool TryEnqueue(uint ack)
{
lock (m_hashSet)
{
if (m_hashSet.Add(ack))
{
m_items[m_next] = ack;
m_next = (m_next + 1) % m_capacity;
if (m_next == m_first)
{
m_hashSet.Remove(m_items[m_first]);
m_first = (m_first + 1) % m_capacity;
}
public override void FromBytes(byte[] bytes, ref int i, ref int packetEnd, byte[] zeroBuffer)
{
return true;
}
}
public override byte[] ToBytes()
{
return new byte[0];
}
public override byte[][] ToBytesMultiple()
{
return new byte[][] { new byte[0] };
}
public KillPacket()
{
Type = PacketType.UseCircuitCode;
Header = new Header();
Header.Frequency = OpenMetaverse.PacketFrequency.Low;
Header.ID = 65531;
Header.Reliable = true;
return false;
}
}
}

29
OpenSim/Region/ClientStack/LindenUDP/J2KImage.cs
View File

@ -76,27 +76,27 @@ namespace OpenSim.Region.ClientStack.LindenUDP
{
if (m_currentPacket <= m_stopPacket)
{
bool SendMore = true;
int count = 0;
bool sendMore = true;
if (!m_sentInfo || (m_currentPacket == 0))
{
if (SendFirstPacket(client))
{
SendMore = false;
}
sendMore = !SendFirstPacket(client);
m_sentInfo = true;
m_currentPacket++;
++m_currentPacket;
++count;
}
if (m_currentPacket < 2)
{
m_currentPacket = 2;
}
int count = 0;
while (SendMore && count < maxpack && m_currentPacket <= m_stopPacket)
while (sendMore && count < maxpack && m_currentPacket <= m_stopPacket)
{
count++;
SendMore = SendPacket(client);
m_currentPacket++;
sendMore = SendPacket(client);
++m_currentPacket;
++count;
}
if (m_currentPacket > m_stopPacket)
@ -196,13 +196,6 @@ namespace OpenSim.Region.ClientStack.LindenUDP
m_currentPacket = StartPacket;
}
if ((m_imageManager != null) && (m_imageManager.Client != null) && (m_imageManager.Client.PacketHandler != null))
if (m_imageManager.Client.PacketHandler.GetQueueCount(ThrottleOutPacketType.Texture) == 0)
{
//m_log.Debug("No textures queued, sending one packet to kickstart it");
SendPacket(m_imageManager.Client);
}
}
}
}

15090
OpenSim/Region/ClientStack/LindenUDP/LLClientView.cs
View File

File diff suppressed because it is too large Load Diff

2
OpenSim/Region/ClientStack/LindenUDP/LLImageManager.cs
View File

@ -172,7 +172,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP
m_lastloopprocessed = DateTime.Now.Ticks;
// This can happen during Close()
if (m_client == null || m_client.PacketHandler == null || m_client.PacketHandler.PacketQueue == null)
if (m_client == null)
return false;
while ((imagereq = GetHighestPriorityImage()) != null)

870
OpenSim/Region/ClientStack/LindenUDP/LLPacketHandler.cs
View File

@ -1,870 +0,0 @@
/*
* 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.Reflection;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Threading;
using System.Timers;
using OpenMetaverse;
using OpenMetaverse.Packets;
using log4net;
using OpenSim.Framework;
using Timer=System.Timers.Timer;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public class LLPacketHandler : ILLPacketHandler
{
private static readonly ILog m_log
= LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
//private int m_resentCount;
// Packet queues
//
LLPacketQueue m_PacketQueue;
public LLPacketQueue PacketQueue
{
get { return m_PacketQueue; }
}
// Timer to run stats and acks on
//
private Timer m_AckTimer = new Timer(250);
// A list of the packets we haven't acked yet
//
private List<uint> m_PendingAcks = new List<uint>();
private Dictionary<uint, uint> m_PendingAcksMap = new Dictionary<uint, uint>();
private Dictionary<uint, LLQueItem> m_NeedAck =
new Dictionary<uint, LLQueItem>();
/// <summary>
/// The number of milliseconds that can pass before a packet that needs an ack is resent.
/// </param>
private uint m_ResendTimeout = 4000;
public uint ResendTimeout
{
get { return m_ResendTimeout; }
set { m_ResendTimeout = value; }
}
private int m_MaxReliableResends = 3;
public int MaxReliableResends
{
get { return m_MaxReliableResends; }
set { m_MaxReliableResends = value; }
}
// Track duplicated packets. This uses a Dictionary. Both insertion
// and lookup are common operations and need to take advantage of
// the hashing. Expiration is less common and can be allowed the
// time for a linear scan.
//
private List<uint> m_alreadySeenList = new List<uint>();
private Dictionary<uint, int>m_alreadySeenTracker = new Dictionary<uint, int>();
private int m_alreadySeenWindow = 30000;
private int m_lastAlreadySeenCheck = Environment.TickCount & Int32.MaxValue;
// private Dictionary<uint, int> m_DupeTracker =
// new Dictionary<uint, int>();
// private uint m_DupeTrackerWindow = 30;
// private int m_DupeTrackerLastCheck = Environment.TickCount;
// Values for the SimStatsReporter
//
private int m_PacketsReceived = 0;
private int m_PacketsReceivedReported = 0;
private int m_PacketsSent = 0;
private int m_PacketsSentReported = 0;
private int m_UnackedBytes = 0;
private int m_LastResend = 0;
public int PacketsReceived
{
get { return m_PacketsReceived; }
}
public int PacketsReceivedReported
{
get { return m_PacketsReceivedReported; }
}
// The client we are working for
//
private IClientAPI m_Client;
// Some events
//
public event PacketStats OnPacketStats;
public event PacketDrop OnPacketDrop;
public event QueueEmpty OnQueueEmpty;
//private SynchronizeClientHandler m_SynchronizeClient = null;
public SynchronizeClientHandler SynchronizeClient
{
set { /* m_SynchronizeClient = value; */ }
}
// Packet sequencing
//
private uint m_Sequence = 0;
private object m_SequenceLock = new object();
private const int MAX_SEQUENCE = 0xFFFFFF;
// Packet dropping
//
List<PacketType> m_ImportantPackets = new List<PacketType>();
private bool m_ReliableIsImportant = false;
public bool ReliableIsImportant
{
get { return m_ReliableIsImportant; }
set { m_ReliableIsImportant = value; }
}
private int m_DropSafeTimeout;
LLPacketServer m_PacketServer;
private byte[] m_ZeroOutBuffer = new byte[4096];
////////////////////////////////////////////////////////////////////
// Constructors
//
public LLPacketHandler(IClientAPI client, LLPacketServer server, ClientStackUserSettings userSettings)
{
m_Client = client;
m_PacketServer = server;
m_DropSafeTimeout = Environment.TickCount + 15000;
m_PacketQueue = new LLPacketQueue(client.AgentId, userSettings);
m_PacketQueue.OnQueueEmpty += TriggerOnQueueEmpty;
m_AckTimer.Elapsed += AckTimerElapsed;
m_AckTimer.Start();
}
public void Dispose()
{
m_AckTimer.Stop();
m_AckTimer.Close();
m_PacketQueue.Enqueue(null);
m_PacketQueue.Close();
m_Client = null;
}
// Send one packet. This actually doesn't send anything, it queues
// it. Designed to be fire-and-forget, but there is an optional
// notifier.
//
public void OutPacket(
Packet packet, ThrottleOutPacketType throttlePacketType)
{
OutPacket(packet, throttlePacketType, null);
}
public void OutPacket(
Packet packet, ThrottleOutPacketType throttlePacketType,
Object id)
{
// Call the load balancer's hook. If this is not active here
// we defer to the sim server this client is actually connected
// to. Packet drop notifies will not be triggered in this
// configuration!
//
packet.Header.Sequence = 0;
lock (m_NeedAck)
{
DropResend(id);
AddAcks(ref packet);
QueuePacket(packet, throttlePacketType, id);
}
}
private void AddAcks(ref Packet packet)
{
// These packet types have shown to have issues with
// acks being appended to the payload, just don't send
// any with them until libsl is fixed.
//
if (packet is ViewerEffectPacket)
return;
if (packet is SimStatsPacket)
return;
// Add acks to outgoing packets
//
if (m_PendingAcks.Count > 0)
{
int count = m_PendingAcks.Count;
if (count > 10)
count = 10;
packet.Header.AckList = new uint[count];
packet.Header.AppendedAcks = true;
for (int i = 0; i < count; i++)
{
packet.Header.AckList[i] = m_PendingAcks[i];
m_PendingAcksMap.Remove(m_PendingAcks[i]);
}
m_PendingAcks.RemoveRange(0, count);
}
}
private void QueuePacket(
Packet packet, ThrottleOutPacketType throttlePacketType,
Object id)
{
LLQueItem item = new LLQueItem();
item.Packet = packet;
item.Incoming = false;
item.throttleType = throttlePacketType;
item.TickCount = Environment.TickCount;
item.Identifier = id;
item.Resends = 0;
item.Length = packet.Length;
item.Sequence = packet.Header.Sequence;
m_PacketQueue.Enqueue(item);
m_PacketsSent++;
}
private void ResendUnacked()
{
int now = Environment.TickCount;
int intervalMs = 250;
if (m_LastResend != 0)
intervalMs = now - m_LastResend;
lock (m_NeedAck)
{
if (m_DropSafeTimeout > now ||
intervalMs > 500) // We were frozen!
{
foreach (LLQueItem data in m_NeedAck.Values)
{
if (m_DropSafeTimeout > now)
{
m_NeedAck[data.Packet.Header.Sequence].TickCount = now;
}
else
{
m_NeedAck[data.Packet.Header.Sequence].TickCount += intervalMs;
}
}
}
m_LastResend = now;
// Unless we have received at least one ack, don't bother resending
// anything. There may not be a client there, don't clog up the
// pipes.
// Nothing to do
//
if (m_NeedAck.Count == 0)
return;
int resent = 0;
long dueDate = now - m_ResendTimeout;
List<LLQueItem> dropped = new List<LLQueItem>();
foreach (LLQueItem data in m_NeedAck.Values)
{
Packet packet = data.Packet;
// Packets this old get resent
//
if (data.TickCount < dueDate && data.Sequence != 0 && !m_PacketQueue.Contains(data.Sequence))
{
if (resent < 20) // Was 20 (= Max 117kbit/sec resends)
{
m_NeedAck[packet.Header.Sequence].Resends++;
// The client needs to be told that a packet is being resent, otherwise it appears to believe
// that it should reset its sequence to that packet number.
packet.Header.Resent = true;
if ((m_NeedAck[packet.Header.Sequence].Resends >= m_MaxReliableResends) &&
(!m_ReliableIsImportant))
{
dropped.Add(data);
continue;
}
m_NeedAck[packet.Header.Sequence].TickCount = Environment.TickCount;
QueuePacket(packet, ThrottleOutPacketType.Resend, data.Identifier);
resent++;
}
else
{
m_NeedAck[packet.Header.Sequence].TickCount += intervalMs;
}
}
}
foreach (LLQueItem data in dropped)
{
m_NeedAck.Remove(data.Packet.Header.Sequence);
TriggerOnPacketDrop(data.Packet, data.Identifier);
m_PacketQueue.Cancel(data.Packet.Header.Sequence);
PacketPool.Instance.ReturnPacket(data.Packet);
}
}
}
// Send the pending packet acks to the client
// Will send blocks of acks for up to 250 packets
//
private void SendAcks()
{
lock (m_NeedAck)
{
if (m_PendingAcks.Count == 0)
return;
PacketAckPacket acks = (PacketAckPacket)PacketPool.Instance.GetPacket(PacketType.PacketAck);
// The case of equality is more common than one might think,
// because this function will be called unconditionally when
// the counter reaches 250. So there is a good chance another
// packet with 250 blocks exists.
//
if (acks.Packets == null ||
acks.Packets.Length != m_PendingAcks.Count)
acks.Packets = new PacketAckPacket.PacketsBlock[m_PendingAcks.Count];
for (int i = 0; i < m_PendingAcks.Count; i++)
{
acks.Packets[i] = new PacketAckPacket.PacketsBlock();
acks.Packets[i].ID = m_PendingAcks[i];
}
m_PendingAcks.Clear();
m_PendingAcksMap.Clear();
acks.Header.Reliable = false;
OutPacket(acks, ThrottleOutPacketType.Unknown);
}
}
// Queue a packet ack. It will be sent either after 250 acks are
// queued, or when the timer fires.
//
private void AckPacket(Packet packet)
{
lock (m_NeedAck)
{
if (m_PendingAcks.Count < 250)
{
if (!m_PendingAcksMap.ContainsKey(packet.Header.Sequence))
{
m_PendingAcks.Add(packet.Header.Sequence);
m_PendingAcksMap.Add(packet.Header.Sequence,
packet.Header.Sequence);
}
return;
}
}
SendAcks();
lock (m_NeedAck)
{
// If this is still full we have a truly exceptional
// condition (means, can't happen)
//
if (m_PendingAcks.Count < 250)
{
if (!m_PendingAcksMap.ContainsKey(packet.Header.Sequence))
{
m_PendingAcks.Add(packet.Header.Sequence);
m_PendingAcksMap.Add(packet.Header.Sequence,
packet.Header.Sequence);
}
return;
}
}
}
// When the timer elapses, send the pending acks, trigger resends
// and report all the stats.
//
private void AckTimerElapsed(object sender, ElapsedEventArgs ea)
{
SendAcks();
ResendUnacked();
SendPacketStats();
}
// Push out pachet counts for the sim status reporter
//
private void SendPacketStats()
{
PacketStats handlerPacketStats = OnPacketStats;
if (handlerPacketStats != null)
{
handlerPacketStats(
m_PacketsReceived - m_PacketsReceivedReported,
m_PacketsSent - m_PacketsSentReported,
m_UnackedBytes);
m_PacketsReceivedReported = m_PacketsReceived;
m_PacketsSentReported = m_PacketsSent;
}
}
// We can't keep an unlimited record of dupes. This will prune the
// dictionary by age.
//
// NOTE: this needs to be called from within lock
// (m_alreadySeenTracker) context!
private void ExpireSeenPackets()
{
if (m_alreadySeenList.Count < 1024)
return;
int ticks = 0;
int tc = Environment.TickCount & Int32.MaxValue;
if (tc >= m_lastAlreadySeenCheck)
ticks = tc - m_lastAlreadySeenCheck;
else
ticks = Int32.MaxValue - m_lastAlreadySeenCheck + tc;
if (ticks < 2000) return;
m_lastAlreadySeenCheck = tc;
// we calculate the drop dead tick count here instead of
// in the loop: any packet with a timestamp before
// dropDeadTC can be expired
int dropDeadTC = tc - m_alreadySeenWindow;
int i = 0;
while (i < m_alreadySeenList.Count && m_alreadySeenTracker[m_alreadySeenList[i]] < dropDeadTC)
{
m_alreadySeenTracker.Remove(m_alreadySeenList[i]);
i++;
}
// if we dropped packet from m_alreadySeenTracker we need
// to drop them from m_alreadySeenList as well, let's do
// that in one go: the list is ordered after all.
if (i > 0)
{
m_alreadySeenList.RemoveRange(0, i);
// m_log.DebugFormat("[CLIENT]: expired {0} packets, {1}:{2} left", i, m_alreadySeenList.Count, m_alreadySeenTracker.Count);
}
}
public void InPacket(Packet packet)
{
if (packet == null)
return;
// When too many acks are needed to be sent, the client sends
// a packet consisting of acks only
//
if (packet.Type == PacketType.PacketAck)
{
PacketAckPacket ackPacket = (PacketAckPacket)packet;
foreach (PacketAckPacket.PacketsBlock block in ackPacket.Packets)
{
ProcessAck(block.ID);
}
PacketPool.Instance.ReturnPacket(packet);
return;
}
// Any packet can have some packet acks in the header.
// Process them here
//
if (packet.Header.AppendedAcks)
{
foreach (uint id in packet.Header.AckList)
{
ProcessAck(id);
}
}
// If this client is on another partial instance, no need
// to handle packets
//
if (!m_Client.IsActive && packet.Type != PacketType.LogoutRequest)
{
PacketPool.Instance.ReturnPacket(packet);
return;
}
if (packet.Type == PacketType.StartPingCheck)
{
StartPingCheckPacket startPing = (StartPingCheckPacket)packet;
CompletePingCheckPacket endPing
= (CompletePingCheckPacket)PacketPool.Instance.GetPacket(PacketType.CompletePingCheck);
endPing.PingID.PingID = startPing.PingID.PingID;
OutPacket(endPing, ThrottleOutPacketType.Task);
}
else
{
LLQueItem item = new LLQueItem();
item.Packet = packet;
item.Incoming = true;
m_PacketQueue.Enqueue(item);
}
}
public void ProcessInPacket(LLQueItem item)
{
Packet packet = item.Packet;
// Always ack the packet!
//
if (packet.Header.Reliable)
AckPacket(packet);
if (packet.Type != PacketType.AgentUpdate)
m_PacketsReceived++;
// Check for duplicate packets.. packets that the client is
// resending because it didn't receive our ack
//
lock (m_alreadySeenTracker)
{
ExpireSeenPackets();
if (m_alreadySeenTracker.ContainsKey(packet.Header.Sequence))
return;
m_alreadySeenTracker.Add(packet.Header.Sequence, Environment.TickCount & Int32.MaxValue);
m_alreadySeenList.Add(packet.Header.Sequence);
}
m_Client.ProcessInPacket(packet);
}
public void Flush()
{
m_PacketQueue.Flush();
m_UnackedBytes = (-1 * m_UnackedBytes);
SendPacketStats();
}
public void Clear()
{
m_UnackedBytes = (-1 * m_UnackedBytes);
SendPacketStats();
lock (m_NeedAck)
{
m_NeedAck.Clear();
m_PendingAcks.Clear();
m_PendingAcksMap.Clear();
}
m_Sequence += 1000000;
}
private void ProcessAck(uint id)
{
LLQueItem data;
lock (m_NeedAck)
{
//m_log.DebugFormat("[CLIENT]: In {0} received ack for packet {1}", m_Client.Scene.RegionInfo.ExternalEndPoint.Port, id);
if (!m_NeedAck.TryGetValue(id, out data))
return;
m_NeedAck.Remove(id);
m_PacketQueue.Cancel(data.Sequence);
PacketPool.Instance.ReturnPacket(data.Packet);
m_UnackedBytes -= data.Length;
}
}
// Allocate packet sequence numbers in a threadsave manner
//
protected uint NextPacketSequenceNumber()
{
// Set the sequence number
uint seq = 1;
lock (m_SequenceLock)
{
if (m_Sequence >= MAX_SEQUENCE)
{
m_Sequence = 1;
}
else
{
m_Sequence++;
}
seq = m_Sequence;
}
return seq;
}
public ClientInfo GetClientInfo()
{
ClientInfo info = new ClientInfo();
info.pendingAcks = m_PendingAcksMap;
info.needAck = new Dictionary<uint, byte[]>();
lock (m_NeedAck)
{
foreach (uint key in m_NeedAck.Keys)
info.needAck.Add(key, m_NeedAck[key].Packet.ToBytes());
}
LLQueItem[] queitems = m_PacketQueue.GetQueueArray();
for (int i = 0; i < queitems.Length; i++)
{
if (queitems[i].Incoming == false)
info.out_packets.Add(queitems[i].Packet.ToBytes());
}
info.sequence = m_Sequence;
float multiplier = m_PacketQueue.ThrottleMultiplier;
info.resendThrottle = (int) (m_PacketQueue.ResendThrottle.Throttle / multiplier);
info.landThrottle = (int) (m_PacketQueue.LandThrottle.Throttle / multiplier);
info.windThrottle = (int) (m_PacketQueue.WindThrottle.Throttle / multiplier);
info.cloudThrottle = (int) (m_PacketQueue.CloudThrottle.Throttle / multiplier);
info.taskThrottle = (int) (m_PacketQueue.TaskThrottle.Throttle / multiplier);
info.assetThrottle = (int) (m_PacketQueue.AssetThrottle.Throttle / multiplier);
info.textureThrottle = (int) (m_PacketQueue.TextureThrottle.Throttle / multiplier);
info.totalThrottle = (int) (m_PacketQueue.TotalThrottle.Throttle / multiplier);
return info;
}
public void SetClientInfo(ClientInfo info)
{
m_PendingAcksMap = info.pendingAcks;
m_PendingAcks = new List<uint>(m_PendingAcksMap.Keys);
m_NeedAck = new Dictionary<uint, LLQueItem>();
Packet packet = null;
int packetEnd = 0;
byte[] zero = new byte[3000];
foreach (uint key in info.needAck.Keys)
{
byte[] buff = info.needAck[key];
packetEnd = buff.Length - 1;
try
{
packet = PacketPool.Instance.GetPacket(buff, ref packetEnd, zero);
}
catch (Exception)
{
}
LLQueItem item = new LLQueItem();
item.Packet = packet;
item.Incoming = false;
item.throttleType = 0;
item.TickCount = Environment.TickCount;
item.Identifier = 0;
item.Resends = 0;
item.Length = packet.Length;
item.Sequence = packet.Header.Sequence;
m_NeedAck.Add(key, item);
}
m_Sequence = info.sequence;
m_PacketQueue.ResendThrottle.Throttle = info.resendThrottle;
m_PacketQueue.LandThrottle.Throttle = info.landThrottle;
m_PacketQueue.WindThrottle.Throttle = info.windThrottle;
m_PacketQueue.CloudThrottle.Throttle = info.cloudThrottle;
m_PacketQueue.TaskThrottle.Throttle = info.taskThrottle;
m_PacketQueue.AssetThrottle.Throttle = info.assetThrottle;
m_PacketQueue.TextureThrottle.Throttle = info.textureThrottle;
m_PacketQueue.TotalThrottle.Throttle = info.totalThrottle;
}
public void AddImportantPacket(PacketType type)
{
if (m_ImportantPackets.Contains(type))
return;
m_ImportantPackets.Add(type);
}
public void RemoveImportantPacket(PacketType type)
{
if (!m_ImportantPackets.Contains(type))
return;
m_ImportantPackets.Remove(type);
}
private void DropResend(Object id)
{
LLQueItem d = null;
foreach (LLQueItem data in m_NeedAck.Values)
{
if (data.Identifier != null && data.Identifier == id)
{
d = data;
break;
}
}
if (null == d) return;
m_NeedAck.Remove(d.Packet.Header.Sequence);
m_PacketQueue.Cancel(d.Sequence);
PacketPool.Instance.ReturnPacket(d.Packet);
}
private void TriggerOnPacketDrop(Packet packet, Object id)
{
PacketDrop handlerPacketDrop = OnPacketDrop;
if (handlerPacketDrop == null)
return;
handlerPacketDrop(packet, id);
}
private void TriggerOnQueueEmpty(ThrottleOutPacketType queue)
{
QueueEmpty handlerQueueEmpty = OnQueueEmpty;
if (handlerQueueEmpty != null)
handlerQueueEmpty(queue);
}
// Convert the packet to bytes and stuff it onto the send queue
//
public void ProcessOutPacket(LLQueItem item)
{
Packet packet = item.Packet;
// Assign sequence number here to prevent out of order packets
if (packet.Header.Sequence == 0)
{
lock (m_NeedAck)
{
packet.Header.Sequence = NextPacketSequenceNumber();
item.Sequence = packet.Header.Sequence;
item.TickCount = Environment.TickCount;
// We want to see that packet arrive if it's reliable
if (packet.Header.Reliable)
{
m_UnackedBytes += item.Length;
// Keep track of when this packet was sent out
item.TickCount = Environment.TickCount;
m_NeedAck[packet.Header.Sequence] = item;
}
}
}
// If we sent a killpacket
if (packet is KillPacket)
Abort();
try
{
// If this packet has been reused/returned, the ToBytes
// will blow up in our face.
// Fail gracefully.
//
// Actually make the byte array and send it
byte[] sendbuffer = item.Packet.ToBytes();
if (packet.Header.Zerocoded)
{
int packetsize = Helpers.ZeroEncode(sendbuffer,
sendbuffer.Length, m_ZeroOutBuffer);
m_PacketServer.SendPacketTo(m_ZeroOutBuffer, packetsize,
SocketFlags.None, m_Client.CircuitCode);
}
else
{
// Need some extra space in case we need to add proxy
// information to the message later
Buffer.BlockCopy(sendbuffer, 0, m_ZeroOutBuffer, 0,
sendbuffer.Length);
m_PacketServer.SendPacketTo(m_ZeroOutBuffer,
sendbuffer.Length, SocketFlags.None, m_Client.CircuitCode);
}
}
catch (NullReferenceException)
{
m_log.Error("[PACKET]: Detected reuse of a returned packet");
m_PacketQueue.Cancel(item.Sequence);
return;
}
// If this is a reliable packet, we are still holding a ref
// Dont't return in that case
//
if (!packet.Header.Reliable)
{
m_PacketQueue.Cancel(item.Sequence);
PacketPool.Instance.ReturnPacket(packet);
}
}
private void Abort()
{
m_PacketQueue.Close();
Thread.CurrentThread.Abort();
}
public int GetQueueCount(ThrottleOutPacketType queue)
{
return m_PacketQueue.GetQueueCount(queue);
}
}
}

742
OpenSim/Region/ClientStack/LindenUDP/LLPacketQueue.cs
View File

@ -1,742 +0,0 @@
/*
* 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.Reflection;
using System.Threading;
using System.Timers;
using log4net;
using OpenMetaverse;
using OpenSim.Framework;
using OpenSim.Framework.Statistics;
using OpenSim.Framework.Statistics.Interfaces;
using Timer=System.Timers.Timer;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public class LLPacketQueue : IPullStatsProvider, IDisposable
{
private static readonly ILog m_log
= LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
/// <summary>
/// Is queueing enabled at all?
/// </summary>
private bool m_enabled = true;
private OpenSim.Framework.BlockingQueue<LLQueItem> SendQueue;
private Queue<LLQueItem> IncomingPacketQueue;
private Queue<LLQueItem> OutgoingPacketQueue;
private Queue<LLQueItem> ResendOutgoingPacketQueue;
private Queue<LLQueItem> LandOutgoingPacketQueue;
private Queue<LLQueItem> WindOutgoingPacketQueue;
private Queue<LLQueItem> CloudOutgoingPacketQueue;
private Queue<LLQueItem> TaskOutgoingPacketQueue;
private Queue<LLQueItem> TaskLowpriorityPacketQueue;
private Queue<LLQueItem> TextureOutgoingPacketQueue;
private Queue<LLQueItem> AssetOutgoingPacketQueue;
private List<ThrottleOutPacketType> Empty = new List<ThrottleOutPacketType>();
// m_log.Info("[THROTTLE]: Entering Throttle");
// private Dictionary<uint, uint> PendingAcks = new Dictionary<uint, uint>();
// private Dictionary<uint, Packet> NeedAck = new Dictionary<uint, Packet>();
// All throttle times and number of bytes are calculated by dividing by this value
// This value also determines how many times per throttletimems the timer will run
// If throttleimems is 1000 ms, then the timer will fire every 1000/7 milliseconds
private float throttleMultiplier = 2.0f; // Default value really doesn't matter.
private int throttleTimeDivisor = 7;
private int throttletimems = 1000;
internal LLPacketThrottle ResendThrottle;
internal LLPacketThrottle LandThrottle;
internal LLPacketThrottle WindThrottle;
internal LLPacketThrottle CloudThrottle;
internal LLPacketThrottle TaskThrottle;
internal LLPacketThrottle AssetThrottle;
internal LLPacketThrottle TextureThrottle;
internal LLPacketThrottle TotalThrottle;
private Dictionary<uint,int> contents = new Dictionary<uint, int>();
// private long LastThrottle;
// private long ThrottleInterval;
private Timer throttleTimer;
private UUID m_agentId;
public event QueueEmpty OnQueueEmpty;
public LLPacketQueue(UUID agentId, ClientStackUserSettings userSettings)
{
// While working on this, the BlockingQueue had me fooled for a bit.
// The Blocking queue causes the thread to stop until there's something
// in it to process. it's an on-purpose threadlock though because
// without it, the clientloop will suck up all sim resources.
SendQueue = new OpenSim.Framework.BlockingQueue<LLQueItem>();
IncomingPacketQueue = new Queue<LLQueItem>();
OutgoingPacketQueue = new Queue<LLQueItem>();
ResendOutgoingPacketQueue = new Queue<LLQueItem>();
LandOutgoingPacketQueue = new Queue<LLQueItem>();
WindOutgoingPacketQueue = new Queue<LLQueItem>();
CloudOutgoingPacketQueue = new Queue<LLQueItem>();
TaskOutgoingPacketQueue = new Queue<LLQueItem>();
TaskLowpriorityPacketQueue = new Queue<LLQueItem>();
TextureOutgoingPacketQueue = new Queue<LLQueItem>();
AssetOutgoingPacketQueue = new Queue<LLQueItem>();
// Store the throttle multiplier for posterity.
throttleMultiplier = userSettings.ClientThrottleMultipler;
int throttleMaxBPS = 1500000;
if (userSettings.TotalThrottleSettings != null)
throttleMaxBPS = userSettings.TotalThrottleSettings.Max;
// Set up the throttle classes (min, max, current) in bits per second
ResendThrottle = new LLPacketThrottle(5000, throttleMaxBPS / 15, 16000, userSettings.ClientThrottleMultipler);
LandThrottle = new LLPacketThrottle(1000, throttleMaxBPS / 15, 2000, userSettings.ClientThrottleMultipler);
WindThrottle = new LLPacketThrottle(0, throttleMaxBPS / 15, 0, userSettings.ClientThrottleMultipler);
CloudThrottle = new LLPacketThrottle(0, throttleMaxBPS / 15, 0, userSettings.ClientThrottleMultipler);
TaskThrottle = new LLPacketThrottle(1000, throttleMaxBPS / 2, 3000, userSettings.ClientThrottleMultipler);
AssetThrottle = new LLPacketThrottle(1000, throttleMaxBPS / 2, 1000, userSettings.ClientThrottleMultipler);
TextureThrottle = new LLPacketThrottle(1000, throttleMaxBPS / 2, 4000, userSettings.ClientThrottleMultipler);
// Total Throttle trumps all - it is the number of bits in total that are allowed to go out per second.
ThrottleSettings totalThrottleSettings = userSettings.TotalThrottleSettings;
if (null == totalThrottleSettings)
{
totalThrottleSettings = new ThrottleSettings(0, throttleMaxBPS, 28000);
}
TotalThrottle
= new LLPacketThrottle(
totalThrottleSettings.Min, totalThrottleSettings.Max, totalThrottleSettings.Current,
userSettings.ClientThrottleMultipler);
throttleTimer = new Timer((int)(throttletimems / throttleTimeDivisor));
throttleTimer.Elapsed += ThrottleTimerElapsed;
throttleTimer.Start();
// TIMERS needed for this
// LastThrottle = DateTime.Now.Ticks;
// ThrottleInterval = (long)(throttletimems/throttleTimeDivisor);
m_agentId = agentId;
if (StatsManager.SimExtraStats != null)
{
StatsManager.SimExtraStats.RegisterPacketQueueStatsProvider(m_agentId, this);
}
}
/* STANDARD QUEUE MANIPULATION INTERFACES */
public void Enqueue(LLQueItem item)
{
if (!m_enabled)
{
return;
}
// We could micro lock, but that will tend to actually
// probably be worse than just synchronizing on SendQueue
if (item == null)
{
SendQueue.Enqueue(item);
return;
}
if (item.Incoming)
{
SendQueue.PriorityEnqueue(item);
return;
}
if (item.Sequence != 0)
lock (contents)
{
if (contents.ContainsKey(item.Sequence))
contents[item.Sequence] += 1;
else
contents.Add(item.Sequence, 1);
}
lock (this)
{
switch (item.throttleType & ThrottleOutPacketType.TypeMask)
{
case ThrottleOutPacketType.Resend:
ThrottleCheck(ref ResendThrottle, ref ResendOutgoingPacketQueue, item, ThrottleOutPacketType.Resend);
break;
case ThrottleOutPacketType.Texture:
ThrottleCheck(ref TextureThrottle, ref TextureOutgoingPacketQueue, item, ThrottleOutPacketType.Texture);
break;
case ThrottleOutPacketType.Task:
if ((item.throttleType & ThrottleOutPacketType.LowPriority) != 0)
ThrottleCheck(ref TaskThrottle, ref TaskLowpriorityPacketQueue, item, ThrottleOutPacketType.Task);
else
ThrottleCheck(ref TaskThrottle, ref TaskOutgoingPacketQueue, item, ThrottleOutPacketType.Task);
break;
case ThrottleOutPacketType.Land:
ThrottleCheck(ref LandThrottle, ref LandOutgoingPacketQueue, item, ThrottleOutPacketType.Land);
break;
case ThrottleOutPacketType.Asset:
ThrottleCheck(ref AssetThrottle, ref AssetOutgoingPacketQueue, item, ThrottleOutPacketType.Asset);
break;
case ThrottleOutPacketType.Cloud:
ThrottleCheck(ref CloudThrottle, ref CloudOutgoingPacketQueue, item, ThrottleOutPacketType.Cloud);
break;
case ThrottleOutPacketType.Wind:
ThrottleCheck(ref WindThrottle, ref WindOutgoingPacketQueue, item, ThrottleOutPacketType.Wind);
break;
default:
// Acknowledgements and other such stuff should go directly to the blocking Queue
// Throttling them may and likely 'will' be problematic
SendQueue.PriorityEnqueue(item);
break;
}
}
}
public LLQueItem Dequeue()
{
while (true)
{
LLQueItem item = SendQueue.Dequeue();
if (item == null)
return null;
if (item.Incoming)
return item;
item.TickCount = System.Environment.TickCount;
if (item.Sequence == 0)
return item;
lock (contents)
{
if (contents.ContainsKey(item.Sequence))
{
if (contents[item.Sequence] == 1)
contents.Remove(item.Sequence);
else
contents[item.Sequence] -= 1;
return item;
}
}
}
}
public void Cancel(uint sequence)
{
lock (contents) contents.Remove(sequence);
}
public bool Contains(uint sequence)
{
lock (contents) return contents.ContainsKey(sequence);
}
public void Flush()
{
lock (this)
{
// These categories do not contain transactional packets so we can safely drop any pending data in them
LandOutgoingPacketQueue.Clear();
WindOutgoingPacketQueue.Clear();
CloudOutgoingPacketQueue.Clear();
TextureOutgoingPacketQueue.Clear();
AssetOutgoingPacketQueue.Clear();
// Now comes the fun part.. we dump all remaining resend and task packets into the send queue
while (ResendOutgoingPacketQueue.Count > 0 || TaskOutgoingPacketQueue.Count > 0 || TaskLowpriorityPacketQueue.Count > 0)
{
if (ResendOutgoingPacketQueue.Count > 0)
SendQueue.Enqueue(ResendOutgoingPacketQueue.Dequeue());
if (TaskOutgoingPacketQueue.Count > 0)
SendQueue.PriorityEnqueue(TaskOutgoingPacketQueue.Dequeue());
if (TaskLowpriorityPacketQueue.Count > 0)
SendQueue.Enqueue(TaskLowpriorityPacketQueue.Dequeue());
}
}
}
public void WipeClean()
{
lock (this)
{
ResendOutgoingPacketQueue.Clear();
LandOutgoingPacketQueue.Clear();
WindOutgoingPacketQueue.Clear();
CloudOutgoingPacketQueue.Clear();
TaskOutgoingPacketQueue.Clear();
TaskLowpriorityPacketQueue.Clear();
TextureOutgoingPacketQueue.Clear();
AssetOutgoingPacketQueue.Clear();
SendQueue.Clear();
lock (contents) contents.Clear();
}
}
public void Close()
{
Dispose();
}
public void Dispose()
{
Flush();
WipeClean(); // I'm sure there's a dirty joke in here somewhere. -AFrisby
m_enabled = false;
throttleTimer.Stop();
throttleTimer.Close();
if (StatsManager.SimExtraStats != null)
{
StatsManager.SimExtraStats.DeregisterPacketQueueStatsProvider(m_agentId);
}
}
private void ResetCounters()
{
ResendThrottle.Reset();
LandThrottle.Reset();
WindThrottle.Reset();
CloudThrottle.Reset();
TaskThrottle.Reset();
AssetThrottle.Reset();
TextureThrottle.Reset();
TotalThrottle.Reset();
}
private bool PacketsWaiting()
{
return (ResendOutgoingPacketQueue.Count > 0 ||
LandOutgoingPacketQueue.Count > 0 ||
WindOutgoingPacketQueue.Count > 0 ||
CloudOutgoingPacketQueue.Count > 0 ||
TaskOutgoingPacketQueue.Count > 0 ||
TaskLowpriorityPacketQueue.Count > 0 ||
AssetOutgoingPacketQueue.Count > 0 ||
TextureOutgoingPacketQueue.Count > 0);
}
public void ProcessThrottle()
{
// I was considering this.. Will an event fire if the thread it's on is blocked?
// Then I figured out.. it doesn't really matter.. because this thread won't be blocked for long
// The General overhead of the UDP protocol gets sent to the queue un-throttled by this
// so This'll pick up about around the right time.
int MaxThrottleLoops = 4550; // 50*7 packets can be dequeued at once.
int throttleLoops = 0;
List<ThrottleOutPacketType> e;
// We're going to dequeue all of the saved up packets until
// we've hit the throttle limit or there's no more packets to send
lock (this)
{
// this variable will be true if there was work done in the last execution of the
// loop, since each pass through the loop checks the queue length, we no longer
// need the check on entering the loop
bool qchanged = true;
ResetCounters();
while (TotalThrottle.UnderLimit() && qchanged && throttleLoops <= MaxThrottleLoops)
{
qchanged = false; // We will break out of the loop if no work was accomplished
throttleLoops++;
//Now comes the fun part.. we dump all our elements into m_packetQueue that we've saved up.
if ((ResendOutgoingPacketQueue.Count > 0) && ResendThrottle.UnderLimit())
{
LLQueItem qpack = ResendOutgoingPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
TotalThrottle.AddBytes(qpack.Length);
ResendThrottle.AddBytes(qpack.Length);
qchanged = true;
}
if ((LandOutgoingPacketQueue.Count > 0) && LandThrottle.UnderLimit())
{
LLQueItem qpack = LandOutgoingPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
TotalThrottle.AddBytes(qpack.Length);
LandThrottle.AddBytes(qpack.Length);
qchanged = true;
if (LandOutgoingPacketQueue.Count == 0 && !Empty.Contains(ThrottleOutPacketType.Land))
Empty.Add(ThrottleOutPacketType.Land);
}
if ((WindOutgoingPacketQueue.Count > 0) && WindThrottle.UnderLimit())
{
LLQueItem qpack = WindOutgoingPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
TotalThrottle.AddBytes(qpack.Length);
WindThrottle.AddBytes(qpack.Length);
qchanged = true;
if (WindOutgoingPacketQueue.Count == 0 && !Empty.Contains(ThrottleOutPacketType.Wind))
Empty.Add(ThrottleOutPacketType.Wind);
}
if ((CloudOutgoingPacketQueue.Count > 0) && CloudThrottle.UnderLimit())
{
LLQueItem qpack = CloudOutgoingPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
TotalThrottle.AddBytes(qpack.Length);
CloudThrottle.AddBytes(qpack.Length);
qchanged = true;
if (CloudOutgoingPacketQueue.Count == 0 && !Empty.Contains(ThrottleOutPacketType.Cloud))
Empty.Add(ThrottleOutPacketType.Cloud);
}
if ((TaskOutgoingPacketQueue.Count > 0 || TaskLowpriorityPacketQueue.Count > 0) && TaskThrottle.UnderLimit())
{
LLQueItem qpack;
if (TaskOutgoingPacketQueue.Count > 0)
{
qpack = TaskOutgoingPacketQueue.Dequeue();
SendQueue.PriorityEnqueue(qpack);
}
else
{
qpack = TaskLowpriorityPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
}
TotalThrottle.AddBytes(qpack.Length);
TaskThrottle.AddBytes(qpack.Length);
qchanged = true;
if (TaskOutgoingPacketQueue.Count == 0 && TaskLowpriorityPacketQueue.Count == 0 && !Empty.Contains(ThrottleOutPacketType.Task))
Empty.Add(ThrottleOutPacketType.Task);
}
if ((TextureOutgoingPacketQueue.Count > 0) && TextureThrottle.UnderLimit())
{
LLQueItem qpack = TextureOutgoingPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
TotalThrottle.AddBytes(qpack.Length);
TextureThrottle.AddBytes(qpack.Length);
qchanged = true;
if (TextureOutgoingPacketQueue.Count == 0 && !Empty.Contains(ThrottleOutPacketType.Texture))
Empty.Add(ThrottleOutPacketType.Texture);
}
if ((AssetOutgoingPacketQueue.Count > 0) && AssetThrottle.UnderLimit())
{
LLQueItem qpack = AssetOutgoingPacketQueue.Dequeue();
SendQueue.Enqueue(qpack);
TotalThrottle.AddBytes(qpack.Length);
AssetThrottle.AddBytes(qpack.Length);
qchanged = true;
if (AssetOutgoingPacketQueue.Count == 0 && !Empty.Contains(ThrottleOutPacketType.Asset))
Empty.Add(ThrottleOutPacketType.Asset);
}
}
// m_log.Info("[THROTTLE]: Processed " + throttleLoops + " packets");
e = new List<ThrottleOutPacketType>(Empty);
Empty.Clear();
}
foreach (ThrottleOutPacketType t in e)
{
if (GetQueueCount(t) == 0)
TriggerOnQueueEmpty(t);
}
}
private void TriggerOnQueueEmpty(ThrottleOutPacketType queue)
{
QueueEmpty handlerQueueEmpty = OnQueueEmpty;
if (handlerQueueEmpty != null)
handlerQueueEmpty(queue);
}
private void ThrottleTimerElapsed(object sender, ElapsedEventArgs e)
{
// just to change the signature, and that ProcessThrottle
// will be used elsewhere possibly
ProcessThrottle();
}
private void ThrottleCheck(ref LLPacketThrottle throttle, ref Queue<LLQueItem> q, LLQueItem item, ThrottleOutPacketType itemType)
{
// The idea.. is if the packet throttle queues are empty
// and the client is under throttle for the type. Queue
// it up directly. This basically short cuts having to
// wait for the timer to fire to put things into the
// output queue
if ((q.Count == 0) && (throttle.UnderLimit()))
{
try
{
Monitor.Enter(this);
throttle.AddBytes(item.Length);
TotalThrottle.AddBytes(item.Length);
SendQueue.Enqueue(item);
lock (this)
{
if (!Empty.Contains(itemType))
Empty.Add(itemType);
}
}
catch (Exception e)
{
// Probably a serialization exception
m_log.WarnFormat("ThrottleCheck: {0}", e.ToString());
}
finally
{
Monitor.Pulse(this);
Monitor.Exit(this);
}
}
else
{
q.Enqueue(item);
}
}
private static int ScaleThrottle(int value, int curmax, int newmax)
{
return (int)((value / (float)curmax) * newmax);
}
public byte[] GetThrottlesPacked(float multiplier)
{
int singlefloat = 4;
float tResend = ResendThrottle.Throttle*multiplier;
float tLand = LandThrottle.Throttle*multiplier;
float tWind = WindThrottle.Throttle*multiplier;
float tCloud = CloudThrottle.Throttle*multiplier;
float tTask = TaskThrottle.Throttle*multiplier;
float tTexture = TextureThrottle.Throttle*multiplier;
float tAsset = AssetThrottle.Throttle*multiplier;
byte[] throttles = new byte[singlefloat*7];
int i = 0;
Buffer.BlockCopy(BitConverter.GetBytes(tResend), 0, throttles, singlefloat*i, singlefloat);
i++;
Buffer.BlockCopy(BitConverter.GetBytes(tLand), 0, throttles, singlefloat*i, singlefloat);
i++;
Buffer.BlockCopy(BitConverter.GetBytes(tWind), 0, throttles, singlefloat*i, singlefloat);
i++;
Buffer.BlockCopy(BitConverter.GetBytes(tCloud), 0, throttles, singlefloat*i, singlefloat);
i++;
Buffer.BlockCopy(BitConverter.GetBytes(tTask), 0, throttles, singlefloat*i, singlefloat);
i++;
Buffer.BlockCopy(BitConverter.GetBytes(tTexture), 0, throttles, singlefloat*i, singlefloat);
i++;
Buffer.BlockCopy(BitConverter.GetBytes(tAsset), 0, throttles, singlefloat*i, singlefloat);
return throttles;
}
public void SetThrottleFromClient(byte[] throttle)
{
// From mantis http://opensimulator.org/mantis/view.php?id=1374
// it appears that sometimes we are receiving empty throttle byte arrays.
// TODO: Investigate this behaviour
if (throttle.Length == 0)
{
m_log.Warn("[PACKET QUEUE]: SetThrottleFromClient unexpectedly received a throttle byte array containing no elements!");
return;
}
int tResend = -1;
int tLand = -1;
int tWind = -1;
int tCloud = -1;
int tTask = -1;
int tTexture = -1;
int tAsset = -1;
int tall = -1;
int singlefloat = 4;
//Agent Throttle Block contains 7 single floatingpoint values.
int j = 0;
// Some Systems may be big endian...
// it might be smart to do this check more often...
if (!BitConverter.IsLittleEndian)
for (int i = 0; i < 7; i++)
Array.Reverse(throttle, j + i*singlefloat, singlefloat);
// values gotten from OpenMetaverse.org/wiki/Throttle. Thanks MW_
// bytes
// Convert to integer, since.. the full fp space isn't used.
tResend = (int) BitConverter.ToSingle(throttle, j);
j += singlefloat;
tLand = (int) BitConverter.ToSingle(throttle, j);
j += singlefloat;
tWind = (int) BitConverter.ToSingle(throttle, j);
j += singlefloat;
tCloud = (int) BitConverter.ToSingle(throttle, j);
j += singlefloat;
tTask = (int) BitConverter.ToSingle(throttle, j);
j += singlefloat;
tTexture = (int) BitConverter.ToSingle(throttle, j);
j += singlefloat;
tAsset = (int) BitConverter.ToSingle(throttle, j);
tall = tResend + tLand + tWind + tCloud + tTask + tTexture + tAsset;
/*
m_log.Info("[CLIENT]: Client AgentThrottle - Got throttle:resendbits=" + tResend +
" landbits=" + tLand +
" windbits=" + tWind +
" cloudbits=" + tCloud +
" taskbits=" + tTask +
" texturebits=" + tTexture +
" Assetbits=" + tAsset +
" Allbits=" + tall);
*/
// Total Sanity
// Make sure that the client sent sane total values.
// If the client didn't send acceptable values....
// Scale the clients values down until they are acceptable.
if (tall <= TotalThrottle.Max)
{
ResendThrottle.Throttle = tResend;
LandThrottle.Throttle = tLand;
WindThrottle.Throttle = tWind;
CloudThrottle.Throttle = tCloud;
TaskThrottle.Throttle = tTask;
TextureThrottle.Throttle = tTexture;
AssetThrottle.Throttle = tAsset;
TotalThrottle.Throttle = tall;
}
// else if (tall < 1)
// {
// // client is stupid, penalize him by minning everything
// ResendThrottle.Throttle = ResendThrottle.Min;
// LandThrottle.Throttle = LandThrottle.Min;
// WindThrottle.Throttle = WindThrottle.Min;
// CloudThrottle.Throttle = CloudThrottle.Min;
// TaskThrottle.Throttle = TaskThrottle.Min;
// TextureThrottle.Throttle = TextureThrottle.Min;
// AssetThrottle.Throttle = AssetThrottle.Min;
// TotalThrottle.Throttle = TotalThrottle.Min;
// }
else
{
// we're over so figure out percentages and use those
ResendThrottle.Throttle = tResend;
LandThrottle.Throttle = ScaleThrottle(tLand, tall, TotalThrottle.Max);
WindThrottle.Throttle = ScaleThrottle(tWind, tall, TotalThrottle.Max);
CloudThrottle.Throttle = ScaleThrottle(tCloud, tall, TotalThrottle.Max);
TaskThrottle.Throttle = ScaleThrottle(tTask, tall, TotalThrottle.Max);
TextureThrottle.Throttle = ScaleThrottle(tTexture, tall, TotalThrottle.Max);
AssetThrottle.Throttle = ScaleThrottle(tAsset, tall, TotalThrottle.Max);
TotalThrottle.Throttle = TotalThrottle.Max;
}
// effectively wiggling the slider causes things reset
// ResetCounters(); // DO NOT reset, better to send less for one period than more
}
// See IPullStatsProvider
public string GetStats()
{
return string.Format("{0,7} {1,7} {2,7} {3,7} {4,7} {5,7} {6,7} {7,7} {8,7} {9,7}",
SendQueue.Count(),
IncomingPacketQueue.Count,
OutgoingPacketQueue.Count,
ResendOutgoingPacketQueue.Count,
LandOutgoingPacketQueue.Count,
WindOutgoingPacketQueue.Count,
CloudOutgoingPacketQueue.Count,
TaskOutgoingPacketQueue.Count,
TextureOutgoingPacketQueue.Count,
AssetOutgoingPacketQueue.Count);
}
public LLQueItem[] GetQueueArray()
{
return SendQueue.GetQueueArray();
}
public float ThrottleMultiplier
{
get { return throttleMultiplier; }
}
public int GetQueueCount(ThrottleOutPacketType queue)
{
switch (queue)
{
case ThrottleOutPacketType.Land:
return LandOutgoingPacketQueue.Count;
case ThrottleOutPacketType.Wind:
return WindOutgoingPacketQueue.Count;
case ThrottleOutPacketType.Cloud:
return CloudOutgoingPacketQueue.Count;
case ThrottleOutPacketType.Task:
return TaskOutgoingPacketQueue.Count;
case ThrottleOutPacketType.Texture:
return TextureOutgoingPacketQueue.Count;
case ThrottleOutPacketType.Asset:
return AssetOutgoingPacketQueue.Count;
}
return 0;
}
}
}

206
OpenSim/Region/ClientStack/LindenUDP/LLPacketServer.cs
View File

@ -1,206 +0,0 @@
/*
* 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.Net;
using System.Net.Sockets;
using OpenMetaverse;
using OpenMetaverse.Packets;
using OpenSim.Framework;
namespace OpenSim.Region.ClientStack.LindenUDP
{
/// <summary>
/// This class sets up new client stacks. It also handles the immediate distribution of incoming packets to
/// client stacks
/// </summary>
public class LLPacketServer
{
// private static readonly log4net.ILog m_log
// = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
protected readonly ILLClientStackNetworkHandler m_networkHandler;
protected IScene m_scene;
/// <summary>
/// Tweakable user settings
/// </summary>
private ClientStackUserSettings m_userSettings;
public LLPacketServer(ILLClientStackNetworkHandler networkHandler, ClientStackUserSettings userSettings)
{
m_userSettings = userSettings;
m_networkHandler = networkHandler;
m_networkHandler.RegisterPacketServer(this);
}
public IScene LocalScene
{
set { m_scene = value; }
}
/// <summary>
/// Process an incoming packet.
/// </summary>
/// <param name="circuitCode"></param>
/// <param name="packet"></param>
public virtual void InPacket(uint circuitCode, Packet packet)
{
m_scene.ClientManager.InPacket(circuitCode, packet);
}
/// <summary>
/// Create a new client circuit
/// </summary>
/// <param name="remoteEP"></param>
/// <param name="scene"></param>
/// <param name="assetCache"></param>
/// <param name="packServer"></param>
/// <param name="sessionInfo"></param>
/// <param name="agentId"></param>
/// <param name="sessionId"></param>
/// <param name="circuitCode"></param>
/// <param name="proxyEP"></param>
/// <returns></returns>
protected virtual IClientAPI CreateNewCircuit(
EndPoint remoteEP, IScene scene,
LLPacketServer packServer, AuthenticateResponse sessionInfo,
UUID agentId, UUID sessionId, uint circuitCode, EndPoint proxyEP)
{
return
new LLClientView(
remoteEP, scene, packServer, sessionInfo, agentId, sessionId, circuitCode, proxyEP,
m_userSettings);
}
/// <summary>
/// Check whether a given client is authorized to connect.
/// </summary>
/// <param name="useCircuit"></param>
/// <param name="circuitManager"></param>
/// <param name="sessionInfo"></param>
/// <returns></returns>
public virtual bool IsClientAuthorized(
UseCircuitCodePacket useCircuit, AgentCircuitManager circuitManager, out AuthenticateResponse sessionInfo)
{
UUID agentId = useCircuit.CircuitCode.ID;
UUID sessionId = useCircuit.CircuitCode.SessionID;
uint circuitCode = useCircuit.CircuitCode.Code;
sessionInfo = circuitManager.AuthenticateSession(sessionId, agentId, circuitCode);
if (!sessionInfo.Authorised)
return false;
return true;
}
/// <summary>
/// Add a new client circuit. We assume that is has already passed an authorization check
/// </summary>
/// <param name="epSender"></param>
/// <param name="useCircuit"></param>
/// <param name="assetCache"></param>
/// <param name="sessionInfo"></param>
/// <param name="proxyEP"></param>
/// <returns>
/// true if a new circuit was created, false if a circuit with the given circuit code already existed
/// </returns>
public virtual bool AddNewClient(
EndPoint epSender, UseCircuitCodePacket useCircuit,
AuthenticateResponse sessionInfo, EndPoint proxyEP)
{
IClientAPI newuser;
uint circuitCode = useCircuit.CircuitCode.Code;
if (m_scene.ClientManager.TryGetClient(circuitCode, out newuser))
{
// The circuit is already known to the scene. This not actually a problem since this will currently
// occur if a client is crossing borders (hence upgrading its circuit). However, we shouldn't
// really by trying to add a new client if this is the case.
return false;
}
UUID agentId = useCircuit.CircuitCode.ID;
UUID sessionId = useCircuit.CircuitCode.SessionID;
newuser
= CreateNewCircuit(
epSender, m_scene, this, sessionInfo, agentId, sessionId, circuitCode, proxyEP);
m_scene.ClientManager.Add(circuitCode, newuser);
newuser.OnViewerEffect += m_scene.ClientManager.ViewerEffectHandler;
newuser.OnLogout += LogoutHandler;
newuser.OnConnectionClosed += CloseClient;
newuser.Start();
return true;
}
public void LogoutHandler(IClientAPI client)
{
client.SendLogoutPacket();
CloseClient(client);
}
/// <summary>
/// Send a packet to the given circuit
/// </summary>
/// <param name="buffer"></param>
/// <param name="size"></param>
/// <param name="flags"></param>
/// <param name="circuitcode"></param>
public virtual void SendPacketTo(byte[] buffer, int size, SocketFlags flags, uint circuitcode)
{
m_networkHandler.SendPacketTo(buffer, size, flags, circuitcode);
}
/// <summary>
/// Close a client circuit only
/// </summary>
/// <param name="circuitcode"></param>
public virtual void CloseCircuit(uint circuitcode)
{
m_networkHandler.RemoveClientCircuit(circuitcode);
}
/// <summary>
/// Completely close down the given client.
/// </summary>
/// <param name="client"></param>
public virtual void CloseClient(IClientAPI client)
{
//m_log.Info("PacketServer:CloseClient()");
CloseCircuit(client.CircuitCode);
m_scene.ClientManager.Remove(client.CircuitCode);
client.Close(false);
}
}
}

128
OpenSim/Region/ClientStack/LindenUDP/LLPacketThrottle.cs
View File

@ -1,128 +0,0 @@
/*
* 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.
*/
namespace OpenSim.Region.ClientStack.LindenUDP
{
public class LLPacketThrottle
{
private readonly int m_maxAllowableThrottle;
private readonly int m_minAllowableThrottle;
private int m_currentThrottle;
private const int m_throttleTimeDivisor = 7;
private int m_currentBitsSent;
private int m_throttleBits;
/// <value>
/// Value with which to multiply all the throttle fields
/// </value>
private float m_throttleMultiplier;
public int Max
{
get { return m_maxAllowableThrottle; }
}
public int Min
{
get { return m_minAllowableThrottle; }
}
public int Current
{
get { return m_currentThrottle; }
}
/// <summary>
/// Constructor.
/// </summary>
/// <param name="min"></param>
/// <param name="max"></param>
/// <param name="throttle"></param>
/// <param name="throttleMultiplier">
/// A parameter that's ends up multiplying all throttle settings. An alternative solution would have been
/// to multiply all the parameters by this before giving them to the constructor. But doing it this way
/// represents the fact that the multiplier is a hack that pumps data to clients much faster than the actual
/// settings that we are given.
/// </param>
public LLPacketThrottle(int min, int max, int throttle, float throttleMultiplier)
{
m_throttleMultiplier = throttleMultiplier;
m_maxAllowableThrottle = max;
m_minAllowableThrottle = min;
m_currentThrottle = throttle;
m_currentBitsSent = 0;
CalcBits();
}
/// <summary>
/// Calculate the actual throttle required.
/// </summary>
private void CalcBits()
{
m_throttleBits = (int)((float)m_currentThrottle * m_throttleMultiplier / (float)m_throttleTimeDivisor);
}
public void Reset()
{
m_currentBitsSent = 0;
}
public bool UnderLimit()
{
return m_currentBitsSent < m_throttleBits;
}
public int AddBytes(int bytes)
{
m_currentBitsSent += bytes * 8;
return m_currentBitsSent;
}
public int Throttle
{
get { return m_currentThrottle; }
set
{
if (value < m_minAllowableThrottle)
{
m_currentThrottle = m_minAllowableThrottle;
}
else if (value > m_maxAllowableThrottle)
{
m_currentThrottle = m_maxAllowableThrottle;
}
else
{
m_currentThrottle = value;
}
CalcBits();
}
}
}
}

442
OpenSim/Region/ClientStack/LindenUDP/LLUDPClient.cs Normal file
View File

@ -0,0 +1,442 @@
/*
* 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.Net;
using OpenSim.Framework;
using OpenMetaverse;
namespace OpenSim.Region.ClientStack.LindenUDP
{
#region Delegates
/// <summary>
/// Fired when updated networking stats are produced for this client
/// </summary>
/// <param name="inPackets">Number of incoming packets received since this
/// event was last fired</param>
/// <param name="outPackets">Number of outgoing packets sent since this
/// event was last fired</param>
/// <param name="unAckedBytes">Current total number of bytes in packets we
/// are waiting on ACKs for</param>
public delegate void PacketStats(int inPackets, int outPackets, int unAckedBytes);
/// <summary>
/// Fired when the queue for a packet category is empty. This event can be
/// hooked to put more data on the empty queue
/// </summary>
/// <param name="category">Category of the packet queue that is empty</param>
public delegate void QueueEmpty(ThrottleOutPacketType category);
#endregion Delegates
/// <summary>
/// Tracks state for a client UDP connection and provides client-specific methods
/// </summary>
public sealed class LLUDPClient
{
/// <summary>The number of packet categories to throttle on. If a throttle category is added
/// or removed, this number must also change</summary>
const int THROTTLE_CATEGORY_COUNT = 7;
/// <summary>Fired when updated networking stats are produced for this client</summary>
public event PacketStats OnPacketStats;
/// <summary>Fired when the queue for a packet category is empty. This event can be
/// hooked to put more data on the empty queue</summary>
public event QueueEmpty OnQueueEmpty;
/// <summary>AgentID for this client</summary>
public readonly UUID AgentID;
/// <summary>The remote address of the connected client</summary>
public readonly IPEndPoint RemoteEndPoint;
/// <summary>Circuit code that this client is connected on</summary>
public readonly uint CircuitCode;
/// <summary>Sequence numbers of packets we've received (for duplicate checking)</summary>
public readonly IncomingPacketHistoryCollection PacketArchive = new IncomingPacketHistoryCollection(200);
/// <summary>Packets we have sent that need to be ACKed by the client</summary>
public readonly UnackedPacketCollection NeedAcks = new UnackedPacketCollection();
/// <summary>ACKs that are queued up, waiting to be sent to the client</summary>
public readonly LocklessQueue<uint> PendingAcks = new LocklessQueue<uint>();
/// <summary>Reference to the IClientAPI for this client</summary>
public LLClientView ClientAPI;
/// <summary>Current packet sequence number</summary>
public int CurrentSequence;
/// <summary>Current ping sequence number</summary>
public byte CurrentPingSequence;
/// <summary>True when this connection is alive, otherwise false</summary>
public bool IsConnected = true;
/// <summary>True when this connection is paused, otherwise false</summary>
public bool IsPaused = true;
/// <summary>Environment.TickCount when the last packet was received for this client</summary>
public int TickLastPacketReceived;
/// <summary>Timer granularity. This is set to the measured resolution of Environment.TickCount</summary>
public readonly float G;
/// <summary>Smoothed round-trip time. A smoothed average of the round-trip time for sending a
/// reliable packet to the client and receiving an ACK</summary>
public float SRTT;
/// <summary>Round-trip time variance. Measures the consistency of round-trip times</summary>
public float RTTVAR;
/// <summary>Retransmission timeout. Packets that have not been acknowledged in this number of
/// milliseconds or longer will be resent</summary>
/// <remarks>Calculated from <seealso cref="SRTT"/> and <seealso cref="RTTVAR"/> using the
/// guidelines in RFC 2988</remarks>
public int RTO;
/// <summary>Number of bytes received since the last acknowledgement was sent out. This is used
/// to loosely follow the TCP delayed ACK algorithm in RFC 1122 (4.2.3.2)</summary>
public int BytesSinceLastACK;
/// <summary>Number of packets received from this client</summary>
public int PacketsReceived;
/// <summary>Number of packets sent to this client</summary>
public int PacketsSent;
/// <summary>Total byte count of unacked packets sent to this client</summary>
public int UnackedBytes;
/// <summary>Total number of received packets that we have reported to the OnPacketStats event(s)</summary>
private int m_packetsReceivedReported;
/// <summary>Total number of sent packets that we have reported to the OnPacketStats event(s)</summary>
private int m_packetsSentReported;
/// <summary>Throttle bucket for this agent's connection</summary>
private readonly TokenBucket throttle;
/// <summary>Throttle buckets for each packet category</summary>
private readonly TokenBucket[] throttleCategories;
/// <summary>Throttle rate defaults and limits</summary>
private readonly ThrottleRates defaultThrottleRates;
/// <summary>Outgoing queues for throttled packets</summary>
private readonly LocklessQueue<OutgoingPacket>[] packetOutboxes = new LocklessQueue<OutgoingPacket>[THROTTLE_CATEGORY_COUNT];
/// <summary>A container that can hold one packet for each outbox, used to store
/// dequeued packets that are being held for throttling</summary>
private readonly OutgoingPacket[] nextPackets = new OutgoingPacket[THROTTLE_CATEGORY_COUNT];
/// <summary>An optimization to store the length of dequeued packets being held
/// for throttling. This avoids expensive calls to Packet.Length</summary>
private readonly int[] nextPacketLengths = new int[THROTTLE_CATEGORY_COUNT];
/// <summary>A reference to the LLUDPServer that is managing this client</summary>
private readonly LLUDPServer udpServer;
/// <summary>
/// Default constructor
/// </summary>
/// <param name="server">Reference to the UDP server this client is connected to</param>
/// <param name="rates">Default throttling rates and maximum throttle limits</param>
/// <param name="parentThrottle">Parent HTB (hierarchical token bucket)
/// that the child throttles will be governed by</param>
/// <param name="circuitCode">Circuit code for this connection</param>
/// <param name="agentID">AgentID for the connected agent</param>
/// <param name="remoteEndPoint">Remote endpoint for this connection</param>
public LLUDPClient(LLUDPServer server, ThrottleRates rates, TokenBucket parentThrottle, uint circuitCode, UUID agentID, IPEndPoint remoteEndPoint)
{
udpServer = server;
AgentID = agentID;
RemoteEndPoint = remoteEndPoint;
CircuitCode = circuitCode;
defaultThrottleRates = rates;
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
packetOutboxes[i] = new LocklessQueue<OutgoingPacket>();
throttle = new TokenBucket(parentThrottle, 0, 0);
throttleCategories = new TokenBucket[THROTTLE_CATEGORY_COUNT];
throttleCategories[(int)ThrottleOutPacketType.Resend] = new TokenBucket(throttle, rates.ResendLimit, rates.Resend);
throttleCategories[(int)ThrottleOutPacketType.Land] = new TokenBucket(throttle, rates.LandLimit, rates.Land);
throttleCategories[(int)ThrottleOutPacketType.Wind] = new TokenBucket(throttle, rates.WindLimit, rates.Wind);
throttleCategories[(int)ThrottleOutPacketType.Cloud] = new TokenBucket(throttle, rates.CloudLimit, rates.Cloud);
throttleCategories[(int)ThrottleOutPacketType.Task] = new TokenBucket(throttle, rates.TaskLimit, rates.Task);
throttleCategories[(int)ThrottleOutPacketType.Texture] = new TokenBucket(throttle, rates.TextureLimit, rates.Texture);
throttleCategories[(int)ThrottleOutPacketType.Asset] = new TokenBucket(throttle, rates.AssetLimit, rates.Asset);
// Set the granularity variable used for retransmission calculations to
// the measured resolution of Environment.TickCount
G = server.TickCountResolution;
// Default the retransmission timeout to three seconds
RTO = 3000;
}
/// <summary>
/// Shuts down this client connection
/// </summary>
public void Shutdown()
{
// TODO: Do we need to invalidate the circuit?
IsConnected = false;
}
/// <summary>
/// Gets information about this client connection
/// </summary>
/// <returns>Information about the client connection</returns>
public ClientInfo GetClientInfo()
{
// TODO: This data structure is wrong in so many ways. Locking and copying the entire lists
// of pending and needed ACKs for every client every time some method wants information about
// this connection is a recipe for poor performance
ClientInfo info = new ClientInfo();
info.pendingAcks = new Dictionary<uint, uint>();
info.needAck = new Dictionary<uint, byte[]>();
info.resendThrottle = throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate;
info.landThrottle = throttleCategories[(int)ThrottleOutPacketType.Land].DripRate;
info.windThrottle = throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate;
info.cloudThrottle = throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate;
info.taskThrottle = throttleCategories[(int)ThrottleOutPacketType.Task].DripRate;
info.assetThrottle = throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate;
info.textureThrottle = throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate;
info.totalThrottle = info.resendThrottle + info.landThrottle + info.windThrottle + info.cloudThrottle +
info.taskThrottle + info.assetThrottle + info.textureThrottle;
return info;
}
/// <summary>
/// Modifies the UDP throttles
/// </summary>
/// <param name="info">New throttling values</param>
public void SetClientInfo(ClientInfo info)
{
// TODO: Allowing throttles to be manually set from this function seems like a reasonable
// idea. On the other hand, letting external code manipulate our ACK accounting is not
// going to happen
throw new NotImplementedException();
}
public string GetStats()
{
// TODO: ???
return string.Format("{0,7} {1,7} {2,7} {3,7} {4,7} {5,7} {6,7} {7,7} {8,7} {9,7}",
0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
}
public void SendPacketStats()
{
PacketStats callback = OnPacketStats;
if (callback != null)
{
int newPacketsReceived = PacketsReceived - m_packetsReceivedReported;
int newPacketsSent = PacketsSent - m_packetsSentReported;
callback(newPacketsReceived, newPacketsSent, UnackedBytes);
m_packetsReceivedReported += newPacketsReceived;
m_packetsSentReported += newPacketsSent;
}
}
public void SetThrottles(byte[] throttleData)
{
byte[] adjData;
int pos = 0;
if (!BitConverter.IsLittleEndian)
{
byte[] newData = new byte[7 * 4];
Buffer.BlockCopy(throttleData, 0, newData, 0, 7 * 4);
for (int i = 0; i < 7; i++)
Array.Reverse(newData, i * 4, 4);
adjData = newData;
}
else
{
adjData = throttleData;
}
int resend = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int land = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int wind = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int cloud = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int task = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int texture = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f); pos += 4;
int asset = (int)(BitConverter.ToSingle(adjData, pos) * 0.125f);
resend = (resend <= defaultThrottleRates.ResendLimit) ? resend : defaultThrottleRates.ResendLimit;
land = (land <= defaultThrottleRates.LandLimit) ? land : defaultThrottleRates.LandLimit;
wind = (wind <= defaultThrottleRates.WindLimit) ? wind : defaultThrottleRates.WindLimit;
cloud = (cloud <= defaultThrottleRates.CloudLimit) ? cloud : defaultThrottleRates.CloudLimit;
task = (task <= defaultThrottleRates.TaskLimit) ? task : defaultThrottleRates.TaskLimit;
texture = (texture <= defaultThrottleRates.TextureLimit) ? texture : defaultThrottleRates.TextureLimit;
asset = (asset <= defaultThrottleRates.AssetLimit) ? asset : defaultThrottleRates.AssetLimit;
SetThrottle(ThrottleOutPacketType.Resend, resend);
SetThrottle(ThrottleOutPacketType.Land, land);
SetThrottle(ThrottleOutPacketType.Wind, wind);
SetThrottle(ThrottleOutPacketType.Cloud, cloud);
SetThrottle(ThrottleOutPacketType.Task, task);
SetThrottle(ThrottleOutPacketType.Texture, texture);
SetThrottle(ThrottleOutPacketType.Asset, asset);
}
public byte[] GetThrottlesPacked()
{
byte[] data = new byte[7 * 4];
int i = 0;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Resend].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Land].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Wind].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Cloud].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Task].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Texture].DripRate), 0, data, i, 4); i += 4;
Buffer.BlockCopy(Utils.FloatToBytes((float)throttleCategories[(int)ThrottleOutPacketType.Asset].DripRate), 0, data, i, 4); i += 4;
return data;
}
public void SetThrottle(ThrottleOutPacketType category, int rate)
{
int i = (int)category;
if (i >= 0 && i < throttleCategories.Length)
{
TokenBucket bucket = throttleCategories[(int)category];
bucket.MaxBurst = rate;
bucket.DripRate = rate;
}
}
public bool EnqueueOutgoing(OutgoingPacket packet)
{
int category = (int)packet.Category;
if (category >= 0 && category < packetOutboxes.Length)
{
LocklessQueue<OutgoingPacket> queue = packetOutboxes[category];
TokenBucket bucket = throttleCategories[category];
if (throttleCategories[category].RemoveTokens(packet.Buffer.DataLength))
{
// Enough tokens were removed from the bucket, the packet will not be queued
return false;
}
else
{
// Not enough tokens in the bucket, queue this packet
queue.Enqueue(packet);
return true;
}
}
else
{
// We don't have a token bucket for this category, so it will not be queued
return false;
}
}
/// <summary>
/// Loops through all of the packet queues for this client and tries to send
/// any outgoing packets, obeying the throttling bucket limits
/// </summary>
/// <remarks>This function is only called from a synchronous loop in the
/// UDPServer so we don't need to bother making this thread safe</remarks>
/// <returns>True if any packets were sent, otherwise false</returns>
public bool DequeueOutgoing()
{
OutgoingPacket packet;
LocklessQueue<OutgoingPacket> queue;
TokenBucket bucket;
bool packetSent = false;
for (int i = 0; i < THROTTLE_CATEGORY_COUNT; i++)
{
bucket = throttleCategories[i];
if (nextPackets[i] != null)
{
// This bucket was empty the last time we tried to send a packet,
// leaving a dequeued packet still waiting to be sent out. Try to
// send it again
if (bucket.RemoveTokens(nextPacketLengths[i]))
{
// Send the packet
udpServer.SendPacketFinal(nextPackets[i]);
nextPackets[i] = null;
packetSent = true;
}
}
else
{
// No dequeued packet waiting to be sent, try to pull one off
// this queue
queue = packetOutboxes[i];
if (queue.Dequeue(out packet))
{
// A packet was pulled off the queue. See if we have
// enough tokens in the bucket to send it out
if (bucket.RemoveTokens(packet.Buffer.DataLength))
{
// Send the packet
udpServer.SendPacketFinal(packet);
packetSent = true;
}
else
{
// Save the dequeued packet and the length calculation for
// the next iteration
nextPackets[i] = packet;
nextPacketLengths[i] = packet.Buffer.DataLength;
}
}
else
{
// No packets in this queue. Fire the queue empty callback
QueueEmpty callback = OnQueueEmpty;
if (callback != null)
callback((ThrottleOutPacketType)i);
}
}
}
return packetSent;
}
public void UpdateRoundTrip(float r)
{
const float ALPHA = 0.125f;
const float BETA = 0.25f;
const float K = 4.0f;
if (RTTVAR == 0.0f)
{
// First RTT measurement
SRTT = r;
RTTVAR = r * 0.5f;
}
else
{
// Subsequence RTT measurement
RTTVAR = (1.0f - BETA) * RTTVAR + BETA * Math.Abs(SRTT - r);
SRTT = (1.0f - ALPHA) * SRTT + ALPHA * r;
}
// Always round retransmission timeout up to two seconds
RTO = Math.Max(2000, (int)(SRTT + Math.Max(G, K * RTTVAR)));
//Logger.Debug("Setting agent " + this.Agent.FullName + "'s RTO to " + RTO + "ms with an RTTVAR of " +
// RTTVAR + " based on new RTT of " + r + "ms");
}
}
}

282
OpenSim/Region/ClientStack/LindenUDP/LLUDPClientCollection.cs Normal file
View File

@ -0,0 +1,282 @@
/*
* 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.Net;
using OpenSim.Framework;
using OpenMetaverse;
using ReaderWriterLockImpl = OpenMetaverse.ReaderWriterLockSlim;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public sealed class UDPClientCollection
{
Dictionary<UUID, LLUDPClient> Dictionary1;
Dictionary<IPEndPoint, LLUDPClient> Dictionary2;
LLUDPClient[] Array;
ReaderWriterLockImpl rwLock = new ReaderWriterLockImpl();
object m_sync = new object();
public UDPClientCollection()
{
Dictionary1 = new Dictionary<UUID, LLUDPClient>();
Dictionary2 = new Dictionary<IPEndPoint, LLUDPClient>();
Array = new LLUDPClient[0];
}
public UDPClientCollection(int capacity)
{
Dictionary1 = new Dictionary<UUID, LLUDPClient>(capacity);
Dictionary2 = new Dictionary<IPEndPoint, LLUDPClient>(capacity);
Array = new LLUDPClient[0];
}
public void Add(UUID key1, IPEndPoint key2, LLUDPClient value)
{
//rwLock.EnterWriteLock();
//try
//{
// if (Dictionary1.ContainsKey(key1))
// {
// if (!Dictionary2.ContainsKey(key2))
// throw new ArgumentException("key1 exists in the dictionary but not key2");
// }
// else if (Dictionary2.ContainsKey(key2))
// {
// if (!Dictionary1.ContainsKey(key1))
// throw new ArgumentException("key2 exists in the dictionary but not key1");
// }
// Dictionary1[key1] = value;
// Dictionary2[key2] = value;
// LLUDPClient[] oldArray = Array;
// int oldLength = oldArray.Length;
// LLUDPClient[] newArray = new LLUDPClient[oldLength + 1];
// for (int i = 0; i < oldLength; i++)
// newArray[i] = oldArray[i];
// newArray[oldLength] = value;
// Array = newArray;
//}
//finally { rwLock.ExitWriteLock(); }
lock (m_sync)
{
if (Dictionary1.ContainsKey(key1))
{
if (!Dictionary2.ContainsKey(key2))
throw new ArgumentException("key1 exists in the dictionary but not key2");
}
else if (Dictionary2.ContainsKey(key2))
{
if (!Dictionary1.ContainsKey(key1))
throw new ArgumentException("key2 exists in the dictionary but not key1");
}
Dictionary1[key1] = value;
Dictionary2[key2] = value;
LLUDPClient[] oldArray = Array;
int oldLength = oldArray.Length;
LLUDPClient[] newArray = new LLUDPClient[oldLength + 1];
for (int i = 0; i < oldLength; i++)
newArray[i] = oldArray[i];
newArray[oldLength] = value;
Array = newArray;
}
}
public bool Remove(UUID key1, IPEndPoint key2)
{
//rwLock.EnterWriteLock();
//try
//{
// LLUDPClient value;
// if (Dictionary1.TryGetValue(key1, out value))
// {
// Dictionary1.Remove(key1);
// Dictionary2.Remove(key2);
// LLUDPClient[] oldArray = Array;
// int oldLength = oldArray.Length;
// LLUDPClient[] newArray = new LLUDPClient[oldLength - 1];
// int j = 0;
// for (int i = 0; i < oldLength; i++)
// {
// if (oldArray[i] != value)
// newArray[j++] = oldArray[i];
// }
// Array = newArray;
// return true;
// }
//}
//finally { rwLock.ExitWriteLock(); }
//return false;
lock (m_sync)
{
LLUDPClient value;
if (Dictionary1.TryGetValue(key1, out value))
{
Dictionary1.Remove(key1);
Dictionary2.Remove(key2);
LLUDPClient[] oldArray = Array;
int oldLength = oldArray.Length;
LLUDPClient[] newArray = new LLUDPClient[oldLength - 1];
int j = 0;
for (int i = 0; i < oldLength; i++)
{
if (oldArray[i] != value)
newArray[j++] = oldArray[i];
}
Array = newArray;
return true;
}
}
return false;
}
public void Clear()
{
//rwLock.EnterWriteLock();
//try
//{
// Dictionary1.Clear();
// Dictionary2.Clear();
// Array = new LLUDPClient[0];
//}
//finally { rwLock.ExitWriteLock(); }
lock (m_sync)
{
Dictionary1.Clear();
Dictionary2.Clear();
Array = new LLUDPClient[0];
}
}
public int Count
{
get { return Array.Length; }
}
public bool ContainsKey(UUID key)
{
return Dictionary1.ContainsKey(key);
}
public bool ContainsKey(IPEndPoint key)
{
return Dictionary2.ContainsKey(key);
}
public bool TryGetValue(UUID key, out LLUDPClient value)
{
////bool success;
////bool doLock = !rwLock.IsUpgradeableReadLockHeld;
////if (doLock) rwLock.EnterReadLock();
////try { success = Dictionary1.TryGetValue(key, out value); }
////finally { if (doLock) rwLock.ExitReadLock(); }
////return success;
//lock (m_sync)
// return Dictionary1.TryGetValue(key, out value);
try
{
return Dictionary1.TryGetValue(key, out value);
}
catch { }
value = null;
return false;
}
public bool TryGetValue(IPEndPoint key, out LLUDPClient value)
{
////bool success;
////bool doLock = !rwLock.IsUpgradeableReadLockHeld;
////if (doLock) rwLock.EnterReadLock();
////try { success = Dictionary2.TryGetValue(key, out value); }
////finally { if (doLock) rwLock.ExitReadLock(); }
////return success;
lock (m_sync)
return Dictionary2.TryGetValue(key, out value);
//try
//{
// return Dictionary2.TryGetValue(key, out value);
//}
//catch { }
//value = null;
//return false;
}
public void ForEach(Action<LLUDPClient> action)
{
//bool doLock = !rwLock.IsUpgradeableReadLockHeld;
//if (doLock) rwLock.EnterUpgradeableReadLock();
//try { Parallel.ForEach<LLUDPClient>(Array, action); }
//finally { if (doLock) rwLock.ExitUpgradeableReadLock(); }
LLUDPClient[] localArray = null;
lock (m_sync)
{
localArray = new LLUDPClient[Array.Length];
Array.CopyTo(localArray, 0);
}
Parallel.ForEach<LLUDPClient>(localArray, action);
}
}
}

1143
OpenSim/Region/ClientStack/LindenUDP/LLUDPServer.cs
View File

File diff suppressed because it is too large Load Diff

47
OpenSim/Region/ClientStack/LindenUDP/LLUtil.cs → OpenSim/Region/ClientStack/LindenUDP/OutgoingPacket.cs
View File

@ -1,4 +1,4 @@
/*
/*
* Copyright (c) Contributors, http://opensimulator.org/
* See CONTRIBUTORS.TXT for a full list of copyright holders.
*
@ -25,27 +25,46 @@
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
using System;
using OpenSim.Framework;
using OpenMetaverse;
namespace OpenSim.Region.ClientStack.LindenUDP
{
public class LLUtil
{
/// <summary>
/// Convert a string to bytes suitable for use in an LL UDP packet.
/// Holds a reference to the <seealso cref="LLUDPClient"/> this packet is
/// destined for, along with the serialized packet data, sequence number
/// (if this is a resend), number of times this packet has been resent,
/// the time of the last resend, and the throttling category for this
/// packet
/// </summary>
/// <param name="s">Truncated to 254 characters if necessary</param>
/// <returns></returns>
public static byte[] StringToPacketBytes(string s)
public sealed class OutgoingPacket
{
// Anything more than 254 will cause libsecondlife to barf
// (libsl 1550) adds an \0 on the Utils.StringToBytes conversion if it isn't present
if (s.Length > 254)
{
s = s.Remove(254);
}
/// <summary>Client this packet is destined for</summary>
public LLUDPClient Client;
/// <summary>Packet data to send</summary>
public UDPPacketBuffer Buffer;
/// <summary>Sequence number of the wrapped packet</summary>
public uint SequenceNumber;
/// <summary>Number of times this packet has been resent</summary>
public int ResendCount;
/// <summary>Environment.TickCount when this packet was last sent over the wire</summary>
public int TickCount;
/// <summary>Category this packet belongs to</summary>
public ThrottleOutPacketType Category;
return Utils.StringToBytes(s);
/// <summary>
/// Default constructor
/// </summary>
/// <param name="client">Reference to the client this packet is destined for</param>
/// <param name="buffer">Serialized packet data. If the flags or sequence number
/// need to be updated, they will be injected directly into this binary buffer</param>
/// <param name="category">Throttling category for this packet</param>
public OutgoingPacket(LLUDPClient client, UDPPacketBuffer buffer, ThrottleOutPacketType category)
{
Client = client;
Buffer = buffer;
Category = category;
}
}
}

2
OpenSim/Region/ClientStack/LindenUDP/Tests/PacketHandlerTests.cs
View File

@ -70,7 +70,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP.Tests
TestClient testClient = new TestClient(agent, scene);
ILLPacketHandler packetHandler
LLPacketHandler packetHandler
= new LLPacketHandler(testClient, testLLPacketServer, new ClientStackUserSettings());
packetHandler.InPacket(new AgentAnimationPacket());

2
OpenSim/Region/ClientStack/LindenUDP/Tests/TestLLPacketServer.cs
View File

@ -37,7 +37,7 @@ namespace OpenSim.Region.ClientStack.LindenUDP.Tests
/// </summary>
protected Dictionary<PacketType, int> m_packetsReceived = new Dictionary<PacketType, int>();
public TestLLPacketServer(ILLClientStackNetworkHandler networkHandler, ClientStackUserSettings userSettings)
public TestLLPacketServer(LLUDPServer networkHandler, ClientStackUserSettings userSettings)
: base(networkHandler, userSettings)
{}

99
OpenSim/Region/ClientStack/LindenUDP/ThrottleRates.cs Normal file
View File

@ -0,0 +1,99 @@
/*
* 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 Nini.Config;
namespace OpenSim.Region.ClientStack.LindenUDP
{
/// <summary>
/// Holds drip rates and maximum burst rates for throttling with hierarchical
/// token buckets. The maximum burst rates set here are hard limits and can
/// not be overridden by client requests
/// </summary>
public sealed class ThrottleRates
{
/// <summary>Drip rate for resent packets</summary>
public int Resend;
/// <summary>Drip rate for terrain packets</summary>
public int Land;
/// <summary>Drip rate for wind packets</summary>
public int Wind;
/// <summary>Drip rate for cloud packets</summary>
public int Cloud;
/// <summary>Drip rate for task (state and transaction) packets</summary>
public int Task;
/// <summary>Drip rate for texture packets</summary>
public int Texture;
/// <summary>Drip rate for asset packets</summary>
public int Asset;
/// <summary>Maximum burst rate for resent packets</summary>
public int ResendLimit;
/// <summary>Maximum burst rate for land packets</summary>
public int LandLimit;
/// <summary>Maximum burst rate for wind packets</summary>
public int WindLimit;
/// <summary>Maximum burst rate for cloud packets</summary>
public int CloudLimit;
/// <summary>Maximum burst rate for task (state and transaction) packets</summary>
public int TaskLimit;
/// <summary>Maximum burst rate for texture packets</summary>
public int TextureLimit;
/// <summary>Maximum burst rate for asset packets</summary>
public int AssetLimit;
/// <summary>
/// Default constructor
/// </summary>
/// <param name="config">Config source to load defaults from</param>
public ThrottleRates(IConfigSource config)
{
try
{
IConfig throttleConfig = config.Configs["ClientStack.LindenUDP"];
Resend = throttleConfig.GetInt("ResendDefault", 12500);
Land = throttleConfig.GetInt("LandDefault", 500);
Wind = throttleConfig.GetInt("WindDefault", 500);
Cloud = throttleConfig.GetInt("CloudDefault", 500);
Task = throttleConfig.GetInt("TaskDefault", 500);
Texture = throttleConfig.GetInt("TextureDefault", 500);
Asset = throttleConfig.GetInt("AssetDefault", 500);
ResendLimit = throttleConfig.GetInt("ResendLimit", 18750);
LandLimit = throttleConfig.GetInt("LandLimit", 29750);
WindLimit = throttleConfig.GetInt("WindLimit", 18750);
CloudLimit = throttleConfig.GetInt("CloudLimit", 18750);
TaskLimit = throttleConfig.GetInt("TaskLimit", 55750);
TextureLimit = throttleConfig.GetInt("TextureLimit", 55750);
AssetLimit = throttleConfig.GetInt("AssetLimit", 27500);
}
catch (Exception) { }
}
}
}

160
OpenSim/Region/ClientStack/LindenUDP/UnackedPacketCollection.cs Normal file
View File

@ -0,0 +1,160 @@
/*
* 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.Net;
using OpenMetaverse;
namespace OpenSim.Region.ClientStack.LindenUDP
{
/// <summary>
/// Special collection that is optimized for tracking unacknowledged packets
/// </summary>
public sealed class UnackedPacketCollection
{
/// <summary>Synchronization primitive. A lock must be acquired on this
/// object before calling any of the unsafe methods</summary>
public object SyncRoot = new object();
/// <summary>Holds the actual unacked packet data, sorted by sequence number</summary>
private SortedDictionary<uint, OutgoingPacket> packets = new SortedDictionary<uint, OutgoingPacket>();
/// <summary>Gets the total number of unacked packets</summary>
public int Count { get { return packets.Count; } }
/// <summary>
/// Default constructor
/// </summary>
public UnackedPacketCollection()
{
}
/// <summary>
/// Add an unacked packet to the collection
/// </summary>
/// <param name="packet">Packet that is awaiting acknowledgement</param>
/// <returns>True if the packet was successfully added, false if the
/// packet already existed in the collection</returns>
public bool Add(OutgoingPacket packet)
{
lock (SyncRoot)
{
if (!packets.ContainsKey(packet.SequenceNumber))
{
packets.Add(packet.SequenceNumber, packet);
return true;
}
return false;
}
}
/// <summary>
/// Removes a packet from the collection without attempting to obtain a
/// lock first
/// </summary>
/// <param name="sequenceNumber">Sequence number of the packet to remove</param>
/// <returns>True if the packet was found and removed, otherwise false</returns>
public bool RemoveUnsafe(uint sequenceNumber)
{
return packets.Remove(sequenceNumber);
}
/// <summary>
/// Removes a packet from the collection without attempting to obtain a
/// lock first
/// </summary>
/// <param name="sequenceNumber">Sequence number of the packet to remove</param>
/// <param name="packet">Returns the removed packet</param>
/// <returns>True if the packet was found and removed, otherwise false</returns>
public bool RemoveUnsafe(uint sequenceNumber, out OutgoingPacket packet)
{
if (packets.TryGetValue(sequenceNumber, out packet))
{
packets.Remove(sequenceNumber);
return true;
}
return false;
}
/// <summary>
/// Gets the packet with the lowest sequence number
/// </summary>
/// <returns>The packet with the lowest sequence number, or null if the
/// collection is empty</returns>
public OutgoingPacket GetOldest()
{
lock (SyncRoot)
{
using (SortedDictionary<uint, OutgoingPacket>.ValueCollection.Enumerator e = packets.Values.GetEnumerator())
{
if (e.MoveNext())
return e.Current;
else
return null;
}
}
}
/// <summary>
/// Returns a list of all of the packets with a TickCount older than
/// the specified timeout
/// </summary>
/// <param name="timeoutMS">Number of ticks (milliseconds) before a
/// packet is considered expired</param>
/// <returns>A list of all expired packets according to the given
/// expiration timeout</returns>
public List<OutgoingPacket> GetExpiredPackets(int timeoutMS)
{
List<OutgoingPacket> expiredPackets = null;
lock (SyncRoot)
{
int now = Environment.TickCount;
foreach (OutgoingPacket packet in packets.Values)
{
if (packet.TickCount == 0)
continue;
if (now - packet.TickCount >= timeoutMS)
{
if (expiredPackets == null)
expiredPackets = new List<OutgoingPacket>();
expiredPackets.Add(packet);
}
else
{
break;
}
}
}
return expiredPackets;
}
}
}

1
OpenSim/Region/CoreModules/Scripting/HttpRequest/ScriptsHttpRequests.cs
View File

@ -323,7 +323,6 @@ namespace OpenSim.Region.CoreModules.Scripting.HttpRequest
httpThread.IsBackground = true;
_finished = false;
httpThread.Start();
ThreadTracker.Add(httpThread);
}
/*

1
OpenSim/Region/CoreModules/Scripting/XMLRPC/XMLRPCModule.cs
View File

@ -639,7 +639,6 @@ namespace OpenSim.Region.CoreModules.Scripting.XMLRPC
httpThread.IsBackground = true;
_finished = false;
httpThread.Start();
ThreadTracker.Add(httpThread);
}
/*

2
OpenSim/Region/CoreModules/World/WorldMap/WorldMapModule.cs
View File

@ -345,7 +345,6 @@ namespace OpenSim.Region.CoreModules.World.WorldMap
mapItemReqThread.Priority = ThreadPriority.BelowNormal;
mapItemReqThread.SetApartmentState(ApartmentState.MTA);
mapItemReqThread.Start();
ThreadTracker.Add(mapItemReqThread);
}
/// <summary>
@ -447,7 +446,6 @@ namespace OpenSim.Region.CoreModules.World.WorldMap
// end gracefully
if (st.agentID == UUID.Zero)
{
ThreadTracker.Remove(mapItemReqThread);
break;
}

5
OpenSim/Region/Framework/Scenes/Scene.cs
View File

@ -903,7 +903,6 @@ namespace OpenSim.Region.Framework.Scenes
//m_heartbeatTimer.Elapsed += new ElapsedEventHandler(Heartbeat);
if (HeartbeatThread != null)
{
ThreadTracker.Remove(HeartbeatThread);
HeartbeatThread.Abort();
HeartbeatThread = null;
}
@ -912,7 +911,6 @@ namespace OpenSim.Region.Framework.Scenes
HeartbeatThread.SetApartmentState(ApartmentState.MTA);
HeartbeatThread.Name = string.Format("Heartbeat for region {0}", RegionInfo.RegionName);
HeartbeatThread.Priority = ThreadPriority.AboveNormal;
ThreadTracker.Add(HeartbeatThread);
HeartbeatThread.Start();
}
@ -1448,6 +1446,9 @@ namespace OpenSim.Region.Framework.Scenes
m_log.Info("[SCENE]: Loading objects from datastore");
List<SceneObjectGroup> PrimsFromDB = m_storageManager.DataStore.LoadObjects(regionID);
m_log.Info("[SCENE]: Loaded " + PrimsFromDB.Count + " objects from the datastore");
foreach (SceneObjectGroup group in PrimsFromDB)
{
if (group.RootPart == null)

1
OpenSim/Region/OptionalModules/Avatar/Chat/IRCConnector.cs
View File

@ -360,7 +360,6 @@ namespace OpenSim.Region.OptionalModules.Avatar.Chat
m_listener.Name = "IRCConnectorListenerThread";
m_listener.IsBackground = true;
m_listener.Start();
ThreadTracker.Add(m_listener);
// This is the message order recommended by RFC 2812
if (m_password != null)

1
OpenSim/Region/OptionalModules/ContentManagementSystem/CMController.cs
View File

@ -152,7 +152,6 @@ namespace OpenSim.Region.OptionalModules.ContentManagement
m_thread.Name = "Content Management";
m_thread.IsBackground = true;
m_thread.Start();
ThreadTracker.Add(m_thread);
m_state = State.NONE;
}
}

1
OpenSim/Region/ScriptEngine/DotNetEngine/EventQueueThreadClass.cs
View File

@ -138,7 +138,6 @@ namespace OpenSim.Region.ScriptEngine.DotNetEngine
EventQueueThread.Priority = MyThreadPriority;
EventQueueThread.Name = "EventQueueManagerThread_" + ThreadCount;
EventQueueThread.Start();
ThreadTracker.Add(EventQueueThread);
// Look at this... Don't you wish everyone did that solid
// coding everywhere? :P

1
OpenSim/Region/ScriptEngine/DotNetEngine/MaintenanceThread.cs
View File

@ -97,7 +97,6 @@ namespace OpenSim.Region.ScriptEngine.DotNetEngine
MaintenanceThreadThread.Name = "ScriptMaintenanceThread";
MaintenanceThreadThread.IsBackground = true;
MaintenanceThreadThread.Start();
ThreadTracker.Add(MaintenanceThreadThread);
}
}

8
OpenSim/Region/ScriptEngine/DotNetEngine/ScriptEngine.cs
View File

@ -129,10 +129,6 @@ namespace OpenSim.Region.ScriptEngine.DotNetEngine
public void AddRegion(Scene Sceneworld)
{
m_log.Info("[" + ScriptEngineName + "]: ScriptEngine initializing");
m_Scene = Sceneworld;
// Make sure we have config
if (ConfigSource.Configs[ScriptEngineName] == null)
ConfigSource.AddConfig(ScriptEngineName);
@ -143,6 +139,10 @@ namespace OpenSim.Region.ScriptEngine.DotNetEngine
if (!m_enabled)
return;
m_log.Info("[" + ScriptEngineName + "]: ScriptEngine initializing");
m_Scene = Sceneworld;
// Create all objects we'll be using
m_EventQueueManager = new EventQueueManager(this);
m_EventManager = new EventManager(this, true);

1
OpenSim/Region/ScriptEngine/Shared/Api/Implementation/AsyncCommandManager.cs
View File

@ -142,7 +142,6 @@ namespace OpenSim.Region.ScriptEngine.Shared.Api
cmdHandlerThread.Priority = ThreadPriority.BelowNormal;
cmdHandlerThread.IsBackground = true;
cmdHandlerThread.Start();
ThreadTracker.Add(cmdHandlerThread);
}
}

1
OpenSim/TestSuite/BotManager.cs
View File

@ -115,7 +115,6 @@ namespace OpenSim.TestSuite
m_td[pos].IsBackground = true;
m_td[pos].Start();
m_lBot.Add(pb);
ThreadTracker.Add(m_td[pos]);
}
/// <summary>

2
OpenSim/Tests/Clients/Grid/GridClient.cs
View File

@ -55,7 +55,7 @@ namespace OpenSim.Tests.Clients.GridClient
new PatternLayout("%date [%thread] %-5level %logger [%property{NDC}] - %message%newline");
log4net.Config.BasicConfigurator.Configure(consoleAppender);
string serverURI = "http://127.0.0.1:" + ConfigSettings.DefaultGridServerHttpPort;
string serverURI = "http://127.0.0.1:8001";
GridServicesConnector m_Connector = new GridServicesConnector(serverURI);
GridRegion r1 = CreateRegion("Test Region 1", 1000, 1000);

1
OpenSim/Tools/pCampBot/BotManager.cs
View File

@ -149,7 +149,6 @@ namespace pCampBot
m_td[pos].IsBackground = true;
m_td[pos].Start();
m_lBot.Add(pb);
ThreadTracker.Add(m_td[pos]);
}
/// <summary>

4
prebuild.xml
View File

@ -131,6 +131,7 @@
<ReferencePath>../../bin/</ReferencePath>
<Reference name="System"/>
<Reference name="System.Core"/>
<Reference name="System.Xml"/>
<Reference name="System.Data"/>
<Reference name="System.Drawing"/>
@ -1750,6 +1751,7 @@
<ReferencePath>../../../../bin/</ReferencePath>
<Reference name="System"/>
<Reference name="System.Core"/>
<Reference name="System.Xml"/>
<Reference name="OpenMetaverseTypes.dll"/>
<Reference name="OpenMetaverse.StructuredData.dll"/>
@ -3689,6 +3691,7 @@
</Files>
</Project>
<!-- Commented for now until new unit tests are written for the new LLUDP implementation
<Project frameworkVersion="v3_5" name="OpenSim.Region.ClientStack.LindenUDP.Tests" path="OpenSim/Region/ClientStack/LindenUDP/Tests" type="Library">
<Configuration name="Debug">
<Options>
@ -3722,6 +3725,7 @@
<Match pattern="*.cs" recurse="false"/>
</Files>
</Project>
-->
<Project frameworkVersion="v3_5" name="OpenSim.Region.ScriptEngine.Tests" path="OpenSim/Region/ScriptEngine" type="Library">
<Configuration name="Debug">