zookeeper-client-c/src/mt_adaptor.c

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef THREADED
#define THREADED
#endif

#if !defined(DLL_EXPORT) && !defined(USE_STATIC_LIB)
#  define USE_STATIC_LIB
#endif

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include "zk_adaptor.h"
#include "zookeeper_log.h"

#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#include <fcntl.h>
#include <assert.h>
#include <errno.h>

#ifndef WIN32
#include <signal.h>
#include <poll.h>
#include <unistd.h>
#include <sys/time.h>
#endif

int zoo_lock_auth(zhandle_t *zh)
{
    return pthread_mutex_lock(&zh->auth_h.lock);
}
int zoo_unlock_auth(zhandle_t *zh)
{
    return pthread_mutex_unlock(&zh->auth_h.lock);
}
int lock_buffer_list(buffer_head_t *l)
{
    return pthread_mutex_lock(&l->lock);
}
int unlock_buffer_list(buffer_head_t *l)
{
    return pthread_mutex_unlock(&l->lock);
}
int lock_completion_list(completion_head_t *l)
{
    return pthread_mutex_lock(&l->lock);
}
int unlock_completion_list(completion_head_t *l)
{
    pthread_cond_broadcast(&l->cond);
    return pthread_mutex_unlock(&l->lock);
}
struct sync_completion *alloc_sync_completion(void)
{
    struct sync_completion *sc = (struct sync_completion*)calloc(1, sizeof(struct sync_completion));
    if (sc) {
       pthread_cond_init(&sc->cond, 0);
       pthread_mutex_init(&sc->lock, 0);
    }
    return sc;
}
int wait_sync_completion(struct sync_completion *sc)
{
    pthread_mutex_lock(&sc->lock);
    while (!sc->complete) {
        pthread_cond_wait(&sc->cond, &sc->lock);
    }
    pthread_mutex_unlock(&sc->lock);
    return 0;
}

void free_sync_completion(struct sync_completion *sc)
{
    if (sc) {
        pthread_mutex_destroy(&sc->lock);
        pthread_cond_destroy(&sc->cond);
        free(sc);
    }
}

void notify_sync_completion(struct sync_completion *sc)
{
    pthread_mutex_lock(&sc->lock);
    sc->complete = 1;
    pthread_cond_broadcast(&sc->cond);
    pthread_mutex_unlock(&sc->lock);
}

int process_async(int outstanding_sync)
{
    return 0;
}

#ifdef WIN32
unsigned __stdcall do_io( void * );
unsigned __stdcall do_completion( void * );

int handle_error(zhandle_t* zh, SOCKET sock, char* message)
{
       LOG_ERROR(LOGCALLBACK(zh), "%s. %d",message, WSAGetLastError());
       closesocket (sock);
       return -1;
}

//--create socket pair for interupting selects.
int create_socket_pair(zhandle_t* zh, SOCKET fds[2])
{
    struct sockaddr_in inaddr;
    struct sockaddr addr;
    int yes=1;
    int len=0;

    SOCKET lst=socket(AF_INET, SOCK_STREAM,IPPROTO_TCP);
    if (lst ==  INVALID_SOCKET ){
       LOG_ERROR(LOGCALLBACK(zh), "Error creating socket. %d",WSAGetLastError());
       return -1;
    }
    memset(&inaddr, 0, sizeof(inaddr));
    memset(&addr, 0, sizeof(addr));
    inaddr.sin_family = AF_INET;
    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
    inaddr.sin_port = 0; //--system assigns the port

    if ( setsockopt(lst,SOL_SOCKET,SO_REUSEADDR,(char*)&yes,sizeof(yes)) == SOCKET_ERROR  ) {
       return handle_error(zh, lst,"Error trying to set socket option.");
    }
    if (bind(lst,(struct sockaddr *)&inaddr,sizeof(inaddr)) == SOCKET_ERROR){
       return handle_error(zh, lst,"Error trying to bind socket.");
    }
    if (listen(lst,1) == SOCKET_ERROR){
       return handle_error(zh, lst,"Error trying to listen on socket.");
    }
    len=sizeof(inaddr);
    getsockname(lst, &addr,&len);
    fds[0]=socket(AF_INET, SOCK_STREAM,0);
    if (connect(fds[0],&addr,len) == SOCKET_ERROR){
       return handle_error(zh, lst, "Error while connecting to socket.");
    }
    if ((fds[1]=accept(lst,0,0)) == INVALID_SOCKET){
       closesocket(fds[0]);
       return handle_error(zh, lst, "Error while accepting socket connection.");
    }
    closesocket(lst);
    return 0;
}
#else
void *do_io(void *);
void *do_completion(void *);
#endif


int wakeup_io_thread(zhandle_t *zh);

#ifdef WIN32
static int set_nonblock(SOCKET fd){
    ULONG nonblocking_flag = 1;
    if (ioctlsocket(fd, FIONBIO, &nonblocking_flag) == 0)
        return 1;
    else
        return -1;
}
#else
static int set_nonblock(int fd){
    long l = fcntl(fd, F_GETFL);
    if(l & O_NONBLOCK) return 0;
    return fcntl(fd, F_SETFL, l | O_NONBLOCK);
}
#endif

void wait_for_others(zhandle_t* zh)
{
    struct adaptor_threads* adaptor=zh->adaptor_priv;
    pthread_mutex_lock(&adaptor->lock);
    while(adaptor->threadsToWait>0)
        pthread_cond_wait(&adaptor->cond,&adaptor->lock);
    pthread_mutex_unlock(&adaptor->lock);
}

void notify_thread_ready(zhandle_t* zh)
{
    struct adaptor_threads* adaptor=zh->adaptor_priv;
    pthread_mutex_lock(&adaptor->lock);
    adaptor->threadsToWait--;
    pthread_cond_broadcast(&adaptor->cond);
    while(adaptor->threadsToWait>0)
        pthread_cond_wait(&adaptor->cond,&adaptor->lock);
    pthread_mutex_unlock(&adaptor->lock);
}


void start_threads(zhandle_t* zh)
{
    int rc = 0;
    struct adaptor_threads* adaptor=zh->adaptor_priv;
    pthread_cond_init(&adaptor->cond,0);
    pthread_mutex_init(&adaptor->lock,0);
    adaptor->threadsToWait=2;  // wait for 2 threads before opening the barrier

    // use api_prolog() to make sure zhandle doesn't get destroyed
    // while initialization is in progress
    api_prolog(zh);
    LOG_DEBUG(LOGCALLBACK(zh), "starting threads...");
    rc=pthread_create(&adaptor->io, 0, do_io, zh);
    assert("pthread_create() failed for the IO thread"&&!rc);
    rc=pthread_create(&adaptor->completion, 0, do_completion, zh);
    assert("pthread_create() failed for the completion thread"&&!rc);
    wait_for_others(zh);
    api_epilog(zh, 0);
}

int adaptor_init(zhandle_t *zh)
{
    pthread_mutexattr_t recursive_mx_attr;
    struct adaptor_threads *adaptor_threads = calloc(1, sizeof(*adaptor_threads));
    if (!adaptor_threads) {
        LOG_ERROR(LOGCALLBACK(zh), "Out of memory");
        return -1;
    }

    /* We use a pipe for interrupting select() in unix/sol and socketpair in windows. */
#ifdef WIN32
    if (create_socket_pair(zh, adaptor_threads->self_pipe) == -1){
       LOG_ERROR(LOGCALLBACK(zh), "Can't make a socket.");
#else
    if(pipe(adaptor_threads->self_pipe)==-1) {
        LOG_ERROR(LOGCALLBACK(zh), "Can't make a pipe %d",errno);
#endif
        free(adaptor_threads);
        return -1;
    }
    set_nonblock(adaptor_threads->self_pipe[1]);
    set_nonblock(adaptor_threads->self_pipe[0]);

    pthread_mutex_init(&zh->auth_h.lock,0);

    zh->adaptor_priv = adaptor_threads;
    pthread_mutex_init(&zh->to_process.lock,0);
    pthread_mutex_init(&adaptor_threads->zh_lock,0);
    pthread_mutex_init(&adaptor_threads->reconfig_lock,0);
    pthread_mutex_init(&adaptor_threads->watchers_lock,0);
    // to_send must be recursive mutex
    pthread_mutexattr_init(&recursive_mx_attr);
    pthread_mutexattr_settype(&recursive_mx_attr, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&zh->to_send.lock,&recursive_mx_attr);
    pthread_mutexattr_destroy(&recursive_mx_attr);

    pthread_mutex_init(&zh->sent_requests.lock,0);
    pthread_cond_init(&zh->sent_requests.cond,0);
    pthread_mutex_init(&zh->completions_to_process.lock,0);
    pthread_cond_init(&zh->completions_to_process.cond,0);
    start_threads(zh);
    return 0;
}

void adaptor_finish(zhandle_t *zh)
{
    struct adaptor_threads *adaptor_threads;
    // make sure zh doesn't get destroyed until after we're done here
    api_prolog(zh);
    adaptor_threads = zh->adaptor_priv;
    if(adaptor_threads==0) {
        api_epilog(zh,0);
        return;
    }

    if(!pthread_equal(adaptor_threads->io,pthread_self())){
        wakeup_io_thread(zh);
        pthread_join(adaptor_threads->io, 0);
    }else
        pthread_detach(adaptor_threads->io);

    if(!pthread_equal(adaptor_threads->completion,pthread_self())){
        pthread_mutex_lock(&zh->completions_to_process.lock);
        pthread_cond_broadcast(&zh->completions_to_process.cond);
        pthread_mutex_unlock(&zh->completions_to_process.lock);
        pthread_join(adaptor_threads->completion, 0);
    }else
        pthread_detach(adaptor_threads->completion);

    api_epilog(zh,0);
}

void adaptor_destroy(zhandle_t *zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if(adaptor==0) return;

    pthread_cond_destroy(&adaptor->cond);
    pthread_mutex_destroy(&adaptor->lock);
    pthread_mutex_destroy(&zh->to_process.lock);
    pthread_mutex_destroy(&zh->to_send.lock);
    pthread_mutex_destroy(&zh->sent_requests.lock);
    pthread_cond_destroy(&zh->sent_requests.cond);
    pthread_mutex_destroy(&zh->completions_to_process.lock);
    pthread_cond_destroy(&zh->completions_to_process.cond);
    pthread_mutex_destroy(&adaptor->zh_lock);

    pthread_mutex_destroy(&zh->auth_h.lock);

    close(adaptor->self_pipe[0]);
    close(adaptor->self_pipe[1]);
    free(adaptor);
    zh->adaptor_priv=0;
}

int wakeup_io_thread(zhandle_t *zh)
{
    struct adaptor_threads *adaptor_threads = zh->adaptor_priv;
    char c=0;
#ifndef WIN32
    return write(adaptor_threads->self_pipe[1],&c,1)==1? ZOK: ZSYSTEMERROR;
#else
    return send(adaptor_threads->self_pipe[1], &c, 1, 0)==1? ZOK: ZSYSTEMERROR;
#endif
}

int adaptor_send_queue(zhandle_t *zh, int timeout)
{
    if(!zh->close_requested)
        return wakeup_io_thread(zh);
    // don't rely on the IO thread to send the messages if the app has
    // requested to close
    return flush_send_queue(zh, timeout);
}

/* These two are declared here because we will run the event loop
 * and not the client */
#ifdef WIN32
int zookeeper_interest(zhandle_t *zh, SOCKET *fd, int *interest,
        struct timeval *tv);
#else
int zookeeper_interest(zhandle_t *zh, int *fd, int *interest,
        struct timeval *tv);
#endif
int zookeeper_process(zhandle_t *zh, int events);

#ifdef WIN32
unsigned __stdcall do_io( void * v)
#else
void *do_io(void *v)
#endif
{
    zhandle_t *zh = (zhandle_t*)v;
#ifndef WIN32
    struct pollfd fds[2];
    struct adaptor_threads *adaptor_threads = zh->adaptor_priv;

    api_prolog(zh);
    notify_thread_ready(zh);
    LOG_DEBUG(LOGCALLBACK(zh), "started IO thread");
    fds[0].fd=adaptor_threads->self_pipe[0];
    fds[0].events=POLLIN;
    while(!zh->close_requested) {
        struct timeval tv;
        int fd;
        int interest;
        int timeout;
        int maxfd=1;

        zh->io_count++;

        zookeeper_interest(zh, &fd, &interest, &tv);
        if (fd != -1) {
            fds[1].fd=fd;
            fds[1].events=(interest&ZOOKEEPER_READ)?POLLIN:0;
            fds[1].events|=(interest&ZOOKEEPER_WRITE)?POLLOUT:0;
            maxfd=2;
        }
        timeout=tv.tv_sec * 1000 + (tv.tv_usec/1000);

        poll(fds,maxfd,timeout);
        if (fd != -1) {
            interest=(fds[1].revents&POLLIN)?ZOOKEEPER_READ:0;
            interest|=((fds[1].revents&POLLOUT)||(fds[1].revents&POLLHUP))?ZOOKEEPER_WRITE:0;
        }
        if(fds[0].revents&POLLIN){
            // flush the pipe
            char b[128];
            while(read(adaptor_threads->self_pipe[0],b,sizeof(b))==sizeof(b)){}
        }
#else
    fd_set rfds, wfds;
    struct adaptor_threads *adaptor_threads = zh->adaptor_priv;
    api_prolog(zh);
    notify_thread_ready(zh);
    LOG_DEBUG(LOGCALLBACK(zh), "started IO thread");

    while(!zh->close_requested) {
        struct timeval tv;
        SOCKET fd;
        int interest;
        int rc;

        zookeeper_interest(zh, &fd, &interest, &tv);

        // FD_ZERO is cheap on Win32, it just sets count of elements to zero.
        // It needs to be done to ensure no stale entries.
        FD_ZERO(&rfds);
        FD_ZERO(&wfds);

        if (fd != -1) {
            if (interest&ZOOKEEPER_READ) {
                FD_SET(fd, &rfds);
            }

            if (interest&ZOOKEEPER_WRITE) {
                FD_SET(fd, &wfds);
            }
        }

        // Always interested in self_pipe.
        FD_SET(adaptor_threads->self_pipe[0], &rfds);

        rc = select(/* unused */0, &rfds, &wfds, NULL, &tv);
        if (rc > 0) {
            interest=(FD_ISSET(fd, &rfds))? ZOOKEEPER_READ: 0;
            interest|=(FD_ISSET(fd, &wfds))? ZOOKEEPER_WRITE: 0;

            if (FD_ISSET(adaptor_threads->self_pipe[0], &rfds)){
                // flush the pipe/socket
                char b[128];
                while(recv(adaptor_threads->self_pipe[0],b,sizeof(b), 0)==sizeof(b)){}
            }
        }
        else if (rc < 0) {
            LOG_ERROR(LOGCALLBACK(zh), ("select() failed %d [%d].", rc, WSAGetLastError()));

            // Clear interest events for zookeeper_process if select() fails.
            interest = 0;
        }

#endif
        // dispatch zookeeper events
        zookeeper_process(zh, interest);
        // check the current state of the zhandle and terminate
        // if it is_unrecoverable()
        if(is_unrecoverable(zh))
            break;
    }
    api_epilog(zh, 0);
    LOG_DEBUG(LOGCALLBACK(zh), "IO thread terminated");
    return 0;
}

#ifdef WIN32
unsigned __stdcall do_completion( void * v)
#else
void *do_completion(void *v)
#endif
{
    zhandle_t *zh = v;
    api_prolog(zh);
    notify_thread_ready(zh);
    LOG_DEBUG(LOGCALLBACK(zh), "started completion thread");
    while(!zh->close_requested) {
        pthread_mutex_lock(&zh->completions_to_process.lock);
        while(!zh->completions_to_process.head && !zh->close_requested) {
            pthread_cond_wait(&zh->completions_to_process.cond, &zh->completions_to_process.lock);
        }
        pthread_mutex_unlock(&zh->completions_to_process.lock);
        process_completions(zh);
    }
    api_epilog(zh, 0);
    LOG_DEBUG(LOGCALLBACK(zh), "completion thread terminated");
    return 0;
}

int32_t inc_ref_counter(zhandle_t* zh,int i)
{
    int incr=(i<0?-1:(i>0?1:0));
    // fetch_and_add implements atomic post-increment
    int v=fetch_and_add(&zh->ref_counter,incr);
    // inc_ref_counter wants pre-increment
    v+=incr;   // simulate pre-increment
    return v;
}

int32_t fetch_and_add(volatile int32_t* operand, int incr)
{
#ifndef WIN32
    return __sync_fetch_and_add(operand, incr);
#else
    return InterlockedExchangeAdd(operand, incr);
#endif
}

// make sure the static xid is initialized before any threads started
__attribute__((constructor)) int32_t get_xid()
{
    static int32_t xid = -1;
    if (xid == -1) {
        xid = time(0);
    }
    return fetch_and_add(&xid,1);
}

int lock_reconfig(struct _zhandle *zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if (adaptor) {
        return pthread_mutex_lock(&adaptor->reconfig_lock);
    } else {
        return 0;
    }
}
int unlock_reconfig(struct _zhandle *zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if (adaptor) {
        return pthread_mutex_unlock(&adaptor->reconfig_lock);
    } else {
        return 0;
    }
}

int lock_watchers(struct _zhandle *zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if (adaptor) {
        return pthread_mutex_lock(&adaptor->watchers_lock);
    } else {
        return 0;
    }
}
int unlock_watchers(struct _zhandle *zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if (adaptor) {
        return pthread_mutex_unlock(&adaptor->watchers_lock);
    } else {
        return 0;
    }
}

int enter_critical(zhandle_t* zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if (adaptor) {
        return pthread_mutex_lock(&adaptor->zh_lock);
    } else {
        return 0;
    }
}

int leave_critical(zhandle_t* zh)
{
    struct adaptor_threads *adaptor = zh->adaptor_priv;
    if (adaptor) {
        return pthread_mutex_unlock(&adaptor->zh_lock);
    } else {
        return 0;
    }
}