natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'fpagliughi-evt-sync' into develop

This commit is contained in:
Ian Craggs 2026-01-29 16:33:43 +00:00
parent 47c8d7ef87 c484bd985d
commit 9c2ee681cf
No known key found for this signature in database
GPG Key ID: A7AE1A8F2CCAB186
8 changed files with 310 additions and 815 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
src/MQTTAsync.c
View File

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

49
src/MQTTAsyncUtils.c
View File

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

121
src/MQTTClient.c
View File

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

4
src/Socket.h
View File

@ -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

633
src/Thread.c
View File

@ -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, "<failure type=\"%s\">file %s, line %d </failure>\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 <stdio.h>
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;
}

43
src/Thread.h
View File

@ -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 <windows.h>
#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 <pthread.h>
#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 <dispatch/dispatch.h>
typedef dispatch_semaphore_t sem_type;
#else
#include <semaphore.h>
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

25
src/mutex_type.h
View File

@ -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 <windows.h>
#define mutex_type HANDLE
#else
#include <pthread.h>
#define mutex_type pthread_mutex_t*
#endif
#endif /* _MUTEX_TYPE_H_ */

185
test/thread.c
View File

@ -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, "<testcase classname=\"test\" name=\"%s\"", testname);
global_start_time = start_clock();
MyLog(LOGA_DEBUG, "Primary semaphore pointer %p\n", sem);
/* The semaphore should be created non-signaled */
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d\n", rc);
MyLog(LOGA_DEBUG, "Post semaphore 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);
/* should be 1, and then reset to 0 */
rc = Thread_check_sem(sem);
assert("rc 1 from check_sem", rc == 1, "rc was %d", rc);
/* so now it'll be 0 */
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
/* multiple posts */
for (i = 0; i < 10; ++i)
{
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);
// Binary sem, so additional checks should be zero
for (i = 0; i < 10; ++i)
{
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
}
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Check timeout");
start = start_clock();
rc = Thread_wait_sem(sem, 1500);
duration = elapsed(start);
#if defined(EAGAIN)
assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT || rc == EAGAIN, "rc was %d", rc);
#else
assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT, "rc was %d", rc);
#endif
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
assert("duration is 2s", duration >= 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, "<testcase classname=\"cond\" name=\"%s\"", testname);
MyLog(LOGA_INFO, "Starting event test");
fprintf(xml, "<testcase classname=\"evt\" name=\"%s\"", testname);
global_start_time = start_clock();
/* The semaphore should be created non-signaled */
rc = Thread_wait_cond(cond, 0);
assert("rc 0 from wait_cond", rc == ETIMEDOUT, "rc was %d", rc);
rc = Thread_wait_evt(evt, 0);
assert("rc 0 from wait_evt", rc == ETIMEDOUT, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Check timeout");
start = start_clock();
rc = Thread_wait_cond(cond, 2000);
rc = Thread_wait_evt(evt, 2000);
duration = elapsed(start);
assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT, "rc was %d", rc);
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
@ -387,42 +281,44 @@ 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);
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);
MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
(failures == 0) ? "passed" : "failed", testname, tests, failures);
write_test_result();
return failures;
}
#endif
static thread_return_type WINAPI mutex_secondary(void* n)
@ -458,6 +354,7 @@ int test_mutex(struct Options options)
MyLog(LOGA_INFO, "Starting mutex test");
fprintf(xml, "<testcase classname=\"test\" name=\"%s\"", testname);
global_start_time = start_clock();
assert("rc 0 from create mutex", rc == 0, "rc was %d", rc);
/* this should happen immediately, as there is no other lock held */
start = start_clock();
@ -497,12 +394,10 @@ 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,
test_sem,
#if !defined(_WIN32)
test_cond
#endif
test_evt
}; /* indexed starting from 1 */
int i;
@ -532,11 +427,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, "</testsuite>\n");
fclose(xml);