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

change Xfermanager file sending

LSLKeyTest
UbitUmarov 2016-07-10 15:25:05 +01:00
parent 8d27abd3b4
commit e6d57a1e49
1 changed files with 194 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

294
OpenSim/Region/CoreModules/Agent/Xfer/XferModule.cs
View File

@ -49,12 +49,11 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
private Dictionary<ulong, XferDownLoad> Transfers = new Dictionary<ulong, XferDownLoad>(); private Dictionary<ulong, XferDownLoad> Transfers = new Dictionary<ulong, XferDownLoad>();
private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType); private static readonly ILog m_log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);
private object timeTickLock = new object();
private object timeTickLock = new object(); private double lastTimeTick = 0.0;
private double lastTimeTick = 0.0; private double lastFilesExpire = 0.0;
private bool inTimeTick = false; private bool inTimeTick = false;
private double lastFilesExpire = 0.0;
public struct XferRequest public struct XferRequest
{ {
public IClientAPI remoteClient; public IClientAPI remoteClient;
@ -81,15 +80,15 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
public void AddRegion(Scene scene) public void AddRegion(Scene scene)
{ {
m_scene = scene; m_scene = scene;
m_scene.RegisterModuleInterface<IXfer>(this);
m_scene.EventManager.OnNewClient += NewClient; m_scene.EventManager.OnNewClient += NewClient;
m_scene.EventManager.OnRegionHeartbeatEnd += OnTimeTick; m_scene.EventManager.OnRegionHeartbeatEnd += OnTimeTick;
m_scene.RegisterModuleInterface<IXfer>(this);
} }
public void RemoveRegion(Scene scene) public void RemoveRegion(Scene scene)
{ {
m_scene.EventManager.OnNewClient -= NewClient; m_scene.EventManager.OnNewClient -= NewClient;
m_scene.EventManager.OnRegionHeartbeatEnd -= OnTimeTick; m_scene.EventManager.OnRegionHeartbeatEnd -= OnTimeTick;
m_scene.UnregisterModuleInterface<IXfer>(this); m_scene.UnregisterModuleInterface<IXfer>(this);
m_scene = null; m_scene = null;
@ -123,7 +122,7 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
if(!inTimeTick) if(!inTimeTick)
{ {
double now = Util.GetTimeStampMS(); double now = Util.GetTimeStampMS();
if(now - lastTimeTick > 1500.0) // 1.5 second if(now - lastTimeTick > 1750.0)
{ {
inTimeTick = true; inTimeTick = true;
@ -197,7 +196,7 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
} }
} }
} }
public void NewClient(IClientAPI client) public void NewClient(IClientAPI client)
{ {
client.OnRequestXfer += RequestXfer; client.OnRequestXfer += RequestXfer;
@ -205,9 +204,49 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
client.OnAbortXfer += AbortXfer; client.OnAbortXfer += AbortXfer;
} }
public void OnClientClosed(IClientAPI client)
{
client.OnRequestXfer -= RequestXfer;
client.OnConfirmXfer -= AckPacket;
client.OnAbortXfer -= AbortXfer;
}
private void RemoveOrDecrementFile(string fileName)
{
// NewFiles must be locked
if (NewFiles.ContainsKey(fileName))
{
if (NewFiles[fileName].refsCount == 1)
NewFiles.Remove(fileName);
else
NewFiles[fileName].refsCount--;
}
}
public void transfersTimeTick(double now) public void transfersTimeTick(double now)
{ {
XferDownLoad[] xfrs;
lock(Transfers)
{
if(Transfers.Count == 0)
return;
xfrs = new XferDownLoad[Transfers.Count];
Transfers.Values.CopyTo(xfrs,0);
}
foreach(XferDownLoad xfr in xfrs)
{
if(xfr.checkTime(now))
{
ulong xfrID = xfr.XferID;
lock(Transfers)
{
if(Transfers.ContainsKey(xfrID))
Transfers.Remove(xfrID);
}
}
}
} }
/// <summary> /// <summary>
@ -222,80 +261,49 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
{ {
if (NewFiles.ContainsKey(fileName)) if (NewFiles.ContainsKey(fileName))
{ {
if (!Transfers.ContainsKey(xferID)) lock(Transfers)
{ {
byte[] fileData = NewFiles[fileName].Data; if (!Transfers.ContainsKey(xferID))
XferDownLoad transaction = new XferDownLoad(fileName, fileData, xferID, remoteClient); {
if (fileName.StartsWith("inventory_")) byte[] fileData = NewFiles[fileName].Data;
transaction.isTaskInventory = true; XferDownLoad transaction =
new XferDownLoad(fileName, fileData, xferID, remoteClient);
Transfers.Add(xferID, transaction); Transfers.Add(xferID, transaction);
if (transaction.StartSend()) transaction.StartSend();
RemoveXferData(xferID);
// The transaction for this file is either complete or on its way
RemoveOrDecrement(fileName);
// The transaction for this file is on its way
RemoveOrDecrementFile(fileName);
}
} }
} }
else else
m_log.WarnFormat("[Xfer]: {0} not found", fileName); m_log.WarnFormat("[Xfer]: {0} not found", fileName);
} }
} }
public void AckPacket(IClientAPI remoteClient, ulong xferID, uint packet) public void AckPacket(IClientAPI remoteClient, ulong xferID, uint packet)
{ {
lock (NewFiles) // This is actually to lock Transfers lock (Transfers)
{ {
if (Transfers.ContainsKey(xferID)) if (Transfers.ContainsKey(xferID))
{ {
XferDownLoad dl = Transfers[xferID];
if (Transfers[xferID].AckPacket(packet)) if (Transfers[xferID].AckPacket(packet))
{ Transfers.Remove(xferID);
RemoveXferData(xferID);
RemoveOrDecrement(dl.FileName);
}
} }
} }
} }
private void RemoveXferData(ulong xferID)
{
// NewFiles must be locked!
if (Transfers.ContainsKey(xferID))
{
XferModule.XferDownLoad xferItem = Transfers[xferID];
//string filename = xferItem.FileName;
Transfers.Remove(xferID);
xferItem.Data = new byte[0]; // Clear the data
xferItem.DataPointer = 0;
}
}
public void AbortXfer(IClientAPI remoteClient, ulong xferID) public void AbortXfer(IClientAPI remoteClient, ulong xferID)
{ {
lock (NewFiles) lock (Transfers)
{ {
if (Transfers.ContainsKey(xferID)) if (Transfers.ContainsKey(xferID))
RemoveOrDecrement(Transfers[xferID].FileName); {
Transfers[xferID].done();
RemoveXferData(xferID); Transfers.Remove(xferID);
} }
}
private void RemoveOrDecrement(string fileName)
{
// NewFiles must be locked
if (NewFiles.ContainsKey(fileName))
{
if (NewFiles[fileName].refsCount == 1)
NewFiles.Remove(fileName);
else
NewFiles[fileName].refsCount--;
} }
} }
@ -304,14 +312,19 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
public class XferDownLoad public class XferDownLoad
{ {
public IClientAPI Client; public IClientAPI Client;
private bool complete;
public byte[] Data = new byte[0]; public byte[] Data = new byte[0];
public int DataPointer = 0;
public string FileName = String.Empty; public string FileName = String.Empty;
public uint Packet = 0;
public uint Serial = 1;
public ulong XferID = 0; public ulong XferID = 0;
public bool isTaskInventory = false; public bool isDeleted = false;
private object myLock = new object();
private double lastsendTimeMS;
private int LastPacket;
private int lastBytes;
private int lastSentPacket;
private int lastAckPacket;
private int burstSize;
private int retries = 0;
public XferDownLoad(string fileName, byte[] data, ulong xferID, IClientAPI client) public XferDownLoad(string fileName, byte[] data, ulong xferID, IClientAPI client)
{ {
@ -325,32 +338,97 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
{ {
} }
public void done()
{
if(!isDeleted)
{
Data = new byte[0];
isDeleted = true;
}
}
/// <summary> /// <summary>
/// Start a transfer /// Start a transfer
/// </summary> /// </summary>
/// <returns>True if the transfer is complete, false if not</returns> /// <returns>True if the transfer is complete, false if not</returns>
public bool StartSend() public void StartSend()
{ {
if (Data.Length < 1000) lock(myLock)
{ {
// for now (testing) we only support files under 1000 bytes if(Data.Length == 0) //??
byte[] transferData = new byte[Data.Length + 4]; {
Array.Copy(Utils.IntToBytes(Data.Length), 0, transferData, 0, 4); LastPacket = 0;
Array.Copy(Data, 0, transferData, 4, Data.Length); lastBytes = 0;
Client.SendXferPacket(XferID, 0 + 0x80000000, transferData, isTaskInventory); burstSize = 0;
complete = true; }
else
{
// payload of 1024bytes
LastPacket = Data.Length >> 10;
lastBytes = Data.Length & 0x3ff;
if(lastBytes == 0)
{
lastBytes = 1024;
LastPacket--;
}
burstSize = Client.GetAgentThrottleSilent((int)ThrottleOutPacketType.Asset) >> 11;
}
lastAckPacket = -1;
lastSentPacket = -1;
double now = Util.GetTimeStampMS();
SendBurst(now);
return;
}
}
private void SendBurst(double now)
{
int start = lastAckPacket + 1;
int end = start + burstSize;
if (end > LastPacket)
end = LastPacket;
while(start <= end)
SendPacket(start++ , now);
}
private void SendPacket(int pkt, double now)
{
if(pkt > LastPacket)
return;
int pktsize;
uint pktid;
if (pkt == LastPacket)
{
pktsize = lastBytes;
pktid = (uint)pkt | 0x80000000u;
} }
else else
{ {
byte[] transferData = new byte[1000 + 4]; pktsize = 1024;
Array.Copy(Utils.IntToBytes(Data.Length), 0, transferData, 0, 4); pktid = (uint)pkt;
Array.Copy(Data, 0, transferData, 4, 1000);
Client.SendXferPacket(XferID, 0, transferData, isTaskInventory);
Packet++;
DataPointer = 1000;
} }
return complete; byte[] transferData;
if(pkt == 0)
{
transferData = new byte[pktsize + 4];
Array.Copy(Utils.IntToBytes(Data.Length), 0, transferData, 0, 4);
Array.Copy(Data, 0, transferData, 4, pktsize);
}
else
{
transferData = new byte[pktsize];
Array.Copy(Data, pkt << 10, transferData, 0, pktsize);
}
Client.SendXferPacket(XferID, pktid, transferData, false);
lastSentPacket = pkt;
lastsendTimeMS = now;
} }
/// <summary> /// <summary>
@ -360,30 +438,46 @@ namespace OpenSim.Region.CoreModules.Agent.Xfer
/// <returns>True if the transfer is complete, false otherwise</returns> /// <returns>True if the transfer is complete, false otherwise</returns>
public bool AckPacket(uint packet) public bool AckPacket(uint packet)
{ {
if (!complete) lock(myLock)
{ {
if ((Data.Length - DataPointer) > 1000) if(isDeleted)
{ return true;
byte[] transferData = new byte[1000];
Array.Copy(Data, DataPointer, transferData, 0, 1000);
Client.SendXferPacket(XferID, Packet, transferData, isTaskInventory);
Packet++;
DataPointer += 1000;
}
else
{
byte[] transferData = new byte[Data.Length - DataPointer];
Array.Copy(Data, DataPointer, transferData, 0, Data.Length - DataPointer);
uint endPacket = Packet |= (uint) 0x80000000;
Client.SendXferPacket(XferID, endPacket, transferData, isTaskInventory);
Packet++;
DataPointer += (Data.Length - DataPointer);
complete = true; packet &= 0x7fffffff;
if(lastAckPacket < packet)
lastAckPacket = (int)packet;
if(lastAckPacket == LastPacket)
{
done();
return true;
} }
double now = Util.GetTimeStampMS();
SendPacket(lastSentPacket + 1, now);
return false;
} }
}
return complete; public bool checkTime(double now)
{
if(Monitor.TryEnter(myLock))
{
if(!isDeleted)
{
double timeMS = now - lastsendTimeMS;
if(timeMS > 60000.0)
done();
else if(timeMS > 3500.0 && retries++ < 3)
{
burstSize >>= 1;
SendBurst(now);
}
}
Monitor.Exit(myLock);
return isDeleted;
}
return false;
} }
} }