at work

cworker.c

@@ -4,15 +4,26 @@
  *
  */
 
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/wait.h>
+#include <sys/time.h>
+#include <sys/event.h>
+#include <sys/socket.h>
+#include <arpa/inet.h>
+#include <netinet/in.h>
+
+#include <limits.h>
 #include <unistd.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <fcntl.h>
 #include <string.h>
 #include <errno.h>
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <sys/wait.h>
+#include <pthread.h>
+#include <poll.h>
+#include <signal.h>
 
 #include <cllexer.h>
 #include <clparser.h>
@@ -20,16 +31,312 @@
 #include <cfparser.h>
 #include <massert.h>
 #include <rcache.h>
+#include <jlexer.h>
+#include <jparser.h>
+
 
 #include <config.h>
 #include <cworker.h>
 #include <logger.h>
 
-int log_fd = 0;
+static const int64_t default_port = 9701;
+static cworker_t* pworker = NULL;
+
+static int mkdirall(const char* path, mode_t mode);
+void sighandler(int signum);
+
+static int cworker_readconf(const cworker_t* worker) {
+    log_debug("Configuration reading");
+
+    int conffd = -1;
+    if ((conffd = open(srv_configpath, O_RDONLY)) < 0) {
+        log_error("Configuration file opening error: %s %s", strerror(errno), srv_configpath);
+        return -1;
+    }
+
+    rcache_t cache;
+    cflexer_t lexer;
+    cfparser_t parser;
+
+    rcache_init(&cache, conffd);
+    cflexer_init(&lexer, &cache);
+    cfparser_init(&parser, &lexer);
+
+    if (cfparser_parse(&parser) < 0) {
+        log_error("Configuration parsing error");
+        return -1;
+    }
+
+    cfparser_bind(&parser, CFVALTYPE_INT, "port", (void *)&(worker->port));
+    cfparser_bind(&parser, CFVALTYPE_BOOL, "nofork", (void *)&(worker->nofork));
+
+    cfparser_destroy(&parser);
+    cflexer_destroy(&lexer);
+    rcache_destroy(&cache);
+
+    return 0;
+}
+
+static int cworker_readopts(const cworker_t* worker, char** argv, int argc) {
+    log_debug("Reading options");
+
+    cllexer_t lexer;
+    clparser_t parser;
+
+    cllexer_init(&lexer);
+    clparser_init(&parser, &lexer);
+
+    clparser_bind(&parser, CLVALTYPE_INT, "port", (void *)&(worker->port));
+    clparser_bind(&parser, CLVALTYPE_BOOL, "nofork", (void *)&(worker->nofork));
+
+    if (clparser_parse(&parser, &argv[1], argc - 1) < 0) {
+        log_error("Args parsing error");
+        return -1;
+    }
+
+    return 0;
+}
+
+static int cworker_openlog(cworker_t* worker) {
+    if (!worker->nofork) {
+        log_debug("Redirect output");
+        if (mkdirall(srv_logpath, S_IRWXU|S_IRGRP|S_IXGRP) < 0) {
+            log_error("Log dir creating error: %s %s", strerror(errno), srv_logpath);
+        }
+
+        if ((worker->logfd = open(srv_logpath, O_WRONLY|O_APPEND|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP)) < 0) {
+            log_error("Log file opening error: %s %s", strerror(errno), srv_logpath);
+            return -1;
+        }
+        dup2(worker->logfd, STDOUT_FILENO);
+        dup2(worker->logfd, STDERR_FILENO);
+    }
+    return 0;
+}
+
+int cworker_writepid(const cworker_t* worker) {
+    if (!worker->nofork) {
+
+        log_debug("Write pid file");
+
+        if (mkdirall(srv_runpath, S_IRWXU|S_IRGRP|S_IXGRP) < 0) {
+            log_error("Run directory creating error: %s %s", strerror(errno), srv_runpath);
+        }
+
+        int pid_fd = -1;
+        if ((pid_fd = open(srv_runpath, O_WRONLY|O_TRUNC|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP)) < 0) {
+            log_error("Process ID file opening error: %s %s", strerror(errno), srv_runpath);
+            return -1;
+        }
+        if (dprintf(pid_fd, "%d", getpid()) < 0) {
+            log_error("Process ID file writing error: %s %s", strerror(errno), srv_runpath);
+            close(pid_fd);
+            return -1;
+        }
+        close(pid_fd);
+    }
+    return 0;
+}
+
+
+int cworker_init(cworker_t* worker, char** argv, int argc) {
+    log_init();
+
+    log_debug("Init service");
+    worker->port = default_port;
+    worker->nofork = true;
+    worker->socket = 0;
+    worker->logfd = 0;
+    pworker = worker;
+
+    if (cworker_readconf(worker) < 0) {
+        log_error("Config reading  error");
+        return -1;
+    }
+    if (cworker_readopts(worker, argv, argc) < 0) {
+        log_error("Options reading error");
+        return -1;
+    }
+    if (!worker->nofork) {
+        if (cworker_openlog(worker) < 0) {
+            log_error("Log opening error");
+            return -1;
+        }
+    }
+    log_debug("Listening port: %d", worker->port);
+
+    return 0;
+}
+
+int cworker_detach(const cworker_t* worker) {
+    if (!worker->nofork) {
+        log_debug("Service detaching");
+        int childpid = -1;
+        if((childpid = fork()) < 0) {
+            log_error("Fork error %d: %s ", errno, strerror(errno));
+            return -1;
+        }
+        if (childpid == 0) {
+            log_debug("Service forked");
+            return 0;
+            // child
+        } else {
+            // parent
+            exit(0);
+        }
+    }
+    return 0;
+}
+
+int cworker_build(cworker_t* worker) {
+    log_debug("Service building");
+
+    signal(SIGCHLD, SIG_IGN);
+    signal(SIGHUP, sighandler);
+    signal(SIGTERM, sighandler);
+
+    if((worker->socket = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
+        log_debug("Socket creating error %d: %s", errno, strerror(errno));
+        return -1;
+    }
+    int optval = 1;
+    if (setsockopt(worker->socket, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)) < 0) {
+        log_debug("Socket option setting error %d: %s", errno, strerror(errno));
+        return -1;
+    }
+
+    struct sockaddr addr;
+    struct sockaddr_in* paddr = (struct sockaddr_in*)&addr;
+    paddr->sin_family = AF_INET;
+    paddr->sin_addr.s_addr = INADDR_ANY;
+    paddr->sin_port = htons(worker->port);
+    paddr->sin_len = sizeof(struct sockaddr_in);
+
+    if (bind(worker->socket, (struct sockaddr*)paddr, paddr->sin_len) < 0) {
+        log_debug("Socket binding error %d: %s ", errno, strerror(errno));
+        return -1;
+    }
+    int backlog = 100;
+    if (listen(worker->socket, backlog) < 0) {
+        log_debug("Socket listening error %d: %s ", errno, strerror(errno));
+        return -1;
+    }
+    return 0;
+}
+
+int cworker_handler(const cworker_t* worker, int socket);
+
+int cworker_run(const cworker_t* worker) {
+    log_debug("Service running");
+
+    while (1) {
+        int newsocket = 0;
+        if ((newsocket = accept(worker->socket, NULL, 0)) > 3) {
+
+            int childpid = -1;
+            if((childpid = fork()) < 0) {
+                log_error("Fork error %d: %s ", errno, strerror(errno));
+                continue;
+            }
+            if (childpid == 0) {
+                // child
+                log_debug("Service %d forked", getpid());
+                signal(SIGHUP, SIG_IGN);
+                signal(SIGTERM, SIG_IGN);
+
+                struct timeval tv = { .tv_sec = 5, .tv_usec = 0 };
+                if (setsockopt(newsocket, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(struct timeval)) < 0) {
+                    log_debug("Socket option setting error %d: %s ", errno, strerror(errno));
+                    return -1;
+                }
+                cworker_handler(worker, newsocket);
+                log_debug("Handler %d done", getpid());
+                close(newsocket);
+                return 0;
+            } else {
+                // parent
+                close(newsocket);
+            }
+        }
+    }
+    close(worker->socket);
+
+    return 0;
+}
+
+
+int cworker_handler(const cworker_t* worker, int socket) {
+    int err = 0;
+
+    rcache_t cache;
+    jlexer_t lexer;
+    jparser_t parser;
+
+    rcache_init(&cache, socket);
+    jlexer_init(&lexer, &cache);
+    jparser_init(&parser, &lexer);
+
+    if (jparser_parse(&parser) < 0) {
+        log_error("Cannot parse json");
+        err = -1;
+        goto exit;
+    }
+
+    int64_t id = 0;
+    char* name = "none";
+
+    if (jparser_bind(&parser, JVALTYPE_NUM, "id", (void *)&id) < 0) {
+        log_error("Cannot bind id");
+    }
+    if (jparser_bind(&parser, JVALTYPE_STR, "name", (void *)&name) < 0) {
+        log_error("Cannot bind name");
+    }
+    dprintf(socket, "{}");
+
+    free(name);
+    exit:
+        jparser_destroy(&parser);
+        jlexer_destroy(&lexer);
+        rcache_destroy(&cache);
+        return err;
+}
+
+void cworker_shutdown(cworker_t* worker) {
+    log_warning("Shutdown service %d", getpid());
+    if (worker->socket != 0) {
+        close(worker->socket);
+        worker->logfd = 0;
+    }
+    if (worker->logfd != 0) {
+        close(worker->logfd);
+        worker->logfd = 0;
+    }
+    wait(NULL);
+    log_destroy();
+}
+
+
+void sighandler(int signum) {
+    log_warning("Handle signal %d", signum);
+    if (signum == SIGHUP) {
+       log_warning("Handle HUP signal");
+    } else if (signum == SIGTERM) {
+        log_warning("Handle TERM signal");
+        if (pworker != NULL) {
+            cworker_shutdown(pworker);
+            sleep(1);
+            exit(0);
+        }
+    }
+}
+
 
 static int mkdirall(const char* path, mode_t mode) {
-    char buffer[1024];
+    char buffer[PATH_MAX];
     size_t psize = strlen(path);
+    if (psize > PATH_MAX) {
+        return -1;
+    }
     if (psize == 0) return 0;
     if (psize == 1 && path[0] == '/') {
         return 0;
@@ -47,187 +354,3 @@ static int mkdirall(const char* path, mode_t mode) {
     }
     return 0;
 }
-
-char* path_last(const char* path) {
-    char buffer[1024];
-    size_t psize = strlen(path);
-    strcpy(buffer, path);
-    buffer[psize] = '\0';
-    if (buffer[psize - 1] == '/') {
-        buffer[--psize] = '\0';
-    }
-    size_t pos = psize;
-    for (size_t i = 1; i < psize; i++) {
-        if (buffer[i] == '/') pos = i;
-    }
-    char* b = &buffer[++pos];
-    int bsize = strlen(b);
-    char* res = malloc(bsize + 1);
-    strcpy(res, b);
-    //printf("\n%s\n", res);
-    return res;
-}
-
-char* path_file(const char* path) {
-    char buffer[1024];
-    size_t psize = strlen(path);
-    strcpy(buffer, path);
-    buffer[psize] = '\0';
-    //if (buffer[psize - 1] == '/') {
-    //    buffer[--psize] = '\0';
-    //}
-    size_t pos = psize;
-    for (size_t i = 1; i < psize; i++) {
-        if (buffer[i] == '/') pos = i;
-    }
-    char* b = &buffer[++pos];
-    int bsize = strlen(b);
-    char* res = malloc(bsize + 1);
-    strcpy(res, b);
-    //printf("\n%s\n", res);
-    return res;
-}
-
-
-static int cworker_readconf(const cworker_t* worker) {
-    log_debug("reading configiration");
-
-    int conf_fd = -1;
-    if ((conf_fd = open(srv_configpath, O_RDONLY)) < 0) {
-        log_error("cannot open config file %s", srv_configpath);
-        return -1;
-    }
-
-    rcache_t cache;
-    cflexer_t lexer;
-    cfparser_t parser;
-
-    rcache_init(&cache, conf_fd);
-    cflexer_init(&lexer, &cache);
-    cfparser_init(&parser, &lexer);
-
-    if (cfparser_parse(&parser) < 0) {
-        log_error("parse config error\n");
-        return -1;
-    }
-
-    cfparser_bind(&parser, CFVALTYPE_INT, "port", (void *)&(worker->port));
-
-    cfparser_destroy(&parser);
-    cflexer_destroy(&lexer);
-    rcache_destroy(&cache);
-
-    return 0;
-}
-
-static int cworker_readopts(const cworker_t* worker, char** argv, int argc) {
-    log_debug("reading options");
-
-    cllexer_t lexer;
-    clparser_t parser;
-
-    cllexer_init(&lexer);
-    clparser_init(&parser, &lexer);
-
-    clparser_bind(&parser, CLVALTYPE_INT, "port", (void *)&(worker->port));
-
-    if (clparser_parse(&parser, &argv[1], argc - 1) < 0) {
-        log_error("parse args error");
-        return -1;
-    }
-
-    return 0;
-}
-
-static int cworker_openlog(const cworker_t* worker) {
-    log_debug("redirect output");
-
-    if (mkdirall(srv_logpath, S_IRWXU|S_IRGRP|S_IXGRP) < 0) {
-        log_error("creating log dir error: %s %s", strerror(errno), srv_logpath);
-    }
-
-    if ((log_fd = open(srv_logpath, O_WRONLY|O_APPEND|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP)) < 0) {
-        log_error("open log file error %s %s", strerror(errno), srv_logpath);
-        return -1;
-    }
-    dup2(log_fd, STDOUT_FILENO);
-    dup2(log_fd, STDERR_FILENO);
-
-    return 0;
-}
-
-int cworker_writepid(const cworker_t* worker) {
-    log_debug("write pid file");
-
-    if (mkdirall(srv_runpath, S_IRWXU|S_IRGRP|S_IXGRP) < 0) {
-        log_error("createing run dir error: %s %s", strerror(errno), srv_runpath);
-    }
-
-    int pid_fd = -1;
-    if ((pid_fd = open(srv_runpath, O_WRONLY|O_TRUNC|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP)) < 0) {
-        log_error("open pid file error %s %s", strerror(errno), srv_runpath);
-        return -1;
-    }
-    if (dprintf(pid_fd, "%d", getpid()) < 0) {
-        log_error("cannon write pid, error %s %s", strerror(errno), srv_runpath);
-        close(pid_fd);
-        return -1;
-    }
-    close(pid_fd);
-    return 0;
-}
-
-const int64_t default_port = 9701;
-
-int cworker_init(cworker_t* worker, char** argv, int argc) {
-    if (cworker_openlog(worker) < 0) {
-        log_error("openlog error");
-        return -1;
-    }
-    log_debug("init service");
-    worker->port = default_port;
-
-    if (cworker_readconf(worker) < 0) {
-        log_error("reading config error");
-        return -1;
-    }
-
-    if (cworker_readopts(worker, argv, argc) < 0) {
-        log_error("reading config error");
-        return -1;
-    }
-
-    log_debug("port: %d", worker->port);
-
-    return 0;
-}
-
-int cworker_detach(const cworker_t* worker) {
-    log_debug("detach service");
-
-    int childpid = -1;
-    if((childpid = fork()) < 0) {
-        log_error("fork error: %s", strerror(errno));
-        return -1;
-    }
-
-    if (childpid == 0) {
-        log_debug("service forkerd");
-        return 0;
-        // child
-    } else {
-        // parent
-        exit(0);
-    }
-    return 0;
-}
-
-int cworker_build(const cworker_t* worker) {
-    log_debug("build service");
-    return 0;
-}
-
-int cworker_run(const cworker_t* worker) {
-    log_debug("run service");
-    return 0;
-}