andre@0: /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
andre@0: /* This Source Code Form is subject to the terms of the Mozilla Public
andre@0:  * License, v. 2.0. If a copy of the MPL was not distributed with this
andre@0:  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
andre@0: 
andre@0: #include "primpl.h"
andre@0: #include "pprmwait.h"
andre@0: 
andre@0: #define _MW_REHASH_MAX 11
andre@0: 
andre@0: static PRLock *mw_lock = NULL;
andre@0: static _PRGlobalState *mw_state = NULL;
andre@0: 
andre@0: static PRIntervalTime max_polling_interval;
andre@0: 
andre@0: #ifdef WINNT
andre@0: 
andre@0: typedef struct TimerEvent {
andre@0:     PRIntervalTime absolute;
andre@0:     void (*func)(void *);
andre@0:     void *arg;
andre@0:     LONG ref_count;
andre@0:     PRCList links;
andre@0: } TimerEvent;
andre@0: 
andre@0: #define TIMER_EVENT_PTR(_qp) \
andre@0:     ((TimerEvent *) ((char *) (_qp) - offsetof(TimerEvent, links)))
andre@0: 
andre@0: struct {
andre@0:     PRLock *ml;
andre@0:     PRCondVar *new_timer;
andre@0:     PRCondVar *cancel_timer;
andre@0:     PRThread *manager_thread;
andre@0:     PRCList timer_queue;
andre@0: } tm_vars;
andre@0: 
andre@0: static PRStatus TimerInit(void);
andre@0: static void TimerManager(void *arg);
andre@0: static TimerEvent *CreateTimer(PRIntervalTime timeout,
andre@0:     void (*func)(void *), void *arg);
andre@0: static PRBool CancelTimer(TimerEvent *timer);
andre@0: 
andre@0: static void TimerManager(void *arg)
andre@0: {
andre@0:     PRIntervalTime now;
andre@0:     PRIntervalTime timeout;
andre@0:     PRCList *head;
andre@0:     TimerEvent *timer;
andre@0: 
andre@0:     PR_Lock(tm_vars.ml);
andre@0:     while (1)
andre@0:     {
andre@0:         if (PR_CLIST_IS_EMPTY(&tm_vars.timer_queue))
andre@0:         {
andre@0:             PR_WaitCondVar(tm_vars.new_timer, PR_INTERVAL_NO_TIMEOUT);
andre@0:         }
andre@0:         else
andre@0:         {
andre@0:             now = PR_IntervalNow();
andre@0:             head = PR_LIST_HEAD(&tm_vars.timer_queue);
andre@0:             timer = TIMER_EVENT_PTR(head);
andre@0:             if ((PRInt32) (now - timer->absolute) >= 0)
andre@0:             {
andre@0:                 PR_REMOVE_LINK(head);
andre@0:                 /*
andre@0:                  * make its prev and next point to itself so that
andre@0:                  * it's obvious that it's not on the timer_queue.
andre@0:                  */
andre@0:                 PR_INIT_CLIST(head);
andre@0:                 PR_ASSERT(2 == timer->ref_count);
andre@0:                 PR_Unlock(tm_vars.ml);
andre@0:                 timer->func(timer->arg);
andre@0:                 PR_Lock(tm_vars.ml);
andre@0:                 timer->ref_count -= 1;
andre@0:                 if (0 == timer->ref_count)
andre@0:                 {
andre@0:                     PR_NotifyAllCondVar(tm_vars.cancel_timer);
andre@0:                 }
andre@0:             }
andre@0:             else
andre@0:             {
andre@0:                 timeout = (PRIntervalTime)(timer->absolute - now);
andre@0:                 PR_WaitCondVar(tm_vars.new_timer, timeout);
andre@0:             } 
andre@0:         }
andre@0:     }
andre@0:     PR_Unlock(tm_vars.ml);
andre@0: }
andre@0: 
andre@0: static TimerEvent *CreateTimer(
andre@0:     PRIntervalTime timeout,
andre@0:     void (*func)(void *),
andre@0:     void *arg)
andre@0: {
andre@0:     TimerEvent *timer;
andre@0:     PRCList *links, *tail;
andre@0:     TimerEvent *elem;
andre@0: 
andre@0:     timer = PR_NEW(TimerEvent);
andre@0:     if (NULL == timer)
andre@0:     {
andre@0:         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:         return timer;
andre@0:     }
andre@0:     timer->absolute = PR_IntervalNow() + timeout;
andre@0:     timer->func = func;
andre@0:     timer->arg = arg;
andre@0:     timer->ref_count = 2;
andre@0:     PR_Lock(tm_vars.ml);
andre@0:     tail = links = PR_LIST_TAIL(&tm_vars.timer_queue);
andre@0:     while (links->prev != tail)
andre@0:     {
andre@0:         elem = TIMER_EVENT_PTR(links);
andre@0:         if ((PRInt32)(timer->absolute - elem->absolute) >= 0)
andre@0:         {
andre@0:             break;
andre@0:         }
andre@0:         links = links->prev;
andre@0:     }
andre@0:     PR_INSERT_AFTER(&timer->links, links);
andre@0:     PR_NotifyCondVar(tm_vars.new_timer);
andre@0:     PR_Unlock(tm_vars.ml);
andre@0:     return timer;
andre@0: }
andre@0: 
andre@0: static PRBool CancelTimer(TimerEvent *timer)
andre@0: {
andre@0:     PRBool canceled = PR_FALSE;
andre@0: 
andre@0:     PR_Lock(tm_vars.ml);
andre@0:     timer->ref_count -= 1;
andre@0:     if (timer->links.prev == &timer->links)
andre@0:     {
andre@0:         while (timer->ref_count == 1)
andre@0:         {
andre@0:             PR_WaitCondVar(tm_vars.cancel_timer, PR_INTERVAL_NO_TIMEOUT);
andre@0:         }
andre@0:     }
andre@0:     else
andre@0:     {
andre@0:         PR_REMOVE_LINK(&timer->links);
andre@0:         canceled = PR_TRUE;
andre@0:     }
andre@0:     PR_Unlock(tm_vars.ml);
andre@0:     PR_DELETE(timer);
andre@0:     return canceled; 
andre@0: }
andre@0: 
andre@0: static PRStatus TimerInit(void)
andre@0: {
andre@0:     tm_vars.ml = PR_NewLock();
andre@0:     if (NULL == tm_vars.ml)
andre@0:     {
andre@0:         goto failed;
andre@0:     }
andre@0:     tm_vars.new_timer = PR_NewCondVar(tm_vars.ml);
andre@0:     if (NULL == tm_vars.new_timer)
andre@0:     {
andre@0:         goto failed;
andre@0:     }
andre@0:     tm_vars.cancel_timer = PR_NewCondVar(tm_vars.ml);
andre@0:     if (NULL == tm_vars.cancel_timer)
andre@0:     {
andre@0:         goto failed;
andre@0:     }
andre@0:     PR_INIT_CLIST(&tm_vars.timer_queue);
andre@0:     tm_vars.manager_thread = PR_CreateThread(
andre@0:         PR_SYSTEM_THREAD, TimerManager, NULL, PR_PRIORITY_NORMAL,
andre@0:         PR_LOCAL_THREAD, PR_UNJOINABLE_THREAD, 0);
andre@0:     if (NULL == tm_vars.manager_thread)
andre@0:     {
andre@0:         goto failed;
andre@0:     }
andre@0:     return PR_SUCCESS;
andre@0: 
andre@0: failed:
andre@0:     if (NULL != tm_vars.cancel_timer)
andre@0:     {
andre@0:         PR_DestroyCondVar(tm_vars.cancel_timer);
andre@0:     }
andre@0:     if (NULL != tm_vars.new_timer)
andre@0:     {
andre@0:         PR_DestroyCondVar(tm_vars.new_timer);
andre@0:     }
andre@0:     if (NULL != tm_vars.ml)
andre@0:     {
andre@0:         PR_DestroyLock(tm_vars.ml);
andre@0:     }
andre@0:     return PR_FAILURE;
andre@0: }
andre@0: 
andre@0: #endif /* WINNT */
andre@0: 
andre@0: /******************************************************************/
andre@0: /******************************************************************/
andre@0: /************************ The private portion *********************/
andre@0: /******************************************************************/
andre@0: /******************************************************************/
andre@0: void _PR_InitMW(void)
andre@0: {
andre@0: #ifdef WINNT
andre@0:     /*
andre@0:      * We use NT 4's InterlockedCompareExchange() to operate
andre@0:      * on PRMWStatus variables.
andre@0:      */
andre@0:     PR_ASSERT(sizeof(LONG) == sizeof(PRMWStatus));
andre@0:     TimerInit();
andre@0: #endif
andre@0:     mw_lock = PR_NewLock();
andre@0:     PR_ASSERT(NULL != mw_lock);
andre@0:     mw_state = PR_NEWZAP(_PRGlobalState);
andre@0:     PR_ASSERT(NULL != mw_state);
andre@0:     PR_INIT_CLIST(&mw_state->group_list);
andre@0:     max_polling_interval = PR_MillisecondsToInterval(MAX_POLLING_INTERVAL);
andre@0: }  /* _PR_InitMW */
andre@0: 
andre@0: void _PR_CleanupMW(void)
andre@0: {
andre@0:     PR_DestroyLock(mw_lock);
andre@0:     mw_lock = NULL;
andre@0:     if (mw_state->group) {
andre@0:         PR_DestroyWaitGroup(mw_state->group);
andre@0:         /* mw_state->group is set to NULL as a side effect. */
andre@0:     }
andre@0:     PR_DELETE(mw_state);
andre@0: }  /* _PR_CleanupMW */
andre@0: 
andre@0: static PRWaitGroup *MW_Init2(void)
andre@0: {
andre@0:     PRWaitGroup *group = mw_state->group;  /* it's the null group */
andre@0:     if (NULL == group)  /* there is this special case */
andre@0:     {
andre@0:         group = PR_CreateWaitGroup(_PR_DEFAULT_HASH_LENGTH);
andre@0:         if (NULL == group) goto failed_alloc;
andre@0:         PR_Lock(mw_lock);
andre@0:         if (NULL == mw_state->group)
andre@0:         {
andre@0:             mw_state->group = group;
andre@0:             group = NULL;
andre@0:         }
andre@0:         PR_Unlock(mw_lock);
andre@0:         if (group != NULL) (void)PR_DestroyWaitGroup(group);
andre@0:         group = mw_state->group;  /* somebody beat us to it */
andre@0:     }
andre@0: failed_alloc:
andre@0:     return group;  /* whatever */
andre@0: }  /* MW_Init2 */
andre@0: 
andre@0: static _PR_HashStory MW_AddHashInternal(PRRecvWait *desc, _PRWaiterHash *hash)
andre@0: {
andre@0:     /*
andre@0:     ** The entries are put in the table using the fd (PRFileDesc*) of
andre@0:     ** the receive descriptor as the key. This allows us to locate
andre@0:     ** the appropriate entry aqain when the poll operation finishes.
andre@0:     **
andre@0:     ** The pointer to the file descriptor object is first divided by
andre@0:     ** the natural alignment of a pointer in the belief that object
andre@0:     ** will have at least that many zeros in the low order bits.
andre@0:     ** This may not be a good assuption.
andre@0:     **
andre@0:     ** We try to put the entry in by rehashing _MW_REHASH_MAX times. After
andre@0:     ** that we declare defeat and force the table to be reconstructed.
andre@0:     ** Since some fds might be added more than once, won't that cause
andre@0:     ** collisions even in an empty table?
andre@0:     */
andre@0:     PRIntn rehash = _MW_REHASH_MAX;
andre@0:     PRRecvWait **waiter;
andre@0:     PRUintn hidx = _MW_HASH(desc->fd, hash->length);
andre@0:     PRUintn hoffset = 0;
andre@0: 
andre@0:     while (rehash-- > 0)
andre@0:     {
andre@0:         waiter = &hash->recv_wait;
andre@0:         if (NULL == waiter[hidx])
andre@0:         {
andre@0:             waiter[hidx] = desc;
andre@0:             hash->count += 1;
andre@0: #if 0
andre@0:             printf("Adding 0x%x->0x%x ", desc, desc->fd);
andre@0:             printf(
andre@0:                 "table[%u:%u:*%u]: 0x%x->0x%x\n",
andre@0:                 hidx, hash->count, hash->length, waiter[hidx], waiter[hidx]->fd);
andre@0: #endif
andre@0:             return _prmw_success;
andre@0:         }
andre@0:         if (desc == waiter[hidx])
andre@0:         {
andre@0:             PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);  /* desc already in table */
andre@0:             return _prmw_error;
andre@0:         }
andre@0: #if 0
andre@0:         printf("Failing 0x%x->0x%x ", desc, desc->fd);
andre@0:         printf(
andre@0:             "table[*%u:%u:%u]: 0x%x->0x%x\n",
andre@0:             hidx, hash->count, hash->length, waiter[hidx], waiter[hidx]->fd);
andre@0: #endif
andre@0:         if (0 == hoffset)
andre@0:         {
andre@0:             hoffset = _MW_HASH2(desc->fd, hash->length);
andre@0:             PR_ASSERT(0 != hoffset);
andre@0:         }
andre@0:         hidx = (hidx + hoffset) % (hash->length);
andre@0:     }
andre@0:     return _prmw_rehash;    
andre@0: }  /* MW_AddHashInternal */
andre@0: 
andre@0: static _PR_HashStory MW_ExpandHashInternal(PRWaitGroup *group)
andre@0: {
andre@0:     PRRecvWait **desc;
andre@0:     PRUint32 pidx, length;
andre@0:     _PRWaiterHash *newHash, *oldHash = group->waiter;
andre@0:     PRBool retry;
andre@0:     _PR_HashStory hrv;
andre@0: 
andre@0:     static const PRInt32 prime_number[] = {
andre@0:         _PR_DEFAULT_HASH_LENGTH, 179, 521, 907, 1427,
andre@0:         2711, 3917, 5021, 8219, 11549, 18911, 26711, 33749, 44771};
andre@0:     PRUintn primes = (sizeof(prime_number) / sizeof(PRInt32));
andre@0: 
andre@0:     /* look up the next size we'd like to use for the hash table */
andre@0:     for (pidx = 0; pidx < primes; ++pidx)
andre@0:     {
andre@0:         if (prime_number[pidx] == oldHash->length)
andre@0:         {
andre@0:             break;
andre@0:         }
andre@0:     }
andre@0:     /* table size must be one of the prime numbers */
andre@0:     PR_ASSERT(pidx < primes);
andre@0: 
andre@0:     /* if pidx == primes - 1, we can't expand the table any more */
andre@0:     while (pidx < primes - 1)
andre@0:     {
andre@0:         /* next size */
andre@0:         ++pidx;
andre@0:         length = prime_number[pidx];
andre@0: 
andre@0:         /* allocate the new hash table and fill it in with the old */
andre@0:         newHash = (_PRWaiterHash*)PR_CALLOC(
andre@0:             sizeof(_PRWaiterHash) + (length * sizeof(PRRecvWait*)));
andre@0:         if (NULL == newHash)
andre@0:         {
andre@0:             PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:             return _prmw_error;
andre@0:         }
andre@0: 
andre@0:         newHash->length = length;
andre@0:         retry = PR_FALSE;
andre@0:         for (desc = &oldHash->recv_wait;
andre@0:             newHash->count < oldHash->count; ++desc)
andre@0:         {
andre@0:             PR_ASSERT(desc < &oldHash->recv_wait + oldHash->length);
andre@0:             if (NULL != *desc)
andre@0:             {
andre@0:                 hrv = MW_AddHashInternal(*desc, newHash);
andre@0:                 PR_ASSERT(_prmw_error != hrv);
andre@0:                 if (_prmw_success != hrv)
andre@0:                 {
andre@0:                     PR_DELETE(newHash);
andre@0:                     retry = PR_TRUE;
andre@0:                     break;
andre@0:                 }
andre@0:             }
andre@0:         }
andre@0:         if (retry) continue;
andre@0: 
andre@0:         PR_DELETE(group->waiter);
andre@0:         group->waiter = newHash;
andre@0:         group->p_timestamp += 1;
andre@0:         return _prmw_success;
andre@0:     }
andre@0: 
andre@0:     PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:     return _prmw_error;  /* we're hosed */
andre@0: }  /* MW_ExpandHashInternal */
andre@0: 
andre@0: #ifndef WINNT
andre@0: static void _MW_DoneInternal(
andre@0:     PRWaitGroup *group, PRRecvWait **waiter, PRMWStatus outcome)
andre@0: {
andre@0:     /*
andre@0:     ** Add this receive wait object to the list of finished I/O
andre@0:     ** operations for this particular group. If there are other
andre@0:     ** threads waiting on the group, notify one. If not, arrange
andre@0:     ** for this thread to return.
andre@0:     */
andre@0: 
andre@0: #if 0
andre@0:     printf("Removing 0x%x->0x%x\n", *waiter, (*waiter)->fd);
andre@0: #endif
andre@0:     (*waiter)->outcome = outcome;
andre@0:     PR_APPEND_LINK(&((*waiter)->internal), &group->io_ready);
andre@0:     PR_NotifyCondVar(group->io_complete);
andre@0:     PR_ASSERT(0 != group->waiter->count);
andre@0:     group->waiter->count -= 1;
andre@0:     *waiter = NULL;
andre@0: }  /* _MW_DoneInternal */
andre@0: #endif /* WINNT */
andre@0: 
andre@0: static PRRecvWait **_MW_LookupInternal(PRWaitGroup *group, PRFileDesc *fd)
andre@0: {
andre@0:     /*
andre@0:     ** Find the receive wait object corresponding to the file descriptor.
andre@0:     ** Only search the wait group specified.
andre@0:     */
andre@0:     PRRecvWait **desc;
andre@0:     PRIntn rehash = _MW_REHASH_MAX;
andre@0:     _PRWaiterHash *hash = group->waiter;
andre@0:     PRUintn hidx = _MW_HASH(fd, hash->length);
andre@0:     PRUintn hoffset = 0;
andre@0:     
andre@0:     while (rehash-- > 0)
andre@0:     {
andre@0:         desc = (&hash->recv_wait) + hidx;
andre@0:         if ((*desc != NULL) && ((*desc)->fd == fd)) return desc;
andre@0:         if (0 == hoffset)
andre@0:         {
andre@0:             hoffset = _MW_HASH2(fd, hash->length);
andre@0:             PR_ASSERT(0 != hoffset);
andre@0:         }
andre@0:         hidx = (hidx + hoffset) % (hash->length);
andre@0:     }
andre@0:     return NULL;
andre@0: }  /* _MW_LookupInternal */
andre@0: 
andre@0: #ifndef WINNT
andre@0: static PRStatus _MW_PollInternal(PRWaitGroup *group)
andre@0: {
andre@0:     PRRecvWait **waiter;
andre@0:     PRStatus rv = PR_FAILURE;
andre@0:     PRInt32 count, count_ready;
andre@0:     PRIntervalTime polling_interval;
andre@0: 
andre@0:     group->poller = PR_GetCurrentThread();
andre@0: 
andre@0:     while (PR_TRUE)
andre@0:     {
andre@0:         PRIntervalTime now, since_last_poll;
andre@0:         PRPollDesc *poll_list;
andre@0: 
andre@0:         while (0 == group->waiter->count)
andre@0:         {
andre@0:             PRStatus st;
andre@0:             st = PR_WaitCondVar(group->new_business, PR_INTERVAL_NO_TIMEOUT);
andre@0:             if (_prmw_running != group->state)
andre@0:             {
andre@0:                 PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:                 goto aborted;
andre@0:             }
andre@0:             if (_MW_ABORTED(st)) goto aborted;
andre@0:         }
andre@0: 
andre@0:         /*
andre@0:         ** There's something to do. See if our existing polling list
andre@0:         ** is large enough for what we have to do?
andre@0:         */
andre@0: 
andre@0:         while (group->polling_count < group->waiter->count)
andre@0:         {
andre@0:             PRUint32 old_count = group->waiter->count;
andre@0:             PRUint32 new_count = PR_ROUNDUP(old_count, _PR_POLL_COUNT_FUDGE);
andre@0:             PRSize new_size = sizeof(PRPollDesc) * new_count;
andre@0:             PRPollDesc *old_polling_list = group->polling_list;
andre@0: 
andre@0:             PR_Unlock(group->ml);
andre@0:             poll_list = (PRPollDesc*)PR_CALLOC(new_size);
andre@0:             if (NULL == poll_list)
andre@0:             {
andre@0:                 PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:                 PR_Lock(group->ml);
andre@0:                 goto failed_alloc;
andre@0:             }
andre@0:             if (NULL != old_polling_list)
andre@0:                 PR_DELETE(old_polling_list);
andre@0:             PR_Lock(group->ml);
andre@0:             if (_prmw_running != group->state)
andre@0:             {
andre@0:                 PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:                 goto aborted;
andre@0:             }
andre@0:             group->polling_list = poll_list;
andre@0:             group->polling_count = new_count;
andre@0:         }
andre@0: 
andre@0:         now = PR_IntervalNow();
andre@0:         polling_interval = max_polling_interval;
andre@0:         since_last_poll = now - group->last_poll;
andre@0: 
andre@0:         waiter = &group->waiter->recv_wait;
andre@0:         poll_list = group->polling_list;
andre@0:         for (count = 0; count < group->waiter->count; ++waiter)
andre@0:         {
andre@0:             PR_ASSERT(waiter < &group->waiter->recv_wait
andre@0:                 + group->waiter->length);
andre@0:             if (NULL != *waiter)  /* a live one! */
andre@0:             {
andre@0:                 if ((PR_INTERVAL_NO_TIMEOUT != (*waiter)->timeout)
andre@0:                 && (since_last_poll >= (*waiter)->timeout))
andre@0:                     _MW_DoneInternal(group, waiter, PR_MW_TIMEOUT);
andre@0:                 else
andre@0:                 {
andre@0:                     if (PR_INTERVAL_NO_TIMEOUT != (*waiter)->timeout)
andre@0:                     {
andre@0:                         (*waiter)->timeout -= since_last_poll;
andre@0:                         if ((*waiter)->timeout < polling_interval)
andre@0:                             polling_interval = (*waiter)->timeout;
andre@0:                     }
andre@0:                     PR_ASSERT(poll_list < group->polling_list
andre@0:                         + group->polling_count);
andre@0:                     poll_list->fd = (*waiter)->fd;
andre@0:                     poll_list->in_flags = PR_POLL_READ;
andre@0:                     poll_list->out_flags = 0;
andre@0: #if 0
andre@0:                     printf(
andre@0:                         "Polling 0x%x[%d]: [fd: 0x%x, tmo: %u]\n",
andre@0:                         poll_list, count, poll_list->fd, (*waiter)->timeout);
andre@0: #endif
andre@0:                     poll_list += 1;
andre@0:                     count += 1;
andre@0:                 }
andre@0:             }
andre@0:         } 
andre@0: 
andre@0:         PR_ASSERT(count == group->waiter->count);
andre@0: 
andre@0:         /*
andre@0:         ** If there are no more threads waiting for completion,
andre@0:         ** we need to return.
andre@0:         */
andre@0:         if ((!PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:         && (1 == group->waiting_threads)) break;
andre@0: 
andre@0:         if (0 == count) continue;  /* wait for new business */
andre@0: 
andre@0:         group->last_poll = now;
andre@0: 
andre@0:         PR_Unlock(group->ml);
andre@0: 
andre@0:         count_ready = PR_Poll(group->polling_list, count, polling_interval);
andre@0: 
andre@0:         PR_Lock(group->ml);
andre@0: 
andre@0:         if (_prmw_running != group->state)
andre@0:         {
andre@0:             PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:             goto aborted;
andre@0:         }
andre@0:         if (-1 == count_ready)
andre@0:         {
andre@0:             goto failed_poll;  /* that's a shame */
andre@0:         }
andre@0:         else if (0 < count_ready)
andre@0:         {
andre@0:             for (poll_list = group->polling_list; count > 0;
andre@0:             poll_list++, count--)
andre@0:             {
andre@0:                 PR_ASSERT(
andre@0:                     poll_list < group->polling_list + group->polling_count);
andre@0:                 if (poll_list->out_flags != 0)
andre@0:                 {
andre@0:                     waiter = _MW_LookupInternal(group, poll_list->fd);
andre@0:                     /*
andre@0:                     ** If 'waiter' is NULL, that means the wait receive
andre@0:                     ** descriptor has been canceled.
andre@0:                     */
andre@0:                     if (NULL != waiter)
andre@0:                         _MW_DoneInternal(group, waiter, PR_MW_SUCCESS);
andre@0:                 }
andre@0:             }
andre@0:         }
andre@0:         /*
andre@0:         ** If there are no more threads waiting for completion,
andre@0:         ** we need to return.
andre@0:         ** This thread was "borrowed" to do the polling, but it really
andre@0:         ** belongs to the client.
andre@0:         */
andre@0:         if ((!PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:         && (1 == group->waiting_threads)) break;
andre@0:     }
andre@0: 
andre@0:     rv = PR_SUCCESS;
andre@0: 
andre@0: aborted:
andre@0: failed_poll:
andre@0: failed_alloc:
andre@0:     group->poller = NULL;  /* we were that, not we ain't */
andre@0:     if ((_prmw_running == group->state) && (group->waiting_threads > 1))
andre@0:     {
andre@0:         /* Wake up one thread to become the new poller. */
andre@0:         PR_NotifyCondVar(group->io_complete);
andre@0:     }
andre@0:     return rv;  /* we return with the lock held */
andre@0: }  /* _MW_PollInternal */
andre@0: #endif /* !WINNT */
andre@0: 
andre@0: static PRMWGroupState MW_TestForShutdownInternal(PRWaitGroup *group)
andre@0: {
andre@0:     PRMWGroupState rv = group->state;
andre@0:     /*
andre@0:     ** Looking at the group's fields is safe because
andre@0:     ** once the group's state is no longer running, it
andre@0:     ** cannot revert and there is a safe check on entry
andre@0:     ** to make sure no more threads are made to wait.
andre@0:     */
andre@0:     if ((_prmw_stopping == rv)
andre@0:     && (0 == group->waiting_threads))
andre@0:     {
andre@0:         rv = group->state = _prmw_stopped;
andre@0:         PR_NotifyCondVar(group->mw_manage);
andre@0:     }
andre@0:     return rv;
andre@0: }  /* MW_TestForShutdownInternal */
andre@0: 
andre@0: #ifndef WINNT
andre@0: static void _MW_InitialRecv(PRCList *io_ready)
andre@0: {
andre@0:     PRRecvWait *desc = (PRRecvWait*)io_ready;
andre@0:     if ((NULL == desc->buffer.start)
andre@0:     || (0 == desc->buffer.length))
andre@0:         desc->bytesRecv = 0;
andre@0:     else
andre@0:     {
andre@0:         desc->bytesRecv = (desc->fd->methods->recv)(
andre@0:             desc->fd, desc->buffer.start,
andre@0:             desc->buffer.length, 0, desc->timeout);
andre@0:         if (desc->bytesRecv < 0)  /* SetError should already be there */
andre@0:             desc->outcome = PR_MW_FAILURE;
andre@0:     }
andre@0: }  /* _MW_InitialRecv */
andre@0: #endif
andre@0: 
andre@0: #ifdef WINNT
andre@0: static void NT_TimeProc(void *arg)
andre@0: {
andre@0:     _MDOverlapped *overlapped = (_MDOverlapped *)arg;
andre@0:     PRRecvWait *desc =  overlapped->data.mw.desc;
andre@0:     PRFileDesc *bottom;
andre@0:     
andre@0:     if (InterlockedCompareExchange((LONG *)&desc->outcome,
andre@0:         (LONG)PR_MW_TIMEOUT, (LONG)PR_MW_PENDING) != (LONG)PR_MW_PENDING)
andre@0:     {
andre@0:         /* This wait recv descriptor has already completed. */
andre@0:         return;
andre@0:     }
andre@0: 
andre@0:     /* close the osfd to abort the outstanding async io request */
andre@0:     /* $$$$
andre@0:     ** Little late to be checking if NSPR's on the bottom of stack,
andre@0:     ** but if we don't check, we can't assert that the private data
andre@0:     ** is what we think it is.
andre@0:     ** $$$$
andre@0:     */
andre@0:     bottom = PR_GetIdentitiesLayer(desc->fd, PR_NSPR_IO_LAYER);
andre@0:     PR_ASSERT(NULL != bottom);
andre@0:     if (NULL != bottom)  /* now what!?!?! */
andre@0:     {
andre@0:         bottom->secret->state = _PR_FILEDESC_CLOSED;
andre@0:         if (closesocket(bottom->secret->md.osfd) == SOCKET_ERROR)
andre@0:         {
andre@0:             fprintf(stderr, "closesocket failed: %d\n", WSAGetLastError());
andre@0:             PR_ASSERT(!"What shall I do?");
andre@0:         }
andre@0:     }
andre@0:     return;
andre@0: }  /* NT_TimeProc */
andre@0: 
andre@0: static PRStatus NT_HashRemove(PRWaitGroup *group, PRFileDesc *fd)
andre@0: {
andre@0:     PRRecvWait **waiter;
andre@0: 
andre@0:     _PR_MD_LOCK(&group->mdlock);
andre@0:     waiter = _MW_LookupInternal(group, fd);
andre@0:     if (NULL != waiter)
andre@0:     {
andre@0:         group->waiter->count -= 1;
andre@0:         *waiter = NULL;
andre@0:     }
andre@0:     _PR_MD_UNLOCK(&group->mdlock);
andre@0:     return (NULL != waiter) ? PR_SUCCESS : PR_FAILURE;
andre@0: }
andre@0: 
andre@0: PRStatus NT_HashRemoveInternal(PRWaitGroup *group, PRFileDesc *fd)
andre@0: {
andre@0:     PRRecvWait **waiter;
andre@0: 
andre@0:     waiter = _MW_LookupInternal(group, fd);
andre@0:     if (NULL != waiter)
andre@0:     {
andre@0:         group->waiter->count -= 1;
andre@0:         *waiter = NULL;
andre@0:     }
andre@0:     return (NULL != waiter) ? PR_SUCCESS : PR_FAILURE;
andre@0: }
andre@0: #endif /* WINNT */
andre@0: 
andre@0: /******************************************************************/
andre@0: /******************************************************************/
andre@0: /********************** The public API portion ********************/
andre@0: /******************************************************************/
andre@0: /******************************************************************/
andre@0: PR_IMPLEMENT(PRStatus) PR_AddWaitFileDesc(
andre@0:     PRWaitGroup *group, PRRecvWait *desc)
andre@0: {
andre@0:     _PR_HashStory hrv;
andre@0:     PRStatus rv = PR_FAILURE;
andre@0: #ifdef WINNT
andre@0:     _MDOverlapped *overlapped;
andre@0:     HANDLE hFile;
andre@0:     BOOL bResult;
andre@0:     DWORD dwError;
andre@0:     PRFileDesc *bottom;
andre@0: #endif
andre@0: 
andre@0:     if (!_pr_initialized) _PR_ImplicitInitialization();
andre@0:     if ((NULL == group) && (NULL == (group = MW_Init2())))
andre@0:     {
andre@0:         return rv;
andre@0:     }
andre@0: 
andre@0:     PR_ASSERT(NULL != desc->fd);
andre@0: 
andre@0:     desc->outcome = PR_MW_PENDING;  /* nice, well known value */
andre@0:     desc->bytesRecv = 0;  /* likewise, though this value is ambiguious */
andre@0: 
andre@0:     PR_Lock(group->ml);
andre@0: 
andre@0:     if (_prmw_running != group->state)
andre@0:     {
andre@0:         /* Not allowed to add after cancelling the group */
andre@0:         desc->outcome = PR_MW_INTERRUPT;
andre@0:         PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:         PR_Unlock(group->ml);
andre@0:         return rv;
andre@0:     }
andre@0: 
andre@0: #ifdef WINNT
andre@0:     _PR_MD_LOCK(&group->mdlock);
andre@0: #endif
andre@0: 
andre@0:     /*
andre@0:     ** If the waiter count is zero at this point, there's no telling
andre@0:     ** how long we've been idle. Therefore, initialize the beginning
andre@0:     ** of the timing interval. As long as the list doesn't go empty,
andre@0:     ** it will maintain itself.
andre@0:     */
andre@0:     if (0 == group->waiter->count)
andre@0:         group->last_poll = PR_IntervalNow();
andre@0: 
andre@0:     do
andre@0:     {
andre@0:         hrv = MW_AddHashInternal(desc, group->waiter);
andre@0:         if (_prmw_rehash != hrv) break;
andre@0:         hrv = MW_ExpandHashInternal(group);  /* gruesome */
andre@0:         if (_prmw_success != hrv) break;
andre@0:     } while (PR_TRUE);
andre@0: 
andre@0: #ifdef WINNT
andre@0:     _PR_MD_UNLOCK(&group->mdlock);
andre@0: #endif
andre@0: 
andre@0:     PR_NotifyCondVar(group->new_business);  /* tell the world */
andre@0:     rv = (_prmw_success == hrv) ? PR_SUCCESS : PR_FAILURE;
andre@0:     PR_Unlock(group->ml);
andre@0: 
andre@0: #ifdef WINNT
andre@0:     overlapped = PR_NEWZAP(_MDOverlapped);
andre@0:     if (NULL == overlapped)
andre@0:     {
andre@0:         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:         NT_HashRemove(group, desc->fd);
andre@0:         return rv;
andre@0:     }
andre@0:     overlapped->ioModel = _MD_MultiWaitIO;
andre@0:     overlapped->data.mw.desc = desc;
andre@0:     overlapped->data.mw.group = group;
andre@0:     if (desc->timeout != PR_INTERVAL_NO_TIMEOUT)
andre@0:     {
andre@0:         overlapped->data.mw.timer = CreateTimer(
andre@0:             desc->timeout,
andre@0:             NT_TimeProc,
andre@0:             overlapped);
andre@0:         if (0 == overlapped->data.mw.timer)
andre@0:         {
andre@0:             NT_HashRemove(group, desc->fd);
andre@0:             PR_DELETE(overlapped);
andre@0:             /*
andre@0:              * XXX It appears that a maximum of 16 timer events can
andre@0:              * be outstanding. GetLastError() returns 0 when I try it.
andre@0:              */
andre@0:             PR_SetError(PR_INSUFFICIENT_RESOURCES_ERROR, GetLastError());
andre@0:             return PR_FAILURE;
andre@0:         }
andre@0:     }
andre@0: 
andre@0:     /* Reach to the bottom layer to get the OS fd */
andre@0:     bottom = PR_GetIdentitiesLayer(desc->fd, PR_NSPR_IO_LAYER);
andre@0:     PR_ASSERT(NULL != bottom);
andre@0:     if (NULL == bottom)
andre@0:     {
andre@0:         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:         return PR_FAILURE;
andre@0:     }
andre@0:     hFile = (HANDLE)bottom->secret->md.osfd; 
andre@0:     if (!bottom->secret->md.io_model_committed)
andre@0:     {
andre@0:         PRInt32 st;
andre@0:         st = _md_Associate(hFile);
andre@0:         PR_ASSERT(0 != st);
andre@0:         bottom->secret->md.io_model_committed = PR_TRUE;
andre@0:     }
andre@0:     bResult = ReadFile(hFile,
andre@0:         desc->buffer.start,
andre@0:         (DWORD)desc->buffer.length,
andre@0:         NULL,
andre@0:         &overlapped->overlapped);
andre@0:     if (FALSE == bResult && (dwError = GetLastError()) != ERROR_IO_PENDING)
andre@0:     {
andre@0:         if (desc->timeout != PR_INTERVAL_NO_TIMEOUT)
andre@0:         {
andre@0:             if (InterlockedCompareExchange((LONG *)&desc->outcome,
andre@0:                 (LONG)PR_MW_FAILURE, (LONG)PR_MW_PENDING)
andre@0:                 == (LONG)PR_MW_PENDING)
andre@0:             {
andre@0:                 CancelTimer(overlapped->data.mw.timer);
andre@0:             }
andre@0:             NT_HashRemove(group, desc->fd);
andre@0:             PR_DELETE(overlapped);
andre@0:         }
andre@0:         _PR_MD_MAP_READ_ERROR(dwError);
andre@0:         rv = PR_FAILURE;
andre@0:     }
andre@0: #endif
andre@0: 
andre@0:     return rv;
andre@0: }  /* PR_AddWaitFileDesc */
andre@0: 
andre@0: PR_IMPLEMENT(PRRecvWait*) PR_WaitRecvReady(PRWaitGroup *group)
andre@0: {
andre@0:     PRCList *io_ready = NULL;
andre@0: #ifdef WINNT
andre@0:     PRThread *me = _PR_MD_CURRENT_THREAD();
andre@0:     _MDOverlapped *overlapped;    
andre@0: #endif
andre@0: 
andre@0:     if (!_pr_initialized) _PR_ImplicitInitialization();
andre@0:     if ((NULL == group) && (NULL == (group = MW_Init2()))) goto failed_init;
andre@0: 
andre@0:     PR_Lock(group->ml);
andre@0: 
andre@0:     if (_prmw_running != group->state)
andre@0:     {
andre@0:         PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:         goto invalid_state;
andre@0:     }
andre@0: 
andre@0:     group->waiting_threads += 1;  /* the polling thread is counted */
andre@0: 
andre@0: #ifdef WINNT
andre@0:     _PR_MD_LOCK(&group->mdlock);
andre@0:     while (PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:     {
andre@0:         _PR_THREAD_LOCK(me);
andre@0:         me->state = _PR_IO_WAIT;
andre@0:         PR_APPEND_LINK(&me->waitQLinks, &group->wait_list);
andre@0:         if (!_PR_IS_NATIVE_THREAD(me))
andre@0:         {
andre@0:             _PR_SLEEPQ_LOCK(me->cpu);
andre@0:             _PR_ADD_SLEEPQ(me, PR_INTERVAL_NO_TIMEOUT);
andre@0:             _PR_SLEEPQ_UNLOCK(me->cpu);
andre@0:         }
andre@0:         _PR_THREAD_UNLOCK(me);
andre@0:         _PR_MD_UNLOCK(&group->mdlock);
andre@0:         PR_Unlock(group->ml);
andre@0:         _PR_MD_WAIT(me, PR_INTERVAL_NO_TIMEOUT);
andre@0:         me->state = _PR_RUNNING;
andre@0:         PR_Lock(group->ml);
andre@0:         _PR_MD_LOCK(&group->mdlock);
andre@0:         if (_PR_PENDING_INTERRUPT(me)) {
andre@0:             PR_REMOVE_LINK(&me->waitQLinks);
andre@0:             _PR_MD_UNLOCK(&group->mdlock);
andre@0:             me->flags &= ~_PR_INTERRUPT;
andre@0:             me->io_suspended = PR_FALSE;
andre@0:             PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
andre@0:             goto aborted;
andre@0:         }
andre@0:     }
andre@0:     io_ready = PR_LIST_HEAD(&group->io_ready);
andre@0:     PR_ASSERT(io_ready != NULL);
andre@0:     PR_REMOVE_LINK(io_ready);
andre@0:     _PR_MD_UNLOCK(&group->mdlock);
andre@0:     overlapped = (_MDOverlapped *)
andre@0:         ((char *)io_ready - offsetof(_MDOverlapped, data));
andre@0:     io_ready = &overlapped->data.mw.desc->internal;
andre@0: #else
andre@0:     do
andre@0:     {
andre@0:         /*
andre@0:         ** If the I/O ready list isn't empty, have this thread
andre@0:         ** return with the first receive wait object that's available.
andre@0:         */
andre@0:         if (PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:         {
andre@0:             /*
andre@0:             ** Is there a polling thread yet? If not, grab this thread
andre@0:             ** and use it.
andre@0:             */
andre@0:             if (NULL == group->poller)
andre@0:             {
andre@0:                 /*
andre@0:                 ** This thread will stay do polling until it becomes the only one
andre@0:                 ** left to service a completion. Then it will return and there will
andre@0:                 ** be none left to actually poll or to run completions.
andre@0:                 **
andre@0:                 ** The polling function should only return w/ failure or
andre@0:                 ** with some I/O ready.
andre@0:                 */
andre@0:                 if (PR_FAILURE == _MW_PollInternal(group)) goto failed_poll;
andre@0:             }
andre@0:             else
andre@0:             {
andre@0:                 /*
andre@0:                 ** There are four reasons a thread can be awakened from
andre@0:                 ** a wait on the io_complete condition variable.
andre@0:                 ** 1. Some I/O has completed, i.e., the io_ready list
andre@0:                 **    is nonempty.
andre@0:                 ** 2. The wait group is canceled.
andre@0:                 ** 3. The thread is interrupted.
andre@0:                 ** 4. The current polling thread has to leave and needs
andre@0:                 **    a replacement.
andre@0:                 ** The logic to find a new polling thread is made more
andre@0:                 ** complicated by all the other possible events.
andre@0:                 ** I tried my best to write the logic clearly, but
andre@0:                 ** it is still full of if's with continue and goto.
andre@0:                 */
andre@0:                 PRStatus st;
andre@0:                 do 
andre@0:                 {
andre@0:                     st = PR_WaitCondVar(group->io_complete, PR_INTERVAL_NO_TIMEOUT);
andre@0:                     if (_prmw_running != group->state)
andre@0:                     {
andre@0:                         PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:                         goto aborted;
andre@0:                     }
andre@0:                     if (_MW_ABORTED(st) || (NULL == group->poller)) break;
andre@0:                 } while (PR_CLIST_IS_EMPTY(&group->io_ready));
andre@0: 
andre@0:                 /*
andre@0:                 ** The thread is interrupted and has to leave.  It might
andre@0:                 ** have also been awakened to process ready i/o or be the
andre@0:                 ** new poller.  To be safe, if either condition is true,
andre@0:                 ** we awaken another thread to take its place.
andre@0:                 */
andre@0:                 if (_MW_ABORTED(st))
andre@0:                 {
andre@0:                     if ((NULL == group->poller
andre@0:                     || !PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:                     && group->waiting_threads > 1)
andre@0:                         PR_NotifyCondVar(group->io_complete);
andre@0:                     goto aborted;
andre@0:                 }
andre@0: 
andre@0:                 /*
andre@0:                 ** A new poller is needed, but can I be the new poller?
andre@0:                 ** If there is no i/o ready, sure.  But if there is any
andre@0:                 ** i/o ready, it has a higher priority.  I want to
andre@0:                 ** process the ready i/o first and wake up another
andre@0:                 ** thread to be the new poller.
andre@0:                 */ 
andre@0:                 if (NULL == group->poller)
andre@0:                 {
andre@0:                     if (PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:                         continue;
andre@0:                     if (group->waiting_threads > 1)
andre@0:                         PR_NotifyCondVar(group->io_complete);
andre@0:                 }
andre@0:             }
andre@0:             PR_ASSERT(!PR_CLIST_IS_EMPTY(&group->io_ready));
andre@0:         }
andre@0:         io_ready = PR_LIST_HEAD(&group->io_ready);
andre@0:         PR_NotifyCondVar(group->io_taken);
andre@0:         PR_ASSERT(io_ready != NULL);
andre@0:         PR_REMOVE_LINK(io_ready);
andre@0:     } while (NULL == io_ready);
andre@0: 
andre@0: failed_poll:
andre@0: 
andre@0: #endif
andre@0: 
andre@0: aborted:
andre@0: 
andre@0:     group->waiting_threads -= 1;
andre@0: invalid_state:
andre@0:     (void)MW_TestForShutdownInternal(group);
andre@0:     PR_Unlock(group->ml);
andre@0: 
andre@0: failed_init:
andre@0:     if (NULL != io_ready)
andre@0:     {
andre@0:         /* If the operation failed, record the reason why */
andre@0:         switch (((PRRecvWait*)io_ready)->outcome)
andre@0:         {
andre@0:             case PR_MW_PENDING:
andre@0:                 PR_ASSERT(0);
andre@0:                 break;
andre@0:             case PR_MW_SUCCESS:
andre@0: #ifndef WINNT
andre@0:                 _MW_InitialRecv(io_ready);
andre@0: #endif
andre@0:                 break;
andre@0: #ifdef WINNT
andre@0:             case PR_MW_FAILURE:
andre@0:                 _PR_MD_MAP_READ_ERROR(overlapped->data.mw.error);
andre@0:                 break;
andre@0: #endif
andre@0:             case PR_MW_TIMEOUT:
andre@0:                 PR_SetError(PR_IO_TIMEOUT_ERROR, 0);
andre@0:                 break;
andre@0:             case PR_MW_INTERRUPT:
andre@0:                 PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
andre@0:                 break;
andre@0:             default: break;
andre@0:         }
andre@0: #ifdef WINNT
andre@0:         if (NULL != overlapped->data.mw.timer)
andre@0:         {
andre@0:             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT
andre@0:                 != overlapped->data.mw.desc->timeout);
andre@0:             CancelTimer(overlapped->data.mw.timer);
andre@0:         }
andre@0:         else
andre@0:         {
andre@0:             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT
andre@0:                 == overlapped->data.mw.desc->timeout);
andre@0:         }
andre@0:         PR_DELETE(overlapped);
andre@0: #endif
andre@0:     }
andre@0:     return (PRRecvWait*)io_ready;
andre@0: }  /* PR_WaitRecvReady */
andre@0: 
andre@0: PR_IMPLEMENT(PRStatus) PR_CancelWaitFileDesc(PRWaitGroup *group, PRRecvWait *desc)
andre@0: {
andre@0: #if !defined(WINNT)
andre@0:     PRRecvWait **recv_wait;
andre@0: #endif
andre@0:     PRStatus rv = PR_SUCCESS;
andre@0:     if (NULL == group) group = mw_state->group;
andre@0:     PR_ASSERT(NULL != group);
andre@0:     if (NULL == group)
andre@0:     {
andre@0:         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:         return PR_FAILURE;
andre@0:     }
andre@0: 
andre@0:     PR_Lock(group->ml);
andre@0: 
andre@0:     if (_prmw_running != group->state)
andre@0:     {
andre@0:         PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:         rv = PR_FAILURE;
andre@0:         goto unlock;
andre@0:     }
andre@0: 
andre@0: #ifdef WINNT
andre@0:     if (InterlockedCompareExchange((LONG *)&desc->outcome,
andre@0:         (LONG)PR_MW_INTERRUPT, (LONG)PR_MW_PENDING) == (LONG)PR_MW_PENDING)
andre@0:     {
andre@0:         PRFileDesc *bottom = PR_GetIdentitiesLayer(desc->fd, PR_NSPR_IO_LAYER);
andre@0:         PR_ASSERT(NULL != bottom);
andre@0:         if (NULL == bottom)
andre@0:         {
andre@0:             PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:             goto unlock;
andre@0:         }
andre@0:         bottom->secret->state = _PR_FILEDESC_CLOSED;
andre@0: #if 0
andre@0:         fprintf(stderr, "cancel wait recv: closing socket\n");
andre@0: #endif
andre@0:         if (closesocket(bottom->secret->md.osfd) == SOCKET_ERROR)
andre@0:         {
andre@0:             fprintf(stderr, "closesocket failed: %d\n", WSAGetLastError());
andre@0:             exit(1);
andre@0:         }
andre@0:     }
andre@0: #else
andre@0:     if (NULL != (recv_wait = _MW_LookupInternal(group, desc->fd)))
andre@0:     {
andre@0:         /* it was in the wait table */
andre@0:         _MW_DoneInternal(group, recv_wait, PR_MW_INTERRUPT);
andre@0:         goto unlock;
andre@0:     }
andre@0:     if (!PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:     {
andre@0:         /* is it already complete? */
andre@0:         PRCList *head = PR_LIST_HEAD(&group->io_ready);
andre@0:         do
andre@0:         {
andre@0:             PRRecvWait *done = (PRRecvWait*)head;
andre@0:             if (done == desc) goto unlock;
andre@0:             head = PR_NEXT_LINK(head);
andre@0:         } while (head != &group->io_ready);
andre@0:     }
andre@0:     PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:     rv = PR_FAILURE;
andre@0: 
andre@0: #endif
andre@0: unlock:
andre@0:     PR_Unlock(group->ml);
andre@0:     return rv;
andre@0: }  /* PR_CancelWaitFileDesc */
andre@0: 
andre@0: PR_IMPLEMENT(PRRecvWait*) PR_CancelWaitGroup(PRWaitGroup *group)
andre@0: {
andre@0:     PRRecvWait **desc;
andre@0:     PRRecvWait *recv_wait = NULL;
andre@0: #ifdef WINNT
andre@0:     _MDOverlapped *overlapped;
andre@0:     PRRecvWait **end;
andre@0:     PRThread *me = _PR_MD_CURRENT_THREAD();
andre@0: #endif
andre@0: 
andre@0:     if (NULL == group) group = mw_state->group;
andre@0:     PR_ASSERT(NULL != group);
andre@0:     if (NULL == group)
andre@0:     {
andre@0:         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:         return NULL;
andre@0:     }
andre@0: 
andre@0:     PR_Lock(group->ml);
andre@0:     if (_prmw_stopped != group->state)
andre@0:     {
andre@0:         if (_prmw_running == group->state)
andre@0:             group->state = _prmw_stopping;  /* so nothing new comes in */
andre@0:         if (0 == group->waiting_threads)  /* is there anybody else? */
andre@0:             group->state = _prmw_stopped;  /* we can stop right now */
andre@0:         else
andre@0:         {
andre@0:             PR_NotifyAllCondVar(group->new_business);
andre@0:             PR_NotifyAllCondVar(group->io_complete);
andre@0:         }
andre@0:         while (_prmw_stopped != group->state)
andre@0:             (void)PR_WaitCondVar(group->mw_manage, PR_INTERVAL_NO_TIMEOUT);
andre@0:     }
andre@0: 
andre@0: #ifdef WINNT
andre@0:     _PR_MD_LOCK(&group->mdlock);
andre@0: #endif
andre@0:     /* make all the existing descriptors look done/interrupted */
andre@0: #ifdef WINNT
andre@0:     end = &group->waiter->recv_wait + group->waiter->length;
andre@0:     for (desc = &group->waiter->recv_wait; desc < end; ++desc)
andre@0:     {
andre@0:         if (NULL != *desc)
andre@0:         {
andre@0:             if (InterlockedCompareExchange((LONG *)&(*desc)->outcome,
andre@0:                 (LONG)PR_MW_INTERRUPT, (LONG)PR_MW_PENDING)
andre@0:                 == (LONG)PR_MW_PENDING)
andre@0:             {
andre@0:                 PRFileDesc *bottom = PR_GetIdentitiesLayer(
andre@0:                     (*desc)->fd, PR_NSPR_IO_LAYER);
andre@0:                 PR_ASSERT(NULL != bottom);
andre@0:                 if (NULL == bottom)
andre@0:                 {
andre@0:                     PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:                     goto invalid_arg;
andre@0:                 }
andre@0:                 bottom->secret->state = _PR_FILEDESC_CLOSED;
andre@0: #if 0
andre@0:                 fprintf(stderr, "cancel wait group: closing socket\n");
andre@0: #endif
andre@0:                 if (closesocket(bottom->secret->md.osfd) == SOCKET_ERROR)
andre@0:                 {
andre@0:                     fprintf(stderr, "closesocket failed: %d\n",
andre@0:                         WSAGetLastError());
andre@0:                     exit(1);
andre@0:                 }
andre@0:             }
andre@0:         }
andre@0:     }
andre@0:     while (group->waiter->count > 0)
andre@0:     {
andre@0:         _PR_THREAD_LOCK(me);
andre@0:         me->state = _PR_IO_WAIT;
andre@0:         PR_APPEND_LINK(&me->waitQLinks, &group->wait_list);
andre@0:         if (!_PR_IS_NATIVE_THREAD(me))
andre@0:         {
andre@0:             _PR_SLEEPQ_LOCK(me->cpu);
andre@0:             _PR_ADD_SLEEPQ(me, PR_INTERVAL_NO_TIMEOUT);
andre@0:             _PR_SLEEPQ_UNLOCK(me->cpu);
andre@0:         }
andre@0:         _PR_THREAD_UNLOCK(me);
andre@0:         _PR_MD_UNLOCK(&group->mdlock);
andre@0:         PR_Unlock(group->ml);
andre@0:         _PR_MD_WAIT(me, PR_INTERVAL_NO_TIMEOUT);
andre@0:         me->state = _PR_RUNNING;
andre@0:         PR_Lock(group->ml);
andre@0:         _PR_MD_LOCK(&group->mdlock);
andre@0:     }
andre@0: #else
andre@0:     for (desc = &group->waiter->recv_wait; group->waiter->count > 0; ++desc)
andre@0:     {
andre@0:         PR_ASSERT(desc < &group->waiter->recv_wait + group->waiter->length);
andre@0:         if (NULL != *desc)
andre@0:             _MW_DoneInternal(group, desc, PR_MW_INTERRUPT);
andre@0:     }
andre@0: #endif
andre@0: 
andre@0:     /* take first element of finished list and return it or NULL */
andre@0:     if (PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:         PR_SetError(PR_GROUP_EMPTY_ERROR, 0);
andre@0:     else
andre@0:     {
andre@0:         PRCList *head = PR_LIST_HEAD(&group->io_ready);
andre@0:         PR_REMOVE_AND_INIT_LINK(head);
andre@0: #ifdef WINNT
andre@0:         overlapped = (_MDOverlapped *)
andre@0:             ((char *)head - offsetof(_MDOverlapped, data));
andre@0:         head = &overlapped->data.mw.desc->internal;
andre@0:         if (NULL != overlapped->data.mw.timer)
andre@0:         {
andre@0:             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT
andre@0:                 != overlapped->data.mw.desc->timeout);
andre@0:             CancelTimer(overlapped->data.mw.timer);
andre@0:         }
andre@0:         else
andre@0:         {
andre@0:             PR_ASSERT(PR_INTERVAL_NO_TIMEOUT
andre@0:                 == overlapped->data.mw.desc->timeout);
andre@0:         }
andre@0:         PR_DELETE(overlapped);
andre@0: #endif
andre@0:         recv_wait = (PRRecvWait*)head;
andre@0:     }
andre@0: #ifdef WINNT
andre@0: invalid_arg:
andre@0:     _PR_MD_UNLOCK(&group->mdlock);
andre@0: #endif
andre@0:     PR_Unlock(group->ml);
andre@0: 
andre@0:     return recv_wait;
andre@0: }  /* PR_CancelWaitGroup */
andre@0: 
andre@0: PR_IMPLEMENT(PRWaitGroup*) PR_CreateWaitGroup(PRInt32 size /* ignored */)
andre@0: {
andre@0:     PRWaitGroup *wg;
andre@0: 
andre@0:     if (NULL == (wg = PR_NEWZAP(PRWaitGroup)))
andre@0:     {
andre@0:         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:         goto failed;
andre@0:     }
andre@0:     /* the wait group itself */
andre@0:     wg->ml = PR_NewLock();
andre@0:     if (NULL == wg->ml) goto failed_lock;
andre@0:     wg->io_taken = PR_NewCondVar(wg->ml);
andre@0:     if (NULL == wg->io_taken) goto failed_cvar0;
andre@0:     wg->io_complete = PR_NewCondVar(wg->ml);
andre@0:     if (NULL == wg->io_complete) goto failed_cvar1;
andre@0:     wg->new_business = PR_NewCondVar(wg->ml);
andre@0:     if (NULL == wg->new_business) goto failed_cvar2;
andre@0:     wg->mw_manage = PR_NewCondVar(wg->ml);
andre@0:     if (NULL == wg->mw_manage) goto failed_cvar3;
andre@0: 
andre@0:     PR_INIT_CLIST(&wg->group_link);
andre@0:     PR_INIT_CLIST(&wg->io_ready);
andre@0: 
andre@0:     /* the waiters sequence */
andre@0:     wg->waiter = (_PRWaiterHash*)PR_CALLOC(
andre@0:         sizeof(_PRWaiterHash) +
andre@0:         (_PR_DEFAULT_HASH_LENGTH * sizeof(PRRecvWait*)));
andre@0:     if (NULL == wg->waiter)
andre@0:     {
andre@0:         PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:         goto failed_waiter;
andre@0:     }
andre@0:     wg->waiter->count = 0;
andre@0:     wg->waiter->length = _PR_DEFAULT_HASH_LENGTH;
andre@0: 
andre@0: #ifdef WINNT
andre@0:     _PR_MD_NEW_LOCK(&wg->mdlock);
andre@0:     PR_INIT_CLIST(&wg->wait_list);
andre@0: #endif /* WINNT */
andre@0: 
andre@0:     PR_Lock(mw_lock);
andre@0:     PR_APPEND_LINK(&wg->group_link, &mw_state->group_list);
andre@0:     PR_Unlock(mw_lock);
andre@0:     return wg;
andre@0: 
andre@0: failed_waiter:
andre@0:     PR_DestroyCondVar(wg->mw_manage);
andre@0: failed_cvar3:
andre@0:     PR_DestroyCondVar(wg->new_business);
andre@0: failed_cvar2:
andre@0:     PR_DestroyCondVar(wg->io_complete);
andre@0: failed_cvar1:
andre@0:     PR_DestroyCondVar(wg->io_taken);
andre@0: failed_cvar0:
andre@0:     PR_DestroyLock(wg->ml);
andre@0: failed_lock:
andre@0:     PR_DELETE(wg);
andre@0:     wg = NULL;
andre@0: 
andre@0: failed:
andre@0:     return wg;
andre@0: }  /* MW_CreateWaitGroup */
andre@0: 
andre@0: PR_IMPLEMENT(PRStatus) PR_DestroyWaitGroup(PRWaitGroup *group)
andre@0: {
andre@0:     PRStatus rv = PR_SUCCESS;
andre@0:     if (NULL == group) group = mw_state->group;
andre@0:     PR_ASSERT(NULL != group);
andre@0:     if (NULL != group)
andre@0:     {
andre@0:         PR_Lock(group->ml);
andre@0:         if ((group->waiting_threads == 0)
andre@0:         && (group->waiter->count == 0)
andre@0:         && PR_CLIST_IS_EMPTY(&group->io_ready))
andre@0:         {
andre@0:             group->state = _prmw_stopped;
andre@0:         }
andre@0:         else
andre@0:         {
andre@0:             PR_SetError(PR_INVALID_STATE_ERROR, 0);
andre@0:             rv = PR_FAILURE;
andre@0:         }
andre@0:         PR_Unlock(group->ml);
andre@0:         if (PR_FAILURE == rv) return rv;
andre@0: 
andre@0:         PR_Lock(mw_lock);
andre@0:         PR_REMOVE_LINK(&group->group_link);
andre@0:         PR_Unlock(mw_lock);
andre@0: 
andre@0: #ifdef WINNT
andre@0:         /*
andre@0:          * XXX make sure wait_list is empty and waiter is empty.
andre@0:          * These must be checked while holding mdlock.
andre@0:          */
andre@0:         _PR_MD_FREE_LOCK(&group->mdlock);
andre@0: #endif
andre@0: 
andre@0:         PR_DELETE(group->waiter);
andre@0:         PR_DELETE(group->polling_list);
andre@0:         PR_DestroyCondVar(group->mw_manage);
andre@0:         PR_DestroyCondVar(group->new_business);
andre@0:         PR_DestroyCondVar(group->io_complete);
andre@0:         PR_DestroyCondVar(group->io_taken);
andre@0:         PR_DestroyLock(group->ml);
andre@0:         if (group == mw_state->group) mw_state->group = NULL;
andre@0:         PR_DELETE(group);
andre@0:     }
andre@0:     else
andre@0:     {
andre@0:         /* The default wait group is not created yet. */
andre@0:         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:         rv = PR_FAILURE;
andre@0:     }
andre@0:     return rv;
andre@0: }  /* PR_DestroyWaitGroup */
andre@0: 
andre@0: /**********************************************************************
andre@0: ***********************************************************************
andre@0: ******************** Wait group enumerations **************************
andre@0: ***********************************************************************
andre@0: **********************************************************************/
andre@0: 
andre@0: PR_IMPLEMENT(PRMWaitEnumerator*) PR_CreateMWaitEnumerator(PRWaitGroup *group)
andre@0: {
andre@0:     PRMWaitEnumerator *enumerator = PR_NEWZAP(PRMWaitEnumerator);
andre@0:     if (NULL == enumerator) PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
andre@0:     else
andre@0:     {
andre@0:         enumerator->group = group;
andre@0:         enumerator->seal = _PR_ENUM_SEALED;
andre@0:     }
andre@0:     return enumerator;
andre@0: }  /* PR_CreateMWaitEnumerator */
andre@0: 
andre@0: PR_IMPLEMENT(PRStatus) PR_DestroyMWaitEnumerator(PRMWaitEnumerator* enumerator)
andre@0: {
andre@0:     PR_ASSERT(NULL != enumerator);
andre@0:     PR_ASSERT(_PR_ENUM_SEALED == enumerator->seal);
andre@0:     if ((NULL == enumerator) || (_PR_ENUM_SEALED != enumerator->seal))
andre@0:     {
andre@0:         PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:         return PR_FAILURE;
andre@0:     }
andre@0:     enumerator->seal = _PR_ENUM_UNSEALED;
andre@0:     PR_Free(enumerator);
andre@0:     return PR_SUCCESS;
andre@0: }  /* PR_DestroyMWaitEnumerator */
andre@0: 
andre@0: PR_IMPLEMENT(PRRecvWait*) PR_EnumerateWaitGroup(
andre@0:     PRMWaitEnumerator *enumerator, const PRRecvWait *previous)
andre@0: {
andre@0:     PRRecvWait *result = NULL;
andre@0:     
andre@0:     /* entry point sanity checking */
andre@0:     PR_ASSERT(NULL != enumerator);
andre@0:     PR_ASSERT(_PR_ENUM_SEALED == enumerator->seal);
andre@0:     if ((NULL == enumerator)
andre@0:     || (_PR_ENUM_SEALED != enumerator->seal)) goto bad_argument;
andre@0: 
andre@0:     /* beginning of enumeration */
andre@0:     if (NULL == previous)
andre@0:     {
andre@0:         if (NULL == enumerator->group)
andre@0:         {
andre@0:             enumerator->group = mw_state->group;
andre@0:             if (NULL == enumerator->group)
andre@0:             {
andre@0:                 PR_SetError(PR_GROUP_EMPTY_ERROR, 0);
andre@0:                 return NULL;
andre@0:             }
andre@0:         }
andre@0:         enumerator->waiter = &enumerator->group->waiter->recv_wait;
andre@0:         enumerator->p_timestamp = enumerator->group->p_timestamp;
andre@0:         enumerator->thread = PR_GetCurrentThread();
andre@0:         enumerator->index = 0;
andre@0:     }
andre@0:     /* continuing an enumeration */
andre@0:     else
andre@0:     {
andre@0:         PRThread *me = PR_GetCurrentThread();
andre@0:         PR_ASSERT(me == enumerator->thread);
andre@0:         if (me != enumerator->thread) goto bad_argument;
andre@0: 
andre@0:         /* need to restart the enumeration */
andre@0:         if (enumerator->p_timestamp != enumerator->group->p_timestamp)
andre@0:             return PR_EnumerateWaitGroup(enumerator, NULL);
andre@0:     }
andre@0: 
andre@0:     /* actually progress the enumeration */
andre@0: #if defined(WINNT)
andre@0:     _PR_MD_LOCK(&enumerator->group->mdlock);
andre@0: #else
andre@0:     PR_Lock(enumerator->group->ml);
andre@0: #endif
andre@0:     while (enumerator->index++ < enumerator->group->waiter->length)
andre@0:     {
andre@0:         if (NULL != (result = *(enumerator->waiter)++)) break;
andre@0:     }
andre@0: #if defined(WINNT)
andre@0:     _PR_MD_UNLOCK(&enumerator->group->mdlock);
andre@0: #else
andre@0:     PR_Unlock(enumerator->group->ml);
andre@0: #endif
andre@0: 
andre@0:     return result;  /* what we live for */
andre@0: 
andre@0: bad_argument:
andre@0:     PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
andre@0:     return NULL;  /* probably ambiguous */
andre@0: }  /* PR_EnumerateWaitGroup */
andre@0: 
andre@0: /* prmwait.c */