From c63170fc5957bbaceb52175f8c7a252e1788551e Mon Sep 17 00:00:00 2001 From: fpagliughi Date: Sat, 11 Jan 2025 00:06:38 -0500 Subject: [PATCH] 'event' type working on Windows --- src/MQTTAsync.c | 10 ++--- src/MQTTAsyncUtils.c | 24 +++++++++- src/Thread.c | 101 +++++++++---------------------------------- src/Thread.h | 8 +--- 4 files changed, 49 insertions(+), 94 deletions(-) diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 2db7680b..44883af7 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -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; +sem_type send_evt = NULL; #if !defined(NO_HEAP_TRACKING) extern mutex_type stack_mutex; extern mutex_type heap_mutex; @@ -154,7 +154,7 @@ int MQTTAsync_init(void) printf("mqttcommand_mutex error %d\n", rc); goto exit; } - if ((send_sem = CreateEvent( + if ((send_evt = CreateEvent( NULL, /* default security attributes */ FALSE, /* manual-reset event? */ FALSE, /* initial state is nonsignaled */ @@ -162,7 +162,7 @@ int MQTTAsync_init(void) )) == NULL) { rc = GetLastError(); - printf("send_sem error %d\n", rc); + printf("send_evt error %d\n", rc); goto exit; } #if !defined(NO_HEAP_TRACKING) @@ -202,8 +202,8 @@ exit: void MQTTAsync_cleanup(void) { - if (send_sem) - CloseHandle(send_sem); + if (send_evt) + CloseHandle(send_evt); #if !defined(NO_HEAP_TRACKING) if (stack_mutex) CloseHandle(stack_mutex); diff --git a/src/MQTTAsyncUtils.c b/src/MQTTAsyncUtils.c index 93f49e7a..16c46ab7 100644 --- a/src/MQTTAsyncUtils.c +++ b/src/MQTTAsyncUtils.c @@ -88,7 +88,7 @@ extern int MQTTAsync_tostop; extern mutex_type mqttasync_mutex; extern mutex_type socket_mutex; extern mutex_type mqttcommand_mutex; -extern sem_type send_sem; +extern evt_type send_evt; #if !defined(NO_HEAP_TRACKING) extern mutex_type stack_mutex; extern mutex_type heap_mutex; @@ -938,6 +938,13 @@ exit: MQTTAsync_unlock_mutex(mqttcommand_mutex); if ((rc1 = Thread_signal_evt(send_evt)) != 0) Log(LOG_ERROR, 0, "Error %d from signal event", rc1); +<<<<<<< HEAD +======= +#else + if ((rc1 = Thread_signal_evt(send_evt)) != 0) + Log(LOG_ERROR, 0, "Error %d from signal cond", rc1); +#endif +>>>>>>> 3137b57 ('event' type working on Windows) FUNC_EXIT_RC(rc); return rc; } @@ -1857,6 +1864,13 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n) } if ((rc = Thread_wait_evt(send_evt, timeout)) != 0 && rc != ETIMEDOUT) Log(LOG_ERROR, -1, "Error %d waiting for send event", rc); +<<<<<<< HEAD +======= +#else + if ((rc = Thread_wait_evt(send_evt, timeout)) != 0 && rc != ETIMEDOUT) + Log(LOG_ERROR, -1, "Error %d waiting for semaphore", rc); +#endif +>>>>>>> 3137b57 ('event' type working on Windows) timeout = 1000; /* 1 second for follow on waits */ MQTTAsync_checkTimeouts(); } @@ -2062,6 +2076,12 @@ static int MQTTAsync_completeConnection(MQTTAsyncs* m, Connack* connack) } m->pack = NULL; Thread_signal_evt(send_evt); +<<<<<<< HEAD +======= +#else + Thread_signal_evt(send_evt); +#endif +>>>>>>> 3137b57 ('event' type working on Windows) } FUNC_EXIT_RC(rc); return rc; @@ -2392,7 +2412,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) Thread_signal_evt(send_evt); #else if (sendThread_state != STOPPED) - Thread_post_sem(send_sem); + Thread_signal_evt(send_evt); #endif #if defined(OPENSSL) diff --git a/src/Thread.c b/src/Thread.c index e158955f..11d284f2 100644 --- a/src/Thread.c +++ b/src/Thread.c @@ -116,6 +116,7 @@ int Thread_set_name(const char* thread_name) return rc; } +#if !defined(_WIN32) struct timespec Thread_time_from_now(int ms) { struct timespec from_now; @@ -143,6 +144,7 @@ struct timespec Thread_time_from_now(int ms) return from_now; } +#endif /** * Create a new mutex @@ -366,11 +368,11 @@ sem_type Thread_create_evt(int *rc) * @param timeout the maximum time to wait, in milliseconds * @return completion code */ -int Thread_wait_sem(sem_type sem, int timeout) +int Thread_wait_evt(evt_type evt, 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); + int rc = WaitForSingleObject(evt, timeout < 0 ? 0 : timeout); if (rc == WAIT_TIMEOUT) rc = ETIMEDOUT; FUNC_EXIT_RC(rc); @@ -378,15 +380,15 @@ int Thread_wait_sem(sem_type sem, int timeout) } /** - * Post a semaphore - * @param sem the semaphore + * Post an event + * @param evt the event * @return 0 on success */ -int Thread_post_sem(sem_type sem) +int Thread_signal_evt(evt_type evt) { FUNC_ENTRY; int rc = 0; - if (SetEvent(sem) == 0) + if (SetEvent(evt) == 0) rc = GetLastError(); FUNC_EXIT_RC(rc); return rc; @@ -399,7 +401,7 @@ int Thread_post_sem(sem_type sem) int Thread_destroy_evt(evt_type evt) { FUNC_ENTRY; - int rc = CloseHandle(sem); + int rc = CloseHandle(evt); FUNC_EXIT_RC(rc); return rc; } @@ -497,29 +499,29 @@ void myassert(char* filename, int lineno, char* description, int value, char* fo thread_return_type cond_secondary(void* n) { int rc = 0; - cond_type cond = n; + evt_type evt = 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); + rc = Thread_wait_evt(evt, 99999); + assert("rc 1 from wait_evt", 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); + rc = Thread_wait_evt(evt, 99999); + assert("rc 1 from wait_evt", rc == 1, "rc was %d", rc); - printf("Secondary cond thread ending\n"); + printf("Secondary evt thread ending\n"); return 0; } -int cond_test() +int evt_test() { int rc = 0; - cond_type cond = Thread_create_cond(); + evt_type evt = Thread_create_evt(); 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); + rc = Thread_signal_evt(evt); + assert("rc 0 from signal evt", rc == 0, "rc was %d", rc); printf("Starting secondary thread\n"); Thread_start(cond_secondary, (void*)cond); @@ -527,8 +529,8 @@ int cond_test() sleep(3); printf("post secondary\n"); - rc = Thread_signal_cond(cond); - assert("rc 1 from signal cond", rc == 1, "rc was %d", rc); + rc = Thread_signal_evt(evt); + assert("rc 1 from signal evt", rc == 1, "rc was %d", rc); sleep(3); @@ -537,67 +539,6 @@ int cond_test() 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(); diff --git a/src/Thread.h b/src/Thread.h index 924ff181..e682ab7b 100644 --- a/src/Thread.h +++ b/src/Thread.h @@ -42,6 +42,7 @@ #define thread_return_type DWORD #define thread_fn LPTHREAD_START_ROUTINE #define cond_type HANDLE + #define evt_type HANDLE #define sem_type HANDLE #undef ETIMEDOUT #define ETIMEDOUT WSAETIMEDOUT @@ -79,11 +80,4 @@ 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); -int Thread_post_sem(sem_type sem); -int Thread_destroy_sem(sem_type sem); - - #endif