diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 3f310e40..178c1c2e 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others + * Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -129,7 +129,7 @@ void MQTTAsync_init_rand(void) mutex_type mqttasync_mutex = NULL; mutex_type socket_mutex = NULL; mutex_type mqttcommand_mutex = NULL; -sem_type send_sem = NULL; +evt_type send_evt = NULL; #if !defined(NO_HEAP_TRACKING) extern mutex_type stack_mutex; extern mutex_type heap_mutex; @@ -142,52 +142,47 @@ int MQTTAsync_init(void) if (mqttasync_mutex == NULL) { - if ((mqttasync_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + mqttasync_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("mqttasync_mutex error %d\n", rc); goto exit; } - if ((mqttcommand_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + mqttcommand_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("mqttcommand_mutex error %d\n", rc); goto exit; } - if ((send_sem = CreateEvent( - NULL, /* default security attributes */ - FALSE, /* manual-reset event? */ - FALSE, /* initial state is nonsignaled */ - NULL /* object name */ - )) == NULL) + send_evt = Thread_create_evt(&rc); + if (rc != 0) { - rc = GetLastError(); - printf("send_sem error %d\n", rc); + printf("send_evt error %d\n", rc); goto exit; } #if !defined(NO_HEAP_TRACKING) - if ((stack_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + stack_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("stack_mutex error %d\n", rc); goto exit; } - if ((heap_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + heap_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("heap_mutex error %d\n", rc); goto exit; } #endif - if ((log_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + log_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("log_mutex error %d\n", rc); goto exit; } - if ((socket_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + socket_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("socket_mutex error %d\n", rc); goto exit; } @@ -202,20 +197,20 @@ exit: void MQTTAsync_cleanup(void) { - if (send_sem) - CloseHandle(send_sem); + if (send_evt) + Thread_destroy_evt(send_evt); #if !defined(NO_HEAP_TRACKING) if (stack_mutex) - CloseHandle(stack_mutex); + Paho_thread_destroy_mutex(stack_mutex); if (heap_mutex) - CloseHandle(heap_mutex); + Paho_thread_destroy_mutex(heap_mutex); #endif if (log_mutex) - CloseHandle(log_mutex); + Paho_thread_destroy_mutex(log_mutex); if (socket_mutex) - CloseHandle(socket_mutex); + Paho_thread_destroy_mutex(socket_mutex); if (mqttasync_mutex) - CloseHandle(mqttasync_mutex); + Paho_thread_destroy_mutex(mqttasync_mutex); } #if defined(PAHO_MQTT_STATIC) @@ -264,8 +259,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 +279,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..4ca23a65 100644 --- a/src/MQTTAsyncUtils.c +++ b/src/MQTTAsyncUtils.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others + * Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -88,7 +88,6 @@ extern int MQTTAsync_tostop; extern mutex_type mqttasync_mutex; extern mutex_type socket_mutex; extern mutex_type mqttcommand_mutex; -extern sem_type send_sem; #if !defined(NO_HEAP_TRACKING) extern mutex_type stack_mutex; extern mutex_type heap_mutex; @@ -98,9 +97,10 @@ 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; #endif +extern evt_type send_evt; + #if !defined(min) #define min(a, b) (((a) < (b)) ? (a) : (b)) #endif @@ -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; @@ -2401,13 +2387,8 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) receiveThread_state = STOPPED; receiveThread_id = 0; MQTTAsync_unlock_mutex(mqttasync_mutex); -#if !defined(_WIN32) if (sendThread_state != STOPPED) - Thread_signal_cond(send_cond); -#else - if (sendThread_state != STOPPED) - Thread_post_sem(send_sem); -#endif + Thread_signal_evt(send_evt); #if defined(OPENSSL) #if ((OPENSSL_VERSION_NUMBER < 0x1010000fL) || defined(LIBRESSL_VERSION_NUMBER)) @@ -3152,11 +3133,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 +3141,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..555fe9e1 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2024 IBM Corp., Ian Craggs and others + * Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -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; }; @@ -128,51 +128,51 @@ extern mutex_type log_mutex; int MQTTClient_init(void) { - DWORD rc = 0; + int rc = 0; if (mqttclient_mutex == NULL) { - if ((mqttclient_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + mqttclient_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("mqttclient_mutex error %d\n", rc); goto exit; } - if ((subscribe_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + subscribe_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("subscribe_mutex error %d\n", rc); goto exit; } - if ((connect_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + connect_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("connect_mutex error %d\n", rc); goto exit; } #if !defined(NO_HEAP_TRACKING) - if ((stack_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + stack_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("stack_mutex error %d\n", rc); goto exit; } - if ((heap_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + heap_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("heap_mutex error %d\n", rc); goto exit; } #endif - if ((log_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + log_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("log_mutex error %d\n", rc); goto exit; } - if ((socket_mutex = CreateMutex(NULL, 0, NULL)) == NULL) + socket_mutex = Paho_thread_create_mutex(&rc); + if (rc != 0) { - rc = GetLastError(); printf("socket_mutex error %d\n", rc); goto exit; } @@ -184,21 +184,21 @@ exit: void MQTTClient_cleanup(void) { if (connect_mutex) - CloseHandle(connect_mutex); + Paho_thread_destroy_mutex(connect_mutex); if (subscribe_mutex) - CloseHandle(subscribe_mutex); + Paho_thread_destroy_mutex(subscribe_mutex); #if !defined(NO_HEAP_TRACKING) if (stack_mutex) - CloseHandle(stack_mutex); + Paho_thread_destroy_mutex(stack_mutex); if (heap_mutex) - CloseHandle(heap_mutex); + Paho_thread_destroy_mutex(heap_mutex); #endif if (log_mutex) - CloseHandle(log_mutex); + Paho_thread_destroy_mutex(log_mutex); if (socket_mutex) - CloseHandle(socket_mutex); + Paho_thread_destroy_mutex(socket_mutex); if (mqttclient_mutex) - CloseHandle(mqttclient_mutex); + Paho_thread_destroy_mutex(mqttclient_mutex); } #if defined(PAHO_MQTT_STATIC) @@ -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/Socket.h b/src/Socket.h index 5ec65eab..eb89c73a 100644 --- a/src/Socket.h +++ b/src/Socket.h @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2022 IBM Corp., Ian Craggs and others + * Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -70,7 +70,7 @@ #define SOCKET int #endif -#include "mutex_type.h" /* Needed for mutex_type */ +#include "Thread.h" /* Needed for mutex_type */ /** socket operation completed successfully */ #define TCPSOCKET_COMPLETE 0 diff --git a/src/Thread.c b/src/Thread.c index f71858cc..fa8722dc 100644 --- a/src/Thread.c +++ b/src/Thread.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2025 IBM Corp. and Ian Craggs + * Copyright (c) 2009, 2026 IBM Corp. and Ian Craggs * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -17,6 +17,7 @@ * Ian Craggs - fix for bug #420851 * Ian Craggs - change MacOS semaphore implementation * Ian Craggs - fix for clock #284 + * Frank Pagliughi - Consolidated semaphores and conditions into "events" *******************************************************************************/ /** @@ -28,9 +29,6 @@ #include "Thread.h" -#if defined(THREAD_UNIT_TESTS) -#define NOSTACKTRACE -#endif #include "Log.h" #include "StackTrace.h" @@ -87,10 +85,12 @@ void Paho_thread_start(thread_fn fn, void* parameter) int Thread_set_name(const char* thread_name) { int rc = 0; +/* #if defined(_WIN32) #define MAX_THREAD_NAME_LENGTH 30 wchar_t wchars[MAX_THREAD_NAME_LENGTH]; #endif +*/ FUNC_ENTRY; #if defined(_WIN32) @@ -116,6 +116,54 @@ int Thread_set_name(const char* thread_name) return rc; } +/** + * Get the thread id of the thread from which this function is called + * @return thread id, type varying according to OS + */ +thread_id_type Paho_thread_getid(void) +{ + #if defined(_WIN32) + return GetCurrentThreadId(); + #else + return pthread_self(); + #endif +} + +#if !defined(_WIN32) +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; + + /* + * Note that MacOS v10.11 was fully deprecated in Aug 2018. + * So this __APPLE__ path only applies to very old, deprecated systems. + */ +#if defined(__APPLE__) && __MAC_OS_X_VERSION_MIN_REQUIRED < 101200 + 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; + + while (from_now.tv_nsec >= NSEC_PER_SEC) + { + from_now.tv_sec++; + from_now.tv_nsec -= NSEC_PER_SEC; + } + + return from_now; +} +#endif + +/* Mutex Functions */ /** * Create a new mutex @@ -205,495 +253,118 @@ int Paho_thread_destroy_mutex(mutex_type mutex) } -/** - * Get the thread id of the thread from which this function is called - * @return thread id, type varying according to OS - */ -thread_id_type Paho_thread_getid(void) -{ - #if defined(_WIN32) - return GetCurrentThreadId(); - #else - return pthread_self(); - #endif -} - +/* Event functions */ /** - * 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; -} - - + evt_type evt = NULL; #if !defined(_WIN32) - -/** - * Create a new condition variable - * @return the condition variable struct - */ -cond_type Thread_create_cond(int *rc) -{ - cond_type condvar = NULL; pthread_condattr_t attr; +#endif FUNC_ENTRY; *rc = -1; - pthread_condattr_init(&attr); - -#if 0 - /* in theory, a monotonic clock should be able to be used. However on at least - * one system reported, even though setclock() reported success, it didn't work. - */ - if ((rc = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC)) == 0) - use_clock_monotonic = 1; - else - Log(LOG_ERROR, -1, "Error %d calling pthread_condattr_setclock(CLOCK_MONOTONIC)", rc); -#endif - - condvar = malloc(sizeof(cond_type_struct)); - if (condvar) - { - *rc = pthread_cond_init(&condvar->cond, &attr); - *rc = pthread_mutex_init(&condvar->mutex, NULL); - } - - FUNC_EXIT_RC(*rc); - return condvar; -} - -/** - * Signal a condition variable - * @return completion code - */ -int Thread_signal_cond(cond_type condvar) -{ - int rc = 0; - - FUNC_ENTRY; - pthread_mutex_lock(&condvar->mutex); - rc = pthread_cond_signal(&condvar->cond); - pthread_mutex_unlock(&condvar->mutex); - - FUNC_EXIT_RC(rc); - return rc; -} - -/** - * 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 rc = 0; - struct timespec cond_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 - - 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(&condvar->mutex); - rc = pthread_cond_timedwait(&condvar->cond, &condvar->mutex, &cond_timeout); - pthread_mutex_unlock(&condvar->mutex); - - FUNC_EXIT_RC(rc); - return rc; -} - -/** - * Destroy a condition variable - * @return completion code - */ -int Thread_destroy_cond(cond_type condvar) -{ - int rc = 0; - - rc = pthread_mutex_destroy(&condvar->mutex); - rc = pthread_cond_destroy(&condvar->cond); - free(condvar); - - return rc; -} -#endif - - -#if defined(THREAD_UNIT_TESTS) - -#if defined(_WIN32) || defined(_WINDOWS) -#define mqsleep(A) Sleep(1000*A) -#define START_TIME_TYPE DWORD -static DWORD start_time = 0; -START_TIME_TYPE start_clock(void) -{ - return GetTickCount(); -} -#elif defined(AIX) -#define mqsleep sleep -#define START_TIME_TYPE struct timespec -START_TIME_TYPE start_clock(void) -{ - static struct timespec start; - clock_gettime(CLOCK_REALTIME, &start); - return start; -} -#else -#define mqsleep sleep -#define START_TIME_TYPE struct timeval -/* TODO - unused - remove? static struct timeval start_time; */ -START_TIME_TYPE start_clock(void) -{ - struct timeval start_time; - gettimeofday(&start_time, NULL); - return start_time; -} -#endif - - #if defined(_WIN32) -long elapsed(START_TIME_TYPE start_time) -{ - return GetTickCount() - start_time; -} -#elif defined(AIX) -#define assert(a) -long elapsed(struct timespec start) -{ - struct timespec now, res; - - clock_gettime(CLOCK_REALTIME, &now); - ntimersub(now, start, res); - return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L; -} + evt = CreateEvent( + NULL, /* default security attributes */ + FALSE, /* manual-reset event? */ + FALSE, /* initial state is nonsignaled */ + NULL /* object name */ + ); + if (rc) + *rc = (evt == NULL) ? GetLastError() : 0; #else -long elapsed(START_TIME_TYPE start_time) -{ - struct timeval now, res; + pthread_condattr_init(&attr); - gettimeofday(&now, NULL); - timersub(&now, &start_time, &res); - return (res.tv_sec)*1000 + (res.tv_usec)/1000; -} -#endif - - -int tests = 0, failures = 0; - -void myassert(char* filename, int lineno, char* description, int value, char* format, ...) -{ - ++tests; - if (!value) + evt = malloc(sizeof(evt_type_struct)); + if (evt) { - va_list args; - - ++failures; - printf("Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description); - - va_start(args, format); - vprintf(format, args); - va_end(args); - - //cur_output += sprintf(cur_output, "file %s, line %d \n", - // description, filename, lineno); + *rc = pthread_cond_init(&evt->cond, &attr); + *rc = pthread_mutex_init(&evt->mutex, NULL); + evt->val = 0; } - else - printf("Assertion succeeded, file %s, line %d, description: %s\n", filename, lineno, description); -} - -#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d) -#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e) - -#include - -thread_return_type cond_secondary(void* n) -{ - int rc = 0; - cond_type cond = n; - - printf("This should return immediately as it was posted already\n"); - rc = Thread_wait_cond(cond, 99999); - assert("rc 1 from wait_cond", rc == 1, "rc was %d", rc); - - printf("This should hang around a few seconds\n"); - rc = Thread_wait_cond(cond, 99999); - assert("rc 1 from wait_cond", rc == 1, "rc was %d", rc); - - printf("Secondary cond thread ending\n"); - return 0; -} - - -int cond_test() -{ - int rc = 0; - cond_type cond = Thread_create_cond(); - - printf("Post secondary so it should return immediately\n"); - rc = Thread_signal_cond(cond); - assert("rc 0 from signal cond", rc == 0, "rc was %d", rc); - - printf("Starting secondary thread\n"); - Thread_start(cond_secondary, (void*)cond); - - sleep(3); - - printf("post secondary\n"); - rc = Thread_signal_cond(cond); - assert("rc 1 from signal cond", rc == 1, "rc was %d", rc); - - sleep(3); - - printf("Main thread ending\n"); - - return failures; -} - - -thread_return_type sem_secondary(void* n) -{ - int rc = 0; - sem_type sem = n; - - printf("Secondary semaphore pointer %p\n", sem); - - rc = Thread_check_sem(sem); - assert("rc 1 from check_sem", rc == 1, "rc was %d", rc); - - printf("Secondary thread about to wait\n"); - rc = Thread_wait_sem(sem, 99999); - printf("Secondary thread returned from wait %d\n", rc); - - printf("Secondary thread about to wait\n"); - rc = Thread_wait_sem(sem, 99999); - printf("Secondary thread returned from wait %d\n", rc); - printf("Secondary check sem %d\n", Thread_check_sem(sem)); - - printf("Secondary thread ending\n"); - return 0; -} - - -int sem_test() -{ - int rc = 0; - sem_type sem = Thread_create_sem(); - - printf("Primary semaphore pointer %p\n", sem); - - rc = Thread_check_sem(sem); - assert("rc 0 from check_sem", rc == 0, "rc was %d\n", rc); - - printf("post secondary so then check should be 1\n"); - rc = Thread_post_sem(sem); - assert("rc 0 from post_sem", rc == 0, "rc was %d\n", rc); - - rc = Thread_check_sem(sem); - assert("rc 1 from check_sem", rc == 1, "rc was %d", rc); - - printf("Starting secondary thread\n"); - Thread_start(sem_secondary, (void*)sem); - - sleep(3); - rc = Thread_check_sem(sem); - assert("rc 1 from check_sem", rc == 1, "rc was %d", rc); - - printf("post secondary\n"); - rc = Thread_post_sem(sem); - assert("rc 1 from post_sem", rc == 1, "rc was %d", rc); - - sleep(3); - - printf("Main thread ending\n"); - - return failures; -} - - -int main(int argc, char *argv[]) -{ - sem_test(); - //cond_test(); -} - #endif + FUNC_EXIT_RC(*rc); + return evt; +} + +/** + * Signal an event + * @return completion code, 0 is success + */ +int Thread_signal_evt(evt_type evt) +{ + int rc = 0; + + FUNC_ENTRY; +#if defined(_WIN32) + if (SetEvent(evt) == 0) + rc = GetLastError(); +#else + pthread_mutex_lock(&evt->mutex); + evt->val = 1; + // TODO: Determine if this could be more efficient with pthread_cond_signal() + rc = pthread_cond_broadcast(&evt->cond); + pthread_mutex_unlock(&evt->mutex); +#endif + FUNC_EXIT_RC(rc); + return rc; +} + +/** + * Wait with a timeout (ms) for the event to become signaled. + * @return The completion code: 0 for success, ETIMEDOUT otherwise + */ +int Thread_wait_evt(evt_type evt, int timeout_ms) +{ + int rc = 0; + + FUNC_ENTRY; +#if defined(_WIN32) + /* returns 0 (WAIT_OBJECT_0) on success, non-zero (WAIT_TIMEOUT) if timeout occurred */ + rc = WaitForSingleObject(evt, timeout_ms < 0 ? 0 : timeout_ms); + if (rc == WAIT_TIMEOUT) + rc = ETIMEDOUT; +#else + struct timespec evt_timeout; + evt_timeout = Thread_time_from_now(timeout_ms); + + 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_unlock(&evt->mutex); +#endif + FUNC_EXIT_RC(rc); + return rc; +} + +/** + * Destroy an event object + * @return completion code, 0 for success + */ +int Thread_destroy_evt(evt_type evt) +{ + int rc = 0; + + FUNC_ENTRY; +#if defined(_WIN32) + rc = CloseHandle(evt); +#else + rc = pthread_mutex_destroy(&evt->mutex); + int rcc = pthread_cond_destroy(&evt->cond); + if (rcc != 0) + rc = rcc; + free(evt); +#endif + FUNC_EXIT_RC(rc); + return rc; +} + diff --git a/src/Thread.h b/src/Thread.h index bb249aee..0cf428be 100644 --- a/src/Thread.h +++ b/src/Thread.h @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2023 IBM Corp. + * Copyright (c) 2009, 2026 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -15,6 +15,7 @@ * Ian Craggs, Allan Stockdill-Mander - async client updates * Ian Craggs - fix for bug #420851 * Ian Craggs - change MacOS semaphore implementation + * Frank Pagliughi - Consolidated semaphores and conditions into "events" *******************************************************************************/ #if !defined(THREAD_H) @@ -33,56 +34,44 @@ #include "MQTTClient.h" -#include "mutex_type.h" /* Needed for mutex_type */ - #if defined(_WIN32) #include + + #define mutex_type HANDLE #define thread_type HANDLE #define thread_id_type DWORD #define thread_return_type DWORD #define thread_fn LPTHREAD_START_ROUTINE - #define cond_type HANDLE - #define sem_type HANDLE + #define evt_type HANDLE #undef ETIMEDOUT #define ETIMEDOUT WSAETIMEDOUT #else #include + #define mutex_type pthread_mutex_t* #define thread_type pthread_t #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; - #if defined(OSX) - #include - typedef dispatch_semaphore_t sem_type; - #else - #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); + typedef struct { pthread_cond_t cond; pthread_mutex_t mutex; int val; } evt_type_struct; + typedef evt_type_struct *evt_type; #endif +/* Thread functions */ LIBMQTT_API void Paho_thread_start(thread_fn, void*); int Thread_set_name(const char* thread_name); +LIBMQTT_API thread_id_type Paho_thread_getid(); +/* Mutex functions */ LIBMQTT_API mutex_type Paho_thread_create_mutex(int*); LIBMQTT_API int Paho_thread_lock_mutex(mutex_type); LIBMQTT_API int Paho_thread_unlock_mutex(mutex_type); int Paho_thread_destroy_mutex(mutex_type); -LIBMQTT_API thread_id_type Paho_thread_getid(); - -sem_type Thread_create_sem(int*); -int Thread_wait_sem(sem_type sem, int timeout); -int Thread_check_sem(sem_type sem); -int Thread_post_sem(sem_type sem); -int Thread_destroy_sem(sem_type sem); - +/* Event Functions */ +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); #endif diff --git a/src/mutex_type.h b/src/mutex_type.h deleted file mode 100644 index 1203055a..00000000 --- a/src/mutex_type.h +++ /dev/null @@ -1,25 +0,0 @@ -/******************************************************************************* - * Copyright (c) 2009, 2018 IBM Corp. - * - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v2.0 - * and Eclipse Distribution License v1.0 which accompany this distribution. - * - * The Eclipse Public License is available at - * https://www.eclipse.org/legal/epl-2.0/ - * and the Eclipse Distribution License is available at - * http://www.eclipse.org/org/documents/edl-v10.php. - * - *******************************************************************************/ -#if !defined(_MUTEX_TYPE_H_) -#define _MUTEX_TYPE_H_ - -#if defined(_WIN32) - #include - #define mutex_type HANDLE -#else - #include - #define mutex_type pthread_mutex_t* -#endif - -#endif /* _MUTEX_TYPE_H_ */ diff --git a/test/thread.c b/test/thread.c index 825170a2..50a06b98 100644 --- a/test/thread.c +++ b/test/thread.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2025 IBM Corp. + * Copyright (c) 2009, 2026 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -226,159 +226,53 @@ 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, "\n"); fclose(xml);