MQTT Slowdown After Wi-Fi Outage

[ngiustini-invn]

Hello,

Using release version 1.9.105

I am currently facing an issue with adding MQTT functionality to my embedded system built on C/C++. So far the MQTT system seems to be working properly and efficiently under normal operating circumstances but the problem happens when the device loses connection to Wi-Fi temporarily.

During runtime the system will be in a loop where we will publish messages for the MQTT server and expect them to show up with a latency of at most, a couple seconds. In the event of a Wi-Fi outage it seems that our messages are placed in some sort of queue and upon re-connecting to Wi-Fi we can see the messages start getting sent out again. The problem here is that the time needed to see the messages re-sent is quite long for our application (~5 minutes) and when the messages are sent out from the queue again they seem to be sent quite slowly. Our system intends to publish messages to the server at 1Hz so upon an outage this can lead to many messages ending up in the queue (200-1000+ messages) which are very slow to leave the queue. We have done our due diligence and confirmed that the slowdown is not caused within our own code and this slowdown is indeed happening somewhere inside of the MQTT library. (We have confirmed that the messages are leaving our queue and sent to the MQTT library to be published to the server, but the callback function is not reached for many minutes)

Is it possible that there is some configuration in the MQTT library which can speed this process up? We have tried different QoS settings which showed no improvements for us. Alternatively, is it possible to flush the queue at some point or force the library to send all messages from the queue to the server? Is there some standard expected behavior for the library when the Wi-Fi connection is lost and then restored?

If there is anything I can help clear up about our issue or any other details I can provide please let me know.

Thanks in advance,
Noah

[jmklix]

Are you seeing this slowdown with this sdk (aws-sdk-cpp) or with the IoT sdk (aws-iot-device-sdk-cpp-v2)? If it is the latter please open an issue on that repo. Please make sure to include additional information requested in the issue template.

[ngiustini-invn]

Thank you for the quick reply,

It seems that the issue is coming from this sdk (aws-sdk-cpp), unless the IoT sdk (aws-iot-device-sdk-cpp-v2) is included as a submodule of this repository. Our application uses v1.9.105 from this repository (commit hash 2aa01e7).

[jmklix]

Can you provide more details about the embedded device that you are using? and include a code sample with steps that reproduce the slowdown you mention above?

[ngiustini-invn]

The embedded device is running a Rockchip RK3308 SOC with a Quad-Core ARM Cortex-A35 64bit processor. The system is also equipped with 512MB of SDRAM. Please see below a sample of the source code where the problem seems to arise. This code (function) is run once inside of a thread and then the while loop on line 276 is responsible for actually sending the messages. The while loop will basically read messages from the system msg queue and publish them to the server. This msg queue is added to at approx 1Hz.

To reproduce the issue the thread should be allowed to reach the while loop while connected to Wi-Fi. Then when the execution is inside the loop (line 276) Wi-Fi should be disconnected for some amount of time (ie. 1 minute) and then the system should be reconnected to Wi-Fi. At that point the slowdown/backlog should be evident.

Please let me know if there is anything I can clarify or if there is anything you can suggest to help.

Sample Code

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/time.h>
#include <unistd.h>
#include <linux/ioctl.h>
#include <math.h>
#include <sys/ioctl.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <math.h>
#include <curl/curl.h>
#include <json-c/json.h>

#include <algorithm>
#include <aws/cognito-identity/CognitoIdentityClient.h>
#include <aws/cognito-identity/CognitoIdentityErrors.h>
#include <aws/cognito-identity/model/CreateIdentityPoolRequest.h>
#include <aws/cognito-identity/model/DeleteIdentityPoolRequest.h>
#include <aws/cognito-identity/model/DescribeIdentityPoolRequest.h>
#include <aws/cognito-identity/model/GetIdentityPoolRolesRequest.h>
#include <aws/cognito-identity/model/UpdateIdentityPoolRequest.h>
#include <aws/cognito-identity/model/ListIdentityPoolsRequest.h>
#include <aws/cognito-identity/model/GetCredentialsForIdentityRequest.h>
#include <aws/cognito-identity/model/GetIdRequest.h>
#include <aws/cognito-identity/model/ListIdentitiesRequest.h>
#include <aws/cognito-identity/model/GetOpenIdTokenRequest.h>
#include <aws/cognito-identity/model/UnlinkIdentityRequest.h>
#include <aws/cognito-identity/model/GetOpenIdTokenForDeveloperIdentityRequest.h>
#include <aws/cognito-identity/model/LookupDeveloperIdentityRequest.h>
#include <aws/cognito-identity/CognitoIdentityErrors.h>
#include <aws/access-management/AccessManagementClient.h>
#include <aws/iam/IAMClient.h>
#include <aws/core/client/CoreErrors.h>
#include <aws/core/utils/json/JsonSerializer.h>
#include <aws/core/utils/Outcome.h>
#include <aws/crt/auth/Credentials.h>
#include <aws/crt/Types.h>
#include <aws/iot/MqttClient.h>
#include <aws/crt/Api.h>
#include <aws/core/Aws.h>

#define MAX_MQTT_PAYLOAD_LEN (1024)

struct aws_cog_mqtt_msg_t{
	int type;
	int payload_len;
	char payload[MAX_MQTT_PAYLOAD_LEN];
};

int run_flag = 1;

static int aws_cog_mqtt_start(const char* cognitoPoolId, const char* region, const char* iotEndPoint, const char* client_id, const char* current_topic, void* hook_func)
{
	int ret;

	//start mqtt request
	if ((cognitoPoolId == NULL) || (region == NULL) || (iotEndPoint == NULL)) {
		printf("either cognitoPoolId, region or iotEndPoint is NULL! return...\n");
		return -1;
	}

	//get cognito credentials
	GetIdRequest getIdRequest;
	getIdRequest.WithIdentityPoolId(cognitoPoolId);

	ClientConfiguration clientConfig;
	clientConfig.region = Aws::Region::US_EAST_1;

	auto client = Aws::MakeShared<Aws::CognitoIdentity::CognitoIdentityClient>("cognito_client", clientConfig);

	Aws::String access_key;
	Aws::String secret_key;
	Aws::String session_token;

	GetIdOutcome getIdOutcome = client->GetId(getIdRequest);
	if (getIdOutcome.IsSuccess()) {
		GetCredentialsForIdentityRequest getCredentialsRequest;
		getCredentialsRequest.WithIdentityId(getIdOutcome.GetResult().GetIdentityId());
		GetCredentialsForIdentityOutcome getCredentialsOutcome = client->GetCredentialsForIdentity(getCredentialsRequest);
		access_key = getCredentialsOutcome.GetResult().GetCredentials().GetAccessKeyId();
		secret_key = getCredentialsOutcome.GetResult().GetCredentials().GetSecretKey();
		session_token = getCredentialsOutcome.GetResult().GetCredentials().GetSessionToken();
		if (getCredentialsOutcome.IsSuccess()) {
		} else {
			printf("GetCredentials failed : %s\n", getCredentialsOutcome.GetError().GetMessage().c_str());
			return -1;
		}
	} else {
		printf(" GetId failed : %s\n", getIdOutcome.GetError().GetMessage().c_str());
		return -1;
	}

	//build credential handler
	Aws::Crt::ApiHandle apiHandle;
	Aws::Crt::Auth::CredentialsProviderStaticConfig defaultConfig;
	std::shared_ptr<Aws::Crt::Auth::ICredentialsProvider> provider = nullptr;

	defaultConfig.AccessKeyId = Aws::Crt::ByteCursorFromCString(access_key.c_str());
	defaultConfig.SecretAccessKey = Aws::Crt::ByteCursorFromCString(secret_key.c_str());
	defaultConfig.SessionToken = Aws::Crt::ByteCursorFromCString(session_token.c_str());

	provider = Aws::Crt::Auth::CredentialsProvider::CreateCredentialsProviderStatic(defaultConfig);

	Aws::Iot::MqttClientConnectionConfigBuilder builder;
	Aws::Iot::WebsocketConfig config(region, provider);
	builder = Aws::Iot::MqttClientConnectionConfigBuilder(config);

	//set up mqtt client
	/********************** Now Setup an Mqtt Client ******************/
	/*
	* You need an event loop group to process IO events.
	* If you only have a few connections, 1 thread is ideal
	*/
	Aws::Crt::Io::EventLoopGroup eventLoopGroup(1);
	if (!eventLoopGroup) {
		printf("Event Loop Group Creation failed with error %s\n", ErrorDebugString(eventLoopGroup.LastError()));
		return -1;
	}

	Aws::Crt::Io::DefaultHostResolver defaultHostResolver(eventLoopGroup, 1, 5);
	Aws::Crt::Io::ClientBootstrap bootstrap(eventLoopGroup, defaultHostResolver);

	if (!bootstrap) {
		printf("ClientBootstrap failed with error %s\n", ErrorDebugString(bootstrap.LastError()));
		return -1;
	}

	Aws::Crt::String endpoint = iotEndPoint;
	builder.WithEndpoint(endpoint);

	auto clientConfig1 = builder.Build();
	if (!clientConfig1) {
		printf("Client Configuration initialization failed with error %s\n", ErrorDebugString(clientConfig1.LastError()));
		return -1;
	}

	Aws::Iot::MqttClient mqttClient(bootstrap);
	/*
	* Since no exceptions are used, always check the bool operator
	* when an error could have occurred.
	*/
	if (!mqttClient) {
		debug_print("MQTT Client Creation failed with error %s\n", ErrorDebugString(mqttClient.LastError()));
		return -1;
	}

	/*
	* Now create a connection object. Note: This type is move only
	* and its underlying memory is managed by the client.
	*/
	auto connection = mqttClient.NewConnection(clientConfig1);
	if (!connection) {
		printf("MQTT Connection Creation failed with error %s\n", ErrorDebugString(mqttClient.LastError()));
		return -1;
	}

	//connect to server
	/*
	* TODO: Not to enforce synchronous behavior and no to do with a condition variable.
	*/
	std::promise<bool> connectionCompletedPromise;
	std::promise<void> connectionClosedPromise;

	/*
	* This will execute when an mqtt connect has completed or failed.
	*/
	auto onConnectionCompleted = [&](Mqtt::MqttConnection &, int errorCode, Mqtt::ReturnCode returnCode, bool) {
		if (errorCode) {
			printf("Connection failed with error %s\n", ErrorDebugString(errorCode));
			connectionCompletedPromise.set_value(false);
        	} else {
			printf("errorCode 0 good!\n");
			if (returnCode != AWS_MQTT_CONNECT_ACCEPTED) {
				perror("Connection failed with mqtt return code %d\n", (int)returnCode);
				connectionCompletedPromise.set_value(false);
			} else {
				printf("Connection completed successfully.\n");
				connectionCompletedPromise.set_value(true);
			}
        	}
	};

	auto onInterrupted = [&](Mqtt::MqttConnection &, int error) {
		printf("Connection interrupted with error %s\n", ErrorDebugString(error));
		//return -1;	//Don't retry with interrupt, will recover/resume
	};

	auto onResumed = [&](Mqtt::MqttConnection &, Mqtt::ReturnCode, bool) { printf("Connection resumed\n"); };

	/*
	* Invoked when a disconnect message has completed.
	*/
	auto onDisconnect = [&](Mqtt::MqttConnection &) {
		printf("Disconnect completed\n");
		connectionClosedPromise.set_value();
	};

	connection->OnConnectionCompleted = std::move(onConnectionCompleted);
	connection->OnDisconnect = std::move(onDisconnect);
	connection->OnConnectionInterrupted = std::move(onInterrupted);
	connection->OnConnectionResumed = std::move(onResumed);

	/*
	* Actually perform the connect dance.
	* This will use default ping behavior of 1 hour and 3 second timeouts.
	* If you want different behavior, those arguments go into slots 3 & 4.
	*/
	printf("Connecting...\n");
	if (!connection->Connect(client_id, false, 1000)) {
		printf("MQTT Connection failed with error %s\n", ErrorDebugString(connection->LastError()));
		return -1;
	}

	//attach publish/subscribe cb
	String topic(current_topic);
	if (connectionCompletedPromise.get_future().get()) {
		std::mutex receiveMutex;
		std::condition_variable receiveSignal;
		uint32_t receivedCount = 0;

        	/*
         	* This is invoked upon the receipt of a Publish on a subscribed topic.
         	*/
        	auto onMessage = [&](Mqtt::MqttConnection &,
                             const String &topic,
                             const ByteBuf &byteBuf,
                             bool /*dup*/,
                             Mqtt::QOS /*qos*/,
                             bool /*retain*/) {
			{
				std::lock_guard<std::mutex> lock(receiveMutex);
				++receivedCount;
				printf("Message: %s\n", byteBuf.buffer);
			}
			receiveSignal.notify_all();
		};

		/*
		* Subscribe for incoming publish messages on topic.
		*/
		std::promise<void> subscribeFinishedPromise;
		auto onSubAck = [&](Mqtt::MqttConnection &, uint16_t packetId, const String &topic, Mqtt::QOS QoS, int errorCode) {
			if (errorCode) {
				printf("Subscribe failed with error %s\n", aws_error_debug_str(errorCode));
			} else {
				if (!packetId || QoS == AWS_MQTT_QOS_FAILURE) {
					printf("Subscribe rejected by the broker.\n");
				} else {
                    //do nothing
				}
			}
			subscribeFinishedPromise.set_value();
		};

		connection->Subscribe(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, onMessage, onSubAck);
		subscribeFinishedPromise.get_future().wait();
		init_status = 1;

		if (hook_func != NULL) {
			printf("Executing mqtt hook func...\n")
			((void(*)(void))hook_func)();
		}

		//publish and subscribe loop
		{
			struct aws_cog_mqtt_msg_t msg;
			int cnt = 0;
			int ret;

			while (run_flag) {
				memset(&msg, 0, sizeof(msg));
                //recieve data from the msg queue so it can be sent to the mqtt server
				ret = msgrcv(aws_cog_mqtt_msgid, &msg, sizeof(msg), 0, 0);
				if(-1 == ret) {
					printf("######## mqtt msgrcv error\n");
				}
                printf("cnt=%d, payload_len=%d, payload=%s\n", cnt++, msg.payload_len, msg.payload);
                    ByteBuf payload = ByteBufFromArray((const uint8_t *)msg.payload, msg.payload_len);
                    printf("Start publishing %s\n", payload.buffer);
                    auto onPublishComplete = [](Mqtt::MqttConnection &, uint16_t v1, int v2) {printf("Publishing msg complete, v1 = %u, v2 = %d\n", v1, v2);};
                    connection->Publish(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, false, payload, onPublishComplete);
			}

			init_status = 0;
			printf("start mqtt exit wait...\n");
            std::unique_lock<std::mutex> receivedLock(receiveMutex);
            receiveSignal.wait(receivedLock);
		}

		printf("start unsubscribe...\n");
		/*
		* Unsubscribe from the topic.
		*/
		std::promise<void> unsubscribeFinishedPromise;
		connection->Unsubscribe(topic.c_str(), [&](Mqtt::MqttConnection &, uint16_t, int) { unsubscribeFinishedPromise.set_value(); });
		unsubscribeFinishedPromise.get_future().wait();

		printf("start disconnecting...\n");
		/* Disconnect */
		if (connection->Disconnect()) {
			connectionClosedPromise.get_future().wait();
			printf("inside disconnect return\n");
		}

		ret = 0;
	} else {
		init_status = -1;
	}
	//exit mqtt work

 
	return ret;
}