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

'event' type working on Windows

This commit is contained in:
fpagliughi 2025-01-11 00:06:38 -05:00
parent 2b6c15678b
commit c63170fc59
4 changed files with 49 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/MQTTAsync.c
View File

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

24
src/MQTTAsyncUtils.c
View File

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

101
src/Thread.c
View File

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

8
src/Thread.h
View File

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