microser/mms/main.c

/**
* @file:    $RCSfile: main.c,v $
* @brief:   $IEC 61850 Protocol  
*
* @version: $Revision: 1.9 $
* @date:    $Date: 2020/10/28 05:21:18 $
* @author:  $Author: lizhongming $
* @state:   $State: Exp $
*
* @latest:  $Id: main.c,v 1.9 2020/10/28 05:21:18 lizhongming Exp $
*
*/

#include "rdb_client.h"
#include "db_interface.h"
#include "node.h"

/*lnk10-10 */
#include "../rocketmq/SimpleProducer.h"

#include "../log4cplus/log4.h"//lnk添加log4

#include "../cfg_parse/custom_printf.h"//lnk20250225

extern G_TEST_FLAG;
extern pthread_mutex_t mtx;

extern pt61850app_t	*g_pt61850app;
extern node_t	*g_node;
char 		g_my_conf_fname[256];

char			g_onlyIP[255];			//直连某个IP，仅仅为方便测试

apr_pool_t         *g_root_pool;
apr_pool_t         *g_rdb_pool;
apr_pool_t         *g_cfg_pool;
uint16_t			g_client_id = 0;
uint32_t			g_node_id = 0;
uint32_t			g_min_free_size = 1024;
int					g_need_password = 0;

int INITFLAG = 0; //lnk20250304

char	subdir[128] = "cfg_stat_data" ;

int usage();
int parse_param(int argc, const char **argv);

#include "ver_conf.h" 

#ifdef DEBUG_SISCO
SD_CONST static ST_CHAR *SD_CONST thisFileName = __FILE__;
#endif

///////////////////////////////////////////////////////
uint32_t	g_dead_lock_counter = 0;
uint32_t	g_thread_blocked_times = 0;

//给mq生产线程和定时线程都加上死锁计数器
uint32_t	g_mqproducer_blocked_times = 0;
uint32_t	g_ontime_blocked_times = 0;

/////////////////////////////////////////////////////////////////////////////// 
int		three_secs_enabled = 0;
int		auto_register_report_enabled = 0;
int     g_front_seg_index = 0;
int     g_front_seg_num = 0;
int     FRONT_MP_NUM = 0;//终端连接数量
int     g_front_num_count = 0;//终端连接数量过低次数


///////////////////////////////////////////////////////////////////////////////
//根据配置子目录判别全局功能的投退，设定四个分功能性能均衡程序的功能运行划分设定
void init_global_function_enable()
{
	if (strcmp(subdir,"cfg_stat_data")==0) {			//历史稳态
		g_node_id = STAT_DATA_BASE_NODE_ID;
		auto_register_report_enabled = 1;
	}else if (strcmp(subdir,"cfg_3s_data")==0) {		//实时
		g_node_id = THREE_SECS_DATA_BASE_NODE_ID;
		three_secs_enabled = 1;
	}else if (strcmp(subdir,"cfg_soe_comtrade")==0) {	//告警和录波和暂态
		g_node_id = SOE_COMTRADE_BASE_NODE_ID;
	}else if (strcmp(subdir,"cfg_his_data")==0)  {		//不使用
		g_node_id = RECALL_ALL_DATA_BASE_NODE_ID;
		g_node_id = HIS_DATA_BASE_NODE_ID;	
	}
	else if (strcmp(subdir, "cfg_newhis_data") == 0) {	//不使用
		g_node_id = RECALL_ALL_DATA_BASE_NODE_ID;
		g_node_id = NEW_HIS_DATA_BASE_NODE_ID;
	}
	else if (strcmp(subdir, "cfg_recallhis_data") == 0) {	//补招
		g_node_id = RECALL_HIS_DATA_BASE_NODE_ID;
	}
	else if (strcmp(subdir, "cfg_recallall_data") == 0) {	//不使用
		g_node_id = RECALL_ALL_DATA_BASE_NODE_ID;
	}
}
//////////////////////////////////////////////////////////////////////////////////

#ifdef  _OS_UNIX_
void init_daemon(void) 
{ 
    int pid; 
    int i; 

    if( pid = fork() ) 
        exit(0);           /** 是父进程，结束父进程  */
    else if( pid < 0 ) 
        exit(1);           /** fork失败，退出        */

    /** 是第一子进程，后台继续执行  */

    setsid();              /** 第一子进程成为新的会话组长和进程组长并与控制终端分离  */

    if( pid = fork() ) 
        exit(0);           /** 是第一子进程，结束第一子进程    */
    else if( pid < 0) 
        exit(1);           /** fork失败，退出                  */

	chdir("/FeProject/bin/"); //multi process running at same time
    umask(0);              /** 重设文件创建掩码                */

    return; 
} 
#endif

//"--subdir, set the subdir of /CloudForward/ as the working directory, \n"
int usage()
{
	fprintf(stderr,"\nUsage : pt61850netd_pqfe -d [subdir] \n");

    exit(-1);
}


int prepare_entironment_2()
{
    apr_status_t rv;

    /* apr library need call this first. */
    apr_initialize();		//APACHE初始化
    atexit(apr_terminate);

    
    /* Create node root pool */
    rv = apr_pool_create(&g_root_pool, NULL);	//分配根内存
    if(rv != APR_SUCCESS) {
        fprintf(stderr,"%s","Create node root pool failed!\n");
        return (-1);
    }

#ifdef SIGPIPE
    /* Ignore writes to connections that have been closed at the other end. */
    apr_signal(SIGPIPE, SIG_IGN);       
#endif
    
    rv = create_log_handle(g_root_pool);
	if(rv != APR_SUCCESS) {
        printf("init_log failed!\n");
        return (-1);
    }
	else  {
		unsigned int error = 0;
		unsigned int warn = 0;
		unsigned int info = 0;
		log_config_t	*log_handle = get_log_handle();
		parse_log_switch_ini(&error,&warn,&info);
		log_handle->status.debug = info;//0;//1; 
		log_handle->status.warn = warn;//0;//1;
		log_handle->status.error = error;//0;//1;
	}
    echo_msg("==============================================================\n");

    rv = apr_pool_create(&g_rdb_pool, g_root_pool);		//分配RDB内存
    if(rv != APR_SUCCESS) {
        echo_errg("Create system runtime pool failed!\n");
        return (-1);
    }

    rv = apr_pool_create(&g_cfg_pool, g_root_pool);		//分配配置内存
    if(rv != APR_SUCCESS) {
        echo_errg("Create system config pool failed!\n");
        return (-1);
    }
	
    /* Initialize the register table. Call these functions first! */
	echo_msg1("%-60s","Initialize system register......"); 

	echo_msg("OK\n\n");

	initTimezoneOffset();

	return APR_SUCCESS;
}

#ifdef  _OS_UNIX_
#include <stdio.h>
#include <pwd.h>
#include <unistd.h>
void printf_cur_user()
{
	struct passwd *pwd;
	pwd = getpwuid(getuid());
	echo_warn1("Current user: %s\n", pwd->pw_name);
}
#endif

int main(int argc, const char **argv)
{

    uint32_t       stimer     = 1;
    apr_status_t   rv;
	 
    /* Prepare the system context */
	rv=prepare_entironment_2();
	if (rv!=APR_SUCCESS){
		return rv;
	}

	getVersion(argc,argv);

	rv=parse_param(argc, argv);
	if (rv!=APR_SUCCESS){
		return rv;
	}


	//lnk启动进程日志
	init_logger_process();
	
	//char buf[256];
	//format_log_msg(buf,sizeof(buf),"前置的%s%d号进程 进程级日志初始化完毕", get_front_msg_from_subdir(), g_front_seg_index);
	//log_debug("process", buf);
	DIY_WARNLOG("process","【WARN】前置的%s%d号进程 进程级日志初始化完毕", get_front_msg_from_subdir(), g_front_seg_index);

#ifdef  _OS_UNIX_ 
	#ifdef QT_NO_DEBUG 
		if (!g_need_password) 
			init_daemon();
	#endif	 
#endif
 
	init_global_function_enable();
	apr_snprintf(g_my_conf_fname,sizeof(g_my_conf_fname),"../%s/etc/pt61850netd_pqfe.xml",subdir);	

#if defined (DEBUG_SISCO)
	init_log_cfg ();
#endif

	printf("\n\n");
#ifdef  _OS_UNIX_
	printf_cur_user();
#endif

	if (g_need_password) {
		if ( process_login_verify() == 0) {
			echo_warn("User login succeeded!\n"); 
		} else {
			echo_warn("User login failed!\n"); 
			apr_sleep(apr_time_from_sec(3));
			exit(-59);
		}
	}

	rv = init_rdb();
	if (rv!=APR_SUCCESS){
		//char buf[256];
		//format_log_msg(buf,sizeof(buf),"前置的%s%d号进程 台账初始化失败", get_front_msg_from_subdir(), g_front_seg_index);
		//log_error("process", buf);
		DIY_ERRORLOG("process","【ERROR】前置的%s%d号进程 台账初始化失败", get_front_msg_from_subdir(), g_front_seg_index);
		return rv;
	}

	rv = run_protocol();
	if (rv!=APR_SUCCESS){
		//char buf[256];
		//format_log_msg(buf,sizeof(buf),"前置的%s%d号进程 线程初始化失败", get_front_msg_from_subdir(), g_front_seg_index);
		//log_error("process", buf);
		DIY_ERRORLOG("process","【ERROR】前置的%s%d号进程 线程初始化失败", get_front_msg_from_subdir(), g_front_seg_index);
		return rv;
	}	

	//初始化完成标志,防止还未准备完全就接收消息处理导致崩溃
	INITFLAG = 1;
	//调试用
	printf("INITFLAG=%d\n",INITFLAG); 

	if (1 == G_TEST_FLAG) {
		//lnk添加mq模拟测试,这个只能放在主线程
		printf("try_start_mqtest_thread \n");
		int ret = try_start_mqtest_thread(0,NULL);
		// 等待线程退出
		echo_warn1("%-60s","System shutdown now......"); 
    	apr_pool_destroy(g_root_pool);
    	echo_msg("OK\n");
    	return ret;
	}

	while(1) {
		/* sleep 1s, just like 1s timer */
		apr_sleep(apr_time_from_sec(1));
		/* 每30秒钟检查一次状态 */

		if( !(stimer++ % 60) ) {//分钟
			if (g_dead_lock_counter++ >=3) {//三分钟
				g_thread_blocked_times++;
				g_dead_lock_counter = 0;				
			}
			MVL_LOG_ACSE1 ("MYLOG: current g_thread_blocked_times = %u ", g_thread_blocked_times);

			if (FRONT_MP_NUM <= 1) {//监测点数
				g_front_num_count++;
			}
			else {
				g_front_num_count = 0;
			}
		}

		//work线程死了3*13分钟，退出进程
		if (g_thread_blocked_times>=13) {
			MVL_LOG_ACSE0 ("MYLOG: g_thread_blocked_times>=3, so exit to restart ");

			//char buf[256];
			//format_log_msg(buf,sizeof(buf),"前置的业务线程死锁，退出进程");
			//log_error("process", buf);
			DIY_ERRORLOG("process","【ERROR】前置的%s%d号进程 业务线程死锁，退出进程", get_front_msg_from_subdir(), g_front_seg_index);

			apr_sleep(apr_time_from_sec(10));
			exit(-1039);
		}

		//lnk20241211 添加测试开关
		/*pthread_mutex_lock(&mtx);
		if (!G_TEST_FLAG && g_front_num_count >= 30 && g_onlyIP[0] == 0 && g_node->n_clients>10) {//30分钟连接数量过低且不是单连且台账大于十个终端
			MVL_LOG_ACSE0("MYLOG: g_front_num_count>=20, so exit to restart ");

			apr_sleep(apr_time_from_sec(10));
			exit(-1039);
		}
		pthread_mutex_unlock(&mtx);*/
		
	}

    echo_warn1("%-60s","System shutdown now......"); 
    apr_pool_destroy(g_root_pool);
    echo_msg("OK\n");

    return 0;
}


int parse_param(int argc, const char **argv)
{
    apr_status_t rv;
    apr_getopt_t *opt;
	char *opt_arg;
    char ch;

	char *p;
	char temp[128];

    g_node_id     = 0;
	
	echo_warn2("================= compiled@ %s %s =================\n",__DATE__ , __TIME__ );

    /* Parse the command-line parameter */
    if( argc > 1 ) {
        rv = apr_getopt_init(&opt, g_root_pool, argc, argv);

		while (apr_getopt(opt, "q:Q:a:A:d:D:p:P:f:F:r:R:T:t:s:S", &ch, &opt_arg) == APR_SUCCESS) {
            switch (ch) {
              case 'q':
              case 'Q':				  
                  if (opt_arg[0] >= '0' && opt_arg[0] <= '9' ) {
                      g_min_free_size = atoi(opt_arg);
                  }
                  else
                      return (usage());
                  break;
              case 'R':
              case 'r':

                  break;
              case 's':
              case 'S':
				  echo_warn1(">>>>>>>>>>>>>>>>>>>>>parse -s %s ", opt_arg);
				  strcpy(temp, opt_arg);

				  p = strtok(temp, "_"); //1_5
				  echo_warn1("p %s ", p);
				  g_front_seg_index = atoi(p);

				  printf("g_front_seg_index:%d",g_front_seg_index);

				  p = strtok(NULL, "_"); //1_5
				  echo_warn1("p %s ", p);
				  g_front_seg_num = atoi(p);

				  printf("g_front_seg_num:%d",g_front_seg_num);

				  break;
				  break;
              case 'a':
              case 'A':
				  strcpy(g_onlyIP,opt_arg);
                  break;
              case 'd':
              case 'D':
				 strcpy(subdir,opt_arg); 
                break;  
              case 'T':
              case 't':
                  break;
              case 'f':
              case 'F':
                  break;
			  case 'P':
			  case 'p':
				  if (opt_arg[0] >= '0' && opt_arg[0] <= '9' ) {
					  g_need_password = atoi(opt_arg);
				  }
				  else
					  return (usage());
				  break;
              case '?':
              default:
                  return (usage());
            }
        }
    } 

    return 0;
}

///////////////////////////////////////////////////////////////添加测试进程的监控函数lnk20250304
void doMonitorTaskmain(void) {
    static int stimer = 0;
    stimer++;

    if( !(stimer++ % 60) ) {//分钟
		if (g_dead_lock_counter++ >=3) {//三分钟
			g_thread_blocked_times++;
			g_mqproducer_blocked_times++;
			g_ontime_blocked_times++;
			g_dead_lock_counter = 0;				
		}
		MVL_LOG_ACSE1 ("MYLOG: current g_thread_blocked_times = %u ", g_thread_blocked_times);

		if (FRONT_MP_NUM <= 1) {//监测点数
			g_front_num_count++;
		}
		else {
			g_front_num_count = 0;
		}
	}

	//work线程死了3*13分钟，退出进程
	if (g_thread_blocked_times>=13 || g_mqproducer_blocked_times>=13 || g_ontime_blocked_times>=13 ) {
		MVL_LOG_ACSE0 ("MYLOG: g_thread_blocked_times>=3, so exit to restart ");
		apr_sleep(apr_time_from_sec(10));
		exit(-1039);
	}

	//lnk20241211 添加测试开关
	pthread_mutex_lock(&mtx);
	if (!G_TEST_FLAG && g_front_num_count >= 30 && g_onlyIP[0] == 0 && g_node->n_clients>10) {//30分钟连接数量过低且不是单连且台账大于十个终端
		MVL_LOG_ACSE0("MYLOG: g_front_num_count>=20, so exit to restart ");

		apr_sleep(apr_time_from_sec(10));
		exit(-1039);
	}
	pthread_mutex_unlock(&mtx);
}