mcs/class/referencesource/System.ServiceModel.Channels/System/ServiceModel/Channels/UdpChannelBase.cs

   1 // <copyright>
   2 // Copyright (c) Microsoft Corporation.  All rights reserved.
   3 // </copyright>
   4
   5 namespace System.ServiceModel.Channels
   6 {
   7     using System;
   8     using System.Collections.Generic;
   9     using System.Diagnostics.CodeAnalysis;
  10     using System.Globalization;
  11     using System.Net;
  12     using System.Net.Sockets;
  13     using System.Runtime;
  14     using System.Runtime.Diagnostics;
  15     using System.ServiceModel.Diagnostics;
  16     using System.Threading;
  17     using System.Xml;
  18
  19     internal abstract class UdpChannelBase<QueueItemType> : InputQueueChannel<QueueItemType>, IUdpReceiveHandler
  20         where QueueItemType : class, IDisposable
  21     {
  22         private bool cleanedUp;
  23         private long pendingMessagesTotalSize;
  24         private long maxPendingMessagesTotalSize;
  25         private int maxReceivedMessageSize;
  26         private UdpRetransmissionSettings retransmitSettings;
  27         private Uri via;
  28
  29         protected UdpChannelBase(
  30             ChannelManagerBase channelManager,
  31             MessageEncoder encoder,
  32             BufferManager bufferManager,
  33             UdpSocket[] sockets,
  34             UdpRetransmissionSettings retransmissionSettings,
  35             long maxPendingMessagesTotalSize,
  36             EndpointAddress localAddress,
  37             Uri via,
  38             bool isMulticast,
  39             int maxReceivedMessageSize)
  40             : base(channelManager)
  41         {
  42             Fx.Assert(encoder != null, "encoder shouldn't be null");
  43             Fx.Assert(bufferManager != null, "buffer manager shouldn't be null");
  44             Fx.Assert(sockets != null, "sendSockets can't be null");
  45             Fx.Assert(sockets.Length > 0, "sendSockets can't be empty");
  46             Fx.Assert(retransmissionSettings != null, "retransmissionSettings can't be null");
  47             Fx.Assert(maxPendingMessagesTotalSize >= 0, "maxPendingMessagesTotalSize must be >= 0");
  48             Fx.Assert(maxReceivedMessageSize > 0, "maxReceivedMessageSize must be > 0");
  49             Fx.Assert(localAddress != null, "localAddress can't be null");
  50             Fx.Assert(via != null, "via can't be null");
  51
  52             this.maxPendingMessagesTotalSize = maxPendingMessagesTotalSize == UdpConstants.Defaults.DefaultMaxPendingMessagesTotalSize ? UdpConstants.Defaults.MaxPendingMessagesTotalSize : maxPendingMessagesTotalSize;
  53             this.Encoder = encoder;
  54             this.Sockets = sockets;
  55             this.BufferManager = bufferManager;
  56             this.retransmitSettings = retransmissionSettings;
  57             this.IsMulticast = isMulticast;
  58             this.DuplicateDetector = null;
  59             this.ReceiveManager = null;
  60             this.OwnsBufferManager = false;
  61             this.maxReceivedMessageSize = maxReceivedMessageSize;
  62             this.LocalAddress = localAddress;
  63             this.via = via;
  64         }
  65
  66         public EndpointAddress LocalAddress
  67         {
  68             get;
  69             private set;
  70         }
  71
  72         public Uri Via
  73         {
  74             get { return this.via; }
  75         }
  76
  77         int IUdpReceiveHandler.MaxReceivedMessageSize
  78         {
  79             get { return this.maxReceivedMessageSize; }
  80         }
  81
  82         protected abstract bool IgnoreSerializationException { get; }
  83
  84         protected bool OwnsBufferManager { get; set; }
  85
  86         protected DuplicateMessageDetector DuplicateDetector { get; set; }
  87
  88         protected UdpSocketReceiveManager ReceiveManager { get; set; }
  89
  90         protected BufferManager BufferManager
  91         {
  92             get;
  93             private set;
  94         }
  95
  96         protected MessageEncoder Encoder
  97         {
  98             get;
  99             private set;
 100         }
 101
 102         protected bool IsMulticast
 103         {
 104             get;
 105             private set;
 106         }
 107
 108         protected UdpOutputChannel UdpOutputChannel { get; private set; }
 109
 110         protected UdpSocket[] Sockets
 111         {
 112             get;
 113             private set;
 114         }
 115
 116         [SuppressMessage("Microsoft.StyleCop.CSharp.ReadabilityRules", "SA1100:DoNotPrefixCallsWithBaseUnlessLocalImplementationExists", Justification = "StyleCop 4.5 does not validate this rule properly.")]
 117         public override T GetProperty<T>()
 118         {
 119             if (typeof(T) == typeof(IDuplexChannel))
 120             {
 121                 return (T)(object)this;
 122             }
 123
 124             T outputChannelProperty = this.UdpOutputChannel.GetProperty<T>();
 125             if (outputChannelProperty != null)
 126             {
 127                 return outputChannelProperty;
 128             }
 129
 130             T messageEncoderProperty = this.Encoder.GetProperty<T>();
 131             if (messageEncoderProperty != null)
 132             {
 133                 return messageEncoderProperty;
 134             }
 135
 136             return base.GetProperty<T>();
 137         }
 138
 139         // returns false if the message was dropped because the max pending message count was hit.
 140         bool IUdpReceiveHandler.HandleDataReceived(ArraySegment<byte> data, EndPoint remoteEndpoint, int interfaceIndex, Action onMessageDequeuedCallback)
 141         {
 142             bool returnBuffer = true;
 143             string messageHash = null;
 144             Message message = null;
 145             bool continueReceiving = true;
 146
 147             try
 148             {
 149                 IPEndPoint remoteIPEndPoint = (IPEndPoint)remoteEndpoint;
 150
 151                 message = UdpUtility.DecodeMessage(
 152                     this.DuplicateDetector,
 153                     this.Encoder,
 154                     this.BufferManager,
 155                     data,
 156                     remoteIPEndPoint,
 157                     interfaceIndex,
 158                     this.IgnoreSerializationException,
 159                     out messageHash);
 160
 161                 if (message != null)
 162                 {
 163                     // We pass in the length of the message buffer instead of the length of the message to keep track of the amount of memory that's been allocated
 164                     continueReceiving = this.EnqueueMessage(message, data.Array.Length, onMessageDequeuedCallback);
 165                     returnBuffer = !continueReceiving;
 166                 }
 167             }
 168             catch (Exception e)
 169             {
 170                 if (Fx.IsFatal(e))
 171                 {
 172                     returnBuffer = false;
 173                     throw;
 174                 }
 175
 176                 this.HandleReceiveException(e);
 177             }
 178             finally
 179             {
 180                 if (returnBuffer)
 181                 {
 182                     if (message != null)
 183                     {
 184                         if (this.DuplicateDetector != null)
 185                         {
 186                             Fx.Assert(messageHash != null, "message hash should always be available if duplicate detector is enabled");
 187                             this.DuplicateDetector.RemoveEntry(messageHash);
 188                         }
 189
 190                         message.Close(); // implicitly returns the buffer
 191                     }
 192                     else
 193                     {
 194                         this.BufferManager.ReturnBuffer(data.Array);
 195                     }
 196                 }
 197             }
 198
 199             return continueReceiving;
 200         }
 201
 202         void IUdpReceiveHandler.HandleAsyncException(Exception ex)
 203         {
 204             this.HandleReceiveException(ex);
 205         }
 206
 207         internal virtual void HandleReceiveException(Exception ex)
 208         {
 209             this.EnqueueAndDispatch(UdpUtility.WrapAsyncException(ex), null, false);
 210         }
 211
 212         // Since ChannelListener and channel lifetimes can be different, we need a
 213         // way to transfer the socketReceiveManager and DuplicateMessageDetection
 214         // objects to the channel if the listener gets closed.  If this method succeeds, then
 215         // this also indicates that the bufferManager is no longer owned by the channel listener,
 216         // so we have to clean that up also.
 217         internal bool TransferReceiveManagerOwnership(UdpSocketReceiveManager socketReceiveManager, DuplicateMessageDetector duplicateDetector)
 218         {
 219             bool success = false;
 220             if (this.State == CommunicationState.Opened)
 221             {
 222                 lock (ThisLock)
 223                 {
 224                     if (this.State == CommunicationState.Opened)
 225                     {
 226                         Fx.Assert(this.ReceiveManager == null, "ReceiveManager is already set to a non-null value");
 227                         Fx.Assert(this.DuplicateDetector == null, "DuplicateDetector is already set to a non-null value");
 228
 229                         this.ReceiveManager = socketReceiveManager;
 230                         this.OwnsBufferManager = true;
 231                         this.ReceiveManager.SetReceiveHandler(this);
 232                         this.DuplicateDetector = duplicateDetector;
 233                         success = true;
 234                     }
 235                 }
 236             }
 237
 238             return success;
 239         }
 240
 241         // returns false if the max pending messages total size was hit.
 242         internal bool EnqueueMessage(Message message, int messageBufferSize, Action messageDequeuedCallback)
 243         {
 244             Action onMessageDequeuedCallback = () =>
 245             {
 246                 lock (this.ThisLock)
 247                 {
 248                     this.pendingMessagesTotalSize -= messageBufferSize;
 249                     Fx.Assert(this.pendingMessagesTotalSize >= 0, "pendingMessagesTotalSize should not be negative.");
 250                 }
 251
 252                 messageDequeuedCallback();
 253             };
 254
 255             bool success = false;
 256             lock (this.ThisLock)
 257             {
 258                 if (this.pendingMessagesTotalSize + messageBufferSize <= this.maxPendingMessagesTotalSize)
 259                 {
 260                     message.Properties.Via = this.Via;
 261                     this.pendingMessagesTotalSize += messageBufferSize;
 262                     try
 263                     {
 264                         this.FinishEnqueueMessage(message, onMessageDequeuedCallback, false);
 265                         success = true;
 266                     }
 267                     finally
 268                     {
 269                         if (!success)
 270                         {
 271                             this.pendingMessagesTotalSize -= messageBufferSize;
 272                         }
 273                     }
 274                 }
 275                 else
 276                 {
 277                     if (TD.MaxPendingMessagesTotalSizeReachedIsEnabled())
 278                     {
 279                         string messageIdString = string.Empty;
 280                         if (message.Headers.MessageId != null)
 281                         {
 282                             messageIdString = string.Format(CultureInfo.CurrentCulture, "'{0}' ", message.Headers.MessageId.ToString());
 283                         }
 284
 285                         EventTraceActivity eventTraceActivity = EventTraceActivityHelper.TryExtractActivity(message);
 286                         TD.MaxPendingMessagesTotalSizeReached(eventTraceActivity, messageIdString, this.maxPendingMessagesTotalSize, typeof(TransportBindingElement).FullName);
 287                     }
 288                 }
 289             }
 290
 291             return success;
 292         }
 293
 294         internal abstract void FinishEnqueueMessage(Message message, Action dequeuedCallback, bool canDispatchOnThisThread);
 295
 296         protected virtual void AddHeadersTo(Message message)
 297         {
 298             Fx.Assert(message != null, "Message can't be null");
 299
 300             if (message.Version.Addressing != AddressingVersion.None)
 301             {
 302                 if (message.Headers.MessageId == null)
 303                 {
 304                     message.Headers.MessageId = new UniqueId();
 305                 }
 306             }
 307             else
 308             {
 309                 if (this.retransmitSettings.Enabled == true)
 310                 {
 311                     // we should only get here if some channel above us starts producing messages that don't match the encoder's message version.
 312                     throw FxTrace.Exception.AsError(new ProtocolException(SR.RetransmissionRequiresAddressingOnMessage(message.Version.Addressing.ToString())));
 313                 }
 314             }
 315         }
 316
 317         // Closes the channel ungracefully during error conditions.
 318         protected override void OnAbort()
 319         {
 320             this.Cleanup(true, TimeSpan.Zero);
 321         }
 322
 323         protected override IAsyncResult OnBeginOpen(TimeSpan timeout, AsyncCallback callback, object state)
 324         {
 325             this.OnOpen(timeout);
 326             return new CompletedAsyncResult(callback, state);
 327         }
 328
 329         protected override void OnEndOpen(IAsyncResult result)
 330         {
 331             CompletedAsyncResult.End(result);
 332         }
 333
 334         protected override void OnOpen(TimeSpan timeout)
 335         {
 336             this.UdpOutputChannel.Open();
 337         }
 338
 339         protected override IAsyncResult OnBeginClose(TimeSpan timeout, AsyncCallback callback, object state)
 340         {
 341             return new CloseAsyncResult<QueueItemType>(
 342                 this,
 343                 new ChainedBeginHandler(base.OnBeginClose),
 344                 new ChainedEndHandler(base.OnEndClose),
 345                 timeout,
 346                 callback,
 347                 state);
 348         }
 349
 350         protected override void OnEndClose(IAsyncResult result)
 351         {
 352             CloseAsyncResult<QueueItemType>.End(result);
 353         }
 354
 355         // Closes the channel gracefully during normal conditions.
 356         protected override void OnClose(TimeSpan timeout)
 357         {
 358             TimeoutHelper timeoutHelper = new TimeoutHelper(timeout);
 359             this.Cleanup(false, timeoutHelper.RemainingTime());
 360             base.OnClose(timeoutHelper.RemainingTime());
 361         }
 362
 363         protected void SetOutputChannel(UdpOutputChannel udpOutputChannel)
 364         {
 365             Fx.Assert(this.UdpOutputChannel == null, "this.UdpOutputChannel must be null");
 366             Fx.Assert(udpOutputChannel != null, "udpOutputChannel can't be null, since SetOutputChannel should be called only once");
 367
 368             this.UdpOutputChannel = udpOutputChannel;
 369         }
 370
 371         // We're guaranteed by CommunicationObject that at most ONE of Close or BeginClose will be called once.
 372         protected void Cleanup(bool aborting, TimeSpan timeout)
 373         {
 374             if (this.cleanedUp)
 375             {
 376                 return;
 377             }
 378
 379             lock (ThisLock)
 380             {
 381                 if (this.cleanedUp)
 382                 {
 383                     return;
 384                 }
 385
 386                 if (aborting)
 387                 {
 388                     this.UdpOutputChannel.Abort();
 389                 }
 390                 else
 391                 {
 392                     this.UdpOutputChannel.Close(timeout);
 393                 }
 394
 395                 if (this.DuplicateDetector != null)
 396                 {
 397                     this.DuplicateDetector.Dispose();
 398                 }
 399
 400                 if (this.ReceiveManager != null)
 401                 {
 402                     this.ReceiveManager.Close();
 403                 }
 404
 405                 this.CleanupBufferManager();
 406
 407                 this.cleanedUp = true;
 408             }
 409         }
 410
 411         private void CleanupBufferManager()
 412         {
 413             if (this.OwnsBufferManager)
 414             {
 415                 this.BufferManager.Clear();
 416             }
 417         }
 418
 419         // Control flow for async path
 420         // We use this mechanism to avoid initializing two async objects as logically cleanup+close is one operation.
 421         // At any point in the Begin* methods, we may go async. The steps are:
 422         // - Close inner UdpOutputChannel
 423         // - Cleanup channel
 424         // - Close channel
 425         private class CloseAsyncResult<T> : AsyncResult
 426             where T : class, IDisposable
 427         {
 428             private static AsyncCompletion completeCloseOutputChannelCallback = new AsyncCompletion(CompleteCloseOutputChannel);
 429             private static AsyncCompletion completeBaseCloseCallback = new AsyncCompletion(CompleteBaseClose);
 430
 431             private UdpChannelBase<T> channel;
 432             private TimeoutHelper timeoutHelper;
 433             private ChainedBeginHandler baseBeginClose;
 434             private ChainedEndHandler baseEndClose;
 435
 436             public CloseAsyncResult(UdpChannelBase<T> channel, ChainedBeginHandler baseBeginClose, ChainedEndHandler baseEndClose, TimeSpan timeout, AsyncCallback callback, object state)
 437                 : base(callback, state)
 438             {
 439                 this.channel = channel;
 440                 this.baseBeginClose = baseBeginClose;
 441                 this.baseEndClose = baseEndClose;
 442                 this.timeoutHelper = new TimeoutHelper(timeout);
 443
 444                 if (this.BeginCloseOutputChannel())
 445                 {
 446                     this.Complete(true);
 447                 }
 448             }
 449
 450             public static void End(IAsyncResult result)
 451             {
 452                 AsyncResult.End<CloseAsyncResult<T>>(result);
 453             }
 454
 455             private static bool CompleteBaseClose(IAsyncResult result)
 456             {
 457                 // AsyncResult.AsyncCompletionWrapperCallback takes care of catching exceptions for us.
 458                 CloseAsyncResult<T> thisPtr = (CloseAsyncResult<T>)result.AsyncState;
 459
 460                 // We are completing the base class close operation at this point.
 461                 thisPtr.baseEndClose(result);
 462
 463                 return true;
 464             }
 465
 466             private static bool CompleteCloseOutputChannel(IAsyncResult result)
 467             {
 468                 // AsyncResult.AsyncCompletionWrapperCallback takes care of catching exceptions for us.
 469                 CloseAsyncResult<T> thisPtr = (CloseAsyncResult<T>)result.AsyncState;
 470
 471                 // We are completing the base class close operation at this point.
 472                 thisPtr.channel.UdpOutputChannel.EndClose(result);
 473
 474                 thisPtr.channel.Cleanup(false, thisPtr.timeoutHelper.RemainingTime());
 475
 476                 return thisPtr.BeginBaseClose();
 477             }
 478
 479             private bool BeginCloseOutputChannel()
 480             {
 481                 // AsyncResult.AsyncCompletionWrapperCallback takes care of catching the exceptions for us.
 482                 IAsyncResult result = this.channel.UdpOutputChannel.BeginClose(this.timeoutHelper.RemainingTime(), this.PrepareAsyncCompletion(completeCloseOutputChannelCallback), this);
 483
 484                 // SyncContinue calls CompleteCloseOutputChannel for us in sync case.
 485                 return this.SyncContinue(result);
 486             }
 487
 488             private bool BeginBaseClose()
 489             {
 490                 // AsyncResult.AsyncCompletionWrapperCallback takes care of catching the exceptions for us.
 491                 IAsyncResult result = this.baseBeginClose(this.timeoutHelper.RemainingTime(), this.PrepareAsyncCompletion(completeBaseCloseCallback), this);
 492
 493                 // SyncContinue calls CompleteBaseClose for us in sync case.
 494                 return this.SyncContinue(result);
 495             }
 496         }
 497     }
 498 }