gwthread-pthread.c

The Kannel Open Source WAP and SMS gateway works as both an SMS gateway, for implementing keyword b

    }
    return 0;
}

static void restore_user_signals(sigset_t *old_set)
{
    int ret;

    ret = pthread_sigmask(SIG_SETMASK, old_set, NULL);
    if (ret != 0) {
        panic(ret, "gwthread-pthread: Couldn't restore signal set.");
    }
}


static long spawn_thread(gwthread_func_t *func, const char *name, void *arg)
{
    int ret;
    pthread_t id;
    struct new_thread_args *p = NULL;
    long new_thread_id;

    /* We want to pass both these arguments to our wrapper function
     * new_thread, but the pthread_create interface will only let
     * us pass one pointer.  So we wrap them in a little struct. */
    p = gw_malloc(sizeof(*p));
    p->func = func;
    p->arg = arg;
    p->ti = gw_malloc(sizeof(*(p->ti)));

    /* Lock the thread table here, so that new_thread can block
     * on that lock.  That way, the new thread won't start until
     * we have entered it in the thread table. */
    lock();

    if (active_threads >= THREADTABLE_SIZE) {
        unlock();
        warning(0, "Too many threads, could not create new thread.");
        gw_free(p);
        return -1;
    }

    ret = pthread_create(&id, NULL, &new_thread, p);
    if (ret != 0) {
        unlock();
        error(ret, "Could not create new thread.");
        gw_free(p);
        return -1;
    }
    ret = pthread_detach(id);
    if (ret != 0) {
        warning(ret, "Could not detach new thread.");
    }

    new_thread_id = fill_threadinfo(id, name, func, p->ti);
    unlock();
    
    debug("gwlib.gwthread", 0, "Started thread %ld (%s)", new_thread_id, name);

    return new_thread_id;
}

long gwthread_create_real(gwthread_func_t *func, const char *name, void *arg)
{
    int sigtrick = 0;
    sigset_t old_signal_set;
    long thread_id;

    /*
     * We want to make sure that only the main thread handles signals,
     * so that each signal is handled exactly once.  To do this, we
     * make sure that each new thread has all the signals that we
     * handle blocked.  To avoid race conditions, we block them in 
     * the spawning thread first, then create the new thread (which
     * inherits the settings), and then restore the old settings in
     * the spawning thread.  This means that there is a brief period
     * when no signals will be processed, but during that time they
     * should be queued by the operating system.
     */
    if (gwthread_self() == MAIN_THREAD_ID)
	    sigtrick = block_user_signals(&old_signal_set) == 0;

    thread_id = spawn_thread(func, name, arg);

    /*
     * Restore the old signal mask.  The new thread will have
     * inherited the resticted one, but the main thread needs
     * the old one back.
     */
    if (sigtrick)
 	    restore_user_signals(&old_signal_set);
    
    return thread_id;
}

void gwthread_join(long thread)
{
    struct threadinfo *threadinfo;
    pthread_cond_t exit_cond;
    int ret;

    gw_assert(thread >= 0);

    lock();
    threadinfo = THREAD(thread);
    if (threadinfo == NULL || threadinfo->number != thread) {
        /* The other thread has already exited */
        unlock();
        return;
    }

    /* Register our desire to be alerted when that thread exits,
     * and wait for it. */

    ret = pthread_cond_init(&exit_cond, NULL);
    if (ret != 0) {
        warning(ret, "gwthread_join: cannot create condition variable.");
        unlock();
        return;
    }

    if (!threadinfo->joiners)
        threadinfo->joiners = list_create();
    list_append(threadinfo->joiners, &exit_cond);

    /* The wait immediately releases the lock, and reacquires it
     * when the condition is satisfied.  So don't worry, we're not
     * blocking while keeping the table locked. */
    ret = pthread_cond_wait(&exit_cond, &threadtable_lock);
    unlock();

    if (ret != 0)
        warning(ret, "gwthread_join: error in pthread_cond_wait");

    pthread_cond_destroy(&exit_cond);
}

void gwthread_join_all(void)
{
    long i;
    long our_thread = gwthread_self();

    for (i = 0; i < THREADTABLE_SIZE; ++i) {
        if (THREAD(our_thread) != THREAD(i))
            gwthread_join(i);
    }
}

void gwthread_wakeup_all(void)
{
    long i;
    long our_thread = gwthread_self();

    for (i = 0; i < THREADTABLE_SIZE; ++i) {
        if (THREAD(our_thread) != THREAD(i))
            gwthread_wakeup(i);
    }
}

void gwthread_join_every(gwthread_func_t *func)
{
    struct threadinfo *ti;
    pthread_cond_t exit_cond;
    int ret;
    long i;

    ret = pthread_cond_init(&exit_cond, NULL);
    if (ret != 0) {
        warning(ret, "gwthread_join_every: cannot create condition variable.");
        unlock();
        return;
    }

    /*
     * FIXME: To be really safe, this function should keep looping
     * over the table until it does a complete run without having
     * to call pthread_cond_wait.  Otherwise, new threads could
     * start while we wait, and we'll miss them.
     */
    lock();
    for (i = 0; i < THREADTABLE_SIZE; ++i) {
        ti = THREAD(i);
        if (ti == NULL || ti->func != func)
            continue;
        debug("gwlib.gwthread", 0,
              "Waiting for %ld (%s) to terminate",
              ti->number, ti->name);
        if (!ti->joiners)
            ti->joiners = list_create();
        list_append(ti->joiners, &exit_cond);
        ret = pthread_cond_wait(&exit_cond, &threadtable_lock);
        if (ret != 0)
            warning(ret, "gwthread_join_all: error in pthread_cond_wait");
    }
    unlock();

    pthread_cond_destroy(&exit_cond);
}

/* Return the thread id of this thread. */
long gwthread_self(void)
{
    struct threadinfo *threadinfo;
    threadinfo = pthread_getspecific(tsd_key);
    if (threadinfo) 
        return threadinfo->number;
    else
        return -1;
}

/* Return the thread pid of this thread. */
long gwthread_self_pid(void)
{
    struct threadinfo *threadinfo;
    threadinfo = pthread_getspecific(tsd_key);
    if (threadinfo && threadinfo->pid != -1) 
        return (long) threadinfo->pid;
    else
        return (long) getpid();
}

void gwthread_self_ids(long *tid, long *pid)
{
    struct threadinfo *threadinfo;
    threadinfo = pthread_getspecific(tsd_key);
    if (threadinfo) {
        *tid = threadinfo->number;
        *pid = (threadinfo->pid != -1) ? threadinfo->pid : getpid();
    } else {
        *tid = -1;
        *pid = getpid();
    }
}

void gwthread_wakeup(long thread)
{
    unsigned char c = 0;
    struct threadinfo *threadinfo;
    int fd;

    gw_assert(thread >= 0);

    lock();

    threadinfo = THREAD(thread);
    if (threadinfo == NULL || threadinfo->number != thread) {
        unlock();
        return;
    }

    fd = threadinfo->wakefd_send;
    unlock();

    write(fd, &c, 1);
}

int gwthread_pollfd(int fd, int events, double timeout)
{
    struct pollfd pollfd[2];
    struct threadinfo *threadinfo;
    int milliseconds;
    int ret;

    threadinfo = getthreadinfo();

    pollfd[0].fd = threadinfo->wakefd_recv;
    pollfd[0].events = POLLIN;
    pollfd[0].revents = 0;

    pollfd[1].fd = fd;
    pollfd[1].events = events;
    pollfd[1].revents = 0;

    milliseconds = timeout * 1000;
    if (milliseconds < 0)
        milliseconds = POLL_NOTIMEOUT;

    ret = poll(pollfd, 2, milliseconds);
    if (ret < 0) {
        if (errno != EINTR)
            error(errno, "gwthread_pollfd: error in poll");
        return -1;
    }

    if (pollfd[0].revents)
        flushpipe(pollfd[0].fd);

    return pollfd[1].revents;
}

int gwthread_poll(struct pollfd *fds, long numfds, double timeout)
{
    struct pollfd *pollfds;
    struct threadinfo *threadinfo;
    int milliseconds;
    int ret;

    threadinfo = getthreadinfo();

    /* Create a new pollfd array with an extra element for the
     * thread wakeup fd. */

    pollfds = gw_malloc((numfds + 1) * sizeof(*pollfds));
    pollfds[0].fd = threadinfo->wakefd_recv;
    pollfds[0].events = POLLIN;
    memcpy(pollfds + 1, fds, numfds * sizeof(*pollfds));

    milliseconds = timeout * 1000;
    if (milliseconds < 0)
        milliseconds = POLL_NOTIMEOUT;

    ret = poll(pollfds, numfds + 1, milliseconds);
    if (ret < 0) {
        if (errno != EINTR)
            error(errno, "gwthread_poll: error in poll");
        gw_free(pollfds);
        return -1;
    }
    if (pollfds[0].revents)
        flushpipe(pollfds[0].fd);

    /* Copy the results back to the caller */
    memcpy(fds, pollfds + 1, numfds * sizeof(*pollfds));
    gw_free(pollfds);

    return ret;
}


void gwthread_sleep(double seconds)
{
    struct pollfd pollfd;
    struct threadinfo *threadinfo;
    int milliseconds;
    int ret;

    threadinfo = getthreadinfo();

    pollfd.fd = threadinfo->wakefd_recv;
    pollfd.events = POLLIN;

    milliseconds = seconds * 1000;
    if (milliseconds < 0)
        milliseconds = POLL_NOTIMEOUT;

    ret = poll(&pollfd, 1, milliseconds);
    if (ret < 0) {
        if (errno != EINTR && errno != EAGAIN) {
            warning(errno, "gwthread_sleep: error in poll");
        }
    }
    if (ret == 1) {
        flushpipe(pollfd.fd);
    }
}


#ifndef BROKEN_PTHREADS

/* Working pthreads */
int gwthread_shouldhandlesignal(int signal){
    return 1;
}
#else

/* Somewhat broken pthreads */ 
int gwthread_shouldhandlesignal(int signal){
    return (gwthread_self() == MAIN_THREAD_ID);
}
#endif

int gwthread_dumpsigmask(void) {
    sigset_t signal_set;
    int signum;

    /* Grab the signal set data from our thread */
    if (pthread_sigmask(SIG_BLOCK, NULL, &signal_set) != 0) {
	    warning(0, "gwthread_dumpsigmask: Couldn't get signal mask.");
	    return -1;
    }
    
    /* For each signal normally defined (there are usually only 32),
     * print a message if we don't block it. */
    for (signum = 1; signum <= 32; signum++) {
	     if (!sigismember(&signal_set, signum)) {
	         debug("gwlib", 0,
		     "gwthread_dumpsigmask: Signal Number %d will be caught.", 
		     signum);
	     }
    }
    return 0;
}

/* DARWIN alias MacOS X doesnt have pthread_sigmask in its pthreads implementation */

#if defined(DARWIN_OLD)
static int pthread_sigmask()
{
    return 0;
}
#endif