From 2b6c15678b4bf8ce871812dd2e8ccdbd4d349fd9 Mon Sep 17 00:00:00 2001 From: fpagliughi Date: Thu, 9 Jan 2025 23:42:03 -0500 Subject: [PATCH] Started merging semaphores and condition variables into a single 'event' type. --- src/MQTTAsync.c | 12 +- src/MQTTAsyncUtils.c | 40 ++--- src/MQTTClient.c | 75 +++++----- src/Thread.c | 340 ++++++++++++++++--------------------------- src/Thread.h | 14 +- test/thread.c | 179 +++++------------------ 6 files changed, 224 insertions(+), 436 deletions(-) diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 3f310e40..2db7680b 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -264,8 +264,8 @@ mutex_type socket_mutex = &socket_mutex_store; static pthread_mutex_t mqttcommand_mutex_store = PTHREAD_MUTEX_INITIALIZER; mutex_type mqttcommand_mutex = &mqttcommand_mutex_store; -static cond_type_struct send_cond_store = { PTHREAD_COND_INITIALIZER, PTHREAD_MUTEX_INITIALIZER }; -cond_type send_cond = &send_cond_store; +static evt_type_struct send_evt_store = { PTHREAD_COND_INITIALIZER, PTHREAD_MUTEX_INITIALIZER, 0 }; +evt_type send_evt = &send_evt_store; int MQTTAsync_init(void) { @@ -284,10 +284,10 @@ int MQTTAsync_init(void) printf("MQTTAsync: error %d initializing command_mutex\n", rc); else if ((rc = pthread_mutex_init(socket_mutex, &attr)) != 0) printf("MQTTClient: error %d initializing socket_mutex\n", rc); - else if ((rc = pthread_cond_init(&send_cond->cond, NULL)) != 0) - printf("MQTTAsync: error %d initializing send_cond cond\n", rc); - else if ((rc = pthread_mutex_init(&send_cond->mutex, &attr)) != 0) - printf("MQTTAsync: error %d initializing send_cond mutex\n", rc); + else if ((rc = pthread_cond_init(&send_evt->cond, NULL)) != 0) + printf("MQTTAsync: error %d initializing send_evt cond\n", rc); + else if ((rc = pthread_mutex_init(&send_evt->mutex, &attr)) != 0) + printf("MQTTAsync: error %d initializing send_evt mutex\n", rc); return rc; } diff --git a/src/MQTTAsyncUtils.c b/src/MQTTAsyncUtils.c index 2c37973c..93f49e7a 100644 --- a/src/MQTTAsyncUtils.c +++ b/src/MQTTAsyncUtils.c @@ -98,7 +98,7 @@ extern mutex_type log_mutex; extern mutex_type mqttasync_mutex; extern mutex_type socket_mutex; extern mutex_type mqttcommand_mutex; -extern cond_type send_cond; +extern evt_type send_evt; #endif #if !defined(min) @@ -936,13 +936,8 @@ int MQTTAsync_addCommand(MQTTAsync_queuedCommand* command, int command_size) } exit: MQTTAsync_unlock_mutex(mqttcommand_mutex); -#if !defined(_WIN32) - if ((rc1 = Thread_signal_cond(send_cond)) != 0) - Log(LOG_ERROR, 0, "Error %d from signal cond", rc1); -#else - if ((rc1 = Thread_post_sem(send_sem)) != 0) - Log(LOG_ERROR, 0, "Error %d from signal cond", rc1); -#endif + if ((rc1 = Thread_signal_evt(send_evt)) != 0) + Log(LOG_ERROR, 0, "Error %d from signal event", rc1); FUNC_EXIT_RC(rc); return rc; } @@ -1860,13 +1855,8 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n) command_count = MQTTAsync_commands->count; MQTTAsync_unlock_mutex(mqttcommand_mutex); } -#if !defined(_WIN32) - if ((rc = Thread_wait_cond(send_cond, timeout)) != 0 && rc != ETIMEDOUT) - Log(LOG_ERROR, -1, "Error %d waiting for condition variable", rc); -#else - if ((rc = Thread_wait_sem(send_sem, timeout)) != 0 && rc != ETIMEDOUT) - Log(LOG_ERROR, -1, "Error %d waiting for semaphore", rc); -#endif + if ((rc = Thread_wait_evt(send_evt, timeout)) != 0 && rc != ETIMEDOUT) + Log(LOG_ERROR, -1, "Error %d waiting for send event", rc); timeout = 1000; /* 1 second for follow on waits */ MQTTAsync_checkTimeouts(); } @@ -2071,11 +2061,7 @@ static int MQTTAsync_completeConnection(MQTTAsyncs* m, Connack* connack) } } m->pack = NULL; -#if !defined(_WIN32) - Thread_signal_cond(send_cond); -#else - Thread_post_sem(send_sem); -#endif + Thread_signal_evt(send_evt); } FUNC_EXIT_RC(rc); return rc; @@ -2403,7 +2389,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) MQTTAsync_unlock_mutex(mqttasync_mutex); #if !defined(_WIN32) if (sendThread_state != STOPPED) - Thread_signal_cond(send_cond); + Thread_signal_evt(send_evt); #else if (sendThread_state != STOPPED) Thread_post_sem(send_sem); @@ -3152,11 +3138,7 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc) { *rc = MQTTProtocol_handlePubcomps(pack, *sock, &pubToRemove); if (sendThread_state != STOPPED) -#if !defined(_WIN32) - Thread_signal_cond(send_cond); -#else - Thread_post_sem(send_sem); -#endif + Thread_signal_evt(send_evt); } else if (msgtype == PUBREC) *rc = MQTTProtocol_handlePubrecs(pack, *sock, &pubToRemove); @@ -3164,11 +3146,7 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc) { *rc = MQTTProtocol_handlePubacks(pack, *sock, &pubToRemove); if (sendThread_state != STOPPED) -#if !defined(_WIN32) - Thread_signal_cond(send_cond); -#else - Thread_post_sem(send_sem); -#endif + Thread_signal_evt(send_evt); } if (!m) Log(LOG_ERROR, -1, "PUBCOMP, PUBACK or PUBREC received for no client, msgid %d", msgid); diff --git a/src/MQTTClient.c b/src/MQTTClient.c index 6020ce1c..3d9bec56 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -91,7 +91,7 @@ const char *client_timestamp_eye = "MQTTClientV3_Timestamp " BUILD_TIMESTAMP; const char *client_version_eye = "MQTTClientV3_Version " CLIENT_VERSION; struct conlost_sync_data { - sem_type sem; + evt_type evt; void *m; }; @@ -318,11 +318,11 @@ typedef struct void* auth_handle_context; /* the context to be associated with the authHandle callback*/ #endif - sem_type connect_sem; int rc; /* getsockopt return code in connect */ - sem_type connack_sem; - sem_type suback_sem; - sem_type unsuback_sem; + evt_type connect_evt; + evt_type connack_evt; + evt_type suback_evt; + evt_type unsuback_evt; MQTTPacket* pack; unsigned long commandTimeout; @@ -524,10 +524,11 @@ int MQTTClient_createWithOptions(MQTTClient* handle, const char* serverURI, cons m->c->messageQueue = ListInitialize(); m->c->outboundQueue = ListInitialize(); m->c->clientID = MQTTStrdup(clientId); - m->connect_sem = Thread_create_sem(&rc); - m->connack_sem = Thread_create_sem(&rc); - m->suback_sem = Thread_create_sem(&rc); - m->unsuback_sem = Thread_create_sem(&rc); + // TODO (fmp): None of these events are being checked for failure + m->connect_evt = Thread_create_evt(&rc); + m->connack_evt = Thread_create_evt(&rc); + m->suback_evt = Thread_create_evt(&rc); + m->unsuback_evt = Thread_create_evt(&rc); #if !defined(NO_PERSISTENCE) rc = MQTTPersistence_create(&(m->c->persistence), persistence_type, persistence_context); @@ -625,10 +626,10 @@ void MQTTClient_destroy(MQTTClient* handle) } if (m->serverURI) free(m->serverURI); - Thread_destroy_sem(m->connect_sem); - Thread_destroy_sem(m->connack_sem); - Thread_destroy_sem(m->suback_sem); - Thread_destroy_sem(m->unsuback_sem); + Thread_destroy_evt(m->connect_evt); + Thread_destroy_evt(m->connack_evt); + Thread_destroy_evt(m->suback_evt); + Thread_destroy_evt(m->unsuback_evt); if (!ListRemove(handles, m)) Log(LOG_ERROR, -1, "free error"); *handle = NULL; @@ -734,7 +735,7 @@ static thread_return_type WINAPI connectionLost_call(void* context) (*(m->cl))(m->context, NULL); - Thread_post_sem(data->sem); + Thread_signal_evt(data->evt); return 0; } @@ -889,15 +890,15 @@ static thread_return_type WINAPI MQTTClient_run(void* n) { if (m->c->connect_state == SSL_IN_PROGRESS) { - Log(TRACE_MIN, -1, "Posting connect semaphore for client %s", m->c->clientID); + Log(TRACE_MIN, -1, "Signaling connect event for client %s", m->c->clientID); m->c->connect_state = NOT_IN_PROGRESS; - Thread_post_sem(m->connect_sem); + Thread_signal_evt(m->connect_evt); } if (m->c->connect_state == WAIT_FOR_CONNACK) { - Log(TRACE_MIN, -1, "Posting connack semaphore for client %s", m->c->clientID); + Log(TRACE_MIN, -1, "Signaling connack event for client %s", m->c->clientID); m->c->connect_state = NOT_IN_PROGRESS; - Thread_post_sem(m->connack_sem); + Thread_signal_evt(m->connack_evt); } } } @@ -936,21 +937,21 @@ static thread_return_type WINAPI MQTTClient_run(void* n) { if (pack->header.bits.type == CONNACK) { - Log(TRACE_MIN, -1, "Posting connack semaphore for client %s", m->c->clientID); + Log(TRACE_MIN, -1, "Signaling connack event for client %s", m->c->clientID); m->pack = pack; - Thread_post_sem(m->connack_sem); + Thread_signal_evt(m->connack_evt); } else if (pack->header.bits.type == SUBACK) { - Log(TRACE_MIN, -1, "Posting suback semaphore for client %s", m->c->clientID); + Log(TRACE_MIN, -1, "Signaling suback event for client %s", m->c->clientID); m->pack = pack; - Thread_post_sem(m->suback_sem); + Thread_signal_evt(m->suback_evt); } else if (pack->header.bits.type == UNSUBACK) { - Log(TRACE_MIN, -1, "Posting unsuback semaphore for client %s", m->c->clientID); + Log(TRACE_MIN, -1, "Signaling unsuback event for client %s", m->c->clientID); m->pack = pack; - Thread_post_sem(m->unsuback_sem); + Thread_signal_evt(m->unsuback_evt); } else if (m->c->MQTTVersion >= MQTTVERSION_5) { @@ -1000,9 +1001,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n) if ((m->rc = getsockopt(m->c->net.socket, SOL_SOCKET, SO_ERROR, (char*)&error, &len)) == 0) m->rc = error; - Log(TRACE_MIN, -1, "Posting connect semaphore for client %s rc %d", m->c->clientID, m->rc); + Log(TRACE_MIN, -1, "Signaling connect event for client %s rc %d", m->c->clientID, m->rc); m->c->connect_state = NOT_IN_PROGRESS; - Thread_post_sem(m->connect_sem); + Thread_signal_evt(m->connect_evt); } #if defined(OPENSSL) else if (m->c->connect_state == SSL_IN_PROGRESS) @@ -1017,9 +1018,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n) if (rc == 1 && (m->c->cleansession == 0 && m->c->cleanstart == 0) && m->c->session == NULL) m->c->session = SSL_get1_session(m->c->net.ssl); m->rc = rc; - Log(TRACE_MIN, -1, "Posting connect semaphore for SSL client %s rc %d", m->c->clientID, m->rc); + Log(TRACE_MIN, -1, "Signaling connect event for SSL client %s rc %d", m->c->clientID, m->rc); m->c->connect_state = NOT_IN_PROGRESS; - Thread_post_sem(m->connect_sem); + Thread_signal_evt(m->connect_evt); } } #endif @@ -1027,9 +1028,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n) { if (rc != TCPSOCKET_INTERRUPTED) { - Log(TRACE_MIN, -1, "Posting websocket handshake for client %s rc %d", m->c->clientID, m->rc); + Log(TRACE_MIN, -1, "Signaling websocket handshake for client %s rc %d", m->c->clientID, m->rc); m->c->connect_state = WAIT_FOR_CONNACK; - Thread_post_sem(m->connect_sem); + Thread_signal_evt(m->connect_evt); } } } @@ -2008,11 +2009,11 @@ exit: MQTTClient_stop(); if (call_connection_lost && m->cl && was_connected) { - sync.sem = Thread_create_sem(&rc); + sync.evt = Thread_create_evt(&rc); Log(TRACE_MIN, -1, "Calling connectionLost for client %s", m->c->clientID); Paho_thread_start(connectionLost_call, &sync); - Thread_wait_sem(sync.sem, 5000); - Thread_destroy_sem(sync.sem); + Thread_wait_evt(sync.evt, 5000); + Thread_destroy_evt(sync.evt); } FUNC_EXIT_RC(rc); return rc; @@ -2728,15 +2729,15 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r { if (packet_type == CONNECT) { - if ((*rc = Thread_wait_sem(m->connect_sem, (int)timeout)) == 0) + if ((*rc = Thread_wait_evt(m->connect_evt, (int)timeout)) == 0) *rc = m->rc; } else if (packet_type == CONNACK) - *rc = Thread_wait_sem(m->connack_sem, (int)timeout); + *rc = Thread_wait_evt(m->connack_evt, (int)timeout); else if (packet_type == SUBACK) - *rc = Thread_wait_sem(m->suback_sem, (int)timeout); + *rc = Thread_wait_evt(m->suback_evt, (int)timeout); else if (packet_type == UNSUBACK) - *rc = Thread_wait_sem(m->unsuback_sem, (int)timeout); + *rc = Thread_wait_evt(m->unsuback_evt, (int)timeout); if (*rc == 0 && packet_type != CONNECT && m->pack == NULL) Log(LOG_ERROR, -1, "waitfor unexpectedly is NULL for client %s, packet_type %d, timeout %ld", m->c->clientID, packet_type, timeout); pack = m->pack; diff --git a/src/Thread.c b/src/Thread.c index f71858cc..e158955f 100644 --- a/src/Thread.c +++ b/src/Thread.c @@ -116,6 +116,33 @@ int Thread_set_name(const char* thread_name) return rc; } +struct timespec Thread_time_from_now(int ms) +{ + struct timespec from_now; + struct timespec interval; + interval.tv_sec = ms / 1000; + interval.tv_nsec = (ms % 1000) * 1000000L; + +#if defined(__APPLE__) && __MAC_OS_X_VERSION_MIN_REQUIRED < 101200 /* for older versions of MacOS */ + struct timeval cur_time; + gettimeofday(&cur_time, NULL); + from_now.tv_sec = cur_time.tv_sec; + from_now.tv_nsec = cur_time.tv_usec * 1000; +#else + clock_gettime(CLOCK_REALTIME, &from_now); +#endif + + from_now.tv_sec += interval.tv_sec; + from_now.tv_nsec += interval.tv_sec; + + if (from_now.tv_nsec >= 1000000000L) + { + from_now.tv_sec++; + from_now.tv_nsec -= 1000000000L; + } + + return from_now; +} /** * Create a new mutex @@ -218,183 +245,13 @@ thread_id_type Paho_thread_getid(void) #endif } - /** - * Create a new semaphore - * @param rc return code: 0 for success, negative otherwise - * @return the new condition variable + * Create a new event + * @return the event struct */ -sem_type Thread_create_sem(int *rc) +evt_type Thread_create_evt(int *rc) { - sem_type sem = NULL; - - FUNC_ENTRY; - *rc = -1; - #if defined(_WIN32) - sem = CreateEvent( - NULL, /* default security attributes */ - FALSE, /* manual-reset event? */ - FALSE, /* initial state is nonsignaled */ - NULL /* object name */ - ); - *rc = (sem == NULL) ? GetLastError() : 0; - #elif defined(OSX) - sem = dispatch_semaphore_create(0L); - *rc = (sem == NULL) ? -1 : 0; - #else - sem = malloc(sizeof(sem_t)); - if (sem) - *rc = sem_init(sem, 0, 0); - #endif - FUNC_EXIT_RC(*rc); - return sem; -} - - -/** - * Wait for a semaphore to be posted, or timeout. - * @param sem the semaphore - * @param timeout the maximum time to wait, in milliseconds - * @return completion code - */ -int Thread_wait_sem(sem_type sem, int timeout) -{ -/* sem_timedwait is the obvious call to use, but seemed not to work on the Viper, - * so I've used trywait in a loop instead. Ian Craggs 23/7/2010 - */ - int rc = -1; -#if !defined(_WIN32) && !defined(OSX) -#define USE_TRYWAIT -#if defined(USE_TRYWAIT) - int i = 0; - useconds_t interval = 10000; /* 10000 microseconds: 10 milliseconds */ - int count = (1000 * timeout) / interval; /* how many intervals in timeout period */ -#else - struct timespec ts; -#endif -#endif - - FUNC_ENTRY; - #if defined(_WIN32) - /* returns 0 (WAIT_OBJECT_0) on success, non-zero (WAIT_TIMEOUT) if timeout occurred */ - rc = WaitForSingleObject(sem, timeout < 0 ? 0 : timeout); - if (rc == WAIT_TIMEOUT) - rc = ETIMEDOUT; - #elif defined(OSX) - /* returns 0 on success, non-zero if timeout occurred */ - rc = (int)dispatch_semaphore_wait(sem, dispatch_time(DISPATCH_TIME_NOW, (int64_t)timeout*1000000L)); - if (rc != 0) - rc = ETIMEDOUT; - #elif defined(USE_TRYWAIT) - while (++i < count && (rc = sem_trywait(sem)) != 0) - { - if (rc == -1 && ((rc = errno) != EAGAIN)) - { - rc = 0; - break; - } - usleep(interval); /* microseconds - .1 of a second */ - } - #else - /* We have to use CLOCK_REALTIME rather than MONOTONIC for sem_timedwait interval. - * Does this make it susceptible to system clock changes? - * The intervals are small enough, and repeated, that I think it's not an issue. - */ - if (clock_gettime(CLOCK_REALTIME, &ts) != -1) - { - ts.tv_sec += timeout; - rc = sem_timedwait(sem, &ts); - } - #endif - - FUNC_EXIT_RC(rc); - return rc; -} - - -/** - * Check to see if a semaphore has been posted, without waiting - * The semaphore will be unchanged, if the return value is false. - * The semaphore will have been decremented, if the return value is true. - * @param sem the semaphore - * @return 0 (false) or 1 (true) - */ -int Thread_check_sem(sem_type sem) -{ -#if defined(_WIN32) - /* if the return value is not 0, the semaphore will not have been decremented */ - return WaitForSingleObject(sem, 0) == WAIT_OBJECT_0; -#elif defined(OSX) - /* if the return value is not 0, the semaphore will not have been decremented */ - return dispatch_semaphore_wait(sem, DISPATCH_TIME_NOW) == 0; -#else - /* If the call was unsuccessful, the state of the semaphore shall be unchanged, - * and the function shall return a value of -1 */ - return sem_trywait(sem) == 0; -#endif -} - - -/** - * Post a semaphore - * @param sem the semaphore - * @return 0 on success - */ -int Thread_post_sem(sem_type sem) -{ - int rc = 0; - - FUNC_ENTRY; - #if defined(_WIN32) - if (SetEvent(sem) == 0) - rc = GetLastError(); - #elif defined(OSX) - rc = (int)dispatch_semaphore_signal(sem); - #else - int val; - int rc1 = sem_getvalue(sem, &val); - if (rc1 != 0) - rc = errno; - else if (val == 0 && sem_post(sem) == -1) - rc = errno; - #endif - - FUNC_EXIT_RC(rc); - return rc; -} - - -/** - * Destroy a semaphore which has already been created - * @param sem the semaphore - */ -int Thread_destroy_sem(sem_type sem) -{ - int rc = 0; - - FUNC_ENTRY; - #if defined(_WIN32) - rc = CloseHandle(sem); - #elif defined(OSX) - dispatch_release(sem); - #else - rc = sem_destroy(sem); - free(sem); - #endif - FUNC_EXIT_RC(rc); - return rc; -} - - -#if !defined(_WIN32) - -/** - * Create a new condition variable - * @return the condition variable struct - */ -cond_type Thread_create_cond(int *rc) -{ - cond_type condvar = NULL; + evt_type evt = NULL; pthread_condattr_t attr; FUNC_ENTRY; @@ -411,29 +268,31 @@ cond_type Thread_create_cond(int *rc) Log(LOG_ERROR, -1, "Error %d calling pthread_condattr_setclock(CLOCK_MONOTONIC)", rc); #endif - condvar = malloc(sizeof(cond_type_struct)); - if (condvar) + evt = malloc(sizeof(evt_type_struct)); + if (evt) { - *rc = pthread_cond_init(&condvar->cond, &attr); - *rc = pthread_mutex_init(&condvar->mutex, NULL); + *rc = pthread_cond_init(&evt->cond, &attr); + *rc = pthread_mutex_init(&evt->mutex, NULL); + evt->val = 0; } FUNC_EXIT_RC(*rc); - return condvar; + return evt; } /** - * Signal a condition variable + * Signal an event * @return completion code */ -int Thread_signal_cond(cond_type condvar) +int Thread_signal_evt(evt_type evt) { int rc = 0; FUNC_ENTRY; - pthread_mutex_lock(&condvar->mutex); - rc = pthread_cond_signal(&condvar->cond); - pthread_mutex_unlock(&condvar->mutex); + pthread_mutex_lock(&evt->mutex); + evt->val = 1; + rc = pthread_cond_signal(&evt->cond); + pthread_mutex_unlock(&evt->mutex); FUNC_EXIT_RC(rc); return rc; @@ -443,40 +302,22 @@ int Thread_signal_cond(cond_type condvar) * Wait with a timeout (ms) for condition variable * @return 0 for success, ETIMEDOUT otherwise */ -int Thread_wait_cond(cond_type condvar, int timeout_ms) +int Thread_wait_evt(evt_type evt, int timeout_ms) { int rc = 0; - struct timespec cond_timeout; + struct timespec evt_timeout; FUNC_ENTRY; -#if defined(__APPLE__) && __MAC_OS_X_VERSION_MIN_REQUIRED < 101200 /* for older versions of MacOS */ - struct timeval cur_time; - gettimeofday(&cur_time, NULL); - cond_timeout.tv_sec = cur_time.tv_sec; - cond_timeout.tv_nsec = cur_time.tv_usec * 1000; -#else - clock_gettime(CLOCK_REALTIME, &cond_timeout); -#endif + evt_timeout = Thread_time_from_now(timeout_ms); - if (timeout_ms > 0) { - struct timespec interval; - - interval.tv_sec = timeout_ms / 1000; - interval.tv_nsec = (timeout_ms % 1000) * 1000000L; - - cond_timeout.tv_sec += interval.tv_sec; - cond_timeout.tv_nsec += interval.tv_nsec; - - while (cond_timeout.tv_nsec >= NSEC_PER_SEC) - { - cond_timeout.tv_sec++; - cond_timeout.tv_nsec -= NSEC_PER_SEC; - } + pthread_mutex_lock(&evt->mutex); + while (evt->val == 0 && + (rc = pthread_cond_timedwait(&evt->cond, &evt->mutex, &evt_timeout)) == 0) + ; + if (rc == 0) { + evt->val = 0; } - - pthread_mutex_lock(&condvar->mutex); - rc = pthread_cond_timedwait(&condvar->cond, &condvar->mutex, &cond_timeout); - pthread_mutex_unlock(&condvar->mutex); + pthread_mutex_unlock(&evt->mutex); FUNC_EXIT_RC(rc); return rc; @@ -486,16 +327,83 @@ int Thread_wait_cond(cond_type condvar, int timeout_ms) * Destroy a condition variable * @return completion code */ -int Thread_destroy_cond(cond_type condvar) +int Thread_destroy_evt(evt_type evt) { int rc = 0; - rc = pthread_mutex_destroy(&condvar->mutex); - rc = pthread_cond_destroy(&condvar->cond); - free(condvar); + rc = pthread_mutex_destroy(&evt->mutex); + rc = pthread_cond_destroy(&evt->cond); + free(evt); return rc; } + +#else + +/** + * Create a new semaphore + * @param rc return code: 0 for success, negative otherwise + * @return the new condition variable + */ +sem_type Thread_create_evt(int *rc) +{ + FUNC_ENTRY; + sem_type sem = CreateEvent( + NULL, /* default security attributes */ + FALSE, /* manual-reset event? */ + FALSE, /* initial state is nonsignaled */ + NULL /* object name */ + ); + if (rc) + *rc = (sem == NULL) ? GetLastError() : 0; + FUNC_EXIT_RC(*rc); + return sem; +} + +/** + * Wait for an event to be signaled, or timeout. + * @param evt the event + * @param timeout the maximum time to wait, in milliseconds + * @return completion code + */ +int Thread_wait_sem(sem_type sem, int timeout) +{ + FUNC_ENTRY; + /* returns 0 (WAIT_OBJECT_0) on success, non-zero (WAIT_TIMEOUT) if timeout occurred */ + int rc = WaitForSingleObject(sem, timeout < 0 ? 0 : timeout); + if (rc == WAIT_TIMEOUT) + rc = ETIMEDOUT; + FUNC_EXIT_RC(rc); + return rc; +} + +/** + * Post a semaphore + * @param sem the semaphore + * @return 0 on success + */ +int Thread_post_sem(sem_type sem) +{ + FUNC_ENTRY; + int rc = 0; + if (SetEvent(sem) == 0) + rc = GetLastError(); + FUNC_EXIT_RC(rc); + return rc; +} + +/** + * Destroy a semaphore which has already been created + * @param sem the semaphore + */ +int Thread_destroy_evt(evt_type evt) +{ + FUNC_ENTRY; + int rc = CloseHandle(sem); + FUNC_EXIT_RC(rc); + return rc; +} + #endif diff --git a/src/Thread.h b/src/Thread.h index 05f1fb67..924ff181 100644 --- a/src/Thread.h +++ b/src/Thread.h @@ -53,8 +53,8 @@ #define thread_id_type pthread_t #define thread_return_type void* typedef thread_return_type (*thread_fn)(void*); - typedef struct { pthread_cond_t cond; pthread_mutex_t mutex; } cond_type_struct; - typedef cond_type_struct *cond_type; + typedef struct { pthread_cond_t cond; pthread_mutex_t mutex; int val; } evt_type_struct; + typedef evt_type_struct *evt_type; #if defined(OSX) #include typedef dispatch_semaphore_t sem_type; @@ -62,11 +62,6 @@ #include typedef sem_t *sem_type; #endif - - cond_type Thread_create_cond(int*); - int Thread_signal_cond(cond_type); - int Thread_wait_cond(cond_type condvar, int timeout); - int Thread_destroy_cond(cond_type); #endif LIBMQTT_API void Paho_thread_start(thread_fn, void*); @@ -79,6 +74,11 @@ int Paho_thread_destroy_mutex(mutex_type); LIBMQTT_API thread_id_type Paho_thread_getid(); +evt_type Thread_create_evt(int*); +int Thread_signal_evt(evt_type); +int Thread_wait_evt(evt_type condvar, int timeout); +int Thread_destroy_evt(evt_type); + sem_type Thread_create_sem(int*); int Thread_wait_sem(sem_type sem, int timeout); int Thread_check_sem(sem_type sem); diff --git a/test/thread.c b/test/thread.c index 825170a2..835aa55d 100644 --- a/test/thread.c +++ b/test/thread.c @@ -226,159 +226,57 @@ void myassert(char* filename, int lineno, char* description, int value, char* fo MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description); } -static thread_return_type WINAPI sem_secondary(void* n) +thread_return_type evt_secondary(void* n) { int rc = 0; - sem_type sem = n; - START_TIME_TYPE start; - long duration; - - MyLog(LOGA_DEBUG, "Secondary semaphore pointer %p", sem); - - rc = Thread_check_sem(sem); - assert("rc 0 from check_sem", rc == 0, "rc was %d", rc); - - MyLog(LOGA_DEBUG, "Secondary thread about to wait"); - start = start_clock(); - rc = Thread_wait_sem(sem, 99999); - duration = elapsed(start); - assert("rc 0 from lock mutex", rc == 0, "rc was %d", rc); - MyLog(LOGA_INFO, "Lock duration was %ld", duration); - assert("duration is 2s", duration >= 1999L && duration < 2050L, "duration was %ld", duration); - - MyLog(LOGA_DEBUG, "Secondary thread ending"); - return 0; -} - - -int test_sem(struct Options options) -{ - char* testname = "test_sem"; - int rc = 0, i = 0; - START_TIME_TYPE start; - long duration; - sem_type sem = Thread_create_sem(&rc); - - MyLog(LOGA_INFO, "Starting semaphore test"); - fprintf(xml, "= 1500L, "duration was %ld", duration); - - MyLog(LOGA_DEBUG, "Starting secondary thread"); - Paho_thread_start(sem_secondary, (void*)sem); - - mysleep(2); - MyLog(LOGA_DEBUG, "post secondary"); - rc = Thread_post_sem(sem); - assert("rc 0 from post_sem", rc == 0, "rc was %d", rc); - - mysleep(1); - - MyLog(LOGA_DEBUG, "Main thread ending"); - - /*exit: */ MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.", - (failures == 0) ? "passed" : "failed", testname, tests, failures); - write_test_result(); - - return failures; -} - -#if !defined(_WIN32) -thread_return_type cond_secondary(void* n) -{ - int rc = 0; - cond_type cond = n; + evt_type evt = n; START_TIME_TYPE start; long duration; MyLog(LOGA_DEBUG, "This will time out"); start = start_clock(); - rc = Thread_wait_cond(cond, 1000); + rc = Thread_wait_evt(evt, 1000); duration = elapsed(start); MyLog(LOGA_INFO, "Lock duration was %ld", duration); assert("duration is about 1s", duration >= 999L && duration <= 1050L, "duration was %ld", duration); - assert("rc non 0 from wait_cond", rc == ETIMEDOUT, "rc was %d", rc); + assert("rc non 0 from wait_evt", rc == ETIMEDOUT, "rc was %d", rc); MyLog(LOGA_DEBUG, "This should hang around a few seconds"); start = start_clock(); - rc = Thread_wait_cond(cond, 99999); + rc = Thread_wait_evt(evt, 99999); duration = elapsed(start); MyLog(LOGA_INFO, "Lock duration was %ld", duration); assert("duration is around 1s", duration >= 990L && duration <= 1010L, "duration was %ld", duration); - assert("rc 9 from wait_cond", rc == 0, "rc was %d", rc); + assert("rc 9 from wait_evt", rc == 0, "rc was %d", rc); - MyLog(LOGA_DEBUG, "Secondary cond thread ending"); + MyLog(LOGA_DEBUG, "Secondary evt thread ending"); return 0; } -int test_cond(struct Options options) +int test_evt(struct Options options) { - char* testname = "test_cond"; + char* testname = "test_evt"; int rc = 0, i = 0; START_TIME_TYPE start; long duration; - cond_type cond = Thread_create_cond(&rc); + evt_type evt = Thread_create_evt(&rc); - MyLog(LOGA_INFO, "Starting condition variable test"); - fprintf(xml, ">>>>>> 1ed408e (Started merging semaphores and condition variables into a single 'event' type.) duration = elapsed(start); assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT, "rc was %d", rc); MyLog(LOGA_INFO, "Lock duration was %ld", duration); @@ -387,30 +285,33 @@ int test_cond(struct Options options) /* multiple posts */ for (i = 0; i < 10; ++i) { - rc = Thread_signal_cond(cond); - assert("rc 0 from signal cond", rc == 0, "rc was %d\n", rc); + rc = Thread_signal_evt(evt); + assert("rc 0 from signal evt", rc == 0, "rc was %d\n", rc); } - /* the signals are not stored */ - for (i = 0; i < 10; ++i) + rc = Thread_wait_evt(evt, 0); + assert("rc zero from wait_evt", rc == 0, "rc was %d", rc); + + /* the signals are not accumulated */ + for (i = 0; i < 9; ++i) { - rc = Thread_wait_cond(cond, 0); - assert("rc non-zero from wait_cond", rc == ETIMEDOUT, "rc was %d", rc); + rc = Thread_wait_evt(evt, 0); + assert("rc non-zero from wait_evt", rc == ETIMEDOUT, "rc was %d", rc); } - MyLog(LOGA_DEBUG, "Post secondary but it will time out"); - rc = Thread_signal_cond(cond); - assert("rc 0 from signal cond", rc == 0, "rc was %d", rc); +// MyLog(LOGA_DEBUG, "Post secondary but it will time out"); +// rc = Thread_signal_evt(evt); +// assert("rc 0 from signal evt", rc == 0, "rc was %d", rc); MyLog(LOGA_DEBUG, "Starting secondary thread"); - Paho_thread_start(cond_secondary, (void*)cond); + Paho_thread_start(evt_secondary, (void*)evt); MyLog(LOGA_DEBUG, "wait for secondary thread to enter second wait"); mysleep(2); MyLog(LOGA_DEBUG, "post secondary"); - rc = Thread_signal_cond(cond); - assert("rc 0 from signal cond", rc == 0, "rc was %d", rc); + rc = Thread_signal_evt(evt); + assert("rc 0 from signal evt", rc == 0, "rc was %d", rc); mysleep(1); @@ -422,7 +323,6 @@ int test_cond(struct Options options) return failures; } -#endif static thread_return_type WINAPI mutex_secondary(void* n) @@ -497,12 +397,17 @@ int test_mutex(struct Options options) int main(int argc, char** argv) { int rc = -1; - int (*tests[])(struct Options) = {NULL, + int (*tests[])(struct Options) = { + NULL, test_mutex, +<<<<<<< HEAD test_sem, #if !defined(_WIN32) test_cond #endif +======= + test_evt +>>>>>>> 1ed408e (Started merging semaphores and condition variables into a single 'event' type.) }; /* indexed starting from 1 */ int i; @@ -532,11 +437,7 @@ int main(int argc, char** argv) } } - if (rc == 0) - MyLog(LOGA_INFO, "verdict pass"); - else - MyLog(LOGA_INFO, "verdict fail"); - + MyLog(LOGA_INFO, (rc == 0) ? "verdict pass" : "verdict fail"); fprintf(xml, "\n"); fclose(xml);