mcs/class/referencesource/System/net/System/Net/connectionpool.cs

   1 //------------------------------------------------------------------------------
   2 // <copyright file="ConnectionPool.cs" company="Microsoft">
   3 //     Copyright (c) Microsoft Corporation.  All rights reserved.
   4 // </copyright>
   5 //------------------------------------------------------------------------------
   6
   7 namespace System.Net {
   8
   9     using System;
  10     using System.Net.Sockets;
  11     using System.Collections;
  12     using System.Diagnostics;
  13     using System.Globalization;
  14     using System.Runtime.InteropServices;
  15     using System.Security;
  16     using System.Security.Permissions;
  17     using System.Threading;
  18
  19     internal delegate void GeneralAsyncDelegate(object request, object state);
  20     internal delegate PooledStream CreateConnectionDelegate(ConnectionPool pool);
  21
  22     /// <devdoc>
  23     /// <para>
  24     ///     Impliments basic ConnectionPooling by pooling PooledStreams
  25     /// </para>
  26     /// </devdoc>
  27     internal class ConnectionPool {
  28         private enum State {
  29             Initializing,
  30             Running,
  31             ShuttingDown,
  32         }
  33
  34         private static TimerThread.Callback s_CleanupCallback = new TimerThread.Callback(CleanupCallbackWrapper);
  35         private static TimerThread.Callback s_CancelErrorCallback = new TimerThread.Callback(CancelErrorCallbackWrapper);
  36         private static TimerThread.Queue s_CancelErrorQueue = TimerThread.GetOrCreateQueue(ErrorWait);
  37
  38         private const int MaxQueueSize    = (int)0x00100000;
  39
  40         // The order of these is important; we want the WaitAny call to be signaled
  41         // for a free object before a creation signal.  Only the index first signaled
  42         // object is returned from the WaitAny call.
  43         private const int SemaphoreHandleIndex = (int)0x0;
  44         private const int ErrorHandleIndex     = (int)0x1;
  45         private const int CreationHandleIndex  = (int)0x2;
  46
  47         private const int WaitTimeout   = (int)0x102;
  48         private const int WaitAbandoned = (int)0x80;
  49
  50         private const int ErrorWait     = 5 * 1000; // 5 seconds
  51
  52         private readonly TimerThread.Queue m_CleanupQueue;
  53
  54         private State                    m_State;
  55         private InterlockedStack         m_StackOld;
  56         private InterlockedStack         m_StackNew;
  57
  58         private int                      m_WaitCount;
  59         private WaitHandle[]             m_WaitHandles;
  60
  61         private Exception                m_ResError;
  62         private volatile bool            m_ErrorOccured;
  63
  64         private TimerThread.Timer        m_ErrorTimer;
  65
  66         private ArrayList                m_ObjectList;
  67         private int                      m_TotalObjects;
  68
  69         private Queue                    m_QueuedRequests;
  70         private Thread                   m_AsyncThread;
  71
  72         private int                      m_MaxPoolSize;
  73         private int                      m_MinPoolSize;
  74         private ServicePoint             m_ServicePoint;
  75         private CreateConnectionDelegate m_CreateConnectionCallback;
  76
  77         private Mutex CreationMutex {
  78             get {
  79                 return (Mutex) m_WaitHandles[CreationHandleIndex];
  80             }
  81         }
  82
  83         private ManualResetEvent ErrorEvent {
  84             get {
  85                 return (ManualResetEvent) m_WaitHandles[ErrorHandleIndex];
  86             }
  87         }
  88
  89         private Semaphore Semaphore {
  90             get {
  91                 return (Semaphore) m_WaitHandles[SemaphoreHandleIndex];
  92             }
  93         }
  94
  95         /// <summary>
  96         ///    <para>Constructor - binds pool with a servicePoint and sets up a cleanup Timer to remove Idle Connections</para>
  97         /// </summary>
  98         internal ConnectionPool(ServicePoint servicePoint, int maxPoolSize, int minPoolSize, int idleTimeout, CreateConnectionDelegate createConnectionCallback) : base() {
  99             m_State                = State.Initializing;
 100
 101             m_CreateConnectionCallback = createConnectionCallback;
 102             m_MaxPoolSize = maxPoolSize;
 103             m_MinPoolSize = minPoolSize;
 104             m_ServicePoint = servicePoint;
 105
 106             Initialize();
 107
 108             if (idleTimeout > 0) {
 109                 // special case: if the timeout value is 1 then the timer thread should have a duration
 110                 // of 1 to avoid having the timer callback run constantly
 111                 m_CleanupQueue = TimerThread.GetOrCreateQueue(idleTimeout == 1 ? 1 : (idleTimeout / 2));
 112                 m_CleanupQueue.CreateTimer(s_CleanupCallback, this);
 113             }
 114         }
 115
 116         /// <summary>
 117         ///    <para>Internal init stuff, creates stacks, queue, wait handles etc</para>
 118         /// </summary>
 119         private void Initialize() {
 120             m_StackOld          = new InterlockedStack();
 121             m_StackNew          = new InterlockedStack();
 122
 123             m_QueuedRequests = new Queue();
 124
 125             m_WaitHandles     = new WaitHandle[3];
 126             m_WaitHandles[SemaphoreHandleIndex] = new Semaphore(0, MaxQueueSize);
 127             m_WaitHandles[ErrorHandleIndex]     = new ManualResetEvent(false);
 128             m_WaitHandles[CreationHandleIndex]  = new Mutex();
 129
 130             m_ErrorTimer         = null;  // No error yet.
 131
 132             m_ObjectList            = new ArrayList();
 133             m_State = State.Running;
 134         }
 135
 136
 137         /// <summary>
 138         ///    <para>Async state object, used for storing state on async calls</para>
 139         /// </summary>
 140         private class AsyncConnectionPoolRequest {
 141             public AsyncConnectionPoolRequest(ConnectionPool pool, object owningObject, GeneralAsyncDelegate asyncCallback, int creationTimeout) {
 142                 Pool = pool;
 143                 OwningObject = owningObject;
 144                 AsyncCallback = asyncCallback;
 145                 CreationTimeout = creationTimeout;
 146             }
 147             public object OwningObject;
 148             public GeneralAsyncDelegate AsyncCallback;
 149 #if !MONO
 150             public bool Completed;
 151 #endif
 152             public ConnectionPool Pool;
 153             public int CreationTimeout;
 154         }
 155
 156         /// <summary>
 157         ///    <para>Queues a AsyncConnectionPoolRequest to our queue of requests needing
 158         ///     a pooled stream. If an AsyncThread is not created, we create one,
 159         ///     and let it process the queued items</para>
 160         /// </summary>
 161         private void QueueRequest(AsyncConnectionPoolRequest asyncRequest) {
 162             lock(m_QueuedRequests) {
 163                 m_QueuedRequests.Enqueue(asyncRequest);
 164                 if (m_AsyncThread == null) {
 165                     m_AsyncThread = new Thread(new ThreadStart(AsyncThread));
 166                     m_AsyncThread.IsBackground = true;
 167                     m_AsyncThread.Start();
 168                 }
 169             }
 170         }
 171
 172         /// <summary>
 173         ///    <para>Processes async queued requests that are blocked on needing a free pooled stream
 174         ///         works as follows:
 175         ///         1. while there are blocked requests, take one out of the queue
 176         ///         2. Wait for a free connection, when one becomes avail, then notify the request that its there
 177         ///         3. repeat 1 until there are no more queued requests
 178         ///         4. if there are no more requests waiting to for a free stream, then close down this thread
 179         ///</para>
 180         /// </summary>
 181         private void AsyncThread() {
 182             do {
 183                 while (m_QueuedRequests.Count > 0) {
 184                     bool continueLoop = true;
 185                     AsyncConnectionPoolRequest asyncState = null;
 186                     lock (m_QueuedRequests) {
 187                         asyncState = (AsyncConnectionPoolRequest) m_QueuedRequests.Dequeue();
 188                     }
 189
 190                     WaitHandle [] localWaitHandles = m_WaitHandles;
 191                     PooledStream PooledStream = null;
 192                     try {
 193                         while ((PooledStream == null) && continueLoop) {
 194                             int result = WaitHandle.WaitAny(localWaitHandles, asyncState.CreationTimeout, false);
 195                             PooledStream =
 196                                 Get(asyncState.OwningObject, result, ref continueLoop, ref localWaitHandles);
 197                         }
 198
 199                         PooledStream.Activate(asyncState.OwningObject, asyncState.AsyncCallback);
 200                     } catch (Exception e) {
 201                         if(PooledStream != null){
 202                             PutConnection(PooledStream, asyncState.OwningObject, asyncState.CreationTimeout, false);
 203                         }
 204                         asyncState.AsyncCallback(asyncState.OwningObject, e);
 205                     }
 206                 }
 207                 Thread.Sleep(500);
 208                 lock(m_QueuedRequests) {
 209                     if (m_QueuedRequests.Count == 0) {
 210                         m_AsyncThread = null;
 211                         break;
 212                     }
 213                 }
 214             } while (true);
 215         }
 216
 217         /// <summary>
 218         ///    <para>Count of total pooled streams associated with this pool, including streams that are being used</para>
 219         /// </summary>
 220         internal int Count {
 221             get { return(m_TotalObjects); }
 222         }
 223
 224         /// <summary>
 225         ///    <para>Our ServicePoint, used for IP resolution</para>
 226         /// </summary>
 227         internal ServicePoint ServicePoint {
 228             get {
 229                 return m_ServicePoint;
 230             }
 231         }
 232
 233         /// <summary>
 234         ///    <para>Our Max Size of outstanding pooled streams</para>
 235         /// </summary>
 236         internal int MaxPoolSize {
 237             get {
 238                 return m_MaxPoolSize;
 239             }
 240         }
 241
 242         /// <summary>
 243         ///    <para>Our Min Size of the pool to remove idled items down to</para>
 244         /// </summary>
 245         internal int MinPoolSize {
 246             get {
 247                 return m_MinPoolSize;
 248             }
 249         }
 250
 251         /// <summary>
 252         ///    <para>An Error occurred usually due to an abort</para>
 253         /// </summary>
 254         private bool ErrorOccurred {
 255             get { return m_ErrorOccured; }
 256         }
 257
 258         private static void CleanupCallbackWrapper(TimerThread.Timer timer, int timeNoticed, object context)
 259         {
 260             ConnectionPool pThis = (ConnectionPool) context;
 261
 262             try
 263             {
 264                 pThis.CleanupCallback();
 265             }
 266             finally
 267             {
 268                 pThis.m_CleanupQueue.CreateTimer(s_CleanupCallback, context);
 269             }
 270         }
 271
 272         /// <summary>
 273         /// Cleans up everything in both the old and new stack.  If a connection is in use
 274         /// then it will be on neither stack and it is the responsibility of the object
 275         /// using that connection to clean it up when it is finished using it.  This does
 276         /// not clean up the ConnectionPool object and new connections can still be
 277         /// created if needed in the future should this ConnectionPool object be reused
 278         ///
 279         /// preconditions: none
 280         ///
 281         /// postconditions: any connections not currently in use by an object will be
 282         /// gracefully terminated and purged from this connection pool
 283         /// </summary>
 284         internal void ForceCleanup()
 285         {
 286             if (Logging.On) {
 287                 Logging.Enter(Logging.Web, "ConnectionPool::ForceCleanup");
 288             }
 289
 290             // If WaitOne returns false, all connections in the pool are in use
 291             // so no cleanup should be performed. The last object owning
 292             // a connection from the pool will perform final cleanup.
 293             while (Count > 0) {
 294                 if (Semaphore.WaitOne(0, false)) {
 295                     // Try to clean up from new stack first, if there isn't anything on new
 296                     // then try old.  When we lock the Semaphore, it gives us a license to
 297                     // remove only one connection from the pool but it can be from either
 298                     // stack since if the Semaphore is locked by another thread it means that
 299                     // there must have been more than one connection available in either stack
 300                     PooledStream pooledStream = (PooledStream)m_StackNew.Pop();
 301
 302                     // no streams in stack new, there must therefore be one in stack old since we
 303                     // were able to acquire the semaphore
 304                     if(pooledStream == null) {
 305                         pooledStream = (PooledStream)m_StackOld.Pop();
 306                     }
 307
 308                     Debug.Assert(pooledStream != null, "Acquired Semaphore with no connections in either stack");
 309                     Destroy(pooledStream);
 310                 }
 311                 else {
 312                     // couldn't get semaphore, nothing to do here
 313                     break;
 314                 }
 315             }
 316
 317             if (Logging.On) {
 318                 Logging.Exit(Logging.Web, "ConnectionPool::ForceCleanup");
 319             }
 320         }
 321
 322         /// <summary>
 323         ///    <para>This is called by a timer, to check for needed cleanup of idle pooled streams</para>
 324         /// </summary>
 325         private void CleanupCallback()
 326         {
 327             // Called when the cleanup-timer ticks over.
 328             //
 329             // This is the automatic prunning method.  Every period, we will perform a two-step
 330             // process.  First, for the objects above MinPool, we will obtain the semaphore for
 331             // the object and then destroy it if it was on the old stack.  We will continue this
 332             // until we either reach MinPool size, or we are unable to obtain a free object, or
 333             // until we have exhausted all the objects on the old stack.  After that, push all
 334             // objects on the new stack to the old stack.  So, every period the objects on the
 335             // old stack are destroyed and the objects on the new stack are pushed to the old
 336             // stack.  All objects that are currently out and in use are not on either stack.
 337             // With this logic, a object is prunned if unused for at least one period but not
 338             // more than two periods.
 339
 340             // Destroy free objects above MinPool size from old stack.
 341             while(Count > MinPoolSize) { // While above MinPoolSize...
 342
 343                 // acquiring the Semaphore gives us a license to remove one and only
 344                 // one connection from the pool
 345                 if (Semaphore.WaitOne(0, false) ) { // != WaitTimeout
 346                     // We obtained a objects from the semaphore.
 347                     PooledStream pooledStream = (PooledStream) m_StackOld.Pop();
 348
 349                     if (null != pooledStream) {
 350                         // If we obtained one from the old stack, destroy it.
 351                         Destroy(pooledStream);
 352                     }
 353                     else {
 354                         // Else we exhausted the old stack, so break
 355                         // and release the Semaphore to indicate that
 356                         // no connection was actually removed so whatever
 357                         // we had locked is still available.
 358                         Semaphore.ReleaseSemaphore();
 359                         break;
 360                     }
 361                 }
 362                 else break;
 363             }
 364
 365             // Push to the old-stack.  For each free object, move object from new stack
 366             // to old stack.  The Semaphore guarantees that we are allowed to handle
 367             // one connection at a time so moving a connection between stacks is safe since
 368             // one connection is reserved for the duration of this loop and we only touch
 369             // one connection at a time on the new stack
 370             if(Semaphore.WaitOne(0, false)) { //  != WaitTimeout
 371                 for(;;) {
 372                     PooledStream pooledStream = (PooledStream) m_StackNew.Pop();
 373
 374                     if (null == pooledStream)
 375                         break;
 376
 377                     GlobalLog.Assert(!pooledStream.IsEmancipated, "Pooled object not in pool.");
 378                     GlobalLog.Assert(pooledStream.CanBePooled, "Pooled object is not poolable.");
 379
 380                     m_StackOld.Push(pooledStream);
 381                 }
 382                 // no connections were actually destroyed so signal that a connection is now
 383                 // available since we are no longer reserving a connection by holding the
 384                 // Semaphore
 385                 Semaphore.ReleaseSemaphore();
 386             }
 387         }
 388
 389         /// <summary>
 390         ///    <para>Creates a new PooledStream, performs checks as well on the new stream</para>
 391         /// </summary>
 392         private PooledStream Create(CreateConnectionDelegate createConnectionCallback) {
 393             GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::Create");
 394             PooledStream newObj = null;
 395
 396             try {
 397                 newObj = createConnectionCallback(this);
 398
 399                 if (null == newObj)
 400                     throw new InternalException();    // Create succeeded, but null object
 401
 402                 if (!newObj.CanBePooled)
 403                     throw new InternalException();    // Create succeeded, but non-poolable object
 404
 405                 newObj.PrePush(null);
 406
 407                 lock (m_ObjectList.SyncRoot) {
 408                     m_ObjectList.Add(newObj);
 409                     m_TotalObjects = m_ObjectList.Count;
 410                 }
 411
 412                 GlobalLog.Print("Create pooledStream#"+ValidationHelper.HashString(newObj));
 413             }
 414             catch(Exception e)  {
 415                 GlobalLog.Print("Pool Exception: " + e.Message);
 416
 417                 newObj = null; // set to null, so we do not return bad new object
 418                 // Failed to create instance
 419                 m_ResError = e;
 420                 Abort();
 421             }
 422             GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::Create",ValidationHelper.HashString(newObj));
 423             return newObj;
 424         }
 425
 426
 427         /// <summary>
 428         ///    <para>Destroys a pooled stream from the pool</para>
 429         /// </summary>
 430         private void Destroy(PooledStream pooledStream) {
 431             GlobalLog.Print("Destroy pooledStream#"+ValidationHelper.HashString(pooledStream));
 432
 433             if (null != pooledStream) {
 434                 try
 435                 {
 436                     lock (m_ObjectList.SyncRoot) {
 437                         m_ObjectList.Remove(pooledStream);
 438                         m_TotalObjects = m_ObjectList.Count;
 439                     }
 440                 }
 441                 finally
 442                 {
 443                     pooledStream.Dispose();
 444                 }
 445             }
 446         }
 447
 448         private static void CancelErrorCallbackWrapper(TimerThread.Timer timer, int timeNoticed, object context)
 449         {
 450             ((ConnectionPool) context).CancelErrorCallback();
 451         }
 452
 453         /// <summary>
 454         ///    <para>Called on error, after we waited a set amount of time from aborting</para>
 455         /// </summary>
 456         private void CancelErrorCallback()
 457         {
 458             TimerThread.Timer timer = m_ErrorTimer;
 459             if (timer != null && timer.Cancel())
 460             {
 461                 m_ErrorOccured = false;
 462                 ErrorEvent.Reset();
 463                 m_ErrorTimer = null;
 464                 m_ResError = null;
 465             }
 466         }
 467
 468         /// <summary>
 469         ///    <para>Retrieves a pooled stream from the pool proper
 470         ///     this work by first attemting to find something in the pool on the New stack
 471         ///     and then trying the Old stack if something is not there availble </para>
 472         /// </summary>
 473         private PooledStream GetFromPool(object owningObject) {
 474             PooledStream res = null;
 475             GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::GetFromPool");
 476             res = (PooledStream) m_StackNew.Pop();
 477             if (null == res) {
 478                 res = (PooledStream) m_StackOld.Pop();
 479             }
 480
 481             // The semaphore guaranteed that a connection was available so if res is
 482             // null it means that this contract has been violated somewhere
 483             GlobalLog.Assert(res != null, "GetFromPool called with nothing in the pool!");
 484
 485             if (null != res) {
 486                 res.PostPop(owningObject);
 487                 GlobalLog.Print("GetFromGeneralPool pooledStream#"+ValidationHelper.HashString(res));
 488             }
 489
 490             GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::GetFromPool",ValidationHelper.HashString(res));
 491             return(res);
 492         }
 493
 494         /// <summary>
 495         ///    <para>Retrieves the pooled stream out of the pool, does this by using the result
 496         ///    of a WaitAny as input, and then based on whether it has a mutex, event, semaphore,
 497         ///     or timeout decides what action to take</para>
 498         /// </summary>
 499         private PooledStream Get(object owningObject, int result, ref bool continueLoop, ref WaitHandle [] waitHandles) {
 500                 PooledStream pooledStream = null;
 501                 GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::Get", result.ToString());
 502
 503
 504                 // From the WaitAny docs: "If more than one object became signaled during
 505                 // the call, this is the array index of the signaled object with the
 506                 // smallest index value of all the signaled objects."  This is important
 507                 // so that the free object signal will be returned before a creation
 508                 // signal.
 509
 510                 switch (result) {
 511                 case WaitTimeout:
 512                     Interlocked.Decrement(ref m_WaitCount);
 513                     continueLoop = false;
 514                     GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::Get","throw Timeout WebException");
 515                     throw new WebException(NetRes.GetWebStatusString("net_timeout", WebExceptionStatus.ConnectFailure), WebExceptionStatus.Timeout);
 516
 517
 518                 case ErrorHandleIndex:
 519                     // Throw the error that PoolCreateRequest stashed.
 520                     int newWaitCount = Interlocked.Decrement(ref m_WaitCount);
 521                     continueLoop = false;
 522                     Exception exceptionToThrow = m_ResError;
 523                     if (newWaitCount == 0) {
 524                         CancelErrorCallback();
 525                     }
 526                     throw exceptionToThrow;
 527
 528                 // The creation mutex signaled, which means no connections are available in
 529                 // the connection pool.  This means you might be able to create a connection.
 530                 case CreationHandleIndex:
 531                     try {
 532                         continueLoop = true;
 533                         // try creating a new connection
 534                         pooledStream = UserCreateRequest();
 535
 536                         if (null != pooledStream) {
 537                             pooledStream.PostPop(owningObject);
 538                             Interlocked.Decrement(ref m_WaitCount);
 539                             continueLoop = false;
 540
 541                         }
 542                         else {
 543                             // If we were not able to create an object, check to see if
 544                             // we reached MaxPoolSize.  If so, we will no longer wait on
 545                             // the CreationHandle, but instead wait for a free object or
 546                             // the timeout.
 547
 548                             // Consider changing: if we receive the CreationHandle midway into the wait
 549                             // period and re-wait, we will be waiting on the full period
 550                             if (Count >= MaxPoolSize && 0 != MaxPoolSize) {
 551                                 if (!ReclaimEmancipatedObjects()) {
 552                                     // modify handle array not to wait on creation mutex anymore
 553                                     waitHandles    = new WaitHandle[2];
 554                                     waitHandles[0] = m_WaitHandles[0];
 555                                     waitHandles[1] = m_WaitHandles[1];
 556                                 }
 557                             }
 558
 559                         }
 560                     }
 561                     finally {
 562                         CreationMutex.ReleaseMutex();
 563                     }
 564                     break;
 565
 566                 default:
 567                     // the semaphore was signaled which can only happen
 568                     // when a connection has been placed in the pool
 569                     // so there is guaranteed available inventory
 570                     Interlocked.Decrement(ref m_WaitCount);
 571                     pooledStream = GetFromPool(owningObject);
 572                     continueLoop = false;
 573                     break;
 574                 }
 575                 GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::Get",ValidationHelper.HashString(pooledStream));
 576                 return pooledStream;
 577         }
 578
 579         /// <devdoc>
 580         ///    <para>Aborts the queued requests to the pool</para>
 581         /// </devdoc>
 582         internal void Abort() {
 583             if (m_ResError == null) {
 584                 m_ResError = new WebException(
 585                         NetRes.GetWebStatusString("net_requestaborted", WebExceptionStatus.RequestCanceled),
 586                         WebExceptionStatus.RequestCanceled);
 587             }
 588             ErrorEvent.Set();
 589             m_ErrorOccured = true;
 590             m_ErrorTimer = s_CancelErrorQueue.CreateTimer(s_CancelErrorCallback, this);
 591         }
 592
 593         /// <devdoc>
 594         ///    <para>Attempts to create a PooledStream, by trying to get a pooled Connection,
 595         ///         or by creating its own new one</para>
 596         /// </devdoc>
 597         internal PooledStream GetConnection(object owningObject,
 598                                             GeneralAsyncDelegate asyncCallback,
 599                                             int creationTimeout) {
 600             int result;
 601             PooledStream stream = null;
 602             bool continueLoop = true;
 603             bool async = (asyncCallback != null) ? true : false;
 604
 605             GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::GetConnection");
 606
 607             if(m_State != State.Running) {
 608                 throw new InternalException();
 609             }
 610
 611             Interlocked.Increment(ref m_WaitCount);
 612             WaitHandle[] localWaitHandles = m_WaitHandles;
 613
 614             if (async) {
 615                 result = WaitHandle.WaitAny(localWaitHandles, 0, false);
 616                 if (result != WaitTimeout) {
 617                     stream = Get(owningObject, result, ref continueLoop, ref localWaitHandles);
 618                 }
 619                 if (stream == null) {
 620                     GlobalLog.Print("GetConnection:"+ValidationHelper.HashString(this)+" going async");
 621                     AsyncConnectionPoolRequest asyncState = new AsyncConnectionPoolRequest(this, owningObject, asyncCallback, creationTimeout);
 622                     QueueRequest(asyncState);
 623                 }
 624             } else {
 625                 // loop while we don't have an error/timeout and we haven't gotten a stream yet
 626                 while ((stream == null) && continueLoop) {
 627                     result = WaitHandle.WaitAny(localWaitHandles, creationTimeout, false);
 628                     stream = Get(owningObject, result, ref continueLoop, ref localWaitHandles);
 629                 }
 630             }
 631
 632             if (null != stream) {
 633                 // if there is already a stream, then we're not going async
 634                 if (!stream.IsInitalizing) {
 635                     asyncCallback = null;
 636                 }
 637
 638                 try{
 639                     // If activate returns false, it is going to finish asynchronously
 640                     // and therefore the stream will be returned in a callback and
 641                     // we should not return it here (return null)
 642                     if (stream.Activate(owningObject, asyncCallback) == false)
 643                         stream = null;
 644                 }
 645                 catch{
 646                     PutConnection(stream,owningObject,creationTimeout, false);
 647                     throw;
 648                 }
 649             } else if (!async) {
 650                 throw new InternalException();
 651             }
 652
 653             GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::GetConnection", ValidationHelper.HashString(stream));
 654             return(stream);
 655         }
 656
 657         /// <devdoc>
 658         ///     <para>
 659         ///     Attempts to return a PooledStream to the pool.  Default is that it can be reused if it can
 660         ///     also be pooled.
 661         ///     </para>
 662         /// </devdoc>
 663         internal void PutConnection(PooledStream pooledStream, object owningObject, int creationTimeout)
 664         {
 665             // ok to reuse
 666             PutConnection(pooledStream, owningObject, creationTimeout, true);
 667         }
 668
 669         /// <devdoc>
 670         ///    <para>
 671         ///    Attempts to return a PooledStream to the pool.  If canReuse is false, then the
 672         ///    connection will be destroyed even if it is marked as reusable and a new conneciton will
 673         ///    be created.  If it is true, then the connection will still be checked to ensure that
 674         ///    it can be pooled and will be cleaned up if it can not for another reason.
 675         ///    </para>
 676         /// </devdoc>
 677         internal void PutConnection(PooledStream pooledStream, object owningObject, int creationTimeout, bool canReuse) {
 678             GlobalLog.Print("ConnectionPool#" + ValidationHelper.HashString(this) + "::PutConnection");
 679             if (pooledStream == null) {
 680                 throw new ArgumentNullException("pooledStream");
 681             }
 682
 683             pooledStream.PrePush(owningObject);
 684
 685             if (m_State != State.ShuttingDown) {
 686                 pooledStream.Deactivate();
 687
 688                 // cancel our error status, if we have no new requests waiting anymore
 689                 if (m_WaitCount == 0) {
 690                     CancelErrorCallback();
 691                 }
 692
 693                 if (canReuse && pooledStream.CanBePooled) {
 694                     PutNew(pooledStream);
 695                 }
 696                 else {
 697                     try {
 698                         Destroy(pooledStream);
 699                     } finally { // Make sure to release the mutex even under error conditions.
 700                         // Make sure we recreate a new pooled stream, if there are requests for a stream
 701                         // at this point
 702                         if (m_WaitCount > 0) {
 703                             if (!CreationMutex.WaitOne(creationTimeout, false)) {
 704                                 Abort();
 705                             } else {
 706                                 try {
 707                                     pooledStream = UserCreateRequest();
 708                                     if (null != pooledStream) {
 709                                         PutNew(pooledStream);
 710                                     }
 711                                 } finally {
 712                                     CreationMutex.ReleaseMutex();
 713                                 }
 714                             }
 715                         }
 716                     }
 717                 }
 718             }
 719             else {
 720                 // If we're shutting down, we destroy the object.
 721                 Destroy(pooledStream);
 722             }
 723         }
 724
 725
 726         /// <devdoc>
 727         ///    <para>Places a new/reusable stream in the new stack of the pool</para>
 728         /// </devdoc>
 729         private void PutNew(PooledStream pooledStream) {
 730             GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::PutNew", "#"+ValidationHelper.HashString(pooledStream));
 731
 732             GlobalLog.Assert(null != pooledStream, "Why are we adding a null object to the pool?");
 733             GlobalLog.Assert(pooledStream.CanBePooled, "Non-poolable object in pool.");
 734
 735             m_StackNew.Push(pooledStream);
 736             // ensure that the semaphore's count is incremented to signal an available connection is in
 737             // the pool
 738             Semaphore.ReleaseSemaphore();
 739             GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::PutNew");
 740         }
 741
 742
 743         /// <devdoc>
 744         ///    <para>Reclaim any pooled Streams that have seen their users/WebRequests GCed away</para>
 745         /// </devdoc>
 746         private bool ReclaimEmancipatedObjects() {
 747             bool emancipatedObjectFound = false;
 748             GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::ReclaimEmancipatedObjects");
 749
 750             lock(m_ObjectList.SyncRoot) {
 751
 752                 object[] objectList = m_ObjectList.ToArray();
 753                 if (null != objectList) {
 754
 755                     for (int i = 0; i < objectList.Length; ++i) {
 756                         PooledStream pooledStream = (PooledStream) objectList[i];
 757
 758                         if (null != pooledStream) {
 759                             bool locked = false;
 760
 761                             try {
 762                                 Monitor.TryEnter(pooledStream, ref locked);
 763
 764                                 if (locked) {
 765                                     if (pooledStream.IsEmancipated) {
 766
 767                                         GlobalLog.Print("EmancipatedObject pooledStream#"+ValidationHelper.HashString(pooledStream));
 768                                         PutConnection(pooledStream, null, Timeout.Infinite);
 769                                         emancipatedObjectFound = true;
 770                                     }
 771                                 }
 772                             }
 773                             finally {
 774                                 if (locked)
 775                                     Monitor.Exit(pooledStream);
 776                             }
 777                         }
 778                     }
 779                 }
 780             }
 781             GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::ReclaimEmancipatedObjects",emancipatedObjectFound.ToString());
 782             return emancipatedObjectFound;
 783         }
 784
 785         /// <devdoc>
 786         ///    <para>Creates a new PooledStream is allowable</para>
 787         /// </devdoc>
 788         private PooledStream UserCreateRequest() {
 789             // called by user when they were not able to obtain a free object but
 790             // instead obtained creation mutex
 791             GlobalLog.Enter("ConnectionPool#" + ValidationHelper.HashString(this) + "::UserCreateRequest");
 792
 793             PooledStream pooledStream = null;
 794
 795             if (!ErrorOccurred) {
 796                  if (Count < MaxPoolSize || 0 == MaxPoolSize) {
 797                     // If we have an odd number of total objects, reclaim any dead objects.
 798                     // If we did not find any objects to reclaim, create a new one.
 799
 800                     //
 801                      if ((Count & 0x1) == 0x1 || !ReclaimEmancipatedObjects())
 802                         pooledStream = Create(m_CreateConnectionCallback);
 803                 }
 804             }
 805             GlobalLog.Leave("ConnectionPool#" + ValidationHelper.HashString(this) + "::UserCreateRequest", ValidationHelper.HashString(pooledStream));
 806             return pooledStream;
 807         }
 808     }
 809
 810
 811     /// <devdoc>
 812     ///    <para>Used to Pool streams in a thread safe manner</para>
 813     /// </devdoc>
 814     sealed internal class InterlockedStack {
 815         private readonly Stack _stack = new Stack();
 816 #if !MONO
 817         private int _count;
 818 #endif
 819
 820 #if DEBUG
 821         private readonly Hashtable doublepush = new Hashtable();
 822 #endif
 823
 824         internal InterlockedStack() {
 825         }
 826
 827         internal void Push(Object pooledStream) {
 828             GlobalLog.Assert(null != pooledStream, "push null");
 829             if (null == pooledStream) { throw new ArgumentNullException("pooledStream"); }
 830             lock(_stack.SyncRoot) {
 831 #if DEBUG
 832                 GlobalLog.Assert(null == doublepush[pooledStream], "object already in stack");
 833                 doublepush[pooledStream] = _stack.Count;
 834 #endif
 835                 _stack.Push(pooledStream);
 836 #if DEBUG
 837                 GlobalLog.Assert(_count+1 == _stack.Count, "push count mishandle");
 838 #endif
 839 #if !MONO
 840                 _count = _stack.Count;
 841 #endif
 842             }
 843         }
 844
 845         internal Object Pop() {
 846             lock(_stack.SyncRoot) {
 847                 object pooledStream = null;
 848                 if (0 <_stack.Count) {
 849                     pooledStream = _stack.Pop();
 850 #if DEBUG
 851                     GlobalLog.Assert(_count-1 == _stack.Count, "pop count mishandle");
 852                     doublepush.Remove(pooledStream);
 853 #endif
 854 #if !MONO
 855                     _count = _stack.Count;
 856 #endif
 857                 }
 858                 return pooledStream;
 859             }
 860         }
 861     }
 862
 863 }
 864