myvtm/Framework/RvcLogSdk/log_producer_manager.c

#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);
                        /*
                        int i = 0;
                        for (i = 0; i < send_param->log_buf->n_logs; i++)
                        {
                            char uuid[MAX_UUID_LEN];
                            memset(uuid, 0, MAX_UUID_LEN);
                            memcpy(uuid, send_param->log_buf->uuid[i], MAX_UUID_LEN);
                            send_log_data("http://127.0.0.1:9000/read_uuid", uuid);
                        }
                            */

                        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;
}