lib: monkey: scheduler timeout improvement / updates on exception han…

…dling and general reports by Coverity tool

 mk_server: server: fix unitialized variable reuse port
 mk_core: config: fix leak on split list
 mk_server: http_parser: fix method table index
 mk_server: http_parser: remove unnecessary continue() after parse_next()
 mk_core: rconf: do not over validate configuration context
 mk_server: plugin: validate plugin instance
 mk_server: plugin: close handler on exception
 mk_server: plugin: fix stage40 call assignation
 mk_server: http: validate return value of cork_flag func
 mk_server: rework connections timeout handling

Signed-off-by: Eduardo Silva <eduardo@treasure-data.com>

lib/monkey/include/monkey/mk_http.h

@@ -130,9 +130,6 @@ struct mk_http_session
     /* head for mk_http_request list nodes, each request is linked here */
     struct mk_list request_list;
 
-    /* while a http_request don't complete a request, it's linked here */
-    struct mk_list request_incomplete;
-
     /* creation time for this HTTP session */
     time_t init_time;
 
@@ -160,12 +157,12 @@ struct mk_http_session
     struct mk_http_parser parser;
 };
 
-static inline void mk_http_status_completed(struct mk_http_session *cs)
+static inline void mk_http_status_completed(struct mk_http_session *cs,
+                                            struct mk_sched_conn *conn)
 {
     mk_bug(cs->status == MK_REQUEST_STATUS_COMPLETED);
-
     cs->status = MK_REQUEST_STATUS_COMPLETED;
-    mk_list_del(&cs->request_incomplete);
+    mk_list_del(&conn->timeout_head);
 }
 
 int mk_http_error(int http_status, struct mk_http_session *cs,
@@ -197,7 +194,6 @@ int mk_http_handler_write(int socket, struct mk_http_session *cs);
 void mk_http_request_free(struct mk_http_request *sr);
 void mk_http_request_free_list(struct mk_http_session *cs);
 
-void mk_http_request_ka_next(struct mk_http_session *cs);
 void mk_http_request_init(struct mk_http_session *session,
                           struct mk_http_request *request);
 struct mk_http_header *mk_http_header_get(int name, struct mk_http_request *req,

lib/monkey/include/monkey/mk_http_parser.h

@@ -85,8 +85,8 @@ enum mk_request_methods {
     MK_METHOD_PUT        ,
     MK_METHOD_DELETE     ,
     MK_METHOD_OPTIONS    ,
-    MK_METHOD_UNKNOWN    ,
-    MK_METHOD_SIZEOF
+    MK_METHOD_SIZEOF     ,
+    MK_METHOD_UNKNOWN
 };
 
 /*

lib/monkey/include/monkey/mk_scheduler.h

@@ -24,11 +24,11 @@
 #ifndef MK_SCHEDULER_H
 #define MK_SCHEDULER_H
 
-#define MK_SCHEDULER_CONN_AVAILABLE  -1
-#define MK_SCHEDULER_CONN_PENDING     0
-#define MK_SCHEDULER_CONN_PROCESS     1
-#define MK_SCHEDULER_SIGNAL_DEADBEEF  0xDEADBEEF
-#define MK_SCHEDULER_SIGNAL_FREE_ALL  0xFFEE0000
+#define MK_SCHED_CONN_TIMEOUT    -1
+#define MK_SCHED_CONN_CLOSED     -2
+
+#define MK_SCHED_SIGNAL_DEADBEEF  0xDEADBEEF
+#define MK_SCHED_SIGNAL_FREE_ALL  0xFFEE0000
 
 /*
  * Scheduler balancing mode:
@@ -67,12 +67,12 @@ struct mk_sched_worker
     struct rb_root rb_queue;
 
     /*
-     * The incoming queue represents client connections that
+     * The timeout queue represents client connections that
      * have not initiated it requests or the request status
      * is incomplete. This linear lists allows the scheduler
      * to perform a fast check upon every timeout.
      */
-    struct mk_list incoming_queue;
+    struct mk_list timeout_queue;
 
     short int idx;
     unsigned char initialized;
@@ -104,7 +104,7 @@ struct mk_sched_conn
     struct mk_sched_handler *protocol; /* protocol handler           */
     struct mk_plugin_network *net;     /* I/O network layer          */
     struct mk_channel channel;         /* stream channel             */
-    struct mk_list status_queue;       /* link to the incoming queue */
+    struct mk_list timeout_head;       /* link to the incoming queue */
     struct rb_node _rb_head;           /* red-black tree head        */
 };
 
@@ -203,13 +203,15 @@ static inline struct mk_event_loop *mk_sched_loop()
 void mk_sched_update_thread_status(struct mk_sched_worker *sched,
                                    int active, int closed);
 
-int mk_sched_drop_connection(int socket);
+int mk_sched_drop_connection(struct mk_sched_conn *conn,
+                             struct mk_sched_worker *sched);
+
 int mk_sched_check_timeouts(struct mk_sched_worker *sched);
 struct mk_sched_conn *mk_sched_add_connection(int remote_fd,
                                               struct mk_server_listen *listener,
                                               struct mk_sched_worker *sched);
-int mk_sched_remove_client(struct mk_sched_worker *sched,
-                           struct mk_sched_conn *conn);
+int mk_sched_remove_client(struct mk_sched_conn *conn,
+                           struct mk_sched_worker *sched);
 
 struct mk_sched_conn *mk_sched_get_connection(struct mk_sched_worker
                                                      *sched, int remote_fd);
@@ -229,6 +231,18 @@ int mk_sched_event_close(struct mk_sched_conn *conn,
                          struct mk_sched_worker *sched,
                          int type);
 
+static inline void mk_sched_conn_timeout_add(struct mk_sched_conn *conn,
+                                             struct mk_sched_worker *sched)
+{
+    mk_list_add(&conn->timeout_head, &sched->timeout_queue);
+}
+
+static inline void mk_sched_conn_timeout_del(struct mk_sched_conn *conn)
+{
+    mk_list_del(&conn->timeout_head);
+}
+
+
 #define mk_sched_conn_read(conn, buf, s)        \
     conn->net->read(conn->event.fd, buf, s)
 #define mk_sched_conn_write(ch, buf, s)         \

lib/monkey/include/monkey/monkey.h

@@ -63,6 +63,8 @@
 extern const mk_ptr_t mk_monkey_protocol;
 
 struct mk_server_config *mk_server_init();
+
+void mk_server_info();
 int mk_server_setup();
 void mk_thread_keys_init();
 void mk_exit_all();

lib/monkey/mk_bin/monkey.c

@@ -309,7 +309,7 @@ int main(int argc, char **argv)
     mk_utils_register_pid(mk_config->pid_file_path);
 
     /* Print server details */
-    mk_details();
+    mk_server_info();
 
     /* Change process owner */
     mk_user_set_uidgid();

lib/monkey/mk_core/mk_rconf.c

@@ -252,8 +252,10 @@ void mk_rconf_free(struct mk_rconf *conf)
         mk_mem_free(section->name);
         mk_mem_free(section);
     }
-    if (conf->file) mk_mem_free(conf->file);
-    if (conf) mk_mem_free(conf);
+    if (conf->file) {
+        mk_mem_free(conf->file);
+    }
+    mk_mem_free(conf);
 }
 
 void mk_rconf_free_entries(struct mk_rconf_section *section)

lib/monkey/mk_server/mk_config.c

@@ -122,49 +122,6 @@ void mk_config_free_all()
     mk_mem_free(mk_config);
 }
 
-static void mk_details_listen(struct mk_list *listen)
-{
-
-    struct mk_list *head;
-    struct mk_config_listener *l;
-
-    mk_list_foreach(head, listen) {
-        l = mk_list_entry(head, struct mk_config_listener, _head);
-        printf(MK_BANNER_ENTRY "Server listening on %s:%s\n",
-               l->address, l->port);
-    }
-}
-
-void mk_details(void)
-{
-    struct mk_list *head;
-    struct mk_plugin *p;
-
-    printf(MK_BANNER_ENTRY "Process ID is %i\n", getpid());
-    mk_details_listen(&mk_config->listeners);
-    printf(MK_BANNER_ENTRY
-           "%i threads, may handle up to %i client connections\n",
-           mk_config->workers, mk_config->server_capacity);
-
-    /* List loaded plugins */
-    printf(MK_BANNER_ENTRY "Loaded Plugins: ");
-    mk_list_foreach(head, &mk_config->plugins) {
-        p = mk_list_entry(head, struct mk_plugin, _head);
-        printf("%s ", p->shortname);
-    }
-    printf("\n");
-
-#ifdef __linux__
-    char tmp[64];
-
-    if (mk_kernel_features_print(tmp, sizeof(tmp)) > 0) {
-        printf(MK_BANNER_ENTRY "Linux Features: %s\n", tmp);
-    }
-#endif
-
-    fflush(stdout);
-}
-
 /* Print a specific error */
 static void mk_config_print_error_msg(char *variable, char *path)
 {
@@ -211,7 +168,7 @@ static int mk_config_listen_read(struct mk_rconf_section *section)
     char *address = NULL;
     char *port = NULL;
     char *divider;
-    struct mk_list *list;
+    struct mk_list *list = NULL;
     struct mk_list *cur;
     struct mk_string_line *listener;
     struct mk_rconf_entry *entry;
@@ -289,8 +246,15 @@ static int mk_config_listen_read(struct mk_rconf_section *section)
     }
 
 error:
-    if (address) mk_mem_free(address);
-    if (port) mk_mem_free(port);
+    if (address) {
+        mk_mem_free(address);
+    }
+    if (port) {
+        mk_mem_free(port);
+    }
+    if (list) {
+        mk_string_split_free(list);
+    }
 
 
     if (mk_list_is_empty(&mk_config->listeners) == 0) {

lib/monkey/mk_server/mk_http.c

@@ -680,14 +680,18 @@ mk_ptr_t mk_http_index_file(char *pathfile, char *file_aux,
 #if defined (__linux__)
 static inline void mk_http_cb_file_on_consume(struct mk_stream *stream, long bytes)
 {
+    int ret;
     (void) bytes;
 
     /*
      * This callback is invoked just once as we want to turn off
      * the TCP Cork. We do this just overriding the callback for
      * the file stream.
      */
-    mk_server_cork_flag(stream->channel->fd, TCP_CORK_OFF);
+    ret = mk_server_cork_flag(stream->channel->fd, TCP_CORK_OFF);
+    if (ret == -1) {
+        mk_warn("Could not set TCP_CORK/TCP_NOPUSH off");
+    }
     stream->cb_bytes_consumed = NULL;
 }
 #endif
@@ -1043,6 +1047,19 @@ int mk_http_keepalive_check(struct mk_http_session *cs,
     return 0;
 }
 
+static inline void mk_http_request_ka_next(struct mk_http_session *cs)
+{
+    cs->body_length = 0;
+    cs->counter_connections++;
+
+    /* Update data for scheduler */
+    cs->init_time = log_current_utime;
+    cs->status = MK_REQUEST_STATUS_INCOMPLETE;
+
+    /* Initialize parser */
+    mk_http_parser_init(&cs->parser);
+}
+
 static inline int mk_http_request_end(struct mk_sched_conn *conn,
                                       struct mk_sched_worker *sched)
 {
@@ -1089,6 +1106,7 @@ static inline int mk_http_request_end(struct mk_sched_conn *conn,
     else {
         mk_http_request_free_list(cs);
         mk_http_request_ka_next(cs);
+        mk_sched_conn_timeout_add(conn, sched);
         return 0;
     }
 
@@ -1257,9 +1275,6 @@ void mk_http_session_remove(struct mk_http_session *cs)
     if (cs->body != cs->body_fixed) {
         mk_mem_free(cs->body);
     }
-    if (cs->status == MK_REQUEST_STATUS_INCOMPLETE) {
-        mk_list_del(&cs->request_incomplete);
-    }
     mk_http_request_free_list(cs);
     mk_list_del(&cs->request_list);
 
@@ -1296,7 +1311,6 @@ int mk_http_session_init(struct mk_http_session *cs, struct mk_sched_conn *conn)
     cs->close_now = MK_FALSE;
     cs->socket = conn->event.fd;
     cs->status = MK_REQUEST_STATUS_INCOMPLETE;
-    mk_list_add(&cs->request_incomplete, cs_incomplete);
 
     /* Map the channel, just for protocol-handler internal stuff */
     cs->channel = &conn->channel;
@@ -1365,18 +1379,6 @@ void mk_http_request_free_list(struct mk_http_session *cs)
     }
 }
 
-void mk_http_request_ka_next(struct mk_http_session *cs)
-{
-    cs->body_length = 0;
-    cs->counter_connections++;
-
-    /* Update data for scheduler */
-    cs->init_time = log_current_utime;
-    cs->status = MK_REQUEST_STATUS_INCOMPLETE;
-    mk_list_add(&cs->request_incomplete, cs_incomplete);
-    mk_http_parser_init(&cs->parser);
-}
-
 /*
  * Lookup a known header or a non-known header. For unknown headers
  * set the 'key' value wth a lowercase string
@@ -1415,7 +1417,6 @@ struct mk_http_header *mk_http_header_get(int name, struct mk_http_request *req,
 /*
  * Main callbacks for the Scheduler
  */
-
 int mk_http_sched_read(struct mk_sched_conn *conn,
                        struct mk_sched_worker *worker)
 {
@@ -1453,7 +1454,7 @@ int mk_http_sched_read(struct mk_sched_conn *conn,
                                 cs->body, cs->body_length);
         if (status == MK_HTTP_PARSER_OK) {
             MK_TRACE("[FD %i] HTTP_PARSER_OK", socket);
-            mk_http_status_completed(cs);
+            mk_http_status_completed(cs, conn);
             mk_http_request_prepare(cs, sr);
         }
         else if (status == MK_HTTP_PARSER_ERROR) {
@@ -1474,20 +1475,20 @@ int mk_http_sched_read(struct mk_sched_conn *conn,
 }
 
 int mk_http_sched_close(struct mk_sched_conn *conn,
-                        struct mk_sched_worker *worker,
+                        struct mk_sched_worker *sched,
                         int type)
 {
     struct mk_http_session *cs;
-    (void) worker;
+    (void) sched;
 
 #ifdef TRACE
     MK_TRACE("[FD %i] HTTP sched close (type=%i)", conn->event.fd, type);
 #else
     (void) type;
 #endif
-
     cs = mk_http_session_get(conn);
     mk_http_session_remove(cs);
+
     return 0;
 }
 

lib/monkey/mk_server/mk_http_parser.c

@@ -93,7 +93,7 @@ static inline int str_searchr(char *buf, char c, int len)
 static inline int method_lookup(struct mk_http_request *req,
                                 struct mk_http_parser *p, char *buffer)
 {
-    int i;
+    int i = 0;
     int len;
 
     /* Method lenght */
@@ -385,7 +385,6 @@ int mk_http_parser(struct mk_http_request *req, struct mk_http_parser *p,
                     }
                     request_set(&req->uri, p, buffer);
                     parse_next();
-                    continue;
                 }
                 else if (buffer[i] == '?') {
                     mark_end();