From 5c2e1e20943144b12120a744d014e00ef8a80cca Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Wed, 1 Oct 2014 16:28:55 +0100 Subject: [PATCH 1/7] Incorrect free in FreeCommand1 Bug: 444934 --- CONTRIBUTING.md | 32 +++++++++++++------------------- src/MQTTAsync.c | 13 ++++++------- 2 files changed, 19 insertions(+), 26 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 56fbb1eb..a830aafd 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -9,7 +9,8 @@ Project description: The Paho project has been created to provide scalable open-source implementations of open and standard messaging protocols aimed at new, existing, and emerging applications for Machine-to-Machine (M2M) and Internet of Things (IoT). Paho reflects the inherent physical and cost constraints of device connectivity. Its objectives include effective levels of decoupling between devices and applications, designed to keep markets open and encourage the rapid growth of scalable Web and Enterprise middleware and applications. Paho is being kicked off with MQTT publish/subscribe client implementations for use on embedded platforms, along with corresponding server support as determined by the community. -- https://projects.eclipse.org/projects/technology.paho +- [Project web site](https://www.eclipse.org/paho) +- [Project information](https://projects.eclipse.org/projects/iot.paho) Source ------ @@ -19,9 +20,7 @@ The Paho C client is stored in a git repository. The URLs to access it are: ssh://@git.eclipse.org:29418/paho/org.eclipse.paho.mqtt.c https://@git.eclipse.org/r/paho/org.eclipse.paho.mqtt.c -A web browsable repository is available at - -http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt.c.git +A [web browsable repository](http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt.c.git) is available. Contributing a patch -------------------- @@ -32,14 +31,14 @@ changes and push them back for review and eventual acceptance into the project. To do this, you must follow a few steps. The first of these are described at -- https://wiki.eclipse.org/Development_Resources/Contributing_via_Git +- [Contributing via git](https://wiki.eclipse.org/Development_Resources/Contributing_via_Git) * Sign the Eclipse CLA * Use a valid commit record, including a signed-off-by entry. There are further details at -- https://wiki.eclipse.org/Development_Resources/Handling_Git_Contributions +- [Handling Git Contributions](https://wiki.eclipse.org/Development_Resources/Handling_Git_Contributions) Once the patch is pushed back to Gerrit, the project committers will be informed and they will undertake a review of the code. The patch may need @@ -53,41 +52,36 @@ change. What happens next depends on the content of the patch. If it is 100% authored by the contributor and is less than 1000 lines (and meets the needs of the project), then it can be committed to the main repository. If not, more steps -are required. These are detailed in the legal process poster: +are required. These are detailed in the +[legal process poster](http://www.eclipse.org/legal/EclipseLegalProcessPoster.pdf). -- http://www.eclipse.org/legal/EclipseLegalProcessPoster.pdf Developer resources: -------------------- Information regarding source code management, builds, coding standards, and more. -- https://projects.eclipse.org/projects/technology.paho/developer +- [https://projects.eclipse.org/projects/iot.paho/developer](https://projects.eclipse.org/projects/iot.paho/developer) Contributor License Agreement: ------------------------------ -Before your contribution can be accepted by the project, you need to create and electronically sign the Eclipse Foundation Contributor License Agreement (CLA). - -- http://www.eclipse.org/legal/CLA.php +Before your contribution can be accepted by the project, you need to create and electronically sign the Eclipse Foundation [Contributor License Agreement (CLA)](http://www.eclipse.org/legal/CLA.php). Contact: -------- -Contact the project developers via the project's "dev" list. - -- https://dev.eclipse.org/mailman/listinfo/paho-dev +Contact the project developers via the project's development +[mailing list](https://dev.eclipse.org/mailman/listinfo/paho-dev). Search for bugs: ---------------- -This project uses Bugzilla to track ongoing development and issues. - -- https://bugs.eclipse.org/bugs/buglist.cgi?product=Paho +This project uses [Bugzilla](https://bugs.eclipse.org/bugs/buglist.cgi?product=Paho) to track ongoing development and issues. Create a new bug: ----------------- Be sure to search for existing bugs before you create another one. Remember that contributions are always welcome! -- https://bugs.eclipse.org/bugs/enter_bug.cgi?product=Paho +- [Create new Paho bug](https://bugs.eclipse.org/bugs/enter_bug.cgi?product=Paho) diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index c464acd8..5e2d096c 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -22,6 +22,7 @@ * Ian Craggs - MQTT 3.1.1 support * Rong Xiang, Ian Craggs - C++ compatibility * Ian Craggs - fix for bug 442400: reconnecting after network cable unplugged + * Ian Craggs - fix for bug 444934 - incorrect free in freeCommand1 *******************************************************************************/ /** @@ -840,21 +841,19 @@ void MQTTAsync_freeCommand1(MQTTAsync_queuedCommand *command) int i; for (i = 0; i < command->command.details.sub.count; i++) - { free(command->command.details.sub.topics[i]); - free(command->command.details.sub.topics); - free(command->command.details.sub.qoss); - } + + free(command->command.details.sub.topics); + free(command->command.details.sub.qoss); } else if (command->command.type == UNSUBSCRIBE) { int i; for (i = 0; i < command->command.details.unsub.count; i++) - { free(command->command.details.unsub.topics[i]); - free(command->command.details.unsub.topics); - } + + free(command->command.details.unsub.topics); } else if (command->command.type == PUBLISH) { From 55fa0087e6606d4aaf921c3e0ef31e7aec0691cd Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Fri, 3 Oct 2014 16:26:21 +0100 Subject: [PATCH 2/7] Thread safe message id assignment Bug: 445891 --- src/MQTTAsync.c | 67 +++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 62 insertions(+), 5 deletions(-) diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 5e2d096c..512d049d 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -23,6 +23,7 @@ * Rong Xiang, Ian Craggs - C++ compatibility * Ian Craggs - fix for bug 442400: reconnecting after network cable unplugged * Ian Craggs - fix for bug 444934 - incorrect free in freeCommand1 + * Ian Craggs - fix for bug 445891 - assigning msgid is not thread safe *******************************************************************************/ /** @@ -76,6 +77,8 @@ enum MQTTAsync_threadStates enum MQTTAsync_threadStates sendThread_state = STOPPED; enum MQTTAsync_threadStates receiveThread_state = STOPPED; +static thread_id_type sendThread_id = 0, + receiveThread_id = 0; #if defined(WIN32) || defined(WIN64) static mutex_type mqttasync_mutex = NULL; @@ -326,7 +329,7 @@ void MQTTAsync_lock_mutex(mutex_type amutex) { int rc = Thread_lock_mutex(amutex); if (rc != 0) - Log(LOG_ERROR, 0, "Error %d locking mutex", rc); + Log(LOG_ERROR, 0, "Error %s locking mutex", strerror(rc)); } @@ -334,7 +337,7 @@ void MQTTAsync_unlock_mutex(mutex_type amutex) { int rc = Thread_unlock_mutex(amutex); if (rc != 0) - Log(LOG_ERROR, 0, "Error %d unlocking mutex", rc); + Log(LOG_ERROR, 0, "Error %s unlocking mutex", strerror(rc)); } @@ -1254,6 +1257,7 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n) FUNC_ENTRY; MQTTAsync_lock_mutex(mqttasync_mutex); sendThread_state = RUNNING; + sendThread_id = Thread_getid(); MQTTAsync_unlock_mutex(mqttasync_mutex); while (!tostop) { @@ -1280,6 +1284,7 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n) sendThread_state = STOPPING; MQTTAsync_lock_mutex(mqttasync_mutex); sendThread_state = STOPPED; + sendThread_id = 0; MQTTAsync_unlock_mutex(mqttasync_mutex); FUNC_EXIT; return 0; @@ -1454,6 +1459,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) FUNC_ENTRY; MQTTAsync_lock_mutex(mqttasync_mutex); receiveThread_state = RUNNING; + receiveThread_id = Thread_getid(); while (!tostop) { int rc = SOCKET_ERROR; @@ -1670,6 +1676,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n) } } receiveThread_state = STOPPED; + receiveThread_id = 0; MQTTAsync_unlock_mutex(mqttasync_mutex); #if !defined(WIN32) && !defined(WIN64) if (sendThread_state != STOPPED) @@ -2157,6 +2164,56 @@ int MQTTAsync_isConnected(MQTTAsync handle) } +int cmdMessageIDCompare(void* a, void* b) +{ + MQTTAsync_queuedCommand* cmd = (MQTTAsync_queuedCommand*)a; + return cmd->command.token == *(int*)b; +} + + +/** + * Assign a new message id for a client. Make sure it isn't already being used and does + * not exceed the maximum. + * @param m a client structure + * @return the next message id to use, or 0 if none available + */ +int MQTTAsync_assignMsgId(MQTTAsyncs* m) +{ + int start_msgid = m->c->msgID; + int msgid = start_msgid; + thread_id_type thread_id = 0; + int locked = 0; + + /* need to check: commands list and response list for a client */ + FUNC_ENTRY; + /* We might be called in a callback. In which case, this mutex will be already locked. */ + thread_id = Thread_getid(); + if (thread_id != sendThread_id && thread_id != receiveThread_id) + { + MQTTAsync_lock_mutex(mqttasync_mutex); + locked = 1; + } + + msgid = (msgid == MAX_MSG_ID) ? 1 : msgid + 1; + while (ListFindItem(commands, &msgid, cmdMessageIDCompare) || + ListFindItem(m->responses, &msgid, cmdMessageIDCompare)) + { + msgid = (msgid == MAX_MSG_ID) ? 1 : msgid + 1; + if (msgid == start_msgid) + { /* we've tried them all - none free */ + msgid = 0; + break; + } + } + if (msgid != 0) + m->c->msgID = msgid; + if (locked) + MQTTAsync_unlock_mutex(mqttasync_mutex); + FUNC_EXIT_RC(msgid); + return msgid; +} + + int MQTTAsync_subscribeMany(MQTTAsync handle, size_t count, char* const* topic, int* qos, MQTTAsync_responseOptions* response) { MQTTAsyncs* m = handle; @@ -2189,7 +2246,7 @@ int MQTTAsync_subscribeMany(MQTTAsync handle, size_t count, char* const* topic, goto exit; } } - if ((msgid = MQTTProtocol_assignMsgId(m->c)) == 0) + if ((msgid = MQTTAsync_assignMsgId(m)) == 0) { rc = MQTTASYNC_NO_MORE_MSGIDS; goto exit; @@ -2262,7 +2319,7 @@ int MQTTAsync_unsubscribeMany(MQTTAsync handle, size_t count, char* const* topic goto exit; } } - if ((msgid = MQTTProtocol_assignMsgId(m->c)) == 0) + if ((msgid = MQTTAsync_assignMsgId(m)) == 0) { rc = MQTTASYNC_NO_MORE_MSGIDS; goto exit; @@ -2321,7 +2378,7 @@ int MQTTAsync_send(MQTTAsync handle, const char* destinationName, size_t payload rc = MQTTASYNC_BAD_UTF8_STRING; else if (qos < 0 || qos > 2) rc = MQTTASYNC_BAD_QOS; - else if (qos > 0 && (msgid = MQTTProtocol_assignMsgId(m->c)) == 0) + else if (qos > 0 && (msgid = MQTTAsync_assignMsgId(m)) == 0) rc = MQTTASYNC_NO_MORE_MSGIDS; if (rc != MQTTASYNC_SUCCESS) From 5a5b163dedb5856e18cba6a5d8fd39a8c4aa9e1d Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Fri, 3 Oct 2014 16:28:33 +0100 Subject: [PATCH 3/7] No DUP flag on PUBREL Bug: 445890 --- src/MQTTProtocolClient.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index d7314db9..fd97b1fa 100644 --- a/src/MQTTProtocolClient.c +++ b/src/MQTTProtocolClient.c @@ -16,6 +16,7 @@ * Ian Craggs - fix for bug 413429 - connectionLost not called * Ian Craggs - fix for bug 421103 - trying to write to same socket, in retry * Rong Xiang, Ian Craggs - C++ compatibility + * Ian Craggs - turn off DUP flag for PUBREL - MQTT 3.1.1 *******************************************************************************/ /** @@ -596,7 +597,7 @@ void MQTTProtocol_retries(time_t now, Clients* client, int regardless) else if (m->qos && m->nextMessageType == PUBCOMP) { Log(TRACE_MIN, 7, NULL, "PUBREL", client->clientID, client->net.socket, m->msgid); - if (MQTTPacket_send_pubrel(m->msgid, 1, &client->net, client->clientID) != TCPSOCKET_COMPLETE) + if (MQTTPacket_send_pubrel(m->msgid, 0, &client->net, client->clientID) != TCPSOCKET_COMPLETE) { client->good = 0; Log(TRACE_PROTOCOL, 29, NULL, client->clientID, client->net.socket, From cf409bd15eb58248a55239b89f176ca33de7fc9d Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Fri, 3 Oct 2014 16:29:46 +0100 Subject: [PATCH 4/7] Test4.c updates for reliability --- test/test4.c | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/test/test4.c b/test/test4.c index 47a606fb..01bfe5bf 100644 --- a/test/test4.c +++ b/test/test4.c @@ -1143,6 +1143,16 @@ Test7: Persistence char* test7_topic = "C client test7"; int test7_messageCount = 0; +void test7_onDisconnectFailure(void* context, MQTTAsync_failureData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MyLog(LOGA_DEBUG, "In onDisconnect failure callback %p", c); + + assert("Successful disconnect", 0, "disconnect failed", 0); + + test_finished = 1; +} + void test7_onDisconnect(void* context, MQTTAsync_successData* response) { MQTTAsync c = (MQTTAsync)context; @@ -1212,6 +1222,24 @@ void test7_onConnect(void* context, MQTTAsync_successData* response) } +void test7_onConnectOnly(void* context, MQTTAsync_successData* response) +{ + MQTTAsync c = (MQTTAsync)context; + MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer; + int rc; + + MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context); + dopts.context = context; + dopts.timeout = 1000; + dopts.onSuccess = test7_onDisconnect; + MQTTAsync_disconnect(c, &dopts); + + assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + test_finished = 1; +} + + /********************************************************************* Test7: Pending tokens @@ -1248,7 +1276,6 @@ int test7(struct Options options) assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); opts.keepAliveInterval = 20; - opts.cleansession = 0; opts.username = "testuser"; opts.password = "testpassword"; opts.MQTTVersion = options.MQTTVersion; @@ -1259,11 +1286,30 @@ int test7(struct Options options) opts.will->retained = 0; opts.will->topicName = "will topic"; opts.will = NULL; - opts.onSuccess = test7_onConnect; + opts.onFailure = NULL; opts.context = c; + opts.cleansession = 1; + opts.onSuccess = test7_onConnectOnly; + MyLog(LOGA_DEBUG, "Connecting to clean up"); + rc = MQTTAsync_connect(c, &opts); + rc = 0; + assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); + if (rc != MQTTASYNC_SUCCESS) + goto exit; + + while (!test_finished) + #if defined(WIN32) + Sleep(100); + #else + usleep(10000L); + #endif + + test_finished = 0; MyLog(LOGA_DEBUG, "Connecting"); + opts.cleansession = 0; + opts.onSuccess = test7_onConnect; rc = MQTTAsync_connect(c, &opts); rc = 0; assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); @@ -1304,6 +1350,7 @@ int test7(struct Options options) /* disconnect immediately without receiving the incoming messages */ dopts.timeout = 0; dopts.onSuccess = test7_onDisconnect; + dopts.context = c; MQTTAsync_disconnect(c, &dopts); /* now there should be "orphaned" publications */ while (!test_finished) @@ -1371,6 +1418,8 @@ int test7(struct Options options) assertions fail against Mosquitto - needs testing */ + dopts.onFailure = test7_onDisconnectFailure; + dopts.onSuccess = test7_onDisconnect; dopts.timeout = 1000; MQTTAsync_disconnect(c, &dopts); From 04e69596c29b3db4bb4706fa3bb604acc2617dd0 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Fri, 3 Oct 2014 16:33:49 +0100 Subject: [PATCH 5/7] Remove compiler warning in test4.c --- test/test4.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/test4.c b/test/test4.c index 01bfe5bf..d378c736 100644 --- a/test/test4.c +++ b/test/test4.c @@ -1232,7 +1232,7 @@ void test7_onConnectOnly(void* context, MQTTAsync_successData* response) dopts.context = context; dopts.timeout = 1000; dopts.onSuccess = test7_onDisconnect; - MQTTAsync_disconnect(c, &dopts); + rc = MQTTAsync_disconnect(c, &dopts); assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc); if (rc != MQTTASYNC_SUCCESS) From 9909826e9bd3fd9eb089c04369e3e50f90c20b08 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Fri, 3 Oct 2014 16:40:31 +0100 Subject: [PATCH 6/7] Turn off some warnings --- src/Socket.c | 2 +- test/test4.c | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/src/Socket.c b/src/Socket.c index 2a8869a2..8475ae2b 100644 --- a/src/Socket.c +++ b/src/Socket.c @@ -98,7 +98,7 @@ int Socket_error(char* aString, int sock) if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) { if (strcmp(aString, "shutdown") != 0 || (errno != ENOTCONN && errno != ECONNRESET)) - Log(LOG_ERROR, -1, "Socket error %s in %s for socket %d", strerror(errno), aString, sock); + Log(TRACE_MINIMUM, -1, "Socket error %s in %s for socket %d", strerror(errno), aString, sock); } FUNC_EXIT_RC(errno); return errno; diff --git a/test/test4.c b/test/test4.c index d378c736..31edbf48 100644 --- a/test/test4.c +++ b/test/test4.c @@ -1180,7 +1180,6 @@ int test7_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync { MQTTAsync c = (MQTTAsync)context; static int message_count = 0; - int rc; MyLog(LOGA_DEBUG, "Test7: received message id %d", message->msgid); From 6d4cf04c653a0f04e8c2684afdde965a5a084f63 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Sat, 4 Oct 2014 11:57:56 +0100 Subject: [PATCH 7/7] Update version number to 1.0.2 --- build.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build.xml b/build.xml index 5b595fa5..27d89a0b 100644 --- a/build.xml +++ b/build.xml @@ -24,7 +24,7 @@ - +