myvtm/Framework/RvcLogSdk/log_persistent_manager.cpp

#include "log_persistent_manager.h"
#include "log_producer_manager.h"
#include "inner_log.h"
#include "log_builder.h"
#include "sds.h"
#include "baseFun.h"

#ifndef _WIN32
#include <sys/types.h>
#include <dirent.h>
#include <sys/stat.h>
#endif //NOT _WIN32

#define DB_TABLE_NAME  "RvcLogTable" 

#if defined(_MSC_VER)
    #define DB_PATH  "RVC.LogSdk\\"
#else
    #define DB_PATH  "RVC.LogSdk/"
#endif //_MSC_VER


#define SLASH '/'
#define BACK_SLASH '\\'
#define MB_SLASH_STR "/"
#define MB_BACK_SLASH_STR "\\"
#define W_SLASH_STR L"/"
#define W_BACK_SLASH_STR L"\\"

#if defined(UNICODE) || defined(_UNICODE)
#define SLASH_STR W_SLASH_STR
#define BACK_SLASH_STR W_BACK_SLASH_STR
#else 
#define SLASH_STR MB_SLASH_STR
#define BACK_SLASH_STR MB_BACK_SLASH_STR
#endif

#ifdef _WIN32
#define SPLIT_SLASH BACK_SLASH
#define SPLIT_SLASH_STR BACK_SLASH_STR
#define MB_SPLIT_SLASH_STR MB_BACK_SLASH_STR
#define W_SPLIT_SLASH_STR   W_BACK_SLASH_STR

#define LINE_BREAK_STR "\r\n"

#else
#define SPLIT_SLASH SLASH
#define SPLIT_SLASH_STR SLASH_STR
#define MB_SPLIT_SLASH_STR MB_SLASH_STR
#define W_SPLIT_SLASH_STR   W_SLASH_STR

#define LINE_BREAK_STR "\n"

#endif



static void _copy_config_string(const char* value, sds* src_value)
{
    int strLen;
    if (value == NULL || src_value == NULL)
    {
        return;
    }
    strLen = strlen(value);
    if (*src_value == NULL)
    {
        *src_value = sdsnewEmpty(strLen);
    }
    else {
        sdsfree(*src_value);
    }
    *src_value = sdscpylen(*src_value, value, strLen);
}

static log_db_manager* create_db_manager(log_persistent_manager* manager, char * modular) {
    int ret = 0;

    log_db_manager* guard_db_manager = NULL;
    char tmpFilePath[MAX_PATH] = { 0 };
    int len = strlen(manager->config->persistentFilePath);
    strcpy(tmpFilePath, manager->config->persistentFilePath);
    if (manager->config->persistentFilePath[len - 1] == '\\' || manager->config->persistentFilePath[len - 1] == '/') {
        strcat(tmpFilePath, DB_PATH);
    }
    else {
        strcat(tmpFilePath, SPLIT_SLASH_STR);
        strcat(tmpFilePath, DB_PATH);
    }

    guard_db_manager = create_log_db(manager->config, tmpFilePath, modular, DB_TABLE_NAME);
    if (guard_db_manager != NULL) {
        ret = open_db(guard_db_manager);
        if (ret <= 0) {
            destroy_log_db(guard_db_manager);
            guard_db_manager = NULL;
        }
    }
    return guard_db_manager;
}


static int log_persistent_manager_update_guard_log_status(log_persistent_manager* manager, 
    char* modular, int send_count, char uuid[MAX_LOG_COUNT][MAX_UUID_LEN], log_db_status_e status) {

    log_db_manager* guard_db_manager = NULL;
    int i, op_count = 0;

    guard_db_manager = create_db_manager(manager, modular);
    if (guard_db_manager != NULL) {
        db_transaction_begin(guard_db_manager);
        for (i = 0; i < send_count; i++) {
            op_count += db_update_status(guard_db_manager, uuid[i], status);
        }
        db_transaction_commit(guard_db_manager);
        destroy_log_db(guard_db_manager);
    }
    else {
        return FALSE;
    }
    return op_count;
}


static int log_persistent_manager_delete_guard_log(log_persistent_manager* manager,
    char* modular, int send_count, char uuid[MAX_LOG_COUNT][MAX_UUID_LEN]) {

    log_db_manager* guard_db_manager = NULL;
    int ret, i, op_count = 0;

    guard_db_manager = create_db_manager(manager, modular);
    if (guard_db_manager != NULL) {
        db_transaction_begin(guard_db_manager);
        for (i = 0; i < send_count; i++) {
            op_count += db_delete_one(guard_db_manager, uuid[i]);
        }
        db_transaction_commit(guard_db_manager);
        destroy_log_db(guard_db_manager);
    }
    else {
        return FALSE;
    }
    return op_count;
}

void on_log_persistent_manager_send_done_uuid(
    log_producer_result result,
    int log_bytes,
    int compressed_bytes,
    const char* req_id,
    const char* error_message,
    const unsigned char* raw_buffer,
    void* persistent_manager,
    int send_count,
    char uuid[MAX_LOG_COUNT][MAX_UUID_LEN],
    char* modular)
{
    int i;
    char tmpFileName[MAX_PATH] = { 0 };
    log_persistent_manager* manager = (log_persistent_manager*)persistent_manager;
    if (manager == NULL)
    {
        return;
    }

    strcat(tmpFileName, "RvcLogSdk.db"/*manager->config->skyeyeEntityName*/);
    /*如果异常，则将条目恢复成待发送状态*/
    if (result != LOG_PRODUCER_OK && result != LOG_PRODUCER_DROP_ERROR && result != LOG_PRODUCER_INVALID)
    {
        CS_ENTER(manager->lock);
        log_persistent_manager_updata_log_status(manager, modular, send_count, uuid, LOG_DB_STATUS_WAITING);
        CS_LEAVE(manager->lock);
        return;
    }

    CS_ENTER(manager->lock);
    if (strcmp(modular,tmpFileName)==0) {
        db_transaction_begin(manager->db_manager);
        for (i = 0; i < send_count; i++) {
            log_persistent_manager_delete_log(manager, uuid[i]);
        }
        db_transaction_commit(manager->db_manager);
    }
    else {
        log_persistent_manager_delete_guard_log(manager, modular, send_count, uuid);
    }
    CS_LEAVE(manager->lock);
}


static int log_persistent_manager_init(log_persistent_manager* manager, log_producer_config* config)
{
    int ret;
    int len;
    char tmpFilePath[MAX_PATH] = {0};
    char tmpFileName[MAX_PATH] = { 0 };
    if (config->persistentFilePath == NULL) {
        return -1;
    }

    memset(manager, 0, sizeof(log_persistent_manager));
    manager->config = config;
    manager->lock = CreateCriticalSection();

    len = strlen(config->persistentFilePath);
    strcpy(tmpFilePath, config->persistentFilePath);
    if (config->persistentFilePath[len - 1] == '\\' || config->persistentFilePath[len - 1] == '/') {
        strcat(tmpFilePath, DB_PATH);
    }
    else {
        strcat(tmpFilePath, SPLIT_SLASH_STR);
        strcat(tmpFilePath, DB_PATH);
    }
    strcat(tmpFileName, "RvcLogSdk.db");//config->skyeyeEntityName);
    manager->db_manager = create_log_db(config, tmpFilePath, tmpFileName, DB_TABLE_NAME);
    _copy_config_string(tmpFileName, &manager->db_name);

    if (manager->db_manager == NULL) {
        return -1;
    }

    ret = open_db(manager->db_manager);
    if (ret <= 0) {
        return -1;
    }
    if (config->openDB == 0)
        close_logdb(manager->db_manager);
    return 0;
}

static void log_persistent_manager_clear(log_persistent_manager* manager)
{
    ReleaseCriticalSection(manager->lock);
    close_logdb(manager->db_manager);
}

log_persistent_manager* create_log_persistent_manager(log_producer_config* config)
{
    int rst;
    log_persistent_manager* manager;
    if (!log_producer_persistent_config_is_enabled(config))
    {
        return NULL;
    }
    manager = (log_persistent_manager*)malloc(sizeof(log_persistent_manager));
    rst = log_persistent_manager_init(manager, config);
    if (rst < 0) {
        free(manager);
        return NULL;
    }
    return manager;
}


void destroy_log_persistent_manager(log_persistent_manager* manager)
{
    if (manager == NULL)
    {
        return;
    }
    log_persistent_manager_clear(manager);
    sdsfree(manager->db_name);
    free(manager);
    manager = NULL;
}


int log_persistent_manager_save_log_group(log_persistent_manager* manager, log_group_builder* builder)
{
    unsigned long end_time;
    unsigned long start_time = GetTickCount();
    int rst = db_insert_group(manager->db_manager, builder);
    end_time = GetTickCount();
    //aos_debug_log((LB, "write db log cust %d", end_time - start_time));
    if (rst <= 0) {
        aos_error_log((LB, "write db log failed, rst %d",
            rst));
        return LOG_PRODUCER_PERSISTENT_ERROR;
    }
    else {
        return LOG_PRODUCER_OK;
    }
}

int log_persistent_manager_save_log(log_persistent_manager* manager, serialize_buf* buf)
{
    //unsigned long end_time;
    //unsigned long start_time = GetTickCount();
    // it would not generate dirty logs
    int rst = db_insert_one(manager->db_manager, buf);
    //end_time = GetTickCount();
    //aos_debug_log((LB, "write db log cust %d", end_time - start_time));
    if (rst <= 0) {
        aos_error_log((LB, "write db log failed, rst %d",
            rst));
        return LOG_PRODUCER_PERSISTENT_ERROR;
    }
    else {
        return LOG_PRODUCER_OK;
    }
}

static log_group_builder* log_persistent_manager_get_guard_log(log_persistent_manager* manager)
{

#if defined(_MSC_VER)

	static int read_file_index = 0;
	char srcFilePath[MAX_PATH] = { 0 };
	WIN32_FIND_DATA FindFileData;
	HANDLE hFind;
	BOOL fFinished = FALSE;
	log_db_manager* guard_db_manager = NULL;

	int i;
	char tmpFilePath[MAX_PATH] = { 0 };
	int len = strlen(manager->config->persistentFilePath);
	strcpy(tmpFilePath, manager->config->persistentFilePath);
	if (manager->config->persistentFilePath[len - 1] == '\\' || manager->config->persistentFilePath[len - 1] == '/') {
		strcat(tmpFilePath, DB_PATH);
	} else {
		strcat(tmpFilePath, "\\");
		strcat(tmpFilePath, DB_PATH);
	}

	sprintf_s(srcFilePath, MAX_PATH, "%s*.db", tmpFilePath);
	hFind = FindFirstFile(srcFilePath, &FindFileData);
	if (INVALID_HANDLE_VALUE != hFind) {
		for (i = 0; !fFinished; i++) {
			if ((FILE_ATTRIBUTE_DIRECTORY & FindFileData.dwFileAttributes) || (i < read_file_index)) {
				if (!FindNextFile(hFind, &FindFileData)) {
					if (GetLastError() == ERROR_NO_MORE_FILES) {
						read_file_index = 0;
						fFinished = TRUE;
					} else {
						break;
					}
				}
				continue;
			}

			aos_debug_log((LB, "guard read log db name %s", FindFileData.cFileName));
			if (strcmp(manager->db_name, FindFileData.cFileName) != 0) {
				guard_db_manager = create_db_manager(manager, FindFileData.cFileName);
				if (guard_db_manager != NULL) {
					log_group_builder* builder = db_read_table_last_logs(guard_db_manager, manager->config->logCountPerPackage);
					destroy_log_db(guard_db_manager);
					if (builder != NULL) {
						read_file_index++;
						FindClose(hFind);
						return builder;
					}
				}
			} else {
				log_group_builder* builder = db_read_table_last_logs(manager->db_manager, manager->config->logCountPerPackage);
				if (builder != NULL) {
					read_file_index++;
					FindClose(hFind);
					return builder;
				}
			}

			if (!FindNextFile(hFind, &FindFileData)) {
				if (GetLastError() == ERROR_NO_MORE_FILES) {
					read_file_index = 0;
					fFinished = TRUE;
				} else {
					break;
				}
			}
		}
		FindClose(hFind);
	}
	return NULL;

#else

	static int read_file_index = 0;

	DIR* d = NULL;
	struct dirent* dp = NULL;
	struct stat st;

	char srcFilePath[MAX_PATH] = { 0 };
	BOOL fFinished = FALSE;
	log_db_manager* guard_db_manager = NULL;

	int ret, i;
	char tmpFilePath[MAX_PATH] = { 0 };
	const int len = strlen(manager->config->persistentFilePath);
	strcpy(tmpFilePath, manager->config->persistentFilePath);
	if (manager->config->persistentFilePath[len - 1] == '\\' || manager->config->persistentFilePath[len - 1] == '/') {
		strcat(tmpFilePath, DB_PATH);
		tmpFilePath[strlen(tmpFilePath) - 1] = '\0';
	} else {
		strcat(tmpFilePath, SPLIT_SLASH_STR);
		strcat(tmpFilePath, DB_PATH);
		tmpFilePath[strlen(tmpFilePath) - 1] = '\0';
	}

	if (stat(tmpFilePath, &st) < 0 || !S_ISDIR(st.st_mode)) {
		return NULL;
	}

	if (!(d = opendir(tmpFilePath))) {
		return NULL;
	}
	while ((dp = readdir(d)) != NULL) {
		if ((!strncmp(dp->d_name, ".", 1)) || (!strncmp(dp->d_name, "..", 2))) {
			continue;
		}
		int isdir = 0;
		struct stat st1;

		sprintf(srcFilePath, "%s/%s", tmpFilePath, dp->d_name);
		if (stat(srcFilePath, &st1) < 0) {
			continue;
		}
		isdir = !!(S_ISDIR(st1.st_mode));
		if (isdir) {
			continue;
		}

		if (strstr(dp->d_name, ".db") == NULL) {
			continue;
		}

		if (i < read_file_index) {
			i++;
			continue;
		}

		aos_debug_log((LB, "guard read log db name %s", dp->d_name));
		if (strcmp(manager->db_name, dp->d_name) != 0) {
			guard_db_manager = create_db_manager(manager, dp->d_name);
			if (guard_db_manager != NULL) {
				log_group_builder* builder = db_read_table_last_logs(guard_db_manager, manager->config->logCountPerPackage);
				destroy_log_db(guard_db_manager);
				if (builder != NULL) {
					read_file_index++;
					return builder;
				}
			}
		} else {
			log_group_builder* builder = db_read_table_last_logs(manager->db_manager, manager->config->logCountPerPackage);
			if (builder != NULL) {
				read_file_index++;
				return builder;
			}
		}
	}
	closedir(d);
	read_file_index = 0;
	return NULL;
#endif //_MSC_VER
}

log_group_builder* log_persistent_manager_get_log(log_persistent_manager* manager) {
    /*two way get log
    (1) get log from db
    (2) get log from dirty resend frame list
    */

    if (RvcLogSdkManager::getInstance().getResendFrameList().size() > 0)//already send all db logs, deal with dirty logs
    {
        log_group_builder* cur = RvcLogSdkManager::getInstance().getResendFrameList().front();
        RvcLogSdkManager::getInstance().getResendFrameList().pop_front();
        return cur;
    }
    
    unsigned long end_time;
    unsigned long start_time = GetTickCount();
    log_group_builder* builder = db_read_table_last_logs(manager->db_manager, manager->config->logCountPerPackage);
    end_time = GetTickCount();
    //aos_debug_log((LB, "read db log cust %d", end_time - start_time));
    if (builder == NULL) {
        aos_error_log((LB, "get db log failed"));
        
        //add for Guarder upload
        if (manager->config->enableGuarder) {
            return log_persistent_manager_get_guard_log(manager);
        }
    }
    return builder;


    
}


int log_persistent_manager_updata_log_status(log_persistent_manager* manager, char* modular, int count, char uuid[][MAX_UUID_LEN], log_db_status_e status) {
    int i = 0, op_count = 0;
    unsigned long end_time;
    unsigned long start_time = GetTickCount();

    if (strcmp(manager->db_name, modular) == 0) {
        db_transaction_begin(manager->db_manager);
        for (i = 0; i < count; i++) {
            op_count += db_update_status(manager->db_manager, uuid[i], status);
        }
        db_transaction_commit(manager->db_manager);
        end_time = GetTickCount();
        //aos_debug_log((LB, "updata db log status cust %d, op count %d", end_time - start_time, op_count));
        if (op_count <= 0) {
            aos_error_log((LB, "updata_log_status failed"));
            return LOG_PRODUCER_PERSISTENT_ERROR;
        }
        else {
            return LOG_PRODUCER_OK;
        }
    }
    else {
        op_count = log_persistent_manager_update_guard_log_status(manager, modular, count, uuid, status);
        if (op_count <= 0) {
            aos_error_log((LB, "updata_log_status failed"));
            return LOG_PRODUCER_PERSISTENT_ERROR;
        }
        else {
            return LOG_PRODUCER_OK;
        }
    }
}

int log_persistent_manager_delete_log(log_persistent_manager* manager, char* uuid) {
    unsigned long end_time;
    unsigned long start_time = GetTickCount();
    int rst = db_delete_one(manager->db_manager, uuid);
    end_time = GetTickCount();
    //aos_debug_log((LB, "delete db log cust %d", end_time - start_time));
    if (rst <= 0) {
        aos_error_log((LB, "delete_log failed, rst %d",
            rst));
        return LOG_PRODUCER_PERSISTENT_ERROR;
    }
    else {
        return LOG_PRODUCER_OK;
    }
}

/*将数据恢复成为待上传状态*/
int log_persistent_manager_recover_inner(log_persistent_manager* manager, log_db_manager* log_db_manager) {
    unsigned long end_time;
    unsigned long start_time = GetTickCount();
    int rst = db_update_status(log_db_manager, NULL, LOG_DB_STATUS_WAITING);
    end_time = GetTickCount();
    aos_debug_log((LB, "recover db log cust %d", end_time - start_time));
    if (rst <= 0) {
        aos_error_log((LB, "persistent_manager recover failed, rst %d",
            rst));
        return LOG_PRODUCER_PERSISTENT_ERROR;
    }
    else {
        return LOG_PRODUCER_OK;
    }
}

int log_persistent_manager_recover(log_persistent_manager* manager) {

#if defined(_MSC_VER)

	char srcFilePath[MAX_PATH] = { 0 };
	WIN32_FIND_DATA FindFileData;
	HANDLE hFind;
	BOOL fFinished = FALSE;
	log_db_manager* guard_db_manager = NULL;

	int i;
	char tmpFilePath[MAX_PATH] = { 0 };
	int len = strlen(manager->config->persistentFilePath);
	strcpy(tmpFilePath, manager->config->persistentFilePath);
	if (manager->config->persistentFilePath[len - 1] == '\\' || manager->config->persistentFilePath[len - 1] == '/') {
		strcat(tmpFilePath, DB_PATH);
	} else {
		strcat(tmpFilePath, "\\");
		strcat(tmpFilePath, DB_PATH);
	}

	sprintf_s(srcFilePath, MAX_PATH, "%s*.db", tmpFilePath);
	hFind = FindFirstFile(srcFilePath, &FindFileData);
	if (INVALID_HANDLE_VALUE != hFind) {
		for (i = 0; !fFinished; i++) {
			if ((FILE_ATTRIBUTE_DIRECTORY & FindFileData.dwFileAttributes)) {
				if (!FindNextFile(hFind, &FindFileData)) {
					if (GetLastError() == ERROR_NO_MORE_FILES) {
						fFinished = TRUE;
					} else {
						break;
					}
				}
				continue;
			}

			aos_debug_log((LB, "recover guard log db name %s", FindFileData.cFileName));
			//clean
			//log_persistent_manager_clean(manager, FindFileData.cFileName);
			//recover
			if (strcmp(manager->db_name, FindFileData.cFileName) != 0) {
				guard_db_manager = create_db_manager(manager, FindFileData.cFileName);
				if (guard_db_manager != NULL) {
					log_persistent_manager_recover_inner(manager, guard_db_manager);
					destroy_log_db(guard_db_manager);
				}
			} else {
				log_persistent_manager_recover_inner(manager, manager->db_manager);
			}

			if (!FindNextFile(hFind, &FindFileData)) {
				if (GetLastError() == ERROR_NO_MORE_FILES) {
					fFinished = TRUE;
				} else {
					break;
				}
			}
		}
		FindClose(hFind);
	}
	return LOG_PRODUCER_OK;

#else

	DIR* d = NULL;
	struct dirent* dp = NULL;
	struct stat st;

	char srcFilePath[MAX_PATH] = { 0 };
	BOOL fFinished = FALSE;
	log_db_manager* guard_db_manager = NULL;

	int ret, i;
	char tmpFilePath[MAX_PATH] = { 0 };
	int len = strlen(manager->config->persistentFilePath);
	strcpy(tmpFilePath, manager->config->persistentFilePath);
	if (manager->config->persistentFilePath[len - 1] == '\\' || manager->config->persistentFilePath[len - 1] == '/') {
		strcat(tmpFilePath, DB_PATH);
		tmpFilePath[strlen(tmpFilePath) - 1] = '\0';
	} else {
		strcat(tmpFilePath, SPLIT_SLASH_STR);
		strcat(tmpFilePath, DB_PATH);
		tmpFilePath[strlen(tmpFilePath) - 1] = '\0';
	}
	aos_debug_log((LB, "recover log db path %s", tmpFilePath));

	if (stat(tmpFilePath, &st) < 0 || !S_ISDIR(st.st_mode)) {
		aos_warn_log((LB, "stat %s failed: %d", tmpFilePath, errno));
		return NULL;
	}

	if (!(d = opendir(tmpFilePath))) {
		aos_warn_log((LB, "opendir %s failed: %d", tmpFilePath, errno));
		return NULL;
	}
	while ((dp = readdir(d)) != NULL) {
		if ((!strncmp(dp->d_name, ".", 1)) || (!strncmp(dp->d_name, "..", 2))) {
			continue;
		}
		int isdir = 0;
		struct stat st1;
		sprintf(srcFilePath, "%s/%s", tmpFilePath, dp->d_name);
		if (stat(srcFilePath, &st1) < 0) {
			aos_warn_log((LB, "stat %s failed: %d", srcFilePath, errno));
			continue;
		}
		isdir = !!(S_ISDIR(st1.st_mode));
		if (isdir) {
			continue;
		}

		if (strstr(dp->d_name, ".db") == NULL) {
			continue;
		}

		aos_debug_log((LB, "recover log db name %s", dp->d_name));
		//clean
		//log_persistent_manager_clean(manager, FindFileData.cFileName);
		//recover
		if (strcmp(manager->db_name, dp->d_name) != 0) {
			guard_db_manager = create_db_manager(manager, dp->d_name);
			if (guard_db_manager != NULL) {
				log_persistent_manager_recover_inner(manager, guard_db_manager);
				destroy_log_db(guard_db_manager);
			}
		} else {
			log_persistent_manager_recover_inner(manager, manager->db_manager);
		}
	}
	closedir(d);

	return LOG_PRODUCER_OK;

#endif //_MSC_VER


}

int log_persistent_manager_clean(log_persistent_manager* manager, char* modular) {

    if (strcmp(manager->db_name, modular) == 0) {
        int rst = 1;
        unsigned long end_time;
        unsigned long start_time = GetTickCount();

        int count = db_get_count(manager->db_manager);
        if (count > manager->config->maxPersistentLogCount) {//超过最大值，然后清理 2*10000
            rst = db_delete_old_logs(manager->db_manager, count - manager->config->maxPersistentLogCount);
            g_discardMsgNum_since_full += count - manager->config->maxPersistentLogCount;
        }
        db_vacuum(manager->db_manager);
        end_time = GetTickCount();
        aos_debug_log((LB, "clean db log cust %d", end_time - start_time));
        if (rst <= 0) {
            aos_error_log((LB, "persistent_manager clean failed, rst %d",
                rst));
            return LOG_PRODUCER_PERSISTENT_ERROR;
        }
        else {
            return LOG_PRODUCER_OK;
        }
    }
    else {
        log_db_manager* guard_db_manager = create_db_manager(manager, modular);
        if (guard_db_manager != NULL) {
            int rst = 1;
            unsigned long end_time;
            unsigned long start_time = GetTickCount();

            int count = db_get_count(guard_db_manager);
            if (count > manager->config->maxPersistentLogCount) {
                rst = db_delete_old_logs(guard_db_manager, count - manager->config->maxPersistentLogCount);
            }
            db_vacuum(guard_db_manager);
            destroy_log_db(guard_db_manager);
            end_time = GetTickCount();
            aos_debug_log((LB, "clean guard db log cust %d, modular:%s", end_time - start_time, modular));
            if (rst <= 0) {
                aos_error_log((LB, "persistent_manager guard clean failed, rst %d",
                    rst));
                return LOG_PRODUCER_PERSISTENT_ERROR;
            }
            else {
                return LOG_PRODUCER_OK;
            }
        }
    }
    return LOG_PRODUCER_OK;
}


int log_persistent_manager_transaction_begin(log_persistent_manager* manager) {
    int rst = 1;
    unsigned long end_time;
    unsigned long start_time = GetTickCount();

    rst = db_transaction_begin(manager->db_manager);
    end_time = GetTickCount();
    aos_debug_log((LB, "begin db log cust %d", end_time - start_time));
    if (rst <= 0) {
        aos_error_log((LB, "persistent_manager begin db failed, rst %d",
            rst));
        return LOG_PRODUCER_PERSISTENT_ERROR;
    }
    else {
        return LOG_PRODUCER_OK;
    }
}
int log_persistent_manager_transaction_commit(log_persistent_manager* manager) {
    int rst = 1;
    unsigned long end_time;
    unsigned long start_time = GetTickCount();
    rst = db_transaction_commit(manager->db_manager);
    end_time = GetTickCount();
    aos_debug_log((LB, "commit db log cust %d", end_time - start_time));
    if (rst <= 0) {
        aos_error_log((LB, "persistent_manager commit db failed, rst %d",
            rst));
        return LOG_PRODUCER_PERSISTENT_ERROR;
    }
    else {
        return LOG_PRODUCER_OK;
    }
}