#include "log_producer_manager.h" #include "inner_log.h" //#include "md5.h" #include "sds.h" // change from 100ms to 1000s, reduce wake up when app switch to back #define LOG_PRODUCER_FLUSH_INTERVAL_MS 100 #define LOG_PRODUCER_READ_INTERVAL_MS 100 #define MAX_LOGGROUP_QUEUE_SIZE 1024 #define MIN_LOGGROUP_QUEUE_SIZE 128 #define MAX_MANAGER_FLUSH_COUNT 100 // 10MS * 100 #define MAX_SENDER_FLUSH_COUNT 100 // 10ms * 100 log_producer_manager* g_producer_manager[10]; int producer_manager_num = 0; #ifdef WIN32 DWORD WINAPI log_producer_send_thread(LPVOID param); #else void* log_producer_send_thread(void* param); #endif void _try_flush_loggroup(log_producer_manager* producer_manager) { int loggroup_size; int rst; int32_t now_time = time(NULL); CS_ENTER(producer_manager->lock); if (producer_manager->builder != NULL) { log_group_builder* builder = producer_manager->builder; producer_manager->builder = NULL; CS_LEAVE(producer_manager->lock); loggroup_size = builder->loggroup_size; rst = log_queue_push(producer_manager->loggroup_queue, builder); //aos_debug_log((LB, "try push loggroup to flusher, size : %d, status : %d", (int)loggroup_size, rst)); if (rst != 0) { aos_error_log((LB, "try push loggroup to flusher failed, force drop this log group, error code : %d", rst)); if (producer_manager->send_done_function != NULL) { producer_manager->send_done_function(LOG_PRODUCER_DROP_ERROR, loggroup_size, 0, NULL, "try push loggroup to flusher failed, force drop this log group", NULL, producer_manager->user_param); } log_group_destroy(builder); } else { producer_manager->totalBufferSize += loggroup_size; COND_SIGNAL(producer_manager->triger_cond); } } else { CS_LEAVE(producer_manager->lock); } } #ifdef WIN32 //从db中读取log DWORD WINAPI log_producer_read_persistent_thread(LPVOID param) #else //从db中读取log void* log_producer_read_persistent_thread(void* param) #endif { int i = 0; int forceClean = 0; int32_t sleepMs = LOG_PRODUCER_READ_INTERVAL_MS; log_group_builder* builder = NULL; log_producer_manager* producer_manager = (log_producer_manager*)param; log_persistent_manager* persistent_manager; log_producer_config* config = producer_manager->producer_config; aos_info_log((LB, "start run read persistent thread")); while (producer_manager->shutdown == 0) { i = 0; for (i = 0; i < sleepMs; i += LOG_PRODUCER_READ_INTERVAL_MS) { #ifdef WIN32 Sleep(LOG_PRODUCER_READ_INTERVAL_MS); #else usleep(LOG_PRODUCER_READ_INTERVAL_MS * 1000); #endif } do { // if send queue is full, skip pack and send data CS_ENTER(producer_manager->lock); if (producer_manager->send_param_queue_write - producer_manager->send_param_queue_read >= producer_manager->send_param_queue_size) { aos_debug_log((LB, "send queue is full, send queue write : %d, send queue read : %d, skip pack and send data", producer_manager->send_param_queue_write, producer_manager->send_param_queue_read)); CS_LEAVE(producer_manager->lock); sleepMs = LOG_PRODUCER_READ_INTERVAL_MS; break; } CS_LEAVE(producer_manager->lock); if (config->endpoint == NULL || config->skyeye_user_topic == NULL || config->skyeye_sys_topic == NULL || config->beidou_topic == NULL || config->business_user_topic == NULL || config->business_sys_topic == NULL || config->vtmweb_topic == NULL) { aos_debug_log((LB, "endpoint is NULL, skip pack and send data")); sleepMs = LOG_PRODUCER_READ_INTERVAL_MS; break; } //get send data from persistent persistent_manager = producer_manager->persistent_manager; if (persistent_manager != NULL) { CS_ENTER(persistent_manager->lock); builder = log_persistent_manager_get_log(persistent_manager);//从db中读取log，数量为logCountPerPackage CS_LEAVE(persistent_manager->lock); if (builder != NULL) { int i = 0; int n_logs = 0; char modular[MAX_PATH] = { 0 }; char uuid[MAX_LOG_COUNT][MAX_UUID_LEN] = { 0 }; lz4_log_buf* lz4_buf = NULL; //缓存uuid和n_logs、modular for (i = 0; i < builder->grp->n_logs; i++) { strcpy(uuid[i], builder->grp->logs[i].uuid); } n_logs = builder->grp->n_logs; if (n_logs > 0) { sleepMs = config->logCountPerPackage / n_logs * LOG_PRODUCER_READ_INTERVAL_MS; } strcpy(modular, builder->modular); aos_debug_log((LB, "get from persistent log count : %d.", n_logs)); // process data CS_ENTER(producer_manager->lock); producer_manager->totalBufferSize -= builder->loggroup_size; CS_LEAVE(producer_manager->lock); //添加terminalNo, SN, 效率贼低 if (config->skyeyeTerminalNo != NULL) { add_log_terminal_no(builder, config->skyeyeTerminalNo); } if (config->skyeyeSn != NULL) { static char local_ip_str[64] = "0.0.0.0"; static long ipInit = 0; //get ip every 1000 times if (ipInit % 1000 == 0) GetLocalIP(local_ip_str); ipInit++; add_log_terminal_sn_other(builder, config->skyeyeSn, local_ip_str); } clear_errJson_msg(builder); // check compress type if (config->compressType == 1)//是否压缩 lz4_buf = serialize_to_proto_buf_with_malloc_lz4(builder); else lz4_buf = serialize_to_proto_buf_with_malloc_no_lz4(builder); if (lz4_buf == NULL) { aos_error_log((LB, "serialize loggroup to proto buf with lz4 failed")); if (producer_manager->send_done_function) { producer_manager->send_done_function(LOG_PRODUCER_DROP_ERROR, builder->loggroup_size, 0, NULL, "serialize loggroup to proto buf with lz4 failed", NULL, producer_manager->user_param); } if (producer_manager->uuid_send_done_function != NULL) { producer_manager->uuid_send_done_function(LOG_PRODUCER_INVALID, builder->loggroup_size, 0, NULL, "invalid send param, magic num not found", NULL, producer_manager->uuid_user_param, builder->grp->n_logs, uuid, builder->modular);//send_done后执行删除对应uuid的数据 } } else { log_producer_send_param* send_param; CS_ENTER(producer_manager->lock); producer_manager->totalBufferSize += lz4_buf->length;//添加了额外字段、压缩后，最终长度 send_param = create_log_producer_send_param(config, producer_manager, lz4_buf, builder); producer_manager->send_param_queue[producer_manager->send_param_queue_write++ % producer_manager->send_param_queue_size] = send_param;//放入发送队列 CS_LEAVE(producer_manager->lock); } log_group_destroy(builder); //set log status to sending CS_ENTER(persistent_manager->lock); log_persistent_manager_updata_log_status(persistent_manager, modular, n_logs, uuid, LOG_DB_STATUS_SENDING); //更新数据状态为sending CS_LEAVE(persistent_manager->lock); /** 每次都是24 [Gifur@2022720]*/ persistent_manager->SendLogCount += n_logs; aos_debug_log((LB, "get from persistent, send log count : %d.", persistent_manager->SendLogCount)); CS_ENTER(persistent_manager->lock); if (forceClean || persistent_manager->SendLogCount - persistent_manager->LastCleanLogCount > 1000) {//每5000条清理一次 log_persistent_manager_clean(persistent_manager, modular); persistent_manager->LastCleanLogCount = persistent_manager->SendLogCount; forceClean = 0; } else { int count = db_get_count(persistent_manager->db_manager); if(count > (persistent_manager->config->maxPersistentLogCount + 1000)) { log_persistent_manager_clean(persistent_manager, modular); persistent_manager->LastCleanLogCount = persistent_manager->SendLogCount; } } CS_LEAVE(persistent_manager->lock); continue; } else { if ((producer_manager->send_param_queue_write - producer_manager->send_param_queue_read) < producer_manager->send_param_queue_size) { sleepMs = 1200 * LOG_PRODUCER_READ_INTERVAL_MS;//120s send once if can not get enough forceClean = 1; } else sleepMs = 10 * LOG_PRODUCER_READ_INTERVAL_MS;//普通暂停时间，等待发送线程发送 //aos_debug_log((LB, "get from persistent is null.")); break; } } break; } while (1); // send data CS_ENTER(producer_manager->lock); if (producer_manager->send_threads != NULL) {//spshell.exe, with send threads // if send thread count > 0, we just push send_param to sender queue while (producer_manager->send_param_queue_write > producer_manager->send_param_queue_read && !log_queue_isfull(producer_manager->sender_data_queue)) { log_producer_send_param* send_param = producer_manager->send_param_queue[producer_manager->send_param_queue_read++ % producer_manager->send_param_queue_size]; // push always success log_queue_push(producer_manager->sender_data_queue, send_param);//往发送队列存 } } else if (producer_manager->send_param_queue_write > producer_manager->send_param_queue_read) { //sphost.exe ,no send threads // if no sender thread, we send this packet out in flush thread log_producer_send_param* send_param = producer_manager->send_param_queue[producer_manager->send_param_queue_read++ % producer_manager->send_param_queue_size]; log_producer_send_data(send_param); } CS_LEAVE(producer_manager->lock); } aos_info_log((LB, "exit read persistent thread")); return 0; } #ifdef WIN32 DWORD WINAPI log_producer_write_persistent_thread(LPVOID param) #else //队列取数据，写入db void* log_producer_write_persistent_thread(void* param) #endif { int lens; int first_process = 1; int32_t now; int32_t last_write_time = time(NULL); shareq_frame share_frame = {0}; serialize_buf buf = { 0 }; int t_count = 0; int discardMsg = 0; log_producer_manager* producer_manager = (log_producer_manager*)param; log_persistent_manager* persistent_manager; share_frame.data = &buf; aos_info_log((LB, "start run write persistent thread")); while (producer_manager->shutdown == 0) { int count = 0; CS_ENTER(producer_manager->lock); COND_WAIT_TIME(producer_manager->triger_cond, producer_manager->lock,LOG_PRODUCER_FLUSH_INTERVAL_MS); CS_LEAVE(producer_manager->lock); persistent_manager = producer_manager->persistent_manager; if (persistent_manager == NULL) { continue; } //第一次执行清理 if (producer_manager->producer_config->enableGuarder && first_process) { int recoverRst; CS_ENTER(persistent_manager->lock); recoverRst = log_persistent_manager_recover(persistent_manager); if (recoverRst != 0) { aos_error_log((LB, "recover log persistent manager failed, result %d", recoverRst)); } else { aos_info_log((LB, "recover log persistent manager success")); } CS_LEAVE(persistent_manager->lock); first_process = 0; } //huchen add,注意事务不能被其他线程打断，通过lock保证事务执行完成 CS_ENTER(persistent_manager->lock); lens = ClibsharequeueGetLens(producer_manager->share_queue); now = time(NULL); if ((lens < MAX_SHAREQUEUE_LENS / 4) && (now - last_write_time < 3)) {//每3s执行一次事务 CS_LEAVE(persistent_manager->lock); continue; } aos_debug_log((LB, "get from share queue log count : %d.", lens)); t_count = db_get_count(persistent_manager->db_manager); if (t_count > (persistent_manager->config->maxPersistentLogCount + 2000)) { discardMsg = 1; aos_warn_log((LB, "give up the db content, bcz db count %d exceeds max count %d", t_count, persistent_manager->config->maxPersistentLogCount)); } else { discardMsg = 0; } // try read queue do { int result; //get send data from share buffer memset(share_frame.data, 0, sizeof(serialize_buf)); result = ClibsharequeueGetAndDel(producer_manager->share_queue, &share_frame); if (result == 1) { if (discardMsg) continue; if (count == 0) { result = log_persistent_manager_transaction_begin(persistent_manager); if (result != LOG_PRODUCER_OK) { aos_warn_log((LB, "get from share queue, uuid %s begin failed.", buf.uuid)); } result = log_persistent_manager_save_log(persistent_manager, share_frame.data); if (result != LOG_PRODUCER_OK) { aos_warn_log((LB, "get from share queue, uuid %s save failed.", buf.uuid)); } } else { result = log_persistent_manager_save_log(persistent_manager, share_frame.data); if (result != LOG_PRODUCER_OK) { aos_warn_log((LB, "get from share queue, uuid %s save failed.", buf.uuid)); } } count++; continue; } if (count > 0) { result = log_persistent_manager_transaction_commit(persistent_manager); if (result != LOG_PRODUCER_OK) { aos_warn_log((LB, "get from share queue, uuid %s commit failed.", buf.uuid)); } } last_write_time = time(NULL); //aos_debug_log((LB, "get from share queue is null.")); break; } while (1); CS_LEAVE(persistent_manager->lock); // if no job, check now loggroup _try_flush_loggroup(producer_manager); } aos_info_log((LB, "exit write persistent thread")); return 0; } #ifdef WIN32 DWORD WINAPI log_producer_flush_thread(LPVOID param) //flush the log to spshell #else void* log_producer_flush_thread(void* param) #endif { while(1) { int i = 0; for(i =0; i < producer_manager_num; i++) { void* data = NULL; shareq_frame frame = { 0 }; serialize_buf buf = { 0 }; log_producer_manager* producer_manager = (log_producer_manager*)g_producer_manager[i]; log_persistent_manager* persistent_manager; frame.data = &buf; aos_info_log((LB, "start run flusher thread")); persistent_manager = producer_manager->persistent_manager; if (persistent_manager == NULL || producer_manager->shutdown == 1) { continue; } do { //get send data from buffer data = log_queue_trypop(producer_manager->loggroup_queue); if (data != NULL) { int insert_result = 0; log_group_builder* builder = (log_group_builder*)data; memset(frame.data, 0, sizeof(serialize_buf)); serialize_to_buf(&builder->grp->logs[0], frame.data); frame.size = sizeof(serialize_buf); CS_ENTER(persistent_manager->lock); insert_result = ClibsharequeueInsert(producer_manager->share_queue, &frame, 0); CS_LEAVE(persistent_manager->lock); if (insert_result == 0) { aos_warn_log((LB, "get from queue, uuid %s insert failed.", buf.uuid)); } log_group_destroy(builder); continue; } //aos_debug_log((LB, "get from queue is null.")); break; } while (1); // if no job, check now loggroup _try_flush_loggroup(producer_manager); } Sleep(1000); } aos_info_log((LB, "exit flusher thread")); return 0; } log_producer_manager* create_log_producer_manager(log_producer_config* producer_config) { int32_t base_queue_size; log_producer_manager* producer_manager; static int isFlushThreadExist = 0; producer_manager = g_producer_manager[producer_manager_num] = (log_producer_manager*)malloc(sizeof(log_producer_manager)); memset(producer_manager, 0, sizeof(log_producer_manager)); producer_manager->producer_config = producer_config; base_queue_size = producer_config->maxBufferBytes / (producer_config->logBytesPerPackage + 1) + 10; if (base_queue_size < MIN_LOGGROUP_QUEUE_SIZE) { base_queue_size = MIN_LOGGROUP_QUEUE_SIZE; } else if (base_queue_size > MAX_LOGGROUP_QUEUE_SIZE) { base_queue_size = MAX_LOGGROUP_QUEUE_SIZE; } producer_manager->loggroup_queue = log_queue_create(base_queue_size); if (producer_config->useMemoryType == 0) { producer_manager->share_queue = ClibsharequeueCreate2("RVC_LOG_SDK_QUEUE"); } else { producer_manager->share_queue = ClibsharequeueCreate3("RVC_LOG_SDK_QUEUE", 1); } producer_manager->send_param_queue_size = base_queue_size * 2; producer_manager->send_param_queue = malloc(sizeof(log_producer_send_param*) * producer_manager->send_param_queue_size); producer_manager->triger_cond = CreateCond(); producer_manager->lock = CreateCriticalSection(); producer_manager_num++; if (!isFlushThreadExist) { isFlushThreadExist = 1; THREAD_INIT(producer_manager->flush_thread, log_producer_flush_thread, NULL);//以当前情况，无法减少这个flush线程，因为是跟producer_config绑定的 } //huchen add for only Guarder has read 、flush and send thread if (producer_config->enableGuarder){//create upload part aos_debug_log((LB, "to create send&read&write thread %d", producer_config->sendThreadCount)); if (producer_config->sendThreadCount > 0) { int32_t threadId; producer_manager->multi_thread_send_count = 0; producer_manager->send_threads = (THREAD*)malloc(sizeof(THREAD) * producer_config->sendThreadCount); producer_manager->sender_data_queue = log_queue_create(base_queue_size * 2); threadId = 0; for (; threadId < producer_manager->producer_config->sendThreadCount; ++threadId) { THREAD_INIT(producer_manager->send_threads[threadId], log_producer_send_thread, producer_manager); } producer_manager->read_thread = (THREAD*)malloc(sizeof(THREAD)); aos_debug_log((LB, "to create read thread")); THREAD_INIT(producer_manager->read_thread, log_producer_read_persistent_thread, producer_manager); producer_manager->write_thread = (THREAD*)malloc(sizeof(THREAD)); aos_debug_log((LB, "to create write thread")); THREAD_INIT(producer_manager->write_thread, log_producer_write_persistent_thread, producer_manager); } } aos_debug_log((LB, "create log producer manager, base_queue_size %d, name: %s", base_queue_size, producer_manager->producer_config->skyeyeEntityName)); return producer_manager; } void _push_last_loggroup(log_producer_manager* manager) { log_group_builder* builder; CS_ENTER(manager->lock); builder = manager->builder; manager->builder = NULL; if (builder != NULL) { int32_t status; size_t loggroup_size = builder->loggroup_size; aos_debug_log((LB, "try push loggroup to flusher, size : %d", (int)builder->loggroup_size)); status = log_queue_push(manager->loggroup_queue, builder); if (status != 0) { aos_error_log((LB, "try push loggroup to flusher failed, force drop this log group, error code : %d", status)); log_group_destroy(builder); } else { manager->totalBufferSize += loggroup_size; COND_SIGNAL(manager->triger_cond); } } CS_LEAVE(manager->lock); } void destroy_log_producer_manager(log_producer_manager* manager) { int32_t total_wait_count; int waitCount = 0; // when destroy instance, flush last loggroup _push_last_loggroup(manager); aos_info_log((LB, "flush out producer loggroup begin")); total_wait_count = manager->producer_config->destroyFlusherWaitTimeoutSec > 0 ? manager->producer_config->destroyFlusherWaitTimeoutSec * 100 : MAX_MANAGER_FLUSH_COUNT; total_wait_count += manager->producer_config->destroySenderWaitTimeoutSec > 0 ? manager->producer_config->destroySenderWaitTimeoutSec * 100 : MAX_SENDER_FLUSH_COUNT; #ifdef WIN32 Sleep(10); #else usleep(10 * 1000); #endif while (log_queue_size(manager->loggroup_queue) > 0 || manager->send_param_queue_write - manager->send_param_queue_read > 0 || (manager->sender_data_queue != NULL && log_queue_size(manager->sender_data_queue) > 0)) { #ifdef WIN32 Sleep(10); #else usleep(10 * 1000); #endif if (++waitCount == total_wait_count) { break; } } if (waitCount == total_wait_count) { aos_error_log((LB, "try flush out producer loggroup error, force exit, now loggroup %d", (int)(log_queue_size(manager->loggroup_queue)))); } else { aos_info_log((LB, "flush out producer loggroup success")); } manager->shutdown = 1; // destroy root resources COND_SIGNAL(manager->triger_cond); if (manager->write_thread != NULL) { aos_info_log((LB, "join write thread begin")); THREAD_JOIN(manager->write_thread); free(manager->write_thread); manager->write_thread = NULL; aos_info_log((LB, "join write thread success")); } if (manager->read_thread != NULL) { aos_info_log((LB, "join read thread begin")); THREAD_JOIN(manager->read_thread); free(manager->read_thread); manager->read_thread = NULL; aos_info_log((LB, "join read thread success")); } aos_info_log((LB, "join flush thread begin")); THREAD_JOIN(manager->flush_thread); aos_info_log((LB, "join flush thread success")); if (manager->send_threads != NULL) { int32_t threadId; aos_info_log((LB, "join sender thread pool begin")); threadId = 0; for (; threadId < manager->producer_config->sendThreadCount; ++threadId) { THREAD_JOIN(manager->send_threads[threadId]); } free(manager->send_threads); manager->send_threads = NULL; aos_info_log((LB, "join sender thread pool success")); } DeleteCond(manager->triger_cond); log_queue_destroy(manager->loggroup_queue); ClibsharequeueDestroy(manager->share_queue); if (manager->sender_data_queue != NULL) { aos_info_log((LB, "flush out sender queue begin")); while (log_queue_size(manager->sender_data_queue) > 0) { void* send_param = log_queue_trypop(manager->sender_data_queue); if (send_param != NULL) { log_producer_send_fun(send_param); } } log_queue_destroy(manager->sender_data_queue); aos_info_log((LB, "flush out sender queue success")); } ReleaseCriticalSection(manager->lock); if (manager->send_param_queue != NULL) { free(manager->send_param_queue); manager->send_param_queue = NULL; } free(manager); //TODO //g_producer_manager[producer_manager_num] } #define LOG_PRODUCER_MANAGER_ADD_LOG_BEGIN CS_ENTER(producer_manager->lock); \ if (producer_manager->builder == NULL) \ { \ if (log_queue_isfull(producer_manager->loggroup_queue)) \ { \ CS_LEAVE(producer_manager->lock); \ return LOG_PRODUCER_DROP_ERROR; \ } \ producer_manager->builder = log_group_create(producer_manager->producer_config); \ producer_manager->builder->private_value = producer_manager; \ } #define LOG_PRODUCER_MANAGER_ADD_LOG_END { int ret; \ int status; \ log_group_builder * builder = producer_manager->builder; \ if (flush == 0 && producer_manager->builder->loggroup_size < producer_manager->producer_config->logBytesPerPackage && producer_manager->builder->grp->n_logs < producer_manager->producer_config->logCountPerPackage) \ { \ CS_LEAVE(producer_manager->lock); \ return LOG_PRODUCER_OK; \ } \ ret = LOG_PRODUCER_OK; \ producer_manager->builder = NULL; \ status = log_queue_push(producer_manager->loggroup_queue, builder); \ if (status != 0) \ { \ do \ { \ char *szLogBuf = AllocLogBuf(); \ char szLevel[10]; \ char szTime[32]; \ szLogBuf[MFLOG_BUFSIZE] = 0; \ MakeLogPrefix( LM_ERROR, szLevel, szTime ); \ _snprintf(szLogBuf, MFLOG_BUFSIZE, "try push loggroup to flusher failed, force drop this log group, error code : %d", status); \ RvcLog_log( __FILE__,__LINE__, 1, LM_ERROR, szLevel, szTime, szLogBuf); \ }while(0); \ ret = LOG_PRODUCER_DROP_ERROR; \ log_group_destroy(builder); \ } \ else \ { \ COND_SIGNAL(producer_manager->triger_cond); \ } \ CS_LEAVE(producer_manager->lock); \ return ret; } log_producer_result log_producer_manager_add_log(log_producer_manager* producer_manager, log_item* log, int flush, long record_time) { LOG_PRODUCER_MANAGER_ADD_LOG_BEGIN; if (0 != add_log(producer_manager->builder, producer_manager->start_time, producer_manager->pack_index, log, record_time)) { aos_error_log((LB, "add log failed !")); } LOG_PRODUCER_MANAGER_ADD_LOG_END; } log_producer_result log_producer_manager_add_beidou_log(log_producer_manager* producer_manager, beidou_log_item* log, int flush) { LOG_PRODUCER_MANAGER_ADD_LOG_BEGIN; if (0 != add_beidou_log(producer_manager->builder, log)) { aos_error_log((LB, "add log failed !")); } LOG_PRODUCER_MANAGER_ADD_LOG_END; } log_producer_result log_producer_manager_add_log_raw(log_producer_manager* producer_manager, build_item* log, int flush) { LOG_PRODUCER_MANAGER_ADD_LOG_BEGIN; add_log_raw(producer_manager->builder, log); LOG_PRODUCER_MANAGER_ADD_LOG_END; } log_producer_result log_producer_manager_add_log_group(log_producer_manager* producer_manager, log_group_builder* builder, int flush) { int ret; int status; CS_ENTER(producer_manager->lock); if (flush == 0 && builder->loggroup_size < producer_manager->producer_config->logBytesPerPackage && builder->grp->n_logs < producer_manager->producer_config->logCountPerPackage) { CS_LEAVE(producer_manager->lock); return LOG_PRODUCER_OK; } ret = LOG_PRODUCER_OK; builder->private_value = producer_manager; aos_debug_log((LB, "try push loggroup to flusher, size : %d, log count %d", (int)builder->loggroup_size, (int)builder->grp->n_logs)); status = log_queue_push(producer_manager->loggroup_queue, builder); if (status != 0) { aos_error_log((LB, "try push loggroup to flusher failed, force drop this log group, error code : %d", status)); ret = LOG_PRODUCER_DROP_ERROR; } else { COND_SIGNAL(producer_manager->triger_cond); } CS_LEAVE(producer_manager->lock); return ret; }