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

Started merging semaphores and condition variables into a single 'event' type.

This commit is contained in:
fpagliughi 2025-01-09 23:42:03 -05:00
parent b90a389380
commit 2b6c15678b
6 changed files with 224 additions and 436 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/MQTTAsync.c
View File

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

40
src/MQTTAsyncUtils.c
View File

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

75
src/MQTTClient.c
View File

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

340
src/Thread.c
View File

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

14
src/Thread.h
View File

@ -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 <dispatch/dispatch.h>
typedef dispatch_semaphore_t sem_type;
@ -62,11 +62,6 @@
#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);
#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);

179
test/thread.c
View File

@ -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, "<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();
<<<<<<< HEAD
rc = Thread_wait_cond(cond, 2000);
=======
rc = Thread_wait_evt(evt, 2000);
>>>>>>> 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, "</testsuite>\n");
fclose(xml);