#include #include #include #include #include #include #include #include #include #include #include "list.h" #include "MESA_shm_ring_queue.h" #define DEFAUT_BUF_SIZE 256 #define CONSUMER_SUCCESS 0 #define CONSUMER_ERROR -1 struct log_file_list{ char log_file_pre[MESA_SHM_LOG_PATH_LEN]; char real_log_file[MESA_SHM_LOG_PATH_LEN]; struct tm create_date; int fd; dev_t dev; ino_t ino; struct list_head list; }; struct log_file_list g_log_file_list; void init_ring_queue_head_arrray(struct MESA_shm_overview *ovw, struct MESA_shm_queue_head **ring_queue_head) { int i = 0; struct MESA_shm_overview *tmp_ovw = NULL; for(i = 0; i< MESA_SHM_RING_QUEUE_NUM; i++){ tmp_ovw = ovw + i; if(tmp_ovw->shmid == -1){ break; } ring_queue_head[i] = shmat(tmp_ovw->shmid, NULL, 0); } return ; } void consumer_daemo() { int fd = -1; switch (fork()) { case -1: printf("fork error\n"); return ; case 0: break; default: exit(0); } if (setsid() == -1) { return ; } umask(0); fd = open("/dev/null", O_RDWR); if (fd == -1) { return ; } if (dup2(fd, STDIN_FILENO) == -1) { return ; } if (dup2(fd, STDOUT_FILENO) == -1) { return ; } if (fd > STDERR_FILENO) { if (close(fd) == -1) { return ; } } return ; } int consumer_is_running(char *process_name) { FILE *fp = NULL; char buf[DEFAUT_BUF_SIZE] = {0}; int count = 0; char command[DEFAUT_BUF_SIZE] = {0}; if(process_name == NULL){ return 0; } snprintf(command, sizeof(command), "ps -ef | grep %s | grep -v grep | wc -l", process_name); fp = popen(command, "r"); if(fp == NULL){ return 0; } if((fgets(buf, sizeof(buf), fp)) != NULL){ count = atoi(buf); } pclose(fp); if(count > 1){ return 1; }else{ return 0; } return 0; } char *get_exe_name(char *argv) { char * p = NULL; p = rindex(argv, '/'); if(p == NULL){ p = argv; }else{ p = p + 1; } return p; } struct log_file_list *get_log_file_node(char *log_file) { struct log_file_list *tmp; struct log_file_list *n; list_for_each_entry_safe(tmp, n, &g_log_file_list.list, list){ if(strcmp(tmp->log_file_pre, log_file) == 0){ return tmp; } } return NULL; } struct log_file_list * create_log_file_node(char *log_file) { struct log_file_list *node; struct tm date; time_t curtime = 0; struct stat buf; int n = 0; node = (struct log_file_list *)malloc(sizeof(struct log_file_list)); if(node == NULL){ return NULL; } memset(node, 0 ,sizeof(struct log_file_list)); memcpy(node->log_file_pre, log_file, strlen(log_file)); curtime = time(NULL); localtime_r(&curtime, &date); n = snprintf(node->real_log_file, sizeof(node->real_log_file), "%s.%d-%d-%d", node->log_file_pre, date.tm_year + 1900, date.tm_mon, date.tm_mday); if(n >= sizeof(node->real_log_file)){ goto error; } node->fd = open(node->real_log_file, O_RDWR | O_CREAT | O_APPEND, 0666); if(node->fd < 0){ goto error; } if(stat(node->real_log_file, &buf) < 0){ goto error; } node->dev = buf.st_dev; node->ino = buf.st_ino; node->create_date = date; list_add(&node->list, &g_log_file_list.list); return node; error: if(node != NULL){ free(node); } return NULL; } void get_cur_date(struct tm *date) { time_t curtime = 0; curtime = time(NULL); localtime_r(&curtime, date); return ; } void get_cur_strftime(char *buf, int maxlen) { struct tm local; time_t curtime = time(NULL); localtime_r(&curtime, &local); strftime(buf, maxlen, "%c", &local); return ; } int reopen_log_file(struct log_file_list *node) { struct stat buf; close(node->fd); node->fd = open(node->real_log_file, O_RDWR | O_CREAT | O_APPEND, 0666); if(node->fd < 0){ return CONSUMER_ERROR; } if(stat(node->real_log_file, &buf) < 0){ return CONSUMER_ERROR; } node->dev = buf.st_dev; node->ino = buf.st_ino; return CONSUMER_SUCCESS; } int check_reopen_log_file(struct log_file_list *node) { struct stat buf; struct tm date; int n = 0; get_cur_date(&date); if(date.tm_year != node->create_date.tm_year || date.tm_mon != node->create_date.tm_mon || date.tm_mday != node->create_date.tm_mday){ node->create_date = date; n = snprintf(node->real_log_file, sizeof(node->real_log_file), "%s.%d-%d-%d", node->log_file_pre, node->create_date.tm_year, node->create_date.tm_mon, node->create_date.tm_mday); if(n >= sizeof(node->real_log_file)){ return CONSUMER_ERROR; } return reopen_log_file(node); } if(stat(node->real_log_file, &buf) < 0){ return reopen_log_file(node); /* we'll have to restat the newly created file to get the inode info*/ } if(buf.st_dev != node->dev || buf.st_ino != node->ino){ return reopen_log_file(node); } return CONSUMER_SUCCESS; } void consumer_ring_queue_to_file(struct MESA_shm_queue_head *head) { char *p_file = NULL; int *p_payload_len = NULL; char *payload = NULL; int payload_len = 0; char buf[MESA_SHM_LOG_BUF_PREFIX_LEN + MESA_SHM_RING_QUEUE_BLOCK_BUFLEN] = {0}; int n = 0; char strtime[DEFAUT_BUF_SIZE] = {0}; struct log_file_list *node = NULL; get_cur_strftime(strtime, sizeof(strtime)); while(!MESA_shm_ring_queue_is_empty(head)){ p_file = (char *)(head + 1) + (head->blksize * head->rd_idx); node = get_log_file_node(p_file); if(node == NULL){ node = create_log_file_node(p_file); if(node == NULL){ head->rd_idx = (head->rd_idx + 1) % head->blknum; continue ; } }else{ if(check_reopen_log_file(node) != CONSUMER_SUCCESS){ head->rd_idx = (head->rd_idx + 1) % head->blknum; continue ; } } p_payload_len = (int *)(p_file + MESA_SHM_LOG_PATH_LEN); payload_len = *p_payload_len; payload = (char *)(p_payload_len + 1); n = snprintf(buf, sizeof(buf), "%s, %s\n", strtime, payload); write(node->fd, buf, n); head->rd_idx = (head->rd_idx + 1) % head->blknum; } } int main(int argc, char **argv) { char *exe_name = get_exe_name(argv[0]); if(consumer_is_running(exe_name)){ printf("consumer process is already running\n"); return 0; } consumer_daemo(); int i = 0; struct MESA_shm_overview *shm_overview = NULL; struct MESA_shm_overview *tmp_ovw = NULL; struct MESA_shm_queue_head *ring_queue_head[MESA_SHM_RING_QUEUE_NUM] = {NULL}; int log_file_fd = -1; struct log_file_list *log_file_node = NULL; INIT_LIST_HEAD(&g_log_file_list.list); shm_overview = MESA_shm_alloc_overview(); if(shm_overview == NULL){ return 0; } init_ring_queue_head_arrray(shm_overview, ring_queue_head); while(1){ for(i = 0; i< MESA_SHM_RING_QUEUE_NUM; i++){ tmp_ovw = shm_overview + i; if(tmp_ovw->shmid == -1){ break; } if(ring_queue_head[i] == NULL){ ring_queue_head[i] = shmat(tmp_ovw->shmid, NULL, 0); if(ring_queue_head[i] == NULL){ break ; } } if(!MESA_shm_ring_queue_is_empty(ring_queue_head[i])){ consumer_ring_queue_to_file(ring_queue_head[i]); } } usleep(5000); } return 0; }