Skip to content

Fix memory leak #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
peter-wangxu merged 3 commits into from
Sep 10, 2025
Merged

Fix memory leak #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
06cc4f5
fix(mqtt): fix memory leak in publish message
Li-haokai Sep 10, 2025
60a9b53
fix(mqtt): fix memory leak in sub and thingmodel
Li-haokai Sep 10, 2025
96a316c
Merge branch 'main' of https://github.com/volcengine/onesdk
Li-haokai Sep 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 111 additions & 33 deletions src/iot/iot_mqtt.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ static int callback_mqtt(struct lws *wsi, enum lws_callback_reasons reason,
// 如果是重复的，则释放待处理列表中的 topic 内存并跳过
if (is_duplicate) {
free((void*)topic_map->topic); // 释放重复 topic 的内存
topic_map->topic = NULL;
continue;
}

Expand All @@ -153,9 +154,10 @@ static int callback_mqtt(struct lws *wsi, enum lws_callback_reasons reason,
(ctx->sub_topic_count + 1) * sizeof(iot_mqtt_topic_map_t));
if (new_sub_topic_maps == NULL) {
// realloc 失败，保留原有 sub_topic_maps
lwsl_err("%s: realloc sub_topic_maps failed\n", __func__);
lwsl_err("%s: realloc sub_topic_maps failed\\n", __func__);

@peter-wangxu peter-wangxu Sep 10, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

typo？\\n

// 释放当前 topic_map 的 topic 内存，因为它无法被添加
free((void*)topic_map->topic);
topic_map->topic = NULL;
// 这里不直接返回错误，允许处理剩余的 topic，但记录下内存问题
continue; // 继续处理下一个 topic
}
Expand Down Expand Up @@ -354,8 +356,9 @@ static int callback_mqtt(struct lws *wsi, enum lws_callback_reasons reason,
case LWS_CALLBACK_TIMER:
lwsl_debug("timer\n");
if (ctx->is_connected) {
lws_set_timer_usecs(wsi, 10*1000000);
lws_set_timer_usecs(wsi, ctx->config->ping_interval*1000000);
lws_callback_on_writable(wsi);
lws_cancel_service(ctx->context);
}
break;

Expand Down Expand Up @@ -464,13 +467,46 @@ int iot_mqtt_init(iot_mqtt_ctx_t *ctx, iot_mqtt_config_t *config) {
}

void iot_mqtt_deinit(iot_mqtt_ctx_t *ctx) {
if (!ctx) {
return;
}
aws_string_destroy_secure(ctx->config->username);
aws_string_destroy_secure(ctx->config->password);

if (ctx->context) {
lws_context_destroy(ctx->context);
}

// Free subscribed topics
for (size_t i = 0; i < ctx->sub_topic_count; i++) {
free((void *)ctx->sub_topic_maps[i].topic);
}
free(ctx->sub_topic_maps);
ctx->sub_topic_maps = NULL;
ctx->sub_topic_count = 0;

// Free pending subscription list
if (ctx->pending_sub_list) {
for (size_t i = 0; i < ctx->pending_sub_list->count; i++) {
free((void *)ctx->pending_sub_list->topic_maps[i].topic);
}
free(ctx->pending_sub_list->topic_maps);
free(ctx->pending_sub_list);
ctx->pending_sub_list = NULL;
}

// Free pending publish list
if (ctx->pending_pub_list) {
for (size_t i = 0; i < ctx->pending_pub_list->count; i++) {
free((void *)ctx->pending_pub_list->publish_params[i].topic);
free((void *)ctx->pending_pub_list->publish_params[i].payload);
}
free(ctx->pending_pub_list->publish_params);
free(ctx->pending_pub_list->sent_list);
free(ctx->pending_pub_list);
ctx->pending_pub_list = NULL;
}

lws_pthread_mutex_destroy(&ctx->sub_topic_mutex);
lws_pthread_mutex_destroy(&ctx->pub_topic_mutex);
}
Expand Down Expand Up @@ -527,19 +563,34 @@ static void _iot_mqtt_append_sub_topic(iot_mqtt_ctx_t *ctx, iot_mqtt_topic_map_t
// 将topic_map存入待订阅列表
if (ctx->pending_sub_list == NULL) {
ctx->pending_sub_list = (iot_mqtt_pending_sub_list_t *)malloc(sizeof(iot_mqtt_pending_sub_list_t));
if (ctx->pending_sub_list == NULL) {
lws_pthread_mutex_unlock(&ctx->sub_topic_mutex);
return;
}
ctx->pending_sub_list->topic_maps = NULL;
ctx->pending_sub_list->count = 0;
}
// copy topic_map
iot_mqtt_topic_map_t *topic_map_copy = (iot_mqtt_topic_map_t *)malloc(sizeof(iot_mqtt_topic_map_t));
memset(topic_map_copy, 0, sizeof(iot_mqtt_topic_map_t));
topic_map_copy->topic = strdup(topic_map->topic);
topic_map_copy->message_callback = topic_map->message_callback;
topic_map_copy->event_callback = topic_map->event_callback;
topic_map_copy->user_data = topic_map->user_data;
topic_map_copy->qos = topic_map->qos;
ctx->pending_sub_list->topic_maps = (iot_mqtt_topic_map_t *)realloc(ctx->pending_sub_list->topic_maps, (ctx->pending_sub_list->count + 1) * sizeof(iot_mqtt_topic_map_t));
ctx->pending_sub_list->topic_maps[ctx->pending_sub_list->count] = *topic_map_copy;
iot_mqtt_topic_map_t topic_map_copy;
memset(&topic_map_copy, 0, sizeof(iot_mqtt_topic_map_t));
topic_map_copy.topic = strdup(topic_map->topic);
if (topic_map_copy.topic == NULL) {
lws_pthread_mutex_unlock(&ctx->sub_topic_mutex);
return;
}
topic_map_copy.message_callback = topic_map->message_callback;
topic_map_copy.event_callback = topic_map->event_callback;
topic_map_copy.user_data = topic_map->user_data;
topic_map_copy.qos = topic_map->qos;

iot_mqtt_topic_map_t *new_maps = (iot_mqtt_topic_map_t *)realloc(ctx->pending_sub_list->topic_maps, (ctx->pending_sub_list->count + 1) * sizeof(iot_mqtt_topic_map_t));
if (new_maps == NULL) {
free((void*)topic_map_copy.topic);
lws_pthread_mutex_unlock(&ctx->sub_topic_mutex);
return;
}
ctx->pending_sub_list->topic_maps = new_maps;
ctx->pending_sub_list->topic_maps[ctx->pending_sub_list->count] = topic_map_copy;
ctx->pending_sub_list->count++;
lws_pthread_mutex_unlock(&ctx->sub_topic_mutex);
}
Expand Down Expand Up @@ -574,31 +625,37 @@ int iot_mqtt_publish(iot_mqtt_ctx_t *ctx, const char *topic, const uint8_t *payl
int ret = VOLC_OK;

// 组装pub_param，放入pending_pub_list
lws_mqtt_publish_param_t *pub_param = malloc(sizeof(lws_mqtt_publish_param_t));
memset(pub_param, 0, sizeof(lws_mqtt_publish_param_t));
pub_param->topic = strdup(topic);
pub_param->topic_len = strlen(topic);
void* payload_byte = malloc(len);
if (payload_byte == NULL) {
// 内存分配失败，记录错误信息
lwsl_err("%s: Memory allocation failed for payload\n", __func__);
// 可以选择返回错误码或进行其他处理
lws_mqtt_publish_param_t pub_param;
memset(&pub_param, 0, sizeof(lws_mqtt_publish_param_t));
pub_param.topic = strdup(topic);
if (!pub_param.topic) {
lwsl_err("%s: Memory allocation failed for topic\n", __func__);
return VOLC_ERR_MALLOC;
}
memcpy(payload_byte, payload, len);
pub_param->payload = payload_byte;
pub_param->payload_len = len;
pub_param->qos = qos;
pub_param.topic_len = strlen(topic);

if (len > 0) {
pub_param.payload = malloc(len);
if (pub_param.payload == NULL) {
lwsl_err("%s: Memory allocation failed for payload\n", __func__);
free((void*)pub_param.topic);
return VOLC_ERR_MALLOC;
}
memcpy((void*)pub_param.payload, payload, len);
} else {
pub_param.payload = NULL;
}

pub_param.payload_len = len;
pub_param.qos = qos;

lws_pthread_mutex_lock(&ctx->pub_topic_mutex);
if (ctx->pending_pub_list == NULL) {
ctx->pending_pub_list = (iot_mqtt_pending_pub_list_t *)malloc(sizeof(iot_mqtt_pending_pub_list_t));
if (ctx->pending_pub_list == NULL) {
// 内存分配失败，记录错误信息
lwsl_err("%s: Memory allocation failed for pending_pub_list\n", __func__);
// 可以选择返回错误码或进行其他处理
ret = VOLC_ERR_MALLOC;
goto finish;
goto finish_with_free;
}
ctx->pending_pub_list->publish_params = NULL;
ctx->pending_pub_list->sent_list = NULL;
Expand All @@ -609,17 +666,38 @@ int iot_mqtt_publish(iot_mqtt_ctx_t *ctx, const char *topic, const uint8_t *payl
if (ctx->pending_pub_list->count > 10) {
ret = VOLC_ERR_MQTT_PUB;
lwsl_err("%s: publish failed\n", __func__);
lws_pthread_mutex_unlock(&ctx->pub_topic_mutex);
goto finish;
goto finish_with_free;
}

ctx->pending_pub_list->publish_params = (lws_mqtt_publish_param_t *)realloc(ctx->pending_pub_list->publish_params, (ctx->pending_pub_list->count + 1) * sizeof(lws_mqtt_publish_param_t));
ctx->pending_pub_list->publish_params[ctx->pending_pub_list->count] = *pub_param;
ctx->pending_pub_list->sent_list = (bool *)realloc(ctx->pending_pub_list->sent_list, (ctx->pending_pub_list->count + 1) * sizeof(bool));
lws_mqtt_publish_param_t *new_params = (lws_mqtt_publish_param_t *)realloc(ctx->pending_pub_list->publish_params, (ctx->pending_pub_list->count + 1) * sizeof(lws_mqtt_publish_param_t));
if (new_params == NULL) {
ret = VOLC_ERR_MALLOC;
goto finish_with_free;
}
ctx->pending_pub_list->publish_params = new_params;

bool *new_sent_list = (bool *)realloc(ctx->pending_pub_list->sent_list, (ctx->pending_pub_list->count + 1) * sizeof(bool));
if (new_sent_list == NULL) {
ret = VOLC_ERR_MALLOC;
goto finish_with_free;
}
ctx->pending_pub_list->sent_list = new_sent_list;

ctx->pending_pub_list->publish_params[ctx->pending_pub_list->count] = pub_param;
ctx->pending_pub_list->sent_list[ctx->pending_pub_list->count] = false;
ctx->pending_pub_list->count++;

finish:
lws_pthread_mutex_unlock(&ctx->pub_topic_mutex);
if (ctx->is_connected) {
lws_callback_on_writable(ctx->wsi);
lws_cancel_service(ctx->context);
}

return ret;

finish_with_free:
free((void*)pub_param.topic);
free((void*)pub_param.payload);
lws_pthread_mutex_unlock(&ctx->pub_topic_mutex);
if (ctx->is_connected) {
lws_callback_on_writable(ctx->wsi);
Expand Down
1 change: 1 addition & 0 deletions src/iot/thingmodel/custom_topic.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ int32_t tm_sub_custom_topic(void *handler, const char *topic_suffix) {
aws_mem_release(dm_handle->allocator, topic);
aws_string_destroy_secure(product_key);
aws_string_destroy_secure(device_name);
aws_string_destroy_secure(topic_suffix_string);

return 0;
}
Expand Down
1 change: 1 addition & 0 deletions src/iot/thingmodel/iot_tm_api.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,7 @@ int __s_tm_set_up_mqtt_topic(iot_tm_handler_t *iot_tm_handler, struct aws_string
topic_mapping.message_callback = g_dm_recv_topic_mapping[i].func;
topic_mapping.user_data = iot_tm_handler;
if (topic_mapping.message_callback == NULL) {
aws_mem_release(iot_tm_handler->allocator, (void*)topic_mapping.topic);
continue;
}
topic_mapping.qos = IOT_MQTT_QOS1;
Expand Down
Loading