#include #include #include #include #include #include #include #include #include #include #define K22_STR_EXP(__A) #__A #define K22_STR(__A) K22_STR_EXP(__A) #define K22_BEEHUB_VERSION_MAJOR 1 #define K22_BEEHUB_VERSION_MINOR 215 #define K22_BEEHUB_VERSION_REVISION 0 #define K22_BEEHUB_VERSION K22_STR(K22_BEEHUB_VERSION_MAJOR) "." K22_STR(K22_BEEHUB_VERSION_MINOR) "." K22_STR(K22_BEEHUB_VERSION_REVISION) "" /* * @build: make * @usage: datahubs -h * datahubs -v * * datahubs -c datahubs.conf -d * ps aux | grep datahubs * * datahubs -s stop * ps aux | grep datahubs * */ typedef struct conf_ctx_s { IniParser* parser; int loglevel; int worker_processes; int worker_threads; std::string host; int port; } conf_ctx_t; conf_ctx_t g_conf_ctx; inline void conf_ctx_init(conf_ctx_t* ctx) { ctx->parser = new IniParser; ctx->loglevel = LOG_LEVEL_DEBUG; ctx->worker_processes = 0; ctx->worker_threads = 0; ctx->port = 0; } static void print_version(); static void print_help(); static int parse_confile(const char* confile); static void worker_fn(void* userdata); // short options static const char options[] = "hvc:ts:dp:"; // long options static const option_t long_options[] = { {'h', "help", NO_ARGUMENT}, {'v', "version", NO_ARGUMENT}, {'c', "confile", REQUIRED_ARGUMENT}, {'t', "test", NO_ARGUMENT}, {'s', "signal", REQUIRED_ARGUMENT}, {'d', "daemon", NO_ARGUMENT}, {'p', "port", REQUIRED_ARGUMENT} }; static const char detail_options[] = R"( -h|--help Print this information -v|--version Print version -c|--confile Set configure file, default etc/{program}.conf -t|--test Test configure file and exit -s|--signal Send to process, =[start,stop,restart,status,reload] -d|--daemon Daemonize -p|--port Set listen port )"; void print_version() { printf("%s version %s\n", g_main_ctx.program_name, K22_BEEHUB_VERSION); } void print_help() { printf("Usage: %s [%s]\n", g_main_ctx.program_name, options); printf("Options:\n%s\n", detail_options); } int parse_confile(const char* confile) { int ret = g_conf_ctx.parser->LoadFromFile(confile); if (ret != 0) { printf("Load confile [%s] failed: %d\n", confile, ret); exit(-40); } // logfile std::string str = g_conf_ctx.parser->GetValue("logfile"); if (!str.empty()) { strncpy(g_main_ctx.logfile, str.c_str(), sizeof(g_main_ctx.logfile)); } hlog_set_file(g_main_ctx.logfile); // loglevel str = g_conf_ctx.parser->GetValue("loglevel"); if (!str.empty()) { hlog_set_level_by_str(str.c_str()); } // log_filesize str = g_conf_ctx.parser->GetValue("log_filesize"); if (!str.empty()) { hlog_set_max_filesize_by_str(str.c_str()); } // log_remain_days str = g_conf_ctx.parser->GetValue("log_remain_days"); if (!str.empty()) { hlog_set_remain_days(atoi(str.c_str())); } // log_fsync str = g_conf_ctx.parser->GetValue("log_fsync"); if (!str.empty()) { logger_enable_fsync(hlog, hv_getboolean(str.c_str())); } // first log here // first log here hlogi("=========--- Welcome to the Earth ---========="); hlogi("%s version: %s", g_main_ctx.program_name, K22_BEEHUB_VERSION); hlog_fsync(); // worker_processes int worker_processes = 0; #ifdef DEBUG // Disable multi-processes mode for debugging worker_processes = 0; #else str = g_conf_ctx.parser->GetValue("worker_processes"); if (str.size() != 0) { if (strcmp(str.c_str(), "auto") == 0) { worker_processes = get_ncpu(); hlogd("worker_processes=ncpu=%d", worker_processes); } else { worker_processes = atoi(str.c_str()); } } #endif g_conf_ctx.worker_processes = LIMIT(0, worker_processes, MAXNUM_WORKER_PROCESSES); // worker_threads int worker_threads = 0; str = g_conf_ctx.parser->GetValue("worker_threads"); if (str.size() != 0) { if (strcmp(str.c_str(), "auto") == 0) { worker_threads = get_ncpu(); hlogd("worker_threads=ncpu=%d", worker_threads); } else { worker_threads = atoi(str.c_str()); } } g_conf_ctx.worker_threads = LIMIT(0, worker_threads, 64); //host str = g_conf_ctx.parser->GetValue("host"); if (str.size() != 0) { g_conf_ctx.host = str; } else { g_conf_ctx.host = "0.0.0.0"; } // port int port = 0; const char* szPort = get_arg("p"); if (szPort) { port = atoi(szPort); } if (port == 0) { port = g_conf_ctx.parser->Get("port"); } if (port == 0) { printf("Please config listen port!\n"); exit(-10); } g_conf_ctx.port = port; hlogi("parse_confile('%s') OK", confile); return 0; } static void on_reload(void* userdata) { hlogi("reload confile [%s]", g_main_ctx.confile); parse_confile(g_main_ctx.confile); } //////1/////////////////////////////// #if 0 #define LOCKFILE "/var/lock/beehubs2.lock" int lockfile(int fd) { struct flock fl; fl.l_type = F_WRLCK; fl.l_start = 0; fl.l_whence = SEEK_SET; fl.l_len = 0; return fcntl(fd, F_SETLK, &fl); } int already_running(void) { int fd; char buf[16]; fd = open(LOCKFILE, O_RDWR | O_CREAT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); if (fd < 0) { fprintf(stderr, "FAILED TO OPEN FILE" LOCKFILE); exit(1); } if (lockfile(fd) < 0) { if (EACCES == errno || EAGAIN == errno) { close(fd); return 1; } fprintf(stderr, "FAILED TO LOCK FILE" LOCKFILE); exit(1); } ftruncate(fd, 0); sprintf(buf, "%ld", (long)getpid()); write(fd, buf, strlen(buf) + 1); return 0; } #endif /// end of checking only one instance /// //////////////////////////////////////////////// int main(int argc, char** argv) { #if 0 if (already_running()) { printf("PROGRAM ALREADY RUNNING...\n"); exit(0); } #endif // g_main_ctx main_ctx_init(argc, argv); if (argc == 1) { print_help(); exit(10); } // int ret = parse_opt(argc, argv, options); int ret = parse_opt_long(argc, argv, long_options, ARRAY_SIZE(long_options)); if (ret != 0) { print_help(); exit(ret); } /* printf("---------------arg------------------------------\n"); printf("%s\n", g_main_ctx.cmdline); for (int i = 0; i < g_main_ctx.arg_kv_size; ++i) { printf("%s\n", g_main_ctx.arg_kv[i]); } for (int i = 0; i < g_main_ctx.arg_list_size; ++i) { printf("%s\n", g_main_ctx.arg_list[i]); } printf("================================================\n"); printf("---------------env------------------------------\n"); for (int i = 0; i < g_main_ctx.envc; ++i) { printf("%s\n", g_main_ctx.save_envp[i]); } printf("================================================\n"); */ // help if (get_arg("h")) { print_help(); exit(0); } // version if (get_arg("v")) { print_version(); exit(0); } // g_conf_ctx conf_ctx_init(&g_conf_ctx); const char* confile = get_arg("c"); if (confile) { strncpy(g_main_ctx.confile, confile, sizeof(g_main_ctx.confile)); } parse_confile(g_main_ctx.confile); // test if (get_arg("t")) { printf("Test confile [%s] OK!\n", g_main_ctx.confile); exit(0); } // signal signal_init(on_reload); const char* signal = get_arg("s"); if (signal) { signal_handle(signal); } #ifdef OS_UNIX // daemon if (get_arg("d")) { // nochdir, noclose int ret = daemon(1, 1); if (ret != 0) { printf("daemon error: %d\n", ret); exit(-10); } } #endif // pidfile create_pidfile(); master_workers_run(worker_fn, (void*)(intptr_t)&g_conf_ctx, g_conf_ctx.worker_processes, g_conf_ctx.worker_threads); hlogi("=========--- I'll be back! ---========="); return 0; } static void on_close(hio_t* io) { hlogi("on_close fd=%d error=%d", hio_fd(io), hio_error(io)); } std::string get_current_timestamp() { auto now = std::chrono::system_clock::now(); //通过不同精度获取相差的毫秒数 uint64_t dis_millseconds = std::chrono::duration_cast(now.time_since_epoch()).count() - std::chrono::duration_cast(now.time_since_epoch()).count() * 1000; time_t tt = std::chrono::system_clock::to_time_t(now); auto time_tm = localtime(&tt); char strTime[25] = { 0 }; sprintf(strTime, "%d-%02d-%02d %02d:%02d:%02d.%03d", time_tm->tm_year + 1900, time_tm->tm_mon + 1, time_tm->tm_mday, time_tm->tm_hour, time_tm->tm_min, time_tm->tm_sec, (int)dis_millseconds); return std::string(strTime); } //接收到数据 static void on_recv(hio_t* io, void* buf, int readbytes) { printf("< %.*s\n", readbytes, (char*)buf); } void on_timer(htimer_t* timer) { char str[DATETIME_FMT_BUFLEN] = {0}; datetime_t dt = datetime_now(); datetime_fmt(&dt, str); printf("> %s\n", str); // 获取userdata hio_t* io = (hio_t*)hevent_userdata(timer); // 发送当前时间字符串 hio_write(io, str, strlen(str)); } void on_close(hio_t* io) { } void on_connect(hio_t* io) { // 设置read回调 hio_setcb_read(io, on_recv); // 开始读 hio_read(io); // 添加一个定时器 htimer_t* timer = htimer_add(hevent_loop(io), on_timer, 1000, INFINITE); // 设置userdata hevent_set_userdata(timer, io); } void worker_fn(void* userdata) { conf_ctx_t* ptrCtx = (conf_ctx_t*)(intptr_t)(userdata); //long port = (long)(intptr_t)(userdata); long port = ptrCtx->port; // 创建事件循环 hloop_t* loop = hloop_new(0); // 创建TCP客户端 hio_t* io = hloop_create_tcp_client(loop, ptrCtx0->host, port, on_connect, on_close); if (io == NULL) { return ; } // 运行事件循环 hloop_run(loop); // 释放事件循环 hloop_free(&loop); hlogi("port=%ld pid=%ld tid=%ld listenfd=%d", port, hv_getpid(), hv_gettid(), hio_fd(listenio)); hloop_run(loop); hloop_free(&loop); }