ServiceModel/Channels/PeerFlooder.cs

//------------------------------------------------------------
// Copyright (c) Microsoft Corporation.  All rights reserved.
//------------------------------------------------------------
namespace System.ServiceModel.Channels
{
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Runtime;
    using System.Runtime.Serialization;
    using System.Security;
    using System.Security.Permissions;
    using System.ServiceModel;
    using System.ServiceModel.Diagnostics;
    using System.ServiceModel.Dispatcher;
    using System.ServiceModel.Security;
    using System.Threading;
    using System.Xml;

    class PeerFlooder : PeerFlooderSimple
    {
        PeerFlooder(PeerNodeConfig config, PeerNeighborManager neighborManager) : base(config, neighborManager) { }

        public static PeerFlooder CreateFlooder(PeerNodeConfig config, PeerNeighborManager neighborManager, IPeerNodeMessageHandling messageHandler)
        {
            PeerFlooder flooder = new PeerFlooder(config, neighborManager);
            flooder.messageHandler = messageHandler;
            return flooder;
        }
    }

    interface IFlooderForThrottle
    {
        void OnThrottleReached();
        void OnThrottleReleased();
    }

    abstract class PeerFlooderBase<TFloodContract, TLinkContract> : IFlooderForThrottle, IPeerFlooderContract<TFloodContract, TLinkContract> where TFloodContract : Message
    {
        protected PeerNodeConfig config;
        protected PeerNeighborManager neighborManager;
        protected List<IPeerNeighbor> neighbors;
        object thisLock = new object();

        internal IPeerNodeMessageHandling messageHandler;
        internal PeerThrottleHelper quotaHelper;
        long messageSequence;

        public event EventHandler ThrottleReached;
        public event EventHandler SlowNeighborKilled;
        public event EventHandler ThrottleReleased;
        public EventHandler OnMessageSentHandler;


        public PeerFlooderBase(PeerNodeConfig config, PeerNeighborManager neighborManager)
        {
            this.neighborManager = neighborManager;
            this.neighbors = new List<IPeerNeighbor>();
            this.config = config;
            this.neighbors = this.neighborManager.GetConnectedNeighbors();
            this.quotaHelper = new PeerThrottleHelper(this, this.config.MaxPendingOutgoingCalls);
            OnMessageSentHandler = new EventHandler(OnMessageSent);
        }

        void PruneNeighborCallback(IPeerNeighbor peer)
        {
            lock (ThisLock)
            {
                if (this.Neighbors.Count <= 1)
                    return;
                if (DiagnosticUtility.ShouldTraceWarning)
                {
                    string message = SR.GetString(SR.PeerThrottlePruning, this.config.MeshId);
                    PeerThrottleTraceRecord record = new PeerThrottleTraceRecord(this.config.MeshId, message);
                    TraceUtility.TraceEvent(TraceEventType.Warning, TraceCode.PeerFlooderReceiveMessageQuotaExceeded,
                        SR.GetString(SR.TraceCodePeerFlooderReceiveMessageQuotaExceeded), record, this, null);
                }
            }
            try
            {
                peer.Abort(PeerCloseReason.NodeTooSlow, PeerCloseInitiator.LocalNode);
            }
            catch (Exception e)
            {
                if (Fx.IsFatal(e)) throw;
                if (null != CloseNeighborIfKnownException(neighborManager, e, peer))
                {
                    throw;
                }
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
            }
        }

        void IFlooderForThrottle.OnThrottleReached()
        {
            if (DiagnosticUtility.ShouldTraceInformation)
            {
                string message = SR.GetString(SR.PeerThrottleWaiting, this.config.MeshId);
                PeerThrottleTraceRecord record = new PeerThrottleTraceRecord(this.config.MeshId, message);
                TraceUtility.TraceEvent(TraceEventType.Information, TraceCode.PeerFlooderReceiveMessageQuotaExceeded,
                    SR.GetString(SR.TraceCodePeerFlooderReceiveMessageQuotaExceeded), record, this, null);
            }

            IPeerNeighbor peer = this.neighborManager.SlowestNeighbor();
            if (peer == null)
                return;
            UtilityExtension extension = peer.Utility;
            if (peer.IsConnected && extension != null)
            {
                if (extension.PendingMessages > PeerTransportConstants.MessageThreshold)
                {
                    extension.BeginCheckPoint(new UtilityExtension.PruneNeighborCallback(PruneNeighborCallback));
                }
                else
                {
                    Fx.Assert(false, "Neighbor is marked slow with messages " + extension.PendingMessages);
                }
                FireReachedEvent();
            }
        }

        void IFlooderForThrottle.OnThrottleReleased()
        {
            FireDequeuedEvent();
        }

        public void FireDequeuedEvent() { FireEvent(ThrottleReleased); }

        public void FireReachedEvent() { FireEvent(ThrottleReached); }

        public void FireKilledEvent() { FireEvent(SlowNeighborKilled); }

        void FireEvent(EventHandler handler)
        {
            if (handler != null)
                handler(this, EventArgs.Empty);
        }

        [PermissionSet(SecurityAction.Demand, Unrestricted = true), SecuritySafeCritical]
        public virtual IAsyncResult BeginFloodEncodedMessage(byte[] id, MessageBuffer encodedMessage, TimeSpan timeout, AsyncCallback callback, object state)
        {
            RecordOutgoingMessage(id);
            SynchronizationContext syncContext = ThreadBehavior.GetCurrentSynchronizationContext();
            SynchronizationContext.SetSynchronizationContext(null);

            if (neighbors.Count == 0)
            {
                return new CompletedAsyncResult(callback, state);
            }
            try
            {
                return FloodMessageToNeighbors(encodedMessage, timeout, callback, state, -1, null, null, OnMessageSentHandler);
            }
            finally
            {
                SynchronizationContext.SetSynchronizationContext(syncContext);
            }

        }

        protected virtual IAsyncResult BeginFloodReceivedMessage(IPeerNeighbor sender, MessageBuffer messageBuffer,
            TimeSpan timeout, AsyncCallback callback, object state, int index, MessageHeader hopHeader)
        {
            quotaHelper.AcquireNoQueue();

            try
            {
                return FloodMessageToNeighbors(messageBuffer, timeout, callback, state, index, hopHeader, sender, OnMessageSentHandler);
            }
            catch (Exception e)
            {
                if (Fx.IsFatal(e)) throw;
                if (e is QuotaExceededException || (e is CommunicationException && e.InnerException is QuotaExceededException))
                {
                    DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
                    if (DiagnosticUtility.ShouldTraceError)
                    {
                        PeerFlooderTraceRecord record = new PeerFlooderTraceRecord(
                                                            this.config.MeshId,
                                                            sender.ListenAddress,
                                                            e);
                        TraceUtility.TraceEvent(
                                    TraceEventType.Error,
                                    TraceCode.PeerFlooderReceiveMessageQuotaExceeded,
                                    SR.GetString(SR.TraceCodePeerFlooderReceiveMessageQuotaExceeded),
                                    record,
                                    this,
                                    null);
                    }
                    return null;
                }
                throw;
            }
        }

        protected IAsyncResult BeginSendHelper(IPeerNeighbor neighbor, TimeSpan timeout, Message message, FloodAsyncResult fresult)
        {
            IAsyncResult result = null;
            bool fatal = false;
            try
            {
                UtilityExtension.OnMessageSent(neighbor);
                result = neighbor.BeginSend(message, timeout, Fx.ThunkCallback(new AsyncCallback(fresult.OnSendComplete)), message);
                fresult.AddResult(result, neighbor);
                if (result.CompletedSynchronously)
                {
                    neighbor.EndSend(result);
                    UtilityExtension.OnEndSend(neighbor, fresult);
                }
                return result;
            }
            catch (Exception e)
            {
                if (Fx.IsFatal(e))
                {
                    fatal = true;
                    throw;
                }
                if (null != CloseNeighborIfKnownException(neighborManager, e, neighbor))
                {
                    fresult.MarkEnd(false);
                    throw;
                }

                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
                return null;
            }
            finally
            {
                if ((result == null || result.CompletedSynchronously) && !fatal)
                    message.Close();

            }
        }

        public void OnMessageSent(object sender, EventArgs args)
        {
            quotaHelper.ItemDequeued();
        }

        void KillSlowNeighbor()
        {
            IPeerNeighbor neighbor = this.neighborManager.SlowestNeighbor();
            if (neighbor != null)
                neighbor.Abort(PeerCloseReason.NodeTooSlow, PeerCloseInitiator.LocalNode);
        }


        protected virtual IAsyncResult FloodMessageToNeighbors(MessageBuffer messageBuffer,
                                                               TimeSpan timeout, AsyncCallback callback, object state,
                                                               int index, MessageHeader hopHeader, IPeerNeighbor except,
                                                               EventHandler OnMessageSentCallback)
        {
            long temp = Interlocked.Increment(ref messageSequence);
            FloodAsyncResult fresult = new FloodAsyncResult(this.neighborManager, timeout, callback, state);
            fresult.OnMessageSent += OnMessageSentCallback;
            List<IPeerNeighbor> neighbors = this.Neighbors;

            foreach (IPeerNeighbor neighbor in neighbors)
            {
                if (neighbor.Equals(except))
                    continue;
                // Don't do anything if the neighbor is not connected
                if (PeerNeighborStateHelper.IsConnected(neighbor.State))
                {
                    Message fmessage = messageBuffer.CreateMessage();
                    if (index != -1)
                    {
                        fmessage.Headers.ReplaceAt(index, hopHeader);
                    }

                    // Don't do anything if the neighbor is not connected
                    if (PeerNeighborStateHelper.IsConnected(neighbor.State))
                    {
                        BeginSendHelper(neighbor, timeout, fmessage, fresult);
                    }
                }
            }
            fresult.MarkEnd(true);
            return fresult;

        }

        public void Open()
        {
            OnOpen();
        }

        public void Close()
        {
            OnClose();
        }

        public abstract void OnOpen();

        public abstract void OnClose();

        public virtual void OnNeighborConnected(IPeerNeighbor neighbor)
        {
            this.neighbors = this.neighborManager.GetConnectedNeighbors();
        }

        public virtual void OnNeighborClosed(IPeerNeighbor neighbor)
        {
            this.neighbors = this.neighborManager.GetConnectedNeighbors();
        }

        public abstract void ProcessLinkUtility(IPeerNeighbor neighbor, TLinkContract utilityInfo);

        public abstract bool ShouldProcess(TFloodContract floodInfo);
        public abstract void RecordOutgoingMessage(byte[] id);

        int UpdateHopCount(Message message, out MessageHeader hopHeader, out ulong currentValue)
        {
            int index = -1;
            currentValue = PeerTransportConstants.MaxHopCount;
            hopHeader = null;
            try
            {
                // If a message contains multiple Hopcounts with our name and namespace or the message can't deserialize to a ulong then ignore the HopCount
                index = message.Headers.FindHeader(PeerStrings.HopCountElementName, PeerStrings.HopCountElementNamespace);
                if (index != -1)
                {
                    currentValue = PeerMessageHelpers.GetHeaderULong(message.Headers, index);
                    hopHeader = MessageHeader.CreateHeader(PeerStrings.HopCountElementName, PeerStrings.HopCountElementNamespace, --currentValue, false);
                }
            }
            catch (MessageHeaderException e)
            {
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Warning);
            }
            catch (CommunicationException e)
            {
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Warning);
            }
            catch (SerializationException e)
            {
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Warning);
            }
            catch (XmlException e)
            {
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Warning);
            }
            Fx.Assert((index == -1) || (hopHeader != null), "Could not successfully create new HopCount Header!");
            return index;
        }

        public virtual IAsyncResult OnFloodedMessage(IPeerNeighbor neighbor, TFloodContract floodInfo, AsyncCallback callback, object state)
        {
            bool process = false;
            MessageBuffer messageBuffer = null;
            Message message = null;
            Uri via;
            Uri to;
            int index = 0;
            ulong remainingHops = PeerTransportConstants.MaxHopCount;
            MessageHeader hopHeader = null;
            bool fatal = false;
            PeerMessageProperty peerProperty = null;
            IAsyncResult result = null;

            try
            {
                peerProperty = (PeerMessageProperty)floodInfo.Properties[PeerStrings.PeerProperty];
                if (!peerProperty.MessageVerified)
                {
                    if (peerProperty.CacheMiss > UtilityExtension.AcceptableMissDistance)
                    {
                        UtilityExtension.ReportCacheMiss(neighbor, peerProperty.CacheMiss);
                    }
                    result = new CompletedAsyncResult(callback, state);
                }
                else
                {
                    process = true;
                    messageBuffer = floodInfo.CreateBufferedCopy((int)this.config.MaxReceivedMessageSize);
                    message = messageBuffer.CreateMessage();
                    via = peerProperty.PeerVia;
                    to = peerProperty.PeerTo;
                    message.Headers.To = message.Properties.Via = via;

                    index = UpdateHopCount(message, out hopHeader, out remainingHops);

                    PeerMessagePropagation propagateFlags = PeerMessagePropagation.LocalAndRemote;
                    if (peerProperty.SkipLocalChannels)
                        propagateFlags = PeerMessagePropagation.Remote;
                    else if (messageHandler.HasMessagePropagation)
                    {
                        using (Message filterMessage = messageBuffer.CreateMessage())
                        {
                            propagateFlags = messageHandler.DetermineMessagePropagation(filterMessage, PeerMessageOrigination.Remote);
                        }
                    }

                    if ((propagateFlags & PeerMessagePropagation.Remote) != 0)
                    {
                        if (remainingHops == 0)
                            propagateFlags &= ~PeerMessagePropagation.Remote;
                    }
                    if ((propagateFlags & PeerMessagePropagation.Remote) != 0)
                    {
                        result = BeginFloodReceivedMessage(neighbor, messageBuffer, PeerTransportConstants.ForwardTimeout, callback, state, index, hopHeader);
                    }
                    else
                    {
                        result = new CompletedAsyncResult(callback, state);
                    }
                    if ((propagateFlags & PeerMessagePropagation.Local) != 0)
                    {
                        messageHandler.HandleIncomingMessage(messageBuffer, propagateFlags, index, hopHeader, via, to);
                    }
                }
                UtilityExtension.UpdateLinkUtility(neighbor, process);
            }
            catch (Exception e)
            {
                if (Fx.IsFatal(e))
                {
                    fatal = true;
                    throw;
                }
                if (null != CloseNeighborIfKnownException(neighborManager, e, neighbor))
                {
                    throw;
                }
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
            }
            finally
            {
                if (!fatal)
                {
                    if (message != null)
                        message.Close();
                    if (messageBuffer != null)
                        messageBuffer.Close();
                }
            }
            return result;
        }

        public virtual void EndFloodMessage(IAsyncResult result)
        {
            if (result is CompletedAsyncResult)
            {
                CompletedAsyncResult.End(result);
                return;
            }
            FloodAsyncResult fresult = result as FloodAsyncResult;
            Fx.Assert(fresult != null, "Invalid AsyncResult type in EndFloodResult");
            fresult.End();

        }

        protected long MaxReceivedMessageSize
        {
            get { return config.MaxReceivedMessageSize; }
        }

        protected MessageEncoder MessageEncoder
        {
            get { return config.MessageEncoder; }
        }

        protected object ThisLock
        {
            get { return this.thisLock; }
        }

        protected List<IPeerNeighbor> Neighbors
        {
            get { return this.neighbors; }
        }

        // Guaranteed not to throw anything other than fatal exceptions
        static internal Exception CloseNeighborIfKnownException(PeerNeighborManager neighborManager, Exception exception, IPeerNeighbor peer)
        {
            try
            {
                //ignore this one since the channel is already closed.
                if (exception is ObjectDisposedException)
                    return null;
                else if (
                    (exception is CommunicationException && !(exception.InnerException is QuotaExceededException))
                    || (exception is TimeoutException)
                    || (exception is InvalidOperationException)
                    || (exception is MessageSecurityException)
                )
                {
                    //is this the right close reason?
                    neighborManager.CloseNeighbor(peer, PeerCloseReason.InternalFailure, PeerCloseInitiator.LocalNode, exception);
                    return null;
                }
                else
                {
                    //exception that we dont know or cant act on.
                    //we will throw this exception to the user.
                    return exception;
                }
            }
            catch (Exception e)
            {
                if (Fx.IsFatal(e)) throw;
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
                return e;
            }
        }

        public static void EndFloodEncodedMessage(IAsyncResult result)
        {
            CompletedAsyncResult cresult = result as CompletedAsyncResult;
            if (cresult != null)
                CompletedAsyncResult.End(result);
            else
            {
                FloodAsyncResult fresult = result as FloodAsyncResult;
                if (fresult == null)
                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperArgument("result", SR.GetString(SR.InvalidAsyncResult));
                fresult.End();
            }
        }

        public void EndFloodReceivedMessage(IAsyncResult result)
        {
            FloodAsyncResult fresult = result as FloodAsyncResult;
            Fx.Assert(fresult != null, "Invalid FloodAsyncResult instance during EndFloodReceivedMessage");
        }


        public class PeerThrottleHelper
        {
            int outgoingEnqueuedCount = 0;
            int outgoingQuota = 128;
            IFlooderForThrottle flooder;


            public PeerThrottleHelper(IFlooderForThrottle flooder, int outgoingLimit)
            {
                this.outgoingQuota = outgoingLimit;
                this.flooder = flooder;
            }

            public void ItemDequeued()
            {
                Interlocked.Decrement(ref outgoingEnqueuedCount);
            }

            public void AcquireNoQueue()
            {
                int value = Interlocked.Increment(ref outgoingEnqueuedCount);
                if (value >= outgoingQuota)
                {
                    flooder.OnThrottleReached();
                }
            }
        }
    }

    class PeerFlooderSimple : PeerFlooderBase<Message, UtilityInfo>
    {
        ListManager messageIds;
        const uint MaxBuckets = 5;

        internal PeerFlooderSimple(PeerNodeConfig config, PeerNeighborManager neighborManager)
            : base(config, neighborManager)
        {
            //we want a message id cache that holds message ids for atmost 5 mins.
            this.messageIds = new ListManager(MaxBuckets);
        }

        public override bool ShouldProcess(Message message)
        {
            return message.Properties.ContainsKey(PeerStrings.MessageVerified);
        }
        public bool IsNotSeenBefore(Message message, out byte[] id, out int cacheHit)
        {
            cacheHit = -1;
            id = PeerNodeImplementation.DefaultId;
            if (message is SecurityVerifiedMessage)
            {
                id = (message as SecurityVerifiedMessage).PrimarySignatureValue;

            }
            else
            {
                System.Xml.UniqueId messageId = PeerMessageHelpers.GetHeaderUniqueId(message.Headers, PeerStrings.MessageId, PeerStrings.Namespace);
                if (messageId == null)
                    return false;
                if (messageId.IsGuid)
                {
                    id = new byte[16];
                    messageId.TryGetGuid(id, 0);
                }
                else
                    return false;
            }
            cacheHit = messageIds.AddForLookup(id);
            if (cacheHit == -1)
            {
                return true;
            }
            return false;

        }

        public override void RecordOutgoingMessage(byte[] id)
        {
            this.messageIds.AddForFlood(id);
        }

        public override void OnOpen()
        {
        }

        public override void OnClose()
        {
            this.messageIds.Close();
        }


        public override IAsyncResult OnFloodedMessage(IPeerNeighbor neighbor, Message floodInfo, AsyncCallback callback, object state)
        {
            return base.OnFloodedMessage(neighbor, floodInfo, callback, state);
        }

        public override void EndFloodMessage(IAsyncResult result)
        {
            base.EndFloodMessage(result);

        }

        public override void ProcessLinkUtility(IPeerNeighbor neighbor, UtilityInfo utilityInfo)
        {
            if (!PeerNeighborStateHelper.IsConnected(neighbor.State))
            {
                neighbor.Abort(PeerCloseReason.InvalidNeighbor, PeerCloseInitiator.LocalNode);
                return;
            }

            try
            {
                UtilityExtension.ProcessLinkUtility(neighbor, utilityInfo);
            }
            catch (Exception e)
            {
                if (Fx.IsFatal(e)) throw;
                if (null != CloseNeighborIfKnownException(neighborManager, e, neighbor))
                {
                    throw;
                }
                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
            }
        }

        class ListManager
        {
            uint active;            //current bucket.
            readonly uint buckets;

            // Double-checked locking pattern requires volatile for read/write synchronization
            volatile bool disposed = false;
            IOThreadTimer messagePruningTimer;
            //we service the hashtables every one minute
            static readonly int PruningTimout = 60 * 1000;
            static readonly int InitialCount = 1000;
            Dictionary<byte[], bool>[] tables;
            //Hashtable[] tables;
            object thisLock;
            static InMemoryNonceCache.NonceCacheImpl.NonceKeyComparer keyComparer = new InMemoryNonceCache.NonceCacheImpl.NonceKeyComparer();
            const int NotFound = -1;
            //creating this ListManager with n implies that the entries will be available for n minutes atmost.
            //in the n+1 minute, the timer message handler will kick in to clear older messages.
            //every minute, the
            public ListManager(uint buckets)
            {
                if (!(buckets > 1))
                {
                    throw Fx.AssertAndThrow("ListManager should be used atleast with 2 buckets");
                }
                this.buckets = buckets;
                tables = new Dictionary<byte[], bool>[buckets];

                for (uint i = 0; i < buckets; i++)
                {
                    tables[i] = NewCache(InitialCount);
                }
                //create a timer and kickit off for 1 minute
                messagePruningTimer = new IOThreadTimer(new Action<object>(OnTimeout), null, false);
                messagePruningTimer.Set(PruningTimout);
                this.active = 0;
                this.disposed = false;
                this.thisLock = new object();
            }

            object ThisLock
            {
                get
                {
                    return thisLock;
                }
            }

            public int AddForLookup(byte[] key)
            {
                int table = NotFound;
                if (disposed)
                {
                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new InvalidOperationException(SR.GetString(SR.PeerFlooderDisposed)));
                }

                lock (ThisLock)
                {
                    if ((table = Contains(key)) == NotFound)
                    {
                        tables[active].Add(key, false);
                    }
                    return table;
                }
            }

            public bool AddForFlood(byte[] key)
            {
                if (disposed)
                {
                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new InvalidOperationException(SR.GetString(SR.PeerFlooderDisposed)));
                }

                lock (ThisLock)
                {
                    if (UpdateFloodEntry(key))
                    {
                        return true;
                    }
                    else
                    {
                        return false;
                    }
                }
            }

            internal void Close()
            {
                lock (ThisLock)
                {
                    if (disposed)
                        return;
                    messagePruningTimer.Cancel();
                    messagePruningTimer = null;
                    tables = null;
                    disposed = true;
                }
            }

            //it does not use locks and expects the caller to hold the lock.
            internal bool UpdateFloodEntry(byte[] key)
            {
                bool flooded = false;
                //check if the message is present in any of the buckets.
                //assumption is that a hit is likely in the current or most recent bucket.
                //we start looking in the current active table and then in the previous and then backwards ...
                for (uint i = buckets; i > 0; i--)
                {
                    if (tables[(active + i) % buckets].TryGetValue(key, out flooded))
                    {
                        if (!flooded)
                        {
                            tables[(active + i) % buckets][key] = true;
                            return true;
                        }
                        else
                            return false;
                    }
                }
                tables[active].Add(key, true);
                return true;
            }

            //it does not use locks and expects the caller to hold the lock.
            internal int Contains(byte[] key)
            {
                int cache = NotFound;
                uint i = 0;
                //check if the message is present in any of the buckets.
                //assumption is that a hit is likely in the current or most recent bucket.
                //we start looking in the current active table and then in the previous and then backwards ...
                for (i = buckets; i > 0; i--)
                {
                    if (tables[(active + i) % buckets].ContainsKey(key))
                        cache = (int)i;
                }
                if (cache < 0)
                    return cache;
                cache = (int)((active + buckets - i) % buckets);
                return cache;
            }

            void OnTimeout(object state)
            {
                if (disposed)
                    return;
                lock (ThisLock)
                {
                    if (disposed)
                        return;
                    active = (active + 1) % (buckets);
                    tables[active] = NewCache(tables[active].Count);
                    messagePruningTimer.Set(PruningTimout);
                }
            }

            Dictionary<byte[], bool> NewCache(int capacity)
            {
                return new Dictionary<byte[], bool>(capacity, keyComparer);
            }
        }
    }


    // this class should contain a collection of IAsyncResults returned from neighbor.BeginSend
    // and complete once all sends have completed
    class FloodAsyncResult : AsyncResult
    {
        bool doneAdding = false;
        Exception exception;
        PeerNeighborManager pnm;

        // Double-checked locking pattern requires volatile for read/write synchronization
        volatile bool isCompleted = false;
        //async results who signaled completion but we have not called EndSend.
        List<IAsyncResult> pending = new List<IAsyncResult>();
        Dictionary<IAsyncResult, IPeerNeighbor> results = new Dictionary<IAsyncResult, IPeerNeighbor>();
        bool shouldCallComplete = false;
        object thisLock = new object();
        TimeoutHelper timeoutHelper;
        bool offNode = false;
        public event EventHandler OnMessageSent;


        public FloodAsyncResult(PeerNeighborManager owner, TimeSpan timeout, AsyncCallback callback, object state)
            : base(callback, state)
        {
            this.pnm = owner;
            this.timeoutHelper = new TimeoutHelper(timeout);
        }

        object ThisLock
        {
            get
            {
                return thisLock;
            }
        }

        public void AddResult(IAsyncResult result, IPeerNeighbor neighbor)
        {
            lock (ThisLock)
            {
                this.results.Add(result, neighbor);
            }
        }

        //user wants to end business. This method is called as a result of EndSend on the flooder.
        //internal methods do not call this. we are asserting that this method should not be called in case of failed BeginX
        public void End()
        {
            if (!(this.doneAdding && this.shouldCallComplete))
            {
                throw Fx.AssertAndThrow("Unexpected end!");
            }
            if (this.isCompleted)
            {
                return;
            }

            //simply wait on the base's event handle
            bool completed = TimeoutHelper.WaitOne(this.AsyncWaitHandle, this.timeoutHelper.RemainingTime());
            if (!completed)
            {
                // a time out occurred - if mo message went off node then tell AsyncResult to throw.
                if (!offNode)
                {
                    try
                    {
                        throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new TimeoutException());
                    }
                    catch (Exception e)
                    {
                        if (Fx.IsFatal(e)) throw;
                        DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
                        this.exception = e;
                    }
                }
                //otherwise trace that the timeout was not sufficient for complete send
                lock (ThisLock)
                {
                    if (this.isCompleted)
                        return;
                    this.isCompleted = true;
                }
                CompleteOp(false);
            }
            AsyncResult.End<FloodAsyncResult>(this);
        }

        //this method marks the end of BeginX by the flooder.
        //if there were errors during BeginX, this method may be prematurely called
        //in this case, our only job is to call EndX on successful BeginX calls. we do not report back to caller in this case.
        //base.Complete will not be called and End() will not be called. User has already received exception during BeginX
        //if there was no exception during BeginX, excep param is null. In this case, we call base.Complete upon the last EndX
        public void MarkEnd(bool success)
        {
            bool callComplete = false;
            try
            {
                lock (this.ThisLock)
                {
                    foreach (IAsyncResult result in pending)
                    {
                        OnSendComplete(result);
                    }
                    pending.Clear();
                    this.doneAdding = true;
                    this.shouldCallComplete = success; //only call base.Complete if there is no error during BeginX
                    if (this.results.Count == 0)
                    {
                        this.isCompleted = true;
                        callComplete = true;
                    }
                }
            }
            finally
            {
                if (callComplete)
                {
                    CompleteOp(true);
                }
            }

        }


        //this is the callback routine for async completion on channel BeginSend() operations.
        //if we are done, simply return. This can happen if user called sync EndX.
        //if the flooder is still processing BeginSend(), then we probably cant complete. In this case, add the result to pending and return
        //main thread will flush the pending completions in MarkEnd().
        //otherwise, call EndX on the result and remove it from results.
        //if this is the last invoke, signal user using base.Complete AND isCompleted=true
        internal void OnSendComplete(IAsyncResult result)
        {
            bool callComplete = false;
            IPeerNeighbor neighbor = null;
            bool fatal = false;
            if (isCompleted)
                return;
            Message message = (Message)result.AsyncState;

            //wait until flooder had a chance to call all outgoing channels and give us Async results.
            lock (ThisLock)
            {
                if (isCompleted)
                    return;

                if (!this.results.TryGetValue(result, out neighbor))
                {
                    if (!doneAdding)
                        this.pending.Add(result);
                    else
                    {
                        throw Fx.AssertAndThrow("IAsyncResult is un-accounted for.");
                    }
                    return;
                }
                this.results.Remove(result);

                try
                {
                    //try doing this only if the async result is marked !CompletedSynchronously.
                    if (!result.CompletedSynchronously)
                    {
                        neighbor.EndSend(result);
                        offNode = true;
                        UtilityExtension.OnEndSend(neighbor, this);
                    }
                }
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                    {
                        fatal = true;
                        throw;
                    }

                    Exception temp = PeerFlooder.CloseNeighborIfKnownException(pnm, e, neighbor);
                    //we want to return the very first exception to the user.
                    if (temp != null && this.doneAdding && !this.shouldCallComplete)
                        throw;
                    if (this.exception == null)
                    {
                        this.exception = temp;
                    }
                    DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
                }
                finally
                {
                    if (message != null && !result.CompletedSynchronously && !fatal)
                        message.Close();
                }
                //dont want to call Complete from the lock.
                //we just decide if this thread should call complete and call outside the lock.
                if (this.results.Count == 0 && this.doneAdding && this.shouldCallComplete)
                {
                    this.isCompleted = true;
                    callComplete = true;
                }
            }
            //if we are done with callbacks and beginx calls,
            if (callComplete && this.shouldCallComplete)
            {
                CompleteOp(false);
            }
        }

        void CompleteOp(bool sync)
        {
            //call the callback upon finish
            OnMessageSent(this, EventArgs.Empty);
            base.Complete(sync, this.exception);
        }

    }
}