ngx_freebsd_rfork_thread.c

nginx 反向代理0.7.1版本 用于实现反向代理

        ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                       "try lock mutex %p lock:%XD", m, m->lock);
    } else {
        ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                       "lock mutex %p lock:%XD", m, m->lock);
    }
#endif

    old = m->lock;
    tries = 0;

    for ( ;; ) {
        if (old & NGX_MUTEX_LOCK_BUSY) {

            if (try) {
                return NGX_AGAIN;
            }

            if (ngx_ncpu > 1 && tries++ < 1000) {

                /* the spinlock is used only on the SMP system */

                old = m->lock;
                continue;
            }

            if (m->semid == -1) {
                sched_yield();

                tries = 0;
                old = m->lock;
                continue;
            }

            ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                           "mutex %p lock:%XD", m, m->lock);

            /*
             * The mutex is locked so we increase a number
             * of the threads that are waiting on the mutex
             */

            lock = old + 1;

            if ((lock & ~NGX_MUTEX_LOCK_BUSY) > nthreads) {
                ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
                              "%D threads wait for mutex %p, "
                              "while only %ui threads are available",
                              lock & ~NGX_MUTEX_LOCK_BUSY, m, nthreads);
                ngx_abort();
            }

            if (ngx_atomic_cmp_set(&m->lock, old, lock)) {

                ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                               "wait mutex %p lock:%XD", m, m->lock);

                /*
                 * The number of the waiting threads has been increased
                 * and we would wait on the SysV semaphore.
                 * A semaphore should wake up us more efficiently than
                 * a simple sched_yield() or usleep().
                 */

                op.sem_num = 0;
                op.sem_op = -1;
                op.sem_flg = 0;

                if (semop(m->semid, &op, 1) == -1) {
                    ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
                                 "semop() failed while waiting on mutex %p", m);
                    ngx_abort();
                }

                ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                               "mutex waked up %p lock:%XD", m, m->lock);

                tries = 0;
                old = m->lock;
                continue;
            }

            old = m->lock;

        } else {
            lock = old | NGX_MUTEX_LOCK_BUSY;

            if (ngx_atomic_cmp_set(&m->lock, old, lock)) {

                /* we locked the mutex */

                break;
            }

            old = m->lock;
        }

        if (tries++ > 1000) {

            ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                           "mutex %p is contested", m);

            /* the mutex is probably contested so we are giving up now */

            sched_yield();

            tries = 0;
            old = m->lock;
        }
    }

    ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                   "mutex %p is locked, lock:%XD", m, m->lock);

    return NGX_OK;
}


void
ngx_mutex_unlock(ngx_mutex_t *m)
{
    uint32_t       lock, old;
    struct sembuf  op;

    if (!ngx_threaded) {
        return;
    }

    old = m->lock;

    if (!(old & NGX_MUTEX_LOCK_BUSY)) {
        ngx_log_error(NGX_LOG_ALERT, m->log, 0,
                      "trying to unlock the free mutex %p", m);
        ngx_abort();
    }

    /* free the mutex */

#if 0
    ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                   "unlock mutex %p lock:%XD", m, old);
#endif

    for ( ;; ) {
        lock = old & ~NGX_MUTEX_LOCK_BUSY;

        if (ngx_atomic_cmp_set(&m->lock, old, lock)) {
            break;
        }

        old = m->lock;
    }

    if (m->semid == -1) {
        ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                       "mutex %p is unlocked", m);

        return;
    }

    /* check whether we need to wake up a waiting thread */

    old = m->lock;

    for ( ;; ) {
        if (old & NGX_MUTEX_LOCK_BUSY) {

            /* the mutex is just locked by another thread */

            break;
        }

        if (old == 0) {
            break;
        }

        /* there are the waiting threads */

        lock = old - 1;

        if (ngx_atomic_cmp_set(&m->lock, old, lock)) {

            /* wake up the thread that waits on semaphore */

            ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                           "wake up mutex %p", m);

            op.sem_num = 0;
            op.sem_op = 1;
            op.sem_flg = 0;

            if (semop(m->semid, &op, 1) == -1) {
                ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
                              "semop() failed while waking up on mutex %p", m);
                ngx_abort();
            }

            break;
        }

        old = m->lock;
    }

    ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
                   "mutex %p is unlocked", m);

    return;
}


ngx_cond_t *
ngx_cond_init(ngx_log_t *log)
{
    ngx_cond_t  *cv;

    cv = ngx_alloc(sizeof(ngx_cond_t), log);
    if (cv == NULL) {
        return NULL;
    }

    cv->signo = NGX_CV_SIGNAL;
    cv->tid = -1;
    cv->log = log;
    cv->kq = -1;

    return cv;
}


void
ngx_cond_destroy(ngx_cond_t *cv)
{
    if (close(cv->kq) == -1) {
        ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno,
                      "kqueue close() failed");
    }

    ngx_free(cv);
}


ngx_int_t
ngx_cond_wait(ngx_cond_t *cv, ngx_mutex_t *m)
{
    int              n;
    ngx_err_t        err;
    struct kevent    kev;
    struct timespec  ts;

    if (cv->kq == -1) {

        /*
         * We have to add the EVFILT_SIGNAL filter in the rfork()ed thread.
         * Otherwise the thread would not get a signal event.
         *
         * However, we have not to open the kqueue in the thread,
         * it is simply handy do it together.
         */

        cv->kq = kqueue();
        if (cv->kq == -1) {
            ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno, "kqueue() failed");
            return NGX_ERROR;
        }

        ngx_log_debug2(NGX_LOG_DEBUG_CORE, cv->log, 0,
                       "cv kq:%d signo:%d", cv->kq, cv->signo);

        kev.ident = cv->signo;
        kev.filter = EVFILT_SIGNAL;
        kev.flags = EV_ADD;
        kev.fflags = 0;
        kev.data = 0;
        kev.udata = NULL;

        ts.tv_sec = 0;
        ts.tv_nsec = 0;

        if (kevent(cv->kq, &kev, 1, NULL, 0, &ts) == -1) {
            ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno, "kevent() failed");
            return NGX_ERROR;
        }

        cv->tid = ngx_thread_self();
    }

    ngx_mutex_unlock(m);

    ngx_log_debug3(NGX_LOG_DEBUG_CORE, cv->log, 0,
                   "cv %p wait, kq:%d, signo:%d", cv, cv->kq, cv->signo);

    for ( ;; ) {
        n = kevent(cv->kq, NULL, 0, &kev, 1, NULL);

        ngx_log_debug2(NGX_LOG_DEBUG_CORE, cv->log, 0,
                       "cv %p kevent: %d", cv, n);

        if (n == -1) {
            err = ngx_errno;
            ngx_log_error((err == NGX_EINTR) ? NGX_LOG_INFO : NGX_LOG_ALERT,
                          cv->log, ngx_errno,
                          "kevent() failed while waiting condition variable %p",
                          cv);

            if (err == NGX_EINTR) {
                break;
            }

            return NGX_ERROR;
        }

        if (n == 0) {
            ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
                          "kevent() returned no events "
                          "while waiting condition variable %p",
                          cv);
            continue;
        }

        if (kev.filter != EVFILT_SIGNAL) {
            ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
                          "kevent() returned unexpected events: %d "
                          "while waiting condition variable %p",
                          kev.filter, cv);
            continue;
        }

        if (kev.ident != (uintptr_t) cv->signo) {
            ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
                          "kevent() returned unexpected signal: %d ",
                          "while waiting condition variable %p",
                          kev.ident, cv);
            continue;
        }

        break;
    }

    ngx_log_debug1(NGX_LOG_DEBUG_CORE, cv->log, 0, "cv %p is waked up", cv);

    ngx_mutex_lock(m);

    return NGX_OK;
}


ngx_int_t
ngx_cond_signal(ngx_cond_t *cv)
{
    ngx_err_t  err;

    ngx_log_debug3(NGX_LOG_DEBUG_CORE, cv->log, 0,
                   "cv %p to signal %P %d",
                   cv, cv->tid, cv->signo);

    if (cv->tid == -1) {
        return NGX_OK;
    }

    if (kill(cv->tid, cv->signo) == -1) {

        err = ngx_errno;

        ngx_log_error(NGX_LOG_ALERT, cv->log, err,
                     "kill() failed while signaling condition variable %p", cv);

        if (err == NGX_ESRCH) {
            cv->tid = -1;
        }

        return NGX_ERROR;
    }

    ngx_log_debug1(NGX_LOG_DEBUG_CORE, cv->log, 0, "cv %p is signaled", cv);

    return NGX_OK;
}