libevent 结合 pthread-win32 在windows 平台上的使用

libevent 结合 pthread-win32 在windows 平台上的使用

libevent 在windows 平台上的使用，大致过程是：

1）WSAStartup

2）evthread_use_windows_threads

3）event_base_new

4）WSACleanup

其中，1）WSAStartup一定要调用，否则3）调用不成功。如果windows的程序使用了跨平台的 pthread 库，那么就要自定义

定制属于自己的多线程, 锁, 条件变量等，具体就是调用：evthread_set_lock_callbacks。

我将这个过程以头文件 evthread_pthread_w32.h 提供给大家，在我的机器上可以编译，具体效果没测试。

/**
 * @filename   evthread_pthread_w32.h
 *  pthread-win32 for libevent event.
 *
 * @refer
 *   http://libevent.org/
 * NOTES:
 *   How to build libevent for Windows x86 and x64
 *
 * @author     Liang Zhang <[email protected]>
 * @version    0.0.10
 * @create     2019-11-24 12:00:00
 * @update     2019-11-24 11:42:35
 */
#ifndef EVTHREAD_PTHREAD_W32_H_
#define EVTHREAD_PTHREAD_W32_H_

#if defined(__cplusplus)
extern "C" {
#endif

/**
 * pthread-w32
 */
#include <sched.h>
#include <pthread.h>

/**
 * Only one event_base_loop can run on each event_base at once.
 * Libevent can also be used for multi-threaded applications,
 *  either by isolating each event_base so that only a single
 *  thread accesses it, or by locked access to a single shared
 * event_base.
 *
 * link to: libevent.a
 */

#include <event2/event.h>


/**
 *  evthread_use_pthreads()
 *
 *  link to: libevent_pthreads.a
 */
#include <event2/thread.h>


#if defined(_WIN32) && defined(PTW32_VERSION)
    static pthread_mutexattr_t attr_recursive;

    static void * evthread_posix_lock_alloc(unsigned locktype)
    {
        pthread_mutexattr_t *attr = NULL;
        pthread_mutex_t *lock = calloc(1, sizeof(pthread_mutex_t));
        if (!lock) {
            return NULL;
        }
        if (locktype & EVTHREAD_LOCKTYPE_RECURSIVE) {
            attr = &attr_recursive;
        }
        if (pthread_mutex_init(lock, attr)) {
            free(lock);
            return NULL;
        }
        return lock;
    }

    static void evthread_posix_lock_free(void *lock_, unsigned locktype)
    {
        pthread_mutex_t *lock = lock_;
        pthread_mutex_destroy(lock);
        free(lock);
    }

    static int evthread_posix_lock(unsigned mode, void *lock_)
    {
        pthread_mutex_t *lock = lock_;
        if (mode & EVTHREAD_TRY) {
            return pthread_mutex_trylock(lock);
        }
        else {
            return pthread_mutex_lock(lock);
        }
    }

    static int evthread_posix_unlock(unsigned mode, void *lock_)
    {
        pthread_mutex_t *lock = lock_;
        return pthread_mutex_unlock(lock);
    }

    static unsigned long evthread_posix_get_id(void)
    {
        union {
            pthread_t thr;
    # if EVENT__SIZEOF_PTHREAD_T > EVENT__SIZEOF_LONG
            ev_uint64_t id;
    # else
            unsigned long id;
    # endif
        } r;

    # if EVENT__SIZEOF_PTHREAD_T < EVENT__SIZEOF_LONG
        memset(&r, 0, sizeof(r));
    # endif
        r.thr = pthread_self();
        return (unsigned long)r.id;
    }

    static void * evthread_posix_cond_alloc(unsigned condflags)
    {
        pthread_cond_t *cond = calloc(1, sizeof(pthread_cond_t));
        if (!cond) {
            return NULL;
        }
        if (pthread_cond_init(cond, NULL)) {
            free(cond);
            return NULL;
        }
        return cond;
    }

    static void evthread_posix_cond_free(void *cond_)
    {
        pthread_cond_t *cond = cond_;
        pthread_cond_destroy(cond);
        free(cond);
    }

    static int evthread_posix_cond_signal(void *cond_, int broadcast)
    {
        pthread_cond_t *cond = cond_;
        int r;
        if (broadcast) {
            r = pthread_cond_broadcast(cond);
        }
        else {
            r = pthread_cond_signal(cond);
        }
        return r ? -1 : 0;
    }

    static int evthread_posix_cond_wait(void *cond_, void *lock_, const struct timeval *tv)
    {
        int r;
        pthread_cond_t *cond = cond_;
        pthread_mutex_t *lock = lock_;

        if (tv) {
            struct timeval now, abstime;
            struct timespec ts;
            evutil_gettimeofday(&now, NULL);
            evutil_timeradd(&now, tv, &abstime);
            ts.tv_sec = abstime.tv_sec;
            ts.tv_nsec = abstime.tv_usec * 1000;
            r = pthread_cond_timedwait(cond, lock, &ts);

            if (r == ETIMEDOUT) {
                return 1;
            }
            else if (r) {
                return -1;
            }
            else {
                return 0;
            }
        }
        else {
            r = pthread_cond_wait(cond, lock);
            return (r ? -1 : 0);
        }
    }

    static int windows_socket_startup (unsigned char majorVer, unsigned char minorVer)
    {
        /* for Windows must call WSAStartup first */
        int err;

        WSADATA wsaData;
        WORD wVersionRequested = MAKEWORD(majorVer, minorVer);

        err = WSAStartup(wVersionRequested, &wsaData);
        if (err) {
            switch (err) {
            case WSASYSNOTREADY:
                printf("WSAStartup error: The underlying network subsystem is not ready for network communication.\n");
                break;

            case WSAVERNOTSUPPORTED:
                printf("WSAStartup error: The version of Windows Sockets support requested is not provided by this particular Windows Sockets implementation.\n");
                break;

            case WSAEINPROGRESS:
                printf("WSAStartup error: A blocking Windows Sockets 1.1 operation is in progress.\n");
                break;

            case WSAEPROCLIM:
                printf("WSAStartup error: A limit on the number of tasks supported by the Windows Sockets implementation has been reached.\n");
                break;

            case WSAEFAULT:
                printf("WSAStartup error: The lpWSAData parameter is not a valid pointer.\n");
                break;
            }

            return (-1);
        }

        if (LOBYTE(wsaData.wVersion) != majorVer || HIBYTE(wsaData.wVersion) != minorVer) {
            /* Tell the user that we could not find a usable WinSock DLL. */
            printf("WSAStartup error: Could not find a usable WinSock DLL with version %d.%d.\n", majorVer, minorVer);
            WSACleanup();

            return (-1);
        }

        /* success */
        return 0;
    }


    static int evthread_use_pthreads (void)
    {
        struct evthread_lock_callbacks cbs = {
            EVTHREAD_LOCK_API_VERSION,
            EVTHREAD_LOCKTYPE_RECURSIVE,
            evthread_posix_lock_alloc,
            evthread_posix_lock_free,
            evthread_posix_lock,
            evthread_posix_unlock
        };

        struct evthread_condition_callbacks cond_cbs = {
            EVTHREAD_CONDITION_API_VERSION,
            evthread_posix_cond_alloc,
            evthread_posix_cond_free,
            evthread_posix_cond_signal,
            evthread_posix_cond_wait
        };

        int err = windows_socket_startup(2, 2);

        if (err) {
            err = windows_socket_startup(2, 1);
        }
        if (err) {
            err = windows_socket_startup(2, 0);
        }
        if (err) {
            err = windows_socket_startup(1, 1);
        }
        if (err) {
            err = windows_socket_startup(1, 0);
        }
        if (err) {
            return -1;
        }

        /* Set ourselves up to get recursive locks. */
        if (pthread_mutexattr_init(&attr_recursive)) {
            WSACleanup();
            printf("pthread_mutexattr_init() error.\n");
            return -1;
        }

        if (pthread_mutexattr_settype(&attr_recursive, PTHREAD_MUTEX_RECURSIVE)) {
            WSACleanup();
            printf("pthread_mutexattr_settype() error.\n");
            return -1;
        }

        evthread_set_lock_callbacks(&cbs);
        evthread_set_condition_callbacks(&cond_cbs);
        evthread_set_id_callback(evthread_posix_get_id);

        return 0;
    }

    static int evthread_use_win32_threads(void)
    {
        int err = windows_socket_startup(2, 2);

        if (err) {
            err = windows_socket_startup(2, 1);
        }
        if (err) {
            err = windows_socket_startup(2, 0);
        }
        if (err) {
            err = windows_socket_startup(1, 1);
        }
        if (err) {
            err = windows_socket_startup(1, 0);
        }
        if (err) {
            return -1;
        }

        err = evthread_use_windows_threads();
        if (err) {
            printf("evthread_use_windows_threads() error.\n");
            WSACleanup();
        }

        return err;
    }

    static void windows_socket_cleanup()
    {
        WSACleanup();
    }

    # define pthread_check_valid(pthrd)       (pthrd.p != 0)
    # define pthread_set_invalid(pthrd)       pthrd.p = 0; pthrd.x = 0
#else
    # define pthread_check_valid(pthrd)       (pthrd != (pthread_t)(-1))
    # define pthread_set_invalid(pthrd)       pthrd = (pthread_t)(-1)

    #define windows_socket_cleanup()          do{}while(0)
#endif

#if defined(__cplusplus)
}
#endif

#endif /* EVTHREAD_PTHREAD_W32_H_ */

用法很简单：

1）evthread_use_pthreads 或 evthread_use_win32_threads

2）event_base_new

3）windows_socket_cleanup