zookeeper-client/zookeeper-client-c/src/winport.c - zookeeper - Git at Google

 /**
  * 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.
  */

 #ifdef WIN32
 #include "winport.h"
 #include <stdlib.h>
 #include <stdint.h> /* for int64_t */
 #include <winsock2.h> /* must always be included before ws2tcpip.h */
 #include <ws2tcpip.h> /* for SOCKET */

 int pthread_mutex_lock(pthread_mutex_t* _mutex ){
        int rc = WaitForSingleObject( *_mutex,    // handle to mutex
             INFINITE);  // no time-out interval
        return ((rc == WAIT_OBJECT_0) ? 0: rc);
 }

 int pthread_mutex_unlock( pthread_mutex_t* _mutex ){
        int rc = ReleaseMutex(*_mutex);
        return ((rc != 0)? 0: GetLastError());
 }

 int pthread_mutex_init(pthread_mutex_t* _mutex, void* ignoredAttr){
        //use CreateMutex as we are using the HANDLES in pthread_cond
        *_mutex = CreateMutex(
         NULL,              // default security attributes
         FALSE,             // initially not owned
         NULL);             // unnamed mutex
        return ((*_mutex == NULL) ? GetLastError() : 0);
 }

 int pthread_mutex_destroy(pthread_mutex_t* _mutex)
 {
        int rc = CloseHandle(*_mutex);
        return ((rc != 0)? 0: GetLastError());
 }

 int pthread_create(pthread_t *thread, const pthread_attr_t *attr, unsigned  (__stdcall* start_routine)(void* a), void *arg)
 {
    int _intThreadId;
    (*thread).thread_handle = (HANDLE)_beginthreadex( NULL, 0, start_routine , arg, 0, (unsigned int*)&_intThreadId );
    (*thread).thread_id = _intThreadId;
    return (((*thread).thread_handle == 0 ) ? errno : 0 );
 }


 int pthread_equal(pthread_t t1, pthread_t t2){
 //Is there a better way to do this? GetThreadId(handle) is only supported Windows 2003 n above.
        return ((t1.thread_id == t2.thread_id) ? 1:0);
 }

 pthread_t pthread_self(){
     pthread_t thread_self;
     thread_self.thread_handle = GetCurrentThread();
     thread_self.thread_id     = GetCurrentThreadId();
     return thread_self;
 }

 int pthread_join(pthread_t _thread, void** ignore)
 {
        int rc = WaitForSingleObject( _thread.thread_handle, INFINITE );
        return ((rc == WAIT_OBJECT_0) ? 0: rc);
 }

 int pthread_detach(pthread_t _thread)
 {
        int rc = CloseHandle(_thread.thread_handle) ;
        return  (rc != 0) ? 0: GetLastError();
 }

 void pthread_mutexattr_init(pthread_mutexattr_t* ignore){}
 void pthread_mutexattr_settype(pthread_mutexattr_t* ingore_attr, int ignore){}
 void pthread_mutexattr_destroy(pthread_mutexattr_t* ignore_attr){}

 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;

 }


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


 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;
 }


 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_);
 }


 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);
 }

 int pthread_key_create(pthread_key_t *key, void (*destructor)(void *) )
 {
   int result = 0;
   pthread_key_t* newkey;

   if ((newkey = (pthread_key_t*) calloc (1, sizeof (pthread_key_t))) == NULL)
     {
       result = ENOMEM;
     }
   else if ((newkey->key = TlsAlloc ()) == TLS_OUT_OF_INDEXES)
     {
       result = EAGAIN;
       free (newkey);
       newkey = NULL;
     }
   else if (destructor != NULL)
     {
       //--we have to store the function pointer for destructor, so that we can call it
          //--to free up the user allocated storage--
       newkey->destructor = destructor;
     }
   key = newkey;
   return (result);
 }

 int pthread_key_delete(pthread_key_t key)
 {
   int rc = 0;
   LPVOID lpvData =  TlsGetValue(key.key);
   rc = TlsFree (key.key);
   rc = (rc != 0 ) ? 0 : GetLastError();
   if (key.destructor != NULL && lpvData != 0){
        key.destructor(lpvData);         //we take control of calling destructor, instead of calling it on thread exit.
   }
   free (&key);
   return (rc);
 }

 void *pthread_getspecific(pthread_key_t key)
 {
        LPVOID lpvData =  TlsGetValue(key.key);
        if ((lpvData == 0) && (GetLastError() != ERROR_SUCCESS))
                return NULL;
        else
                return lpvData;
 }

 int pthread_setspecific(pthread_key_t key, const void *value)
 {
        int rc = TlsSetValue (key.key, value);
        return ((rc != 0 ) ? 0 : GetLastError());
 }

 int gettimeofday(struct timeval *tp, void *tzp) {
         int64_t now = 0;
         if (tzp != 0) { errno = EINVAL; return -1; }
         GetSystemTimeAsFileTime( (LPFILETIME)&now );
         tp->tv_sec = (long)(now / 10000000 - 11644473600LL);
         tp->tv_usec = (now / 10) % 1000000;
         return 0;
 }

 int close(SOCKET fd) {
         return closesocket(fd);
 }

 int Win32WSAStartup()
 {
        WORD    wVersionRq;
        WSADATA wsaData;
        int             err;

        wVersionRq = MAKEWORD(2,0);
        err = WSAStartup(wVersionRq, &wsaData);
        if (err != 0)
                return 1;

        // confirm the version information
        if ((LOBYTE(wsaData.wVersion) != 2) ||
            (HIBYTE(wsaData.wVersion) != 0))
        {
                Win32WSACleanup();
                return 1;
        }
        return 0;
 }

 void Win32WSACleanup()
 {
        WSACleanup();
 }

 double drand48(void)
 {
     return (double)(rand()) / RAND_MAX;
 }

 #endif //WIN32
	/**
	* 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.
	*/

	#ifdef WIN32
	#include "winport.h"
	#include <stdlib.h>
	#include <stdint.h> /* for int64_t */
	#include <winsock2.h> /* must always be included before ws2tcpip.h */
	#include <ws2tcpip.h> /* for SOCKET */

	int pthread_mutex_lock(pthread_mutex_t* _mutex ){
	int rc = WaitForSingleObject( *_mutex, // handle to mutex
	INFINITE); // no time-out interval
	return ((rc == WAIT_OBJECT_0) ? 0: rc);
	}

	int pthread_mutex_unlock( pthread_mutex_t* _mutex ){
	int rc = ReleaseMutex(*_mutex);
	return ((rc != 0)? 0: GetLastError());
	}

	int pthread_mutex_init(pthread_mutex_t* _mutex, void* ignoredAttr){
	//use CreateMutex as we are using the HANDLES in pthread_cond
	*_mutex = CreateMutex(
	NULL, // default security attributes
	FALSE, // initially not owned
	NULL); // unnamed mutex
	return ((*_mutex == NULL) ? GetLastError() : 0);
	}

	int pthread_mutex_destroy(pthread_mutex_t* _mutex)
	{
	int rc = CloseHandle(*_mutex);
	return ((rc != 0)? 0: GetLastError());
	}

	int pthread_create(pthread_t thread, const pthread_attr_t attr, unsigned (__stdcall* start_routine)(void* a), void *arg)
	{
	int _intThreadId;
	(thread).thread_handle = (HANDLE)_beginthreadex( NULL, 0, start_routine , arg, 0, (unsigned int)&_intThreadId );
	(*thread).thread_id = _intThreadId;
	return (((*thread).thread_handle == 0 ) ? errno : 0 );
	}


	int pthread_equal(pthread_t t1, pthread_t t2){
	//Is there a better way to do this? GetThreadId(handle) is only supported Windows 2003 n above.
	return ((t1.thread_id == t2.thread_id) ? 1:0);
	}

	pthread_t pthread_self(){
	pthread_t thread_self;
	thread_self.thread_handle = GetCurrentThread();
	thread_self.thread_id = GetCurrentThreadId();
	return thread_self;
	}

	int pthread_join(pthread_t _thread, void** ignore)
	{
	int rc = WaitForSingleObject( _thread.thread_handle, INFINITE );
	return ((rc == WAIT_OBJECT_0) ? 0: rc);
	}

	int pthread_detach(pthread_t _thread)
	{
	int rc = CloseHandle(_thread.thread_handle) ;
	return (rc != 0) ? 0: GetLastError();
	}

	void pthread_mutexattr_init(pthread_mutexattr_t* ignore){}
	void pthread_mutexattr_settype(pthread_mutexattr_t* ingore_attr, int ignore){}
	void pthread_mutexattr_destroy(pthread_mutexattr_t* ignore_attr){}

	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;

	}


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


	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;
	}


	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_);
	}


	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);
	}

	int pthread_key_create(pthread_key_t key, void (destructor)(void *) )
	{
	int result = 0;
	pthread_key_t* newkey;

	if ((newkey = (pthread_key_t*) calloc (1, sizeof (pthread_key_t))) == NULL)
	{
	result = ENOMEM;
	}
	else if ((newkey->key = TlsAlloc ()) == TLS_OUT_OF_INDEXES)
	{
	result = EAGAIN;
	free (newkey);
	newkey = NULL;
	}
	else if (destructor != NULL)
	{
	//--we have to store the function pointer for destructor, so that we can call it
	//--to free up the user allocated storage--
	newkey->destructor = destructor;
	}
	key = newkey;
	return (result);
	}

	int pthread_key_delete(pthread_key_t key)
	{
	int rc = 0;
	LPVOID lpvData = TlsGetValue(key.key);
	rc = TlsFree (key.key);
	rc = (rc != 0 ) ? 0 : GetLastError();
	if (key.destructor != NULL && lpvData != 0){
	key.destructor(lpvData); //we take control of calling destructor, instead of calling it on thread exit.
	}
	free (&key);
	return (rc);
	}

	void *pthread_getspecific(pthread_key_t key)
	{
	LPVOID lpvData = TlsGetValue(key.key);
	if ((lpvData == 0) && (GetLastError() != ERROR_SUCCESS))
	return NULL;
	else
	return lpvData;
	}

	int pthread_setspecific(pthread_key_t key, const void *value)
	{
	int rc = TlsSetValue (key.key, value);
	return ((rc != 0 ) ? 0 : GetLastError());
	}

	int gettimeofday(struct timeval tp, void tzp) {
	int64_t now = 0;
	if (tzp != 0) { errno = EINVAL; return -1; }
	GetSystemTimeAsFileTime( (LPFILETIME)&now );
	tp->tv_sec = (long)(now / 10000000 - 11644473600LL);
	tp->tv_usec = (now / 10) % 1000000;
	return 0;
	}

	int close(SOCKET fd) {
	return closesocket(fd);
	}

	int Win32WSAStartup()
	{
	WORD wVersionRq;
	WSADATA wsaData;
	int err;

	wVersionRq = MAKEWORD(2,0);
	err = WSAStartup(wVersionRq, &wsaData);
	if (err != 0)
	return 1;

	// confirm the version information
	if ((LOBYTE(wsaData.wVersion) != 2) \|\|
	(HIBYTE(wsaData.wVersion) != 0))
	{
	Win32WSACleanup();
	return 1;
	}
	return 0;
	}

	void Win32WSACleanup()
	{
	WSACleanup();
	}

	double drand48(void)
	{
	return (double)(rand()) / RAND_MAX;
	}

	#endif //WIN32