The Tor Browser / file revision

 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */

 /* This Source Code Form is subject to the terms of the Mozilla Public

  * License, v. 2.0. If a copy of the MPL was not distributed with this

  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

 #include "primpl.h"

 #include "pprmwait.h"

 #define _MW_REHASH_MAX 11

 static PRLock *mw_lock = NULL;

 static _PRGlobalState *mw_state = NULL;

 static PRIntervalTime max_polling_interval;

 #ifdef WINNT

 typedef struct TimerEvent {

     PRIntervalTime absolute;

     void (*func)(void *);

     void *arg;

     LONG ref_count;

     PRCList links;

 } TimerEvent;

 #define TIMER_EVENT_PTR(_qp) \

     ((TimerEvent *) ((char *) (_qp) - offsetof(TimerEvent, links)))

 struct {

     PRLock *ml;

     PRCondVar *new_timer;

     PRCondVar *cancel_timer;

     PRThread *manager_thread;

     PRCList timer_queue;

 } tm_vars;

 static PRStatus TimerInit(void);

 static void TimerManager(void *arg);

 static TimerEvent *CreateTimer(PRIntervalTime timeout,

     void (*func)(void *), void *arg);

 static PRBool CancelTimer(TimerEvent *timer);

 static void TimerManager(void *arg)

 {

     PRIntervalTime now;

     PRIntervalTime timeout;

     PRCList *head;

     TimerEvent *timer;

     PR_Lock(tm_vars.ml);

     while (1)

     {

         if (PR_CLIST_IS_EMPTY(&tm_vars.timer_queue))

         {

             PR_WaitCondVar(tm_vars.new_timer, PR_INTERVAL_NO_TIMEOUT);

         }

         else

         {

             now = PR_IntervalNow();

             head = PR_LIST_HEAD(&tm_vars.timer_queue);

             timer = TIMER_EVENT_PTR(head);

             if ((PRInt32) (now - timer->absolute) >= 0)

             {

                 PR_REMOVE_LINK(head);

                 /*

                  * make its prev and next point to itself so that

                  * it's obvious that it's not on the timer_queue.

                  */

                 PR_INIT_CLIST(head);

                 PR_ASSERT(2 == timer->ref_count);

                 PR_Unlock(tm_vars.ml);

                 timer->func(timer->arg);

                 PR_Lock(tm_vars.ml);

                 timer->ref_count -= 1;

                 if (0 == timer->ref_count)

                 {

                     PR_NotifyAllCondVar(tm_vars.cancel_timer);

                 }

             }

             else

             {

                 timeout = (PRIntervalTime)(timer->absolute - now);

                 PR_WaitCondVar(tm_vars.new_timer, timeout);

             } 

         }

     }

     PR_Unlock(tm_vars.ml);

 }

 static TimerEvent *CreateTimer(

     PRIntervalTime timeout,

     void (*func)(void *),

     void *arg)

 {

     TimerEvent *timer;

     PRCList *links, *tail;

     TimerEvent *elem;

     timer = PR_NEW(TimerEvent);

     if (NULL == timer)

     {

         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

         return timer;

     }

     timer->absolute = PR_IntervalNow() + timeout;

     timer->func = func;

     timer->arg = arg;

     timer->ref_count = 2;

     PR_Lock(tm_vars.ml);

     tail = links = PR_LIST_TAIL(&tm_vars.timer_queue);

     while (links->prev != tail)

     {

         elem = TIMER_EVENT_PTR(links);

         if ((PRInt32)(timer->absolute - elem->absolute) >= 0)

         {

             break;

         }

         links = links->prev;

     }

     PR_INSERT_AFTER(&timer->links, links);

     PR_NotifyCondVar(tm_vars.new_timer);

     PR_Unlock(tm_vars.ml);

     return timer;

 }

 static PRBool CancelTimer(TimerEvent *timer)

 {

     PRBool canceled = PR_FALSE;

     PR_Lock(tm_vars.ml);

     timer->ref_count -= 1;

     if (timer->links.prev == &timer->links)

     {

         while (timer->ref_count == 1)

         {

             PR_WaitCondVar(tm_vars.cancel_timer, PR_INTERVAL_NO_TIMEOUT);

         }

     }

     else

     {

         PR_REMOVE_LINK(&timer->links);

         canceled = PR_TRUE;

     }

     PR_Unlock(tm_vars.ml);

     PR_DELETE(timer);

     return canceled; 

 }

 static PRStatus TimerInit(void)

 {

     tm_vars.ml = PR_NewLock();

     if (NULL == tm_vars.ml)

     {

         goto failed;

     }

     tm_vars.new_timer = PR_NewCondVar(tm_vars.ml);

     if (NULL == tm_vars.new_timer)

     {

         goto failed;

     }

     tm_vars.cancel_timer = PR_NewCondVar(tm_vars.ml);

     if (NULL == tm_vars.cancel_timer)

     {

         goto failed;

     }

     PR_INIT_CLIST(&tm_vars.timer_queue);

     tm_vars.manager_thread = PR_CreateThread(

         PR_SYSTEM_THREAD, TimerManager, NULL, PR_PRIORITY_NORMAL,

         PR_LOCAL_THREAD, PR_UNJOINABLE_THREAD, 0);

     if (NULL == tm_vars.manager_thread)

     {

         goto failed;

     }

     return PR_SUCCESS;

 failed:

     if (NULL != tm_vars.cancel_timer)

     {

         PR_DestroyCondVar(tm_vars.cancel_timer);

     }

     if (NULL != tm_vars.new_timer)

     {

         PR_DestroyCondVar(tm_vars.new_timer);

     }

     if (NULL != tm_vars.ml)

     {

         PR_DestroyLock(tm_vars.ml);

     }

     return PR_FAILURE;

 }

 #endif /* WINNT */

 /******************************************************************/

 /******************************************************************/

 /************************ The private portion *********************/

 /******************************************************************/

 /******************************************************************/

 void _PR_InitMW(void)

 {

 #ifdef WINNT

     /*

      * We use NT 4's InterlockedCompareExchange() to operate

      * on PRMWStatus variables.

      */

     PR_ASSERT(sizeof(LONG) == sizeof(PRMWStatus));

     TimerInit();

 #endif

     mw_lock = PR_NewLock();

     PR_ASSERT(NULL != mw_lock);

     mw_state = PR_NEWZAP(_PRGlobalState);

     PR_ASSERT(NULL != mw_state);

     PR_INIT_CLIST(&mw_state->group_list);

     max_polling_interval = PR_MillisecondsToInterval(MAX_POLLING_INTERVAL);

 }  /* _PR_InitMW */

 void _PR_CleanupMW(void)

 {

     PR_DestroyLock(mw_lock);

     mw_lock = NULL;

     if (mw_state->group) {

         PR_DestroyWaitGroup(mw_state->group);

         /* mw_state->group is set to NULL as a side effect. */

     }

     PR_DELETE(mw_state);

 }  /* _PR_CleanupMW */

 static PRWaitGroup *MW_Init2(void)

 {

     PRWaitGroup *group = mw_state->group;  /* it's the null group */

     if (NULL == group)  /* there is this special case */

     {

         group = PR_CreateWaitGroup(_PR_DEFAULT_HASH_LENGTH);

         if (NULL == group) goto failed_alloc;

         PR_Lock(mw_lock);

         if (NULL == mw_state->group)

         {

             mw_state->group = group;

             group = NULL;

         }

         PR_Unlock(mw_lock);

         if (group != NULL) (void)PR_DestroyWaitGroup(group);

         group = mw_state->group;  /* somebody beat us to it */

     }

 failed_alloc:

     return group;  /* whatever */

 }  /* MW_Init2 */

 static _PR_HashStory MW_AddHashInternal(PRRecvWait *desc, _PRWaiterHash *hash)

 {

     /*

     ** The entries are put in the table using the fd (PRFileDesc*) of

     ** the receive descriptor as the key. This allows us to locate

     ** the appropriate entry aqain when the poll operation finishes.

     **

     ** The pointer to the file descriptor object is first divided by

     ** the natural alignment of a pointer in the belief that object

     ** will have at least that many zeros in the low order bits.

     ** This may not be a good assuption.

     **

     ** We try to put the entry in by rehashing _MW_REHASH_MAX times. After

     ** that we declare defeat and force the table to be reconstructed.

     ** Since some fds might be added more than once, won't that cause

     ** collisions even in an empty table?

     */

     PRIntn rehash = _MW_REHASH_MAX;

     PRRecvWait **waiter;

     PRUintn hidx = _MW_HASH(desc->fd, hash->length);

     PRUintn hoffset = 0;

     while (rehash-- > 0)

     {

         waiter = &hash->recv_wait;

         if (NULL == waiter[hidx])

         {

             waiter[hidx] = desc;

             hash->count += 1;

 #if 0

             printf("Adding 0x%x->0x%x ", desc, desc->fd);

             printf(

                 "table[%u:%u:*%u]: 0x%x->0x%x\n",

                 hidx, hash->count, hash->length, waiter[hidx], waiter[hidx]->fd);

 #endif

             return _prmw_success;

         }

         if (desc == waiter[hidx])

         {

             PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);  /* desc already in table */

             return _prmw_error;

         }

 #if 0

         printf("Failing 0x%x->0x%x ", desc, desc->fd);

         printf(

             "table[*%u:%u:%u]: 0x%x->0x%x\n",

             hidx, hash->count, hash->length, waiter[hidx], waiter[hidx]->fd);

 #endif

         if (0 == hoffset)

         {

             hoffset = _MW_HASH2(desc->fd, hash->length);

             PR_ASSERT(0 != hoffset);

         }

         hidx = (hidx + hoffset) % (hash->length);

     }

     return _prmw_rehash;    

 }  /* MW_AddHashInternal */

 static _PR_HashStory MW_ExpandHashInternal(PRWaitGroup *group)

 {

     PRRecvWait **desc;

     PRUint32 pidx, length;

     _PRWaiterHash *newHash, *oldHash = group->waiter;

     PRBool retry;

     _PR_HashStory hrv;

     static const PRInt32 prime_number[] = {

         _PR_DEFAULT_HASH_LENGTH, 179, 521, 907, 1427,

         2711, 3917, 5021, 8219, 11549, 18911, 26711, 33749, 44771};

     PRUintn primes = (sizeof(prime_number) / sizeof(PRInt32));

     /* look up the next size we'd like to use for the hash table */

     for (pidx = 0; pidx < primes; ++pidx)

     {

         if (prime_number[pidx] == oldHash->length)

         {

             break;

         }

     }

     /* table size must be one of the prime numbers */

     PR_ASSERT(pidx < primes);

     /* if pidx == primes - 1, we can't expand the table any more */

     while (pidx < primes - 1)

     {

         /* next size */

         ++pidx;

         length = prime_number[pidx];

         /* allocate the new hash table and fill it in with the old */

         newHash = (_PRWaiterHash*)PR_CALLOC(

             sizeof(_PRWaiterHash) + (length * sizeof(PRRecvWait*)));

         if (NULL == newHash)

         {

             PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

             return _prmw_error;

         }

         newHash->length = length;

         retry = PR_FALSE;

         for (desc = &oldHash->recv_wait;

             newHash->count < oldHash->count; ++desc)

         {

             PR_ASSERT(desc < &oldHash->recv_wait + oldHash->length);

             if (NULL != *desc)

             {

                 hrv = MW_AddHashInternal(*desc, newHash);

                 PR_ASSERT(_prmw_error != hrv);

                 if (_prmw_success != hrv)

                 {

                     PR_DELETE(newHash);

                     retry = PR_TRUE;

                     break;

                 }

             }

         }

         if (retry) continue;

         PR_DELETE(group->waiter);

         group->waiter = newHash;

         group->p_timestamp += 1;

         return _prmw_success;

     }

     PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

     return _prmw_error;  /* we're hosed */

 }  /* MW_ExpandHashInternal */

 #ifndef WINNT

 static void _MW_DoneInternal(

     PRWaitGroup *group, PRRecvWait **waiter, PRMWStatus outcome)

 {

     /*

     ** Add this receive wait object to the list of finished I/O

     ** operations for this particular group. If there are other

     ** threads waiting on the group, notify one. If not, arrange

     ** for this thread to return.

     */

 #if 0

     printf("Removing 0x%x->0x%x\n", *waiter, (*waiter)->fd);

 #endif

     (*waiter)->outcome = outcome;

     PR_APPEND_LINK(&((*waiter)->internal), &group->io_ready);

     PR_NotifyCondVar(group->io_complete);

     PR_ASSERT(0 != group->waiter->count);

     group->waiter->count -= 1;

     *waiter = NULL;

 }  /* _MW_DoneInternal */

 #endif /* WINNT */

 static PRRecvWait **_MW_LookupInternal(PRWaitGroup *group, PRFileDesc *fd)

 {

     /*

     ** Find the receive wait object corresponding to the file descriptor.

     ** Only search the wait group specified.

     */

     PRRecvWait **desc;

     PRIntn rehash = _MW_REHASH_MAX;

     _PRWaiterHash *hash = group->waiter;

     PRUintn hidx = _MW_HASH(fd, hash->length);

     PRUintn hoffset = 0;

     while (rehash-- > 0)

     {

         desc = (&hash->recv_wait) + hidx;

         if ((*desc != NULL) && ((*desc)->fd == fd)) return desc;

         if (0 == hoffset)

         {

             hoffset = _MW_HASH2(fd, hash->length);

             PR_ASSERT(0 != hoffset);

         }

         hidx = (hidx + hoffset) % (hash->length);

     }

     return NULL;

 }  /* _MW_LookupInternal */

 #ifndef WINNT

 static PRStatus _MW_PollInternal(PRWaitGroup *group)

 {

     PRRecvWait **waiter;

     PRStatus rv = PR_FAILURE;

     PRInt32 count, count_ready;

     PRIntervalTime polling_interval;

     group->poller = PR_GetCurrentThread();

     while (PR_TRUE)

     {

         PRIntervalTime now, since_last_poll;

         PRPollDesc *poll_list;

         while (0 == group->waiter->count)

         {

             PRStatus st;

             st = PR_WaitCondVar(group->new_business, PR_INTERVAL_NO_TIMEOUT);

             if (_prmw_running != group->state)

             {

                 PR_SetError(PR_INVALID_STATE_ERROR, 0);

                 goto aborted;

             }

             if (_MW_ABORTED(st)) goto aborted;

         }

         /*

         ** There's something to do. See if our existing polling list

         ** is large enough for what we have to do?

         */

         while (group->polling_count < group->waiter->count)

         {

             PRUint32 old_count = group->waiter->count;

             PRUint32 new_count = PR_ROUNDUP(old_count, _PR_POLL_COUNT_FUDGE);

             PRSize new_size = sizeof(PRPollDesc) * new_count;

             PRPollDesc *old_polling_list = group->polling_list;

             PR_Unlock(group->ml);

             poll_list = (PRPollDesc*)PR_CALLOC(new_size);

             if (NULL == poll_list)

             {

                 PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

                 PR_Lock(group->ml);

                 goto failed_alloc;

             }

             if (NULL != old_polling_list)

                 PR_DELETE(old_polling_list);

             PR_Lock(group->ml);

             if (_prmw_running != group->state)

             {

                 PR_SetError(PR_INVALID_STATE_ERROR, 0);

                 goto aborted;

             }

             group->polling_list = poll_list;

             group->polling_count = new_count;

         }

         now = PR_IntervalNow();

         polling_interval = max_polling_interval;

         since_last_poll = now - group->last_poll;

         waiter = &group->waiter->recv_wait;

         poll_list = group->polling_list;

         for (count = 0; count < group->waiter->count; ++waiter)

         {

             PR_ASSERT(waiter < &group->waiter->recv_wait

                 + group->waiter->length);

             if (NULL != *waiter)  /* a live one! */

             {

                 if ((PR_INTERVAL_NO_TIMEOUT != (*waiter)->timeout)

                 && (since_last_poll >= (*waiter)->timeout))

                     _MW_DoneInternal(group, waiter, PR_MW_TIMEOUT);

                 else

                 {

                     if (PR_INTERVAL_NO_TIMEOUT != (*waiter)->timeout)

                     {

                         (*waiter)->timeout -= since_last_poll;

                         if ((*waiter)->timeout < polling_interval)

                             polling_interval = (*waiter)->timeout;

                     }

                     PR_ASSERT(poll_list < group->polling_list

                         + group->polling_count);

                     poll_list->fd = (*waiter)->fd;

                     poll_list->in_flags = PR_POLL_READ;

                     poll_list->out_flags = 0;

 #if 0

                     printf(

                         "Polling 0x%x[%d]: [fd: 0x%x, tmo: %u]\n",

                         poll_list, count, poll_list->fd, (*waiter)->timeout);

 #endif

                     poll_list += 1;

                     count += 1;

                 }

             }

         } 

         PR_ASSERT(count == group->waiter->count);

         /*

         ** If there are no more threads waiting for completion,

         ** we need to return.

         */

         if ((!PR_CLIST_IS_EMPTY(&group->io_ready))

         && (1 == group->waiting_threads)) break;

         if (0 == count) continue;  /* wait for new business */

         group->last_poll = now;

         PR_Unlock(group->ml);

         count_ready = PR_Poll(group->polling_list, count, polling_interval);

         PR_Lock(group->ml);

         if (_prmw_running != group->state)

         {

             PR_SetError(PR_INVALID_STATE_ERROR, 0);

             goto aborted;

         }

         if (-1 == count_ready)

         {

             goto failed_poll;  /* that's a shame */

         }

         else if (0 < count_ready)

         {

             for (poll_list = group->polling_list; count > 0;

             poll_list++, count--)

             {

                 PR_ASSERT(

                     poll_list < group->polling_list + group->polling_count);

                 if (poll_list->out_flags != 0)

                 {

                     waiter = _MW_LookupInternal(group, poll_list->fd);

                     /*

                     ** If 'waiter' is NULL, that means the wait receive

                     ** descriptor has been canceled.

                     */

                     if (NULL != waiter)

                         _MW_DoneInternal(group, waiter, PR_MW_SUCCESS);

                 }

             }

         }

         /*

         ** If there are no more threads waiting for completion,

         ** we need to return.

         ** This thread was "borrowed" to do the polling, but it really

         ** belongs to the client.

         */

         if ((!PR_CLIST_IS_EMPTY(&group->io_ready))

         && (1 == group->waiting_threads)) break;

     }

     rv = PR_SUCCESS;

 aborted:

 failed_poll:

 failed_alloc:

     group->poller = NULL;  /* we were that, not we ain't */

     if ((_prmw_running == group->state) && (group->waiting_threads > 1))

     {

         /* Wake up one thread to become the new poller. */

         PR_NotifyCondVar(group->io_complete);

     }

     return rv;  /* we return with the lock held */

 }  /* _MW_PollInternal */

 #endif /* !WINNT */

 static PRMWGroupState MW_TestForShutdownInternal(PRWaitGroup *group)

 {

     PRMWGroupState rv = group->state;

     /*

     ** Looking at the group's fields is safe because

     ** once the group's state is no longer running, it

     ** cannot revert and there is a safe check on entry

     ** to make sure no more threads are made to wait.

     */

     if ((_prmw_stopping == rv)

     && (0 == group->waiting_threads))

     {

         rv = group->state = _prmw_stopped;

         PR_NotifyCondVar(group->mw_manage);

     }

     return rv;

 }  /* MW_TestForShutdownInternal */

 #ifndef WINNT

 static void _MW_InitialRecv(PRCList *io_ready)

 {

     PRRecvWait *desc = (PRRecvWait*)io_ready;

     if ((NULL == desc->buffer.start)

     || (0 == desc->buffer.length))

         desc->bytesRecv = 0;

     else

     {

         desc->bytesRecv = (desc->fd->methods->recv)(

             desc->fd, desc->buffer.start,

             desc->buffer.length, 0, desc->timeout);

         if (desc->bytesRecv < 0)  /* SetError should already be there */

             desc->outcome = PR_MW_FAILURE;

     }

 }  /* _MW_InitialRecv */

 #endif

 #ifdef WINNT

 static void NT_TimeProc(void *arg)

 {

     _MDOverlapped *overlapped = (_MDOverlapped *)arg;

     PRRecvWait *desc =  overlapped->data.mw.desc;

     PRFileDesc *bottom;

     if (InterlockedCompareExchange((LONG *)&desc->outcome,

         (LONG)PR_MW_TIMEOUT, (LONG)PR_MW_PENDING) != (LONG)PR_MW_PENDING)

     {

         /* This wait recv descriptor has already completed. */

         return;

     }

     /* close the osfd to abort the outstanding async io request */

     /* $$$$

     ** Little late to be checking if NSPR's on the bottom of stack,

     ** but if we don't check, we can't assert that the private data

     ** is what we think it is.

     ** $$$$

     */

     bottom = PR_GetIdentitiesLayer(desc->fd, PR_NSPR_IO_LAYER);

     PR_ASSERT(NULL != bottom);

     if (NULL != bottom)  /* now what!?!?! */

     {

         bottom->secret->state = _PR_FILEDESC_CLOSED;

         if (closesocket(bottom->secret->md.osfd) == SOCKET_ERROR)

         {

             fprintf(stderr, "closesocket failed: %d\n", WSAGetLastError());

             PR_ASSERT(!"What shall I do?");

         }

     }

     return;

 }  /* NT_TimeProc */

 static PRStatus NT_HashRemove(PRWaitGroup *group, PRFileDesc *fd)

 {

     PRRecvWait **waiter;

     _PR_MD_LOCK(&group->mdlock);

     waiter = _MW_LookupInternal(group, fd);

     if (NULL != waiter)

     {

         group->waiter->count -= 1;

         *waiter = NULL;

     }

     _PR_MD_UNLOCK(&group->mdlock);

     return (NULL != waiter) ? PR_SUCCESS : PR_FAILURE;

 }

 PRStatus NT_HashRemoveInternal(PRWaitGroup *group, PRFileDesc *fd)

 {

     PRRecvWait **waiter;

     waiter = _MW_LookupInternal(group, fd);

     if (NULL != waiter)

     {

         group->waiter->count -= 1;

         *waiter = NULL;

     }

     return (NULL != waiter) ? PR_SUCCESS : PR_FAILURE;

 }

 #endif /* WINNT */

 /******************************************************************/

 /******************************************************************/

 /********************** The public API portion ********************/

 /******************************************************************/

 /******************************************************************/

 PR_IMPLEMENT(PRStatus) PR_AddWaitFileDesc(

     PRWaitGroup *group, PRRecvWait *desc)

 {

     _PR_HashStory hrv;

     PRStatus rv = PR_FAILURE;

 #ifdef WINNT

     _MDOverlapped *overlapped;

     HANDLE hFile;

     BOOL bResult;

     DWORD dwError;

     PRFileDesc *bottom;

 #endif

     if (!_pr_initialized) _PR_ImplicitInitialization();

     if ((NULL == group) && (NULL == (group = MW_Init2())))

     {

         return rv;

     }

     PR_ASSERT(NULL != desc->fd);

     desc->outcome = PR_MW_PENDING;  /* nice, well known value */

     desc->bytesRecv = 0;  /* likewise, though this value is ambiguious */

     PR_Lock(group->ml);

     if (_prmw_running != group->state)

     {

         /* Not allowed to add after cancelling the group */

         desc->outcome = PR_MW_INTERRUPT;

         PR_SetError(PR_INVALID_STATE_ERROR, 0);

         PR_Unlock(group->ml);

         return rv;

     }

 #ifdef WINNT

     _PR_MD_LOCK(&group->mdlock);

 #endif

     /*

     ** If the waiter count is zero at this point, there's no telling

     ** how long we've been idle. Therefore, initialize the beginning

     ** of the timing interval. As long as the list doesn't go empty,

     ** it will maintain itself.

     */

     if (0 == group->waiter->count)

         group->last_poll = PR_IntervalNow();

     do

     {

         hrv = MW_AddHashInternal(desc, group->waiter);

         if (_prmw_rehash != hrv) break;

         hrv = MW_ExpandHashInternal(group);  /* gruesome */

         if (_prmw_success != hrv) break;

     } while (PR_TRUE);

 #ifdef WINNT

     _PR_MD_UNLOCK(&group->mdlock);

 #endif

     PR_NotifyCondVar(group->new_business);  /* tell the world */

     rv = (_prmw_success == hrv) ? PR_SUCCESS : PR_FAILURE;

     PR_Unlock(group->ml);

 #ifdef WINNT

     overlapped = PR_NEWZAP(_MDOverlapped);

     if (NULL == overlapped)

     {

         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

         NT_HashRemove(group, desc->fd);

         return rv;

     }

     overlapped->ioModel = _MD_MultiWaitIO;

     overlapped->data.mw.desc = desc;

     overlapped->data.mw.group = group;

     if (desc->timeout != PR_INTERVAL_NO_TIMEOUT)

     {

         overlapped->data.mw.timer = CreateTimer(

             desc->timeout,

             NT_TimeProc,

             overlapped);

         if (0 == overlapped->data.mw.timer)

         {

             NT_HashRemove(group, desc->fd);

             PR_DELETE(overlapped);

             /*

              * XXX It appears that a maximum of 16 timer events can

              * be outstanding. GetLastError() returns 0 when I try it.

              */

             PR_SetError(PR_INSUFFICIENT_RESOURCES_ERROR, GetLastError());

             return PR_FAILURE;

         }

     }

     /* Reach to the bottom layer to get the OS fd */

     bottom = PR_GetIdentitiesLayer(desc->fd, PR_NSPR_IO_LAYER);

     PR_ASSERT(NULL != bottom);

     if (NULL == bottom)

     {

         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

         return PR_FAILURE;

     }

     hFile = (HANDLE)bottom->secret->md.osfd; 

     if (!bottom->secret->md.io_model_committed)

     {

         PRInt32 st;

         st = _md_Associate(hFile);

         PR_ASSERT(0 != st);

         bottom->secret->md.io_model_committed = PR_TRUE;

     }

     bResult = ReadFile(hFile,

         desc->buffer.start,

         (DWORD)desc->buffer.length,

         NULL,

         &overlapped->overlapped);

     if (FALSE == bResult && (dwError = GetLastError()) != ERROR_IO_PENDING)

     {

         if (desc->timeout != PR_INTERVAL_NO_TIMEOUT)

         {

             if (InterlockedCompareExchange((LONG *)&desc->outcome,

                 (LONG)PR_MW_FAILURE, (LONG)PR_MW_PENDING)

                 == (LONG)PR_MW_PENDING)

             {

                 CancelTimer(overlapped->data.mw.timer);

             }

             NT_HashRemove(group, desc->fd);

             PR_DELETE(overlapped);

         }

         _PR_MD_MAP_READ_ERROR(dwError);

         rv = PR_FAILURE;

     }

 #endif

     return rv;

 }  /* PR_AddWaitFileDesc */

 PR_IMPLEMENT(PRRecvWait*) PR_WaitRecvReady(PRWaitGroup *group)

 {

     PRCList *io_ready = NULL;

 #ifdef WINNT

     PRThread *me = _PR_MD_CURRENT_THREAD();

     _MDOverlapped *overlapped;    

 #endif

     if (!_pr_initialized) _PR_ImplicitInitialization();

     if ((NULL == group) && (NULL == (group = MW_Init2()))) goto failed_init;

     PR_Lock(group->ml);

     if (_prmw_running != group->state)

     {

         PR_SetError(PR_INVALID_STATE_ERROR, 0);

         goto invalid_state;

     }

     group->waiting_threads += 1;  /* the polling thread is counted */

 #ifdef WINNT

     _PR_MD_LOCK(&group->mdlock);

     while (PR_CLIST_IS_EMPTY(&group->io_ready))

     {

         _PR_THREAD_LOCK(me);

         me->state = _PR_IO_WAIT;

         PR_APPEND_LINK(&me->waitQLinks, &group->wait_list);

         if (!_PR_IS_NATIVE_THREAD(me))

         {

             _PR_SLEEPQ_LOCK(me->cpu);

             _PR_ADD_SLEEPQ(me, PR_INTERVAL_NO_TIMEOUT);

             _PR_SLEEPQ_UNLOCK(me->cpu);

         }

         _PR_THREAD_UNLOCK(me);

         _PR_MD_UNLOCK(&group->mdlock);

         PR_Unlock(group->ml);

         _PR_MD_WAIT(me, PR_INTERVAL_NO_TIMEOUT);

         me->state = _PR_RUNNING;

         PR_Lock(group->ml);

         _PR_MD_LOCK(&group->mdlock);

         if (_PR_PENDING_INTERRUPT(me)) {

             PR_REMOVE_LINK(&me->waitQLinks);

             _PR_MD_UNLOCK(&group->mdlock);

             me->flags &= ~_PR_INTERRUPT;

             me->io_suspended = PR_FALSE;

             PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);

             goto aborted;

         }

     }

     io_ready = PR_LIST_HEAD(&group->io_ready);

     PR_ASSERT(io_ready != NULL);

     PR_REMOVE_LINK(io_ready);

     _PR_MD_UNLOCK(&group->mdlock);

     overlapped = (_MDOverlapped *)

         ((char *)io_ready - offsetof(_MDOverlapped, data));

     io_ready = &overlapped->data.mw.desc->internal;

 #else

     do

     {

         /*

         ** If the I/O ready list isn't empty, have this thread

         ** return with the first receive wait object that's available.

         */

         if (PR_CLIST_IS_EMPTY(&group->io_ready))

         {

             /*

             ** Is there a polling thread yet? If not, grab this thread

             ** and use it.

             */

             if (NULL == group->poller)

             {

                 /*

                 ** This thread will stay do polling until it becomes the only one

                 ** left to service a completion. Then it will return and there will

                 ** be none left to actually poll or to run completions.

                 **

                 ** The polling function should only return w/ failure or

                 ** with some I/O ready.

                 */

                 if (PR_FAILURE == _MW_PollInternal(group)) goto failed_poll;

             }

             else

             {

                 /*

                 ** There are four reasons a thread can be awakened from

                 ** a wait on the io_complete condition variable.

                 ** 1. Some I/O has completed, i.e., the io_ready list

                 **    is nonempty.

                 ** 2. The wait group is canceled.

                 ** 3. The thread is interrupted.

                 ** 4. The current polling thread has to leave and needs

                 **    a replacement.

                 ** The logic to find a new polling thread is made more

                 ** complicated by all the other possible events.

                 ** I tried my best to write the logic clearly, but

                 ** it is still full of if's with continue and goto.

                 */

                 PRStatus st;

                 do 

                 {

                     st = PR_WaitCondVar(group->io_complete, PR_INTERVAL_NO_TIMEOUT);

                     if (_prmw_running != group->state)

                     {

                         PR_SetError(PR_INVALID_STATE_ERROR, 0);

                         goto aborted;

                     }

                     if (_MW_ABORTED(st) || (NULL == group->poller)) break;

                 } while (PR_CLIST_IS_EMPTY(&group->io_ready));

                 /*

                 ** The thread is interrupted and has to leave.  It might

                 ** have also been awakened to process ready i/o or be the

                 ** new poller.  To be safe, if either condition is true,

                 ** we awaken another thread to take its place.

                 */

                 if (_MW_ABORTED(st))

                 {

                     if ((NULL == group->poller

                     || !PR_CLIST_IS_EMPTY(&group->io_ready))

                     && group->waiting_threads > 1)

                         PR_NotifyCondVar(group->io_complete);

                     goto aborted;

                 }

                 /*

                 ** A new poller is needed, but can I be the new poller?

                 ** If there is no i/o ready, sure.  But if there is any

                 ** i/o ready, it has a higher priority.  I want to

                 ** process the ready i/o first and wake up another

                 ** thread to be the new poller.

                 */ 

                 if (NULL == group->poller)

                 {

                     if (PR_CLIST_IS_EMPTY(&group->io_ready))

                         continue;

                     if (group->waiting_threads > 1)

                         PR_NotifyCondVar(group->io_complete);

                 }

             }

             PR_ASSERT(!PR_CLIST_IS_EMPTY(&group->io_ready));

         }

         io_ready = PR_LIST_HEAD(&group->io_ready);

         PR_NotifyCondVar(group->io_taken);

         PR_ASSERT(io_ready != NULL);

         PR_REMOVE_LINK(io_ready);

     } while (NULL == io_ready);

 failed_poll:

 #endif

 aborted:

     group->waiting_threads -= 1;

 invalid_state:

     (void)MW_TestForShutdownInternal(group);

     PR_Unlock(group->ml);

 failed_init:

     if (NULL != io_ready)

     {

         /* If the operation failed, record the reason why */

         switch (((PRRecvWait*)io_ready)->outcome)

         {

             case PR_MW_PENDING:

                 PR_ASSERT(0);

                 break;

             case PR_MW_SUCCESS:

 #ifndef WINNT

                 _MW_InitialRecv(io_ready);

 #endif

                 break;

 #ifdef WINNT

             case PR_MW_FAILURE:

                 _PR_MD_MAP_READ_ERROR(overlapped->data.mw.error);

                 break;

 #endif

             case PR_MW_TIMEOUT:

                 PR_SetError(PR_IO_TIMEOUT_ERROR, 0);

                 break;

             case PR_MW_INTERRUPT:

                 PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);

                 break;

             default: break;

         }

 #ifdef WINNT

         if (NULL != overlapped->data.mw.timer)

         {

             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT

                 != overlapped->data.mw.desc->timeout);

             CancelTimer(overlapped->data.mw.timer);

         }

         else

         {

             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT

                 == overlapped->data.mw.desc->timeout);

         }

         PR_DELETE(overlapped);

 #endif

     }

     return (PRRecvWait*)io_ready;

 }  /* PR_WaitRecvReady */

 PR_IMPLEMENT(PRStatus) PR_CancelWaitFileDesc(PRWaitGroup *group, PRRecvWait *desc)

 {

 #if !defined(WINNT)

     PRRecvWait **recv_wait;

 #endif

     PRStatus rv = PR_SUCCESS;

     if (NULL == group) group = mw_state->group;

     PR_ASSERT(NULL != group);

     if (NULL == group)

     {

         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

         return PR_FAILURE;

     }

     PR_Lock(group->ml);

     if (_prmw_running != group->state)

     {

         PR_SetError(PR_INVALID_STATE_ERROR, 0);

         rv = PR_FAILURE;

         goto unlock;

     }

 #ifdef WINNT

     if (InterlockedCompareExchange((LONG *)&desc->outcome,

         (LONG)PR_MW_INTERRUPT, (LONG)PR_MW_PENDING) == (LONG)PR_MW_PENDING)

     {

         PRFileDesc *bottom = PR_GetIdentitiesLayer(desc->fd, PR_NSPR_IO_LAYER);

         PR_ASSERT(NULL != bottom);

         if (NULL == bottom)

         {

             PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

             goto unlock;

         }

         bottom->secret->state = _PR_FILEDESC_CLOSED;

 #if 0

         fprintf(stderr, "cancel wait recv: closing socket\n");

 #endif

         if (closesocket(bottom->secret->md.osfd) == SOCKET_ERROR)

         {

             fprintf(stderr, "closesocket failed: %d\n", WSAGetLastError());

             exit(1);

         }

     }

 #else

     if (NULL != (recv_wait = _MW_LookupInternal(group, desc->fd)))

     {

         /* it was in the wait table */

         _MW_DoneInternal(group, recv_wait, PR_MW_INTERRUPT);

         goto unlock;

     }

     if (!PR_CLIST_IS_EMPTY(&group->io_ready))

     {

         /* is it already complete? */

         PRCList *head = PR_LIST_HEAD(&group->io_ready);

         do

         {

             PRRecvWait *done = (PRRecvWait*)head;

             if (done == desc) goto unlock;

             head = PR_NEXT_LINK(head);

         } while (head != &group->io_ready);

     }

     PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

     rv = PR_FAILURE;

 #endif

 unlock:

     PR_Unlock(group->ml);

     return rv;

 }  /* PR_CancelWaitFileDesc */

 PR_IMPLEMENT(PRRecvWait*) PR_CancelWaitGroup(PRWaitGroup *group)

 {

     PRRecvWait **desc;

     PRRecvWait *recv_wait = NULL;

 #ifdef WINNT

     _MDOverlapped *overlapped;

     PRRecvWait **end;

     PRThread *me = _PR_MD_CURRENT_THREAD();

 #endif

     if (NULL == group) group = mw_state->group;

     PR_ASSERT(NULL != group);

     if (NULL == group)

     {

         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

         return NULL;

     }

     PR_Lock(group->ml);

     if (_prmw_stopped != group->state)

     {

         if (_prmw_running == group->state)

             group->state = _prmw_stopping;  /* so nothing new comes in */

         if (0 == group->waiting_threads)  /* is there anybody else? */

             group->state = _prmw_stopped;  /* we can stop right now */

         else

         {

             PR_NotifyAllCondVar(group->new_business);

             PR_NotifyAllCondVar(group->io_complete);

         }

         while (_prmw_stopped != group->state)

             (void)PR_WaitCondVar(group->mw_manage, PR_INTERVAL_NO_TIMEOUT);

     }

 #ifdef WINNT

     _PR_MD_LOCK(&group->mdlock);

 #endif

     /* make all the existing descriptors look done/interrupted */

 #ifdef WINNT

     end = &group->waiter->recv_wait + group->waiter->length;

     for (desc = &group->waiter->recv_wait; desc < end; ++desc)

     {

         if (NULL != *desc)

         {

             if (InterlockedCompareExchange((LONG *)&(*desc)->outcome,

                 (LONG)PR_MW_INTERRUPT, (LONG)PR_MW_PENDING)

                 == (LONG)PR_MW_PENDING)

             {

                 PRFileDesc *bottom = PR_GetIdentitiesLayer(

                     (*desc)->fd, PR_NSPR_IO_LAYER);

                 PR_ASSERT(NULL != bottom);

                 if (NULL == bottom)

                 {

                     PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

                     goto invalid_arg;

                 }

                 bottom->secret->state = _PR_FILEDESC_CLOSED;

 #if 0

                 fprintf(stderr, "cancel wait group: closing socket\n");

 #endif

                 if (closesocket(bottom->secret->md.osfd) == SOCKET_ERROR)

                 {

                     fprintf(stderr, "closesocket failed: %d\n",

                         WSAGetLastError());

                     exit(1);

                 }

             }

         }

     }

     while (group->waiter->count > 0)

     {

         _PR_THREAD_LOCK(me);

         me->state = _PR_IO_WAIT;

         PR_APPEND_LINK(&me->waitQLinks, &group->wait_list);

         if (!_PR_IS_NATIVE_THREAD(me))

         {

             _PR_SLEEPQ_LOCK(me->cpu);

             _PR_ADD_SLEEPQ(me, PR_INTERVAL_NO_TIMEOUT);

             _PR_SLEEPQ_UNLOCK(me->cpu);

         }

         _PR_THREAD_UNLOCK(me);

         _PR_MD_UNLOCK(&group->mdlock);

         PR_Unlock(group->ml);

         _PR_MD_WAIT(me, PR_INTERVAL_NO_TIMEOUT);

         me->state = _PR_RUNNING;

         PR_Lock(group->ml);

         _PR_MD_LOCK(&group->mdlock);

     }

 #else

     for (desc = &group->waiter->recv_wait; group->waiter->count > 0; ++desc)

     {

         PR_ASSERT(desc < &group->waiter->recv_wait + group->waiter->length);

         if (NULL != *desc)

             _MW_DoneInternal(group, desc, PR_MW_INTERRUPT);

     }

 #endif

     /* take first element of finished list and return it or NULL */

     if (PR_CLIST_IS_EMPTY(&group->io_ready))

         PR_SetError(PR_GROUP_EMPTY_ERROR, 0);

     else

     {

         PRCList *head = PR_LIST_HEAD(&group->io_ready);

         PR_REMOVE_AND_INIT_LINK(head);

 #ifdef WINNT

         overlapped = (_MDOverlapped *)

             ((char *)head - offsetof(_MDOverlapped, data));

         head = &overlapped->data.mw.desc->internal;

         if (NULL != overlapped->data.mw.timer)

         {

             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT

                 != overlapped->data.mw.desc->timeout);

             CancelTimer(overlapped->data.mw.timer);

         }

         else

         {

             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT

                 == overlapped->data.mw.desc->timeout);

         }

         PR_DELETE(overlapped);

 #endif

         recv_wait = (PRRecvWait*)head;

     }

 #ifdef WINNT

 invalid_arg:

     _PR_MD_UNLOCK(&group->mdlock);

 #endif

     PR_Unlock(group->ml);

     return recv_wait;

 }  /* PR_CancelWaitGroup */

 PR_IMPLEMENT(PRWaitGroup*) PR_CreateWaitGroup(PRInt32 size /* ignored */)

 {

     PRWaitGroup *wg;

     if (NULL == (wg = PR_NEWZAP(PRWaitGroup)))

     {

         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

         goto failed;

     }

     /* the wait group itself */

     wg->ml = PR_NewLock();

     if (NULL == wg->ml) goto failed_lock;

     wg->io_taken = PR_NewCondVar(wg->ml);

     if (NULL == wg->io_taken) goto failed_cvar0;

     wg->io_complete = PR_NewCondVar(wg->ml);

     if (NULL == wg->io_complete) goto failed_cvar1;

     wg->new_business = PR_NewCondVar(wg->ml);

     if (NULL == wg->new_business) goto failed_cvar2;

     wg->mw_manage = PR_NewCondVar(wg->ml);

     if (NULL == wg->mw_manage) goto failed_cvar3;

     PR_INIT_CLIST(&wg->group_link);

     PR_INIT_CLIST(&wg->io_ready);

     /* the waiters sequence */

     wg->waiter = (_PRWaiterHash*)PR_CALLOC(

         sizeof(_PRWaiterHash) +

         (_PR_DEFAULT_HASH_LENGTH * sizeof(PRRecvWait*)));

     if (NULL == wg->waiter)

     {

         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

         goto failed_waiter;

     }

     wg->waiter->count = 0;

     wg->waiter->length = _PR_DEFAULT_HASH_LENGTH;

 #ifdef WINNT

     _PR_MD_NEW_LOCK(&wg->mdlock);

     PR_INIT_CLIST(&wg->wait_list);

 #endif /* WINNT */

     PR_Lock(mw_lock);

     PR_APPEND_LINK(&wg->group_link, &mw_state->group_list);

     PR_Unlock(mw_lock);

     return wg;

 failed_waiter:

     PR_DestroyCondVar(wg->mw_manage);

 failed_cvar3:

     PR_DestroyCondVar(wg->new_business);

 failed_cvar2:

     PR_DestroyCondVar(wg->io_complete);

 failed_cvar1:

     PR_DestroyCondVar(wg->io_taken);

 failed_cvar0:

     PR_DestroyLock(wg->ml);

 failed_lock:

     PR_DELETE(wg);

     wg = NULL;

 failed:

     return wg;

 }  /* MW_CreateWaitGroup */

 PR_IMPLEMENT(PRStatus) PR_DestroyWaitGroup(PRWaitGroup *group)

 {

     PRStatus rv = PR_SUCCESS;

     if (NULL == group) group = mw_state->group;

     PR_ASSERT(NULL != group);

     if (NULL != group)

     {

         PR_Lock(group->ml);

         if ((group->waiting_threads == 0)

         && (group->waiter->count == 0)

         && PR_CLIST_IS_EMPTY(&group->io_ready))

         {

             group->state = _prmw_stopped;

         }

         else

         {

             PR_SetError(PR_INVALID_STATE_ERROR, 0);

             rv = PR_FAILURE;

         }

         PR_Unlock(group->ml);

         if (PR_FAILURE == rv) return rv;

         PR_Lock(mw_lock);

         PR_REMOVE_LINK(&group->group_link);

         PR_Unlock(mw_lock);

 #ifdef WINNT

         /*

          * XXX make sure wait_list is empty and waiter is empty.

          * These must be checked while holding mdlock.

          */

         _PR_MD_FREE_LOCK(&group->mdlock);

 #endif

         PR_DELETE(group->waiter);

         PR_DELETE(group->polling_list);

         PR_DestroyCondVar(group->mw_manage);

         PR_DestroyCondVar(group->new_business);

         PR_DestroyCondVar(group->io_complete);

         PR_DestroyCondVar(group->io_taken);

         PR_DestroyLock(group->ml);

         if (group == mw_state->group) mw_state->group = NULL;

         PR_DELETE(group);

     }

     else

     {

         /* The default wait group is not created yet. */

         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

         rv = PR_FAILURE;

     }

     return rv;

 }  /* PR_DestroyWaitGroup */

 /**********************************************************************

 ***********************************************************************

 ******************** Wait group enumerations **************************

 ***********************************************************************

 **********************************************************************/

 PR_IMPLEMENT(PRMWaitEnumerator*) PR_CreateMWaitEnumerator(PRWaitGroup *group)

 {

     PRMWaitEnumerator *enumerator = PR_NEWZAP(PRMWaitEnumerator);

     if (NULL == enumerator) PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);

     else

     {

         enumerator->group = group;

         enumerator->seal = _PR_ENUM_SEALED;

     }

     return enumerator;

 }  /* PR_CreateMWaitEnumerator */

 PR_IMPLEMENT(PRStatus) PR_DestroyMWaitEnumerator(PRMWaitEnumerator* enumerator)

 {

     PR_ASSERT(NULL != enumerator);

     PR_ASSERT(_PR_ENUM_SEALED == enumerator->seal);

     if ((NULL == enumerator) || (_PR_ENUM_SEALED != enumerator->seal))

     {

         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

         return PR_FAILURE;

     }

     enumerator->seal = _PR_ENUM_UNSEALED;

     PR_Free(enumerator);

     return PR_SUCCESS;

 }  /* PR_DestroyMWaitEnumerator */

 PR_IMPLEMENT(PRRecvWait*) PR_EnumerateWaitGroup(

     PRMWaitEnumerator *enumerator, const PRRecvWait *previous)

 {

     PRRecvWait *result = NULL;

     /* entry point sanity checking */

     PR_ASSERT(NULL != enumerator);

     PR_ASSERT(_PR_ENUM_SEALED == enumerator->seal);

     if ((NULL == enumerator)

     || (_PR_ENUM_SEALED != enumerator->seal)) goto bad_argument;

     /* beginning of enumeration */

     if (NULL == previous)

     {

         if (NULL == enumerator->group)

         {

             enumerator->group = mw_state->group;

             if (NULL == enumerator->group)

             {

                 PR_SetError(PR_GROUP_EMPTY_ERROR, 0);

                 return NULL;

             }

         }

         enumerator->waiter = &enumerator->group->waiter->recv_wait;

         enumerator->p_timestamp = enumerator->group->p_timestamp;

         enumerator->thread = PR_GetCurrentThread();

         enumerator->index = 0;

     }

     /* continuing an enumeration */

     else

     {

         PRThread *me = PR_GetCurrentThread();

         PR_ASSERT(me == enumerator->thread);

         if (me != enumerator->thread) goto bad_argument;

         /* need to restart the enumeration */

         if (enumerator->p_timestamp != enumerator->group->p_timestamp)

             return PR_EnumerateWaitGroup(enumerator, NULL);

     }

     /* actually progress the enumeration */

 #if defined(WINNT)

     _PR_MD_LOCK(&enumerator->group->mdlock);

 #else

     PR_Lock(enumerator->group->ml);

 #endif

     while (enumerator->index++ < enumerator->group->waiter->length)

     {

         if (NULL != (result = *(enumerator->waiter)++)) break;

     }

 #if defined(WINNT)

     _PR_MD_UNLOCK(&enumerator->group->mdlock);

 #else

     PR_Unlock(enumerator->group->ml);

 #endif

     return result;  /* what we live for */

 bad_argument:

     PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);

     return NULL;  /* probably ambiguous */

 }  /* PR_EnumerateWaitGroup */

 /* prmwait.c */