|
//------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.
//------------------------------------------------------------
namespace System.ServiceModel.Channels
{
using System.Diagnostics;
using System.Runtime;
using System.ServiceModel;
using System.ServiceModel.Description;
using System.Threading;
class UtilityExtension : IExtension<IPeerNeighbor>
{
uint linkUtility;
uint updateCount;
IOThreadTimer ackTimer;
const uint linkUtilityIncrement = 128;
const uint maxLinkUtility = 4096;
int outTotal;
uint inTotal;
uint inUseful;
IPeerNeighbor owner;
object thisLock = new object();
object throttleLock = new object();
public event EventHandler UtilityInfoReceived;
public event EventHandler UtilityInfoSent;
TypedMessageConverter messageConverter;
public const int AcceptableMissDistance = 2;
int pendingSends = 0;
int checkPointPendingSends = 0;
bool isMonitoring = false;
int expectedClearance;
IOThreadTimer pruneTimer;
const int PruneIntervalMilliseconds = 10000;
TimeSpan pruneInterval;
const int MinimumPendingMessages = 8;
public delegate void PruneNeighborCallback(IPeerNeighbor peer);
PruneNeighborCallback pruneNeighbor;
UtilityExtension()
{
ackTimer = new IOThreadTimer(new Action<object>(AcknowledgeLoop), null, false);
pendingSends = 0;
pruneTimer = new IOThreadTimer(new Action<object>(VerifyCheckPoint), null, false);
pruneInterval = TimeSpan.FromMilliseconds(PruneIntervalMilliseconds + new Random(Process.GetCurrentProcess().Id).Next(PruneIntervalMilliseconds));
}
public bool IsAccurate
{
get { return updateCount >= 32; }
}
public uint LinkUtility
{
get
{
return linkUtility;
}
}
internal TypedMessageConverter MessageConverter
{
get
{
if (messageConverter == null)
{
messageConverter = TypedMessageConverter.Create(typeof(UtilityInfo), PeerStrings.LinkUtilityAction);
}
return messageConverter;
}
}
public void Attach(IPeerNeighbor host)
{
this.owner = host;
ackTimer.Set(PeerTransportConstants.AckTimeout);
}
static public void OnNeighborConnected(IPeerNeighbor neighbor)
{
Fx.Assert(neighbor != null, "Neighbor must have a value");
neighbor.Extensions.Add(new UtilityExtension());
}
static public void OnNeighborClosed(IPeerNeighbor neighbor)
{
Fx.Assert(neighbor != null, "Neighbor must have a value");
UtilityExtension ext = neighbor.Extensions.Find<UtilityExtension>();
if (ext != null) neighbor.Extensions.Remove(ext);
}
public void Detach(IPeerNeighbor host)
{
ackTimer.Cancel();
owner = null;
lock (throttleLock)
{
pruneTimer.Cancel();
}
}
public object ThisLock
{
get
{
return thisLock;
}
}
public static void OnMessageSent(IPeerNeighbor neighbor)
{
UtilityExtension ext = neighbor.Extensions.Find<UtilityExtension>();
if (ext != null) ext.OnMessageSent();
}
void OnMessageSent()
{
lock (ThisLock)
{
outTotal++;
}
Interlocked.Increment(ref pendingSends);
}
public static void OnEndSend(IPeerNeighbor neighbor, FloodAsyncResult fresult)
{
if (neighbor.State >= PeerNeighborState.Disconnecting)
return;
UtilityExtension instance = neighbor.Utility;
if (instance == null)
return;
instance.OnEndSend(fresult);
}
public void OnEndSend(FloodAsyncResult fresult)
{
Interlocked.Decrement(ref pendingSends);
}
void AcknowledgeLoop(object state)
{
IPeerNeighbor peer = owner;
if (peer == null || !peer.IsConnected)
return;
FlushAcknowledge();
if (owner != null)
ackTimer.Set(PeerTransportConstants.AckTimeout);
}
static public void ProcessLinkUtility(IPeerNeighbor neighbor, UtilityInfo umessage)
{
Fx.Assert(neighbor != null, "Neighbor must have a value");
UtilityExtension ext = neighbor.Extensions.Find<UtilityExtension>();
if (ext != null)
{
ext.ProcessLinkUtility(umessage.Useful, umessage.Total);
}
}
// Update link utility for the neighbor. received from the neighbor
void ProcessLinkUtility(uint useful, uint total)
{
uint i = 0;
lock (ThisLock)
{
if (total > PeerTransportConstants.AckWindow
|| useful > total
|| (uint)outTotal < total
)
{
throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new InvalidOperationException(SR.GetString(SR.PeerLinkUtilityInvalidValues, useful, total)));
}
//VERIFY with in this range, we are hoping that the order of useful/useless messages doesnt matter much.
for (i = 0; i < useful; i++)
{
this.linkUtility = Calculate(this.linkUtility, true);
}
for (; i < total; i++)
{
this.linkUtility = Calculate(this.linkUtility, false);
}
outTotal -= (int)total;
}
if (UtilityInfoReceived != null)
{
UtilityInfoReceived(this, EventArgs.Empty);
}
}
uint Calculate(uint current, bool increase)
{
uint utility = 0;
// Refer to graph maintenance white paper for explanation of the formula
// used to compute utility index.
utility = (uint)current * 31 / 32;
if (increase)
utility += linkUtilityIncrement;
if (!(utility <= maxLinkUtility))
{
throw Fx.AssertAndThrow("Link utility should not exceed " + maxLinkUtility);
}
if (!IsAccurate)
++updateCount;
return utility;
}
public static uint UpdateLinkUtility(IPeerNeighbor neighbor, bool useful)
{
Fx.Assert(neighbor != null, "Neighbor must have a value");
uint linkUtility = 0;
UtilityExtension ext = neighbor.Extensions.Find<UtilityExtension>();
if (ext != null)
{
// Can happen if the neighbor has been closed for instance
linkUtility = ext.UpdateLinkUtility(useful);
}
return linkUtility;
}
public uint UpdateLinkUtility(bool useful)
{
lock (ThisLock)
{
inTotal++;
if (useful)
inUseful++;
linkUtility = Calculate(linkUtility, useful);
if (inTotal == PeerTransportConstants.AckWindow)
{
FlushAcknowledge();
}
}
return linkUtility;
}
public void FlushAcknowledge()
{
if (inTotal == 0)
return;
uint tempUseful = 0, tempTotal = 0;
lock (ThisLock)
{
tempUseful = inUseful;
tempTotal = inTotal;
inUseful = 0;
inTotal = 0;
}
SendUtilityMessage(tempUseful, tempTotal);
}
class AsyncUtilityState
{
public Message message;
public UtilityInfo info;
public AsyncUtilityState(Message message, UtilityInfo info)
{
this.message = message;
this.info = info;
}
}
void SendUtilityMessage(uint useful, uint total)
{
IPeerNeighbor host = owner;
if (host == null || !PeerNeighborStateHelper.IsConnected(host.State) || total == 0)
return;
UtilityInfo umessage = new UtilityInfo(useful, total);
IAsyncResult result = null;
Message message = MessageConverter.ToMessage(umessage, MessageVersion.Soap12WSAddressing10);
bool fatal = false;
try
{
result = host.BeginSend(message, Fx.ThunkCallback(new AsyncCallback(UtilityMessageSent)), new AsyncUtilityState(message, umessage));
if (result.CompletedSynchronously)
{
host.EndSend(result);
EventHandler handler = UtilityInfoSent;
if (handler != null)
handler(this, EventArgs.Empty);
}
}
catch (Exception e)
{
if (Fx.IsFatal(e))
{
fatal = true;
throw;
}
if (null != HandleSendException(host, e, umessage))
throw;
DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
}
finally
{
if (!fatal && (result == null || result.CompletedSynchronously))
message.Close();
}
}
void UtilityMessageSent(IAsyncResult result)
{
if (result == null || result.AsyncState == null)
return;
IPeerNeighbor host = this.owner;
if (host == null || !PeerNeighborStateHelper.IsConnected(host.State))
return;
if (result.CompletedSynchronously)
return;
AsyncUtilityState state = (AsyncUtilityState)result.AsyncState;
Fx.Assert(state != null, "IAsyncResult.AsyncState does not contain AsyncUtilityState");
Message message = state.message;
UtilityInfo umessage = state.info;
bool fatal = false;
if (!(umessage != null))
{
throw Fx.AssertAndThrow("expecting a UtilityInfo message in the AsyncState!");
}
try
{
host.EndSend(result);
}
catch (Exception e)
{
if (Fx.IsFatal(e))
{
fatal = true;
throw;
}
if (null != HandleSendException(host, e, umessage))
throw;
DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
}
finally
{
if (!fatal)
{
Fx.Assert(!result.CompletedSynchronously, "result.CompletedSynchronously");
message.Close();
}
}
EventHandler handler = UtilityInfoSent;
if (handler != null)
handler(this, EventArgs.Empty);
}
Exception HandleSendException(IPeerNeighbor host, Exception e, UtilityInfo umessage)
{
if ((e is ObjectDisposedException) ||
(e is TimeoutException) ||
(e is CommunicationException))
{
if (!(!(e.InnerException is QuotaExceededException)))
{
throw Fx.AssertAndThrow("insufficient quota for sending messages!");
}
lock (ThisLock)
{
this.inTotal += umessage.Total;
this.inUseful += umessage.Useful;
}
return null;
}
else
{
return e;
}
}
static internal void ReportCacheMiss(IPeerNeighbor neighbor, int missedBy)
{
Fx.Assert(missedBy > AcceptableMissDistance, "Call this method for cache misses ONLY!");
Fx.Assert(neighbor != null, "Neighbor must have a value");
if (!neighbor.IsConnected)
return;
UtilityExtension ext = neighbor.Extensions.Find<UtilityExtension>();
if (ext != null)
{
ext.ReportCacheMiss(missedBy);
}
}
void ReportCacheMiss(int missedBy)
{
lock (ThisLock)
{
for (int i = 0; i < missedBy; i++)
{
this.linkUtility = Calculate(this.linkUtility, false);
}
}
}
public int PendingMessages
{
get
{
return this.pendingSends;
}
}
public void BeginCheckPoint(PruneNeighborCallback pruneCallback)
{
if (this.isMonitoring)
return;
lock (throttleLock)
{
if (this.isMonitoring)
return;
this.checkPointPendingSends = this.pendingSends;
this.pruneNeighbor = pruneCallback;
this.expectedClearance = this.pendingSends / 2;
this.isMonitoring = true;
if (owner == null)
return;
pruneTimer.Set(pruneInterval);
}
}
void VerifyCheckPoint(object state)
{
int lclPendingSends;
int lclCheckPointPendingSends;
IPeerNeighbor peer = (IPeerNeighbor)owner;
if (peer == null || !peer.IsConnected)
return;
lock (throttleLock)
{
lclPendingSends = this.pendingSends;
lclCheckPointPendingSends = this.checkPointPendingSends;
}
if (lclPendingSends <= MinimumPendingMessages)
{
lock (throttleLock)
{
isMonitoring = false;
}
}
else if (lclPendingSends + this.expectedClearance >= lclCheckPointPendingSends)
{
pruneNeighbor(peer);
}
else
{
lock (throttleLock)
{
if (owner == null)
return;
this.checkPointPendingSends = this.pendingSends;
this.expectedClearance = this.expectedClearance / 2;
pruneTimer.Set(pruneInterval);
}
}
}
}
}
|