diff --git a/Makefile b/Makefile index 63d745f5..76d26fba 100755 --- a/Makefile +++ b/Makefile @@ -96,7 +96,7 @@ SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}} TEST_FILES_CS = test3 SYNC_SSL_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_CS}} -TEST_FILES_A = test4 test9 test_mqtt4async +TEST_FILES_A = test4 test6 test9 test_mqtt4async test_issue373 ASYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_A}} TEST_FILES_AS = test5 diff --git a/test/test6.c b/test/test6.c index 492fc4bf..327dc373 100644 --- a/test/test6.c +++ b/test/test6.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2011, 2017 IBM Corp. + * Copyright (c) 2011, 2014 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 @@ -17,11 +17,12 @@ /** * @file - * Async C client program for the MQTT restart/recovery test suite + * Async C client program for the MQTT v3 restart/recovery test suite. */ - #include "MQTTAsync.h" +#define NO_HEAP_TRACKING +#include "Heap.h" #include #include @@ -44,12 +45,12 @@ static char sub_topic[200]; struct { - char* connection; /**< connection to system under test. */ + char* connection; /**< connection to system under test. */ char** connections; /**< HA connection list */ int connection_count; - char* control_connection; /**< MQTT control connection, for test sync */ - char* topic; /**< test message topic */ - char* control_topic; /**< topic for control messages */ + char* control_connection; /**< MQTT control connection, for test sync */ + char* topic; + char* control_topic; char* clientid; int slot_no; int qos; @@ -58,24 +59,22 @@ struct char* password; int verbose; int persistence; - int payload_len; } opts = { - "tcp://localhost:1884", + "tcp://localhost:1885", NULL, 0, "tcp://localhost:7777", "Eclipse/Paho/restart_test", "Eclipse/Paho/restart_test/control", "C_broken_client", - 1, /* slot_no */ - 0, /* QoS */ - 0, /* retained */ + 1, + 2, + 0, NULL, NULL, 0, - 1, - 1000 + 0, }; void getopts(int argc, char** argv) @@ -168,10 +167,6 @@ void getopts(int argc, char** argv) } } -#if 0 -#include /* For general log messages */ -#define MyLog logaLine -#else #define LOGA_DEBUG 0 #define LOGA_ALWAYS 1 #define LOGA_INFO 2 @@ -204,11 +199,17 @@ void MyLog(int log_level, char* format, ...) printf("%s\n", msg_buf); fflush(stdout); } -#endif +void MySleep(long milliseconds) +{ +#if defined(WIN32) || defined(WIN64) + Sleep(milliseconds); +#else + usleep(milliseconds*1000); +#endif +} #if defined(WIN32) || defined(_WINDOWS) -#define mysleep(A) Sleep(1000*A) #define START_TIME_TYPE DWORD static DWORD start_time = 0; START_TIME_TYPE start_clock(void) @@ -216,7 +217,6 @@ START_TIME_TYPE start_clock(void) return GetTickCount(); } #elif defined(AIX) -#define mysleep sleep #define START_TIME_TYPE struct timespec START_TIME_TYPE start_clock(void) { @@ -225,9 +225,8 @@ START_TIME_TYPE start_clock(void) return start; } #else -#define mysleep sleep #define START_TIME_TYPE struct timeval -static struct timeval start_time; +/* TODO - unused - remove? static struct timeval start_time; */ START_TIME_TYPE start_clock(void) { struct timeval start_time; @@ -236,41 +235,6 @@ START_TIME_TYPE start_clock(void) } #endif -int tests = 0; -int failures = 0; -FILE* xml; -START_TIME_TYPE global_start_time; -char output[3000]; -char* cur_output = output; - -#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) - -void myassert(char* filename, int lineno, char* description, int value, - char* format, ...) -{ - ++tests; - if (!value) - { - va_list args; - - ++failures; - MyLog(LOGA_INFO, "Assertion failed, file %s, line %d, description: %s", filename, - lineno, description); - - va_start(args, format); - vprintf(format, args); - va_end(args); - - cur_output += sprintf(cur_output, "file %s, line %d \n", - description, filename, lineno); - } - else - MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", - filename, lineno, description); -} - - #if defined(WIN32) long elapsed(START_TIME_TYPE start_time) { @@ -333,9 +297,13 @@ void control_connectionLost(void* context, char* cause) */ int control_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* m) { - MyLog(LOGA_INFO, "Control message arrived: %.*s", m->payloadlen, m->payload); - if (strcmp(m->payload, "stop") == 0) + MyLog(LOGA_ALWAYS, "Control message arrived: %.*s %s", + m->payloadlen, m->payload, (wait_message == NULL) ? "None" : wait_message); + if (strncmp(m->payload, "stop", 4) == 0) + { + MyLog(LOGA_ALWAYS, "Stop message arrived, stopping..."); stopping = 1; + } else if (wait_message != NULL && strncmp(wait_message, m->payload, strlen(wait_message)) == 0) { @@ -361,8 +329,8 @@ int control_send(char* message) int rc = 0; MQTTAsync_responseOptions ropts = MQTTAsync_responseOptions_initializer; - MyLog(LOGA_ALWAYS, "Sending control message: %s", message); sprintf(buf, "%s: %s", opts.clientid, message); + MyLog(LOGA_ALWAYS, "Sending control message: %s", message); rc = MQTTAsync_send(control_client, pub_topic, (int)strlen(buf), buf, 1, 0, &ropts); MyLog(LOGA_DEBUG, "Control message sent: %s", buf); @@ -381,9 +349,9 @@ int control_wait(char* message) wait_message = message; sprintf(buf, "waiting for: %s", message); - MyLog(LOGA_ALWAYS, "%s", buf); control_send(buf); + MyLog(LOGA_ALWAYS, "waiting for: %s", message); while (control_found == 0 && stopping == 0) { if (++count == 300) @@ -392,10 +360,8 @@ int control_wait(char* message) MyLog(LOGA_ALWAYS, "Failed to receive message %s, stopping ", message); return 0; /* time out and tell the caller the message was not found */ } - mysleep(1); + MySleep(1000); } - MyLog(LOGA_ALWAYS, (control_found == 0) ? - "Waited... not found" : "Waited... found %d" , control_found); return control_found; } @@ -412,7 +378,7 @@ int control_which(char* message1, char* message2) { if (++count == 300) return 0; /* time out and tell the caller the message was not found */ - mysleep(1); + MySleep(1000); } return control_found; } @@ -517,13 +483,14 @@ void connectionLost(void* context, char* cause) { conn_opts.serverURIcount = opts.connection_count; conn_opts.serverURIs = opts.connections; - printf("reconnecting to first serverURI %s\n", conn_opts.serverURIs[0]); } else { conn_opts.serverURIcount = 0; conn_opts.serverURIs = NULL; } + //printf("reconnecting to first serverURI %s\n", conn_opts.serverURIs[0]); + MyLog(LOGA_ALWAYS, "Starting reconnect attempt"); rc = MQTTAsync_connect(context, &conn_opts); if (rc != MQTTASYNC_SUCCESS) { @@ -544,6 +511,12 @@ int recreateReconnect(void) MQTTAsync_destroy(&client); /* destroy the client object so that we force persistence to be read on recreate */ + heap_info* mqtt_mem = 0; + mqtt_mem = Heap_get_info(); + MyLog(LOGA_INFO, "MQTT mem current %ld, max %ld",mqtt_mem->current_size,mqtt_mem->max_size); + //if (mqtt_mem->current_size > 20) + // HeapScan(5); + rc = MQTTAsync_create(&client, opts.connection, opts.clientid, MQTTCLIENT_PERSISTENCE_DEFAULT, NULL); if (rc != MQTTASYNC_SUCCESS) { @@ -612,7 +585,10 @@ int waitForCompletion(START_TIME_TYPE start_time) int wait_count = 0; int limit = 120; - mysleep(1); + MyLog(LOGA_ALWAYS, "Wait for completion"); + if (opts.qos == 0) + limit = 30; /* we aren't going to get back all QoS 0 messages anyway */ + MySleep(1000); while (arrivedCount < expectedCount) { if (arrivedCount > lastreport) @@ -621,15 +597,18 @@ int waitForCompletion(START_TIME_TYPE start_time) arrivedCount, expectedCount, elapsed(start_time) / 1000); lastreport = arrivedCount; } - mysleep(1); + MySleep(1000); if (opts.persistence && connection_lost) recreateReconnect(); if (++wait_count > limit || stopping) break; } last_completion_time = elapsed(start_time) / 1000; - MyLog(LOGA_ALWAYS, "Extra wait to see if any duplicates arrive"); - mysleep(10); /* check if any duplicate messages arrive */ + if (opts.qos > 0) + { + MyLog(LOGA_ALWAYS, "Extra wait to see if any duplicates arrive"); + MySleep(10000); /* check if any duplicate messages arrive */ + } MyLog(LOGA_ALWAYS, "%d messages arrived out of %d expected, in %d seconds", arrivedCount, expectedCount, elapsed(start_time) / 1000); return success(expectedCount); @@ -652,7 +631,6 @@ void one_iteration(void) START_TIME_TYPE start_time; int last_expected_count = expectedCount; int test_interval = 30; - char* payload = malloc(opts.payload_len); if (control_wait("start_measuring") == 0) goto exit; @@ -667,9 +645,11 @@ void one_iteration(void) global_start_time = start_clock(); for (i = 1; i <= test_count; ++i) { + char payload[128]; + sprintf(payload, "message number %d", i); - rc = MQTTAsync_send(client, opts.topic, opts.payload_len, payload, + rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload, opts.qos, opts.retained, NULL); while (rc != MQTTASYNC_SUCCESS) { @@ -677,26 +657,32 @@ void one_iteration(void) recreateReconnect(); if (stopping) goto exit; - mysleep(1); + MySleep(1000); rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload, opts.qos, opts.retained, NULL); + while (seqno - messagesSent > 2000) + MySleep(1000); } } MyLog(LOGA_INFO, "Messages sent... waiting for echoes"); while (arrivedCount < test_count) { + if (opts.persistence && connection_lost) + recreateReconnect(); if (stopping) goto exit; - mysleep(1); - printf("arrivedCount %d\n", arrivedCount); + MySleep(1000); + MyLog(LOGA_ALWAYS, "arrivedCount %d", arrivedCount); } measuring = 0; /* Now set a target of 30 seconds total round trip */ - if (last_completion_time == -1) + if (1) //last_completion_time == -1) { MyLog(LOGA_ALWAYS, "Round trip time for %d messages is %d ms", test_count, roundtrip_time); - expectedCount = 1000 * test_count * test_interval / roundtrip_time / 2; + // test_count messages in 3039 ms: (test_interval * 1000) / roundtrip_time * test_count + //expectedCount = 1000 * test_count * test_interval / roundtrip_time / 2; + expectedCount = (test_interval * 1000) / roundtrip_time * test_count; } else { @@ -722,22 +708,24 @@ void one_iteration(void) sprintf(payload, "message number %d", seqno); rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload, opts.qos, opts.retained, &ropts); - assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); while (rc != MQTTASYNC_SUCCESS) { - MyLog(LOGA_DEBUG, "Rc %d from publish with payload %s, retrying", rc, payload); + MyLog(LOGA_INFO, "Rc %d from publish with payload %s, retrying", rc, payload); if (opts.persistence && (connection_lost || rc == MQTTASYNC_DISCONNECTED)) recreateReconnect(); if (stopping) goto exit; - mysleep(1); + MySleep(1000); rc = MQTTAsync_send(client, opts.topic, (int)(strlen(payload)+1), payload, opts.qos, opts.retained, &ropts); - assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d ", rc); } //MyLog(LOGA_DEBUG, "Successful publish with payload %s", payload); - while (seqno - messagesSent > 2000) - mysleep(1); + //while (seqno - messagesSent > 2000) + //{ + //if (opts.persistence && (connection_lost || rc == MQTTASYNC_DISCONNECTED)) + // recreateReconnect(); + //} + // MySleep(1000); } MyLog(LOGA_ALWAYS, "%d messages sent in %d seconds", expectedCount, elapsed(start_time) / 1000); @@ -840,11 +828,14 @@ int sendAndReceive(void) } /* wait to know that the controlling process is running before connecting to the SUT */ - control_wait("who is ready?"); + if (control_wait("who is ready?") == 0) + { + MyLog(LOGA_ALWAYS, "Wait for controller failed"); + goto exit; + } /* connect cleansession, and then disconnect, to clean up */ conn_opts.keepAliveInterval = 10; - conn_opts.maxInflight = test_count; conn_opts.username = opts.username; conn_opts.password = opts.password; conn_opts.cleansession = 1; @@ -869,7 +860,7 @@ int sendAndReceive(void) } while (client_cleaned == 0) - mysleep(1); + MySleep(1000); MyLog(LOGA_ALWAYS, "Client state cleaned up"); @@ -887,7 +878,7 @@ int sendAndReceive(void) /* wait until subscribed */ while (client_subscribed == 0) - mysleep(1); + MySleep(1000); if (client_subscribed != 1) goto disconnect_exit; @@ -925,6 +916,7 @@ void control_onSubscribe(void* context, MQTTAsync_successData* response) MyLog(LOGA_DEBUG, "In control subscribe onSuccess callback %p granted qos %d", c, response->alt.qos); control_subscribed = 1; + MyLog(LOGA_ALWAYS, "Connected and subscribed to control connection"); } void control_onFailure(void* context, MQTTAsync_failureData* response) @@ -946,6 +938,7 @@ void control_onConnect(void* context, MQTTAsync_successData* response) ropts.onSuccess = control_onSubscribe; ropts.onFailure = control_onFailure; ropts.context = c; + MyLog(LOGA_ALWAYS, "Subscribing to control topic %s", sub_topic); if ((rc = MQTTAsync_subscribe(c, sub_topic, 2, &ropts)) != MQTTASYNC_SUCCESS) { MyLog(LOGA_ALWAYS, "control MQTTAsync_subscribe failed, rc %d", rc); @@ -955,7 +948,7 @@ void control_onConnect(void* context, MQTTAsync_successData* response) void trace_callback(enum MQTTASYNC_TRACE_LEVELS level, char* message) { - if (level == MQTTASYNC_TRACE_ERROR || strstr(message, "Connect") || strstr(message, "failed")) + //if (level == MQTTASYNC_TRACE_ERROR || strstr(message, "Connect") || strstr(message, "failed")) printf("Trace : %d, %s\n", level, message); } @@ -988,7 +981,7 @@ int main(int argc, char** argv) MyLog(LOGA_ALWAYS, "Starting with clientid %s", opts.clientid); - //MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_MAXIMUM); + MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR); MQTTAsync_setTraceCallback(trace_callback); rc = MQTTAsync_create(&control_client, opts.control_connection, @@ -1022,7 +1015,7 @@ int main(int argc, char** argv) } while (control_subscribed == 0) - mysleep(1); + MySleep(1000); if (control_subscribed != 1) goto destroy_exit; @@ -1035,5 +1028,12 @@ exit: destroy_exit: MQTTAsync_destroy(&control_client); +#include "Heap.h" + heap_info* mqtt_mem = 0; + mqtt_mem = Heap_get_info(); + MyLog(LOGA_INFO, "MQTT mem current %ld, max %ld",mqtt_mem->current_size,mqtt_mem->max_size); + if (mqtt_mem->current_size > 0) + /*failures++*/; /* consider any not freed memory as failure */ + return 0; }