Windows和Linux下通用的线程接口

对于多线程开发，Linux下有pthread线程库，使用起来比较方便，而Windows没有，对于涉及到多线程的跨平台代码开发，会带来不便。这里参考网络上的一些文章，整理了在Windows和Linux下通用的线程接口。经过测试，在Windows和Linux下均可正常使用。
Windows和Linux下对应的线程函数列表如下：


Windows和Linux线程通用接口如下：
#ifndef FBC_MESSY_TEST_THREAD_HPP_
#define FBC_MESSY_TEST_THREAD_HPP_

/* reference: http://locklessinc.com/articles/pthreads_on_windows/ http://locklessinc.com/downloads/winpthreads.h http://www.jmhartsoftware.com/Pthreads_Emulation.html https://sourceforge.net/p/log4c/log4c/ci/a4d3e19b3e55c1d5d66a5dc09a2603d9dcfcff52/tree/src/sd/sd_xplatform.h https://git.libav.org/?p=libav.git;a=blob;f=compat/w32pthreads.h;h=2fe2a5ab979ff676d1e7dba4a360306dde29a0f0;hb=HEAD https://github.com/krux/zookeeper-pkg/blob/master/src/c/src/winport.c https://my.oschina.net/mjRao/blog/359870 */

#ifdef _MSC_VER

#include <windows.h>
#include <sys/timeb.h>
#include <process.h>

typedef HANDLE pthread_mutex_t;
typedef int pthread_condattr_t;
typedef HANDLE pthread_t;
typedef DWORD pthread_attr_t;

#define pthread_create(thrp, attr, func, arg) \
(((*(thrp) = CreateThread(NULL, 0, \
(LPTHREAD_START_ROUTINE)(func), (arg), 0, NULL)) == NULL) ? -1 : 0)
#define pthread_join(thr, statusp) \
((WaitForSingleObject((thr), INFINITE) == WAIT_OBJECT_0) && \
((statusp == NULL) ? 0 : \
(GetExitCodeThread((thr), (LPDWORD)(statusp)) ? 0 : -1)))

#define PTHREAD_MUTEX_INITIALIZER {(void*)-1,-1,0,0,0,0}
#define pthread_mutex_lock(pobject) WaitForSingleObject(*pobject,INFINITE)
#define pthread_mutex_unlock(pobject) ReleaseMutex(*pobject)
#define pthread_mutex_init(pobject,pattr) (*pobject=CreateMutex(NULL,FALSE,NULL))
#define pthread_mutex_destroy(pobject) CloseHandle(*pobject)

/* Windows doesn't have this, so declare it ourselves. */
typedef struct timespec {
/* long long in windows is the same as long in unix for 64bit */
long long tv_sec;
long long tv_nsec;
} timespec;

typedef struct {
int waiters_count_;
// Number of waiting threads.

CRITICAL_SECTION waiters_count_lock_;
// Serialize access to <waiters_count_>.

HANDLE sema_;
// Semaphore used to queue up threads waiting for the condition to
// become signaled.

HANDLE waiters_done_;
// An auto-reset event used by the broadcast/signal thread to wait
// for all the waiting thread(s) to wake up and be released from the
// semaphore.

size_t was_broadcast_;
// Keeps track of whether we were broadcasting or signaling. This
// allows us to optimize the code if we're just signaling.
} pthread_cond_t;

static unsigned long long _pthread_time_in_ms(void)
{
struct __timeb64 tb;

_ftime64(&tb);

return tb.time * 1000 + tb.millitm;
}

static unsigned long long _pthread_time_in_ms_from_timespec(const struct timespec *ts)
{
unsigned long long t = ts->tv_sec * 1000;
t += ts->tv_nsec / 1000000;

return t;
}

static unsigned long long _pthread_rel_time_in_ms(const struct timespec *ts)
{
unsigned long long t1 = _pthread_time_in_ms_from_timespec(ts);
unsigned long long t2 = _pthread_time_in_ms();

/* Prevent underflow */
if (t1 < t2) return 0;
return t1 - t2;
}

static int pthread_cond_init(pthread_cond_t *cv, const pthread_condattr_t * ignore)
{
cv->waiters_count_ = 0;
cv->was_broadcast_ = 0;
cv->sema_ = CreateSemaphore(NULL, // no security
0, // initially 0
0x7fffffff, // max count
NULL); // unnamed
if (cv->sema_ == NULL)
return GetLastError();

InitializeCriticalSection(&cv->waiters_count_lock_);
cv->waiters_done_ = CreateEvent(NULL, // no security
FALSE, // auto-reset
FALSE, // non-signaled initially
NULL); // unnamed
return (cv->waiters_done_ == NULL) ? GetLastError() : 0;
}

static int pthread_cond_destroy(pthread_cond_t *cond)
{
CloseHandle(cond->sema_);
DeleteCriticalSection(&cond->waiters_count_lock_);
return (CloseHandle(cond->waiters_done_) == 0) ? GetLastError() : 0;
}

static int pthread_cond_signal(pthread_cond_t *cv)
{
int have_waiters;
EnterCriticalSection(&(cv->waiters_count_lock_));
have_waiters = cv->waiters_count_ > 0;
LeaveCriticalSection(&cv->waiters_count_lock_);

// If there aren't any waiters, then this is a no-op.
if (have_waiters){
return (ReleaseSemaphore(cv->sema_, 1, 0) == 0) ? GetLastError() : 0;
}
else
return 0;
}

static int pthread_cond_broadcast(pthread_cond_t *cv)
{
// This is needed to ensure that <waiters_count_> and <was_broadcast_> are
// consistent relative to each other.
int have_waiters = 0;
EnterCriticalSection(&cv->waiters_count_lock_);

if (cv->waiters_count_ > 0) {
// We are broadcasting, even if there is just one waiter...
// Record that we are broadcasting, which helps optimize
// <pthread_cond_wait> for the non-broadcast case.
cv->was_broadcast_ = 1;
have_waiters = 1;
}

if (have_waiters) {
// Wake up all the waiters atomically.
ReleaseSemaphore(cv->sema_, cv->waiters_count_, 0);

LeaveCriticalSection(&cv->waiters_count_lock_);

// Wait for all the awakened threads to acquire the counting
// semaphore.
WaitForSingleObject(cv->waiters_done_, INFINITE);
// This assignment is okay, even without the <waiters_count_lock_> held
// because no other waiter threads can wake up to access it.
cv->was_broadcast_ = 0;
}
else
LeaveCriticalSection(&cv->waiters_count_lock_);

return 0;
}

static int pthread_cond_wait(pthread_cond_t *cv, pthread_mutex_t *external_mutex)
{
int last_waiter;
// Avoid race conditions.
EnterCriticalSection(&cv->waiters_count_lock_);
cv->waiters_count_++;
LeaveCriticalSection(&cv->waiters_count_lock_);

// This call atomically releases the mutex and waits on the
// semaphore until <pthread_cond_signal> or <pthread_cond_broadcast>
// are called by another thread.
SignalObjectAndWait(*external_mutex, cv->sema_, INFINITE, FALSE);

// Reacquire lock to avoid race conditions.
EnterCriticalSection(&cv->waiters_count_lock_);

// We're no longer waiting...
cv->waiters_count_--;

// Check to see if we're the last waiter after <pthread_cond_broadcast>.
last_waiter = cv->was_broadcast_ && cv->waiters_count_ == 0;

LeaveCriticalSection(&cv->waiters_count_lock_);

// If we're the last waiter thread during this particular broadcast
// then let all the other threads proceed.
if (last_waiter)
// This call atomically signals the <waiters_done_> event and waits until
// it can acquire the <external_mutex>. This is required to ensure fairness.
SignalObjectAndWait(cv->waiters_done_, *external_mutex, INFINITE, FALSE);
else
// Always regain the external mutex since that's the guarantee we
// give to our callers.
WaitForSingleObject(*external_mutex, INFINITE);

return 0;
}

#else // linux

#include <pthread.h>

#endif // _MSC_VER

#endif // FBC_MESSY_TEST_THREAD_HPP_
测试代码如下：
#include "test_thread.hpp"
#include <iostream>
#include "thread.hpp"

// test code reference: http://blog.csdn.net/fengbingchun/article/details/48579725
void* run1(void* para)
{
std::cout << "start new thread!" << std::endl;

//sleep(5);//suspend 5 s，在正式的代码中，一般不要用sleep函数
int* iptr = (int*)((void**)para)[0];
float* fptr = (float*)((void**)para)[1];
char* str = (char*)((void**)para)[2];
std::cout << *iptr << " " << *fptr << " " << str << std::endl;

std::cout << "end new thread!" << std::endl;

return ((void *)0);
}

int test_create_thread()
{
pthread_t pid;//thread handle
int ival = 1;
float fval = 10.0f;
char buf[] = "func";
void* para[3] = { &ival, &fval, buf };

pthread_create(&pid, NULL, run1, para);

// 新线程创建之后主线程如何运行----主线程按顺序继续执行下一行程序
std::cout << "main thread!" << std::endl;

// 新线程结束时如何处理----新线程先停止，然后作为其清理过程的一部分，等待与另一个线程合并或“连接”
pthread_join(pid, NULL);

return 0;
}

/////////////////////////////////////////////////////////
pthread_t tid[2];
int counter = 0;
pthread_mutex_t lock;

void* run2(void* arg)
{
pthread_mutex_lock(&lock);

unsigned long i = 0;
counter += 1;
std::cout << "Job " << counter << " started!" << std::endl;
for (i = 0; i<(0xFFFFFFFF); i++);
std::cout << "Job " << counter << " finished!" << std::endl;

pthread_mutex_unlock(&lock);

return NULL;
}

int test_thread_mutex()
{
int i = 0, err = -1;

pthread_mutex_init(&lock, NULL);

while (i < 2) {
pthread_create(&(tid[i]), NULL, &run2, NULL);
i++;
}

pthread_join(tid[0], NULL);
pthread_join(tid[1], NULL);
pthread_mutex_destroy(&lock);

return 0;
}

/////////////////////////////////////////////////
pthread_mutex_t count_lock1;
pthread_cond_t count_nonzero1;
unsigned count1 = 0;

void* decrement_count1(void* arg)
{
pthread_mutex_lock(&count_lock1);
std::cout << "decrement_count get count_lock" << std::endl;

while (count1 == 0) {
std::cout << "decrement_count count == 0" << std::endl;

std::cout << "decrement_count before cond_wait" << std::endl;
pthread_cond_wait(&count_nonzero1, &count_lock1);
std::cout << "decrement_count after cond_wait" << std::endl;
}

count1 = count1 + 1;

pthread_mutex_unlock(&count_lock1);

return NULL;
}

void* increment_count1(void* arg)
{
pthread_mutex_lock(&count_lock1);
std::cout << "increment_count get count_lock" << std::endl;

if (count1 == 0) {
std::cout << "increment_count before cond_signal" << std::endl;
pthread_cond_signal(&count_nonzero1);
std::cout << "increment_count after cond_signal" << std::endl;
}

count1 = count1 + 1;

pthread_mutex_unlock(&count_lock1);

return NULL;
}

int test_thread_cond1()
{
pthread_t tid1, tid2;

pthread_mutex_init(&count_lock1, NULL);
pthread_cond_init(&count_nonzero1, NULL);

pthread_create(&tid1, NULL, decrement_count1, NULL);
Sleep(2000); // == linux sleep(2)

pthread_create(&tid2, NULL, increment_count1, NULL);
Sleep(2000);

pthread_join(tid1, NULL);
pthread_join(tid2, NULL);
pthread_mutex_destroy(&count_lock1);
pthread_cond_destroy(&count_nonzero1);

return 0;
}

////////////////////////////////////////////////////////////
pthread_mutex_t counter_lock2;
pthread_cond_t counter_nonzero2;
int counter2 = 0;

void* decrement_counter2(void* argv)
{
std::cout << "counter(decrement): " << counter2 << std::endl;

pthread_mutex_lock(&counter_lock2);
while (counter2 == 0)
pthread_cond_wait(&counter_nonzero2, &counter_lock2); //进入阻塞(wait)，等待激活(signal)

std::cout << "counter--(decrement, before): " << counter2 << std::endl;
counter2--; //等待signal激活后再执行
std::cout << "counter--(decrement, after): " << counter2 << std::endl;
pthread_mutex_unlock(&counter_lock2);

return NULL;
}

void* increment_counter2(void* argv)
{
std::cout << "counter(increment): " << counter2 << std::endl;

pthread_mutex_lock(&counter_lock2);
if (counter2 == 0)
pthread_cond_signal(&counter_nonzero2); //激活(signal)阻塞(wait)的线程(先执行完signal线程，然后再执行wait线程)

std::cout << "counter++(increment, before): " << counter2 << std::endl;
counter2++;
std::cout << "counter++(increment, after): " << counter2 << std::endl;
pthread_mutex_unlock(&counter_lock2);

return NULL;
}

int test_thread_cond2()
{
std::cout << "counter: " << counter2 << std::endl;
pthread_mutex_init(&counter_lock2, NULL);
pthread_cond_init(&counter_nonzero2, NULL);

pthread_t thd1, thd2;

pthread_create(&thd1, NULL, decrement_counter2, NULL);

pthread_create(&thd2, NULL, increment_counter2, NULL);

int counter2 = 0;
while (counter2 != 10) {
std::cout << "counter(main): " << counter2 << std::endl;
Sleep(1);
counter2++;
}

pthread_join(thd1, NULL);
pthread_join(thd2, NULL);
pthread_mutex_destroy(&counter_lock2);
pthread_cond_destroy(&counter_nonzero2);

return 0;
}

///////////////////////////////////////////////////
pthread_mutex_t counter_lock3_1, counter_lock3_2;
pthread_cond_t counter_nonzero3_1, counter_nonzero3_2;
int counter3 = 0;

void* decrement_increment_counter3(void* argv)
{
std::cout << "start counter: " << counter3 << std::endl;

pthread_mutex_lock(&counter_lock3_1);
std::cout << "counter(decrement): " << counter3 << std::endl;
while (counter3 == 1)
pthread_cond_wait(&counter_nonzero3_1, &counter_lock3_1); //进入阻塞(wait)，等待激活(signal)

std::cout << "counter--(decrement, before): " << counter3 << std::endl;
counter3--; //等待signal激活后再执行
std::cout << "counter--(decrement, after): " << counter3 << std::endl;
pthread_mutex_unlock(&counter_lock3_1);

pthread_mutex_lock(&counter_lock3_2);
std::cout << "counter(increment): " << counter3 << std::endl;
if (counter3 == 0)
pthread_cond_signal(&counter_nonzero3_2); //激活(signal)阻塞(wait)的线程(先执行完signal线程，然后再执行wait线程)

std::cout << "counter++(increment, before): " << counter3 << std::endl;
counter3++;
std::cout << "counter++(increment, after): " << counter3 << std::endl;
pthread_mutex_unlock(&counter_lock3_2);

return NULL;
}

int test_thread_cond3()
{
std::cout << "counter: " << counter3 << std::endl;
pthread_mutex_init(&counter_lock3_1, NULL);
pthread_mutex_init(&counter_lock3_2, NULL);
pthread_cond_init(&counter_nonzero3_1, NULL);
pthread_cond_init(&counter_nonzero3_2, NULL);

pthread_t thd;

pthread_create(&thd, NULL, decrement_increment_counter3, NULL);

counter3 = 0;
while (counter3 != 10) {
std::cout << "counter(main): " << counter3 << std::endl;
Sleep(1000);
counter3++;
}

pthread_join(thd, NULL);
pthread_mutex_destroy(&counter_lock3_1);
pthread_mutex_destroy(&counter_lock3_2);
pthread_cond_destroy(&counter_nonzero3_1);
pthread_cond_destroy(&counter_nonzero3_2);

return 0;
}

GitHub：https://github.com/fengbingchun/Messy_Test