Change DP termination logic #2

- Move destory operation to event loop thread
- Move deinit operation to event loop thread
- Destroy a vhost using lws_vhost_destroy()
- Check a fd of websocket_op_s before processing op queue

Change-Id: I572b2c7449362e47c6506fc121f4fcfdd2de0d65

diff --git a/plugins/libwebsockets/libwebsockets-plugin.cpp b/plugins/libwebsockets/libwebsockets-plugin.cpp
index 60933ee..748a117 100755 (executable)
--- a/plugins/libwebsockets/libwebsockets-plugin.cpp
+++ b/plugins/libwebsockets/libwebsockets-plugin.cpp
@@ -86,6 +86,8 @@ typedef struct {
        char *iface_name; // open, connect only
        int max_conn; // open only
        vine_dp_ssl ssl; // open, connect only
+
+       int fd; // eventfd
 } websocket_op_s;
 
 static VineQueue<websocket_op_s *> op_queue;
@@ -220,24 +222,25 @@ static void _get_peer_network_info(struct lws *wsi,
        VINE_LOGD("address family[%d] peer ip[%s], peer port[%d]\n", addr.ss_family, ip, *port);
 }
 
-static void _notify_websocket_op_request()
+static void _notify_websocket_op_request(websocket_op_s *op)
 {
        uint64_t v = 1;
        int _eventfd = eventfd(0, 0);
 
+       op->fd = _eventfd;
        g_callbacks.pollfd_cb(VINE_DATA_PATH_POLLFD_OP, _eventfd, LWS_POLLOUT, NULL);
        if (write(_eventfd, &v, sizeof(v)) == -1)
                VINE_LOGE("Write error(%d)", errno);
 }
 
-static int _add_websocket_op_request(websocket_op_code_e code,
+static websocket_op_s *_add_websocket_op_request(websocket_op_code_e code,
                websocket_s *ws, struct lws_vhost *vh, int addr_family,
                const char *ip, int port, const char *iface_name,
                int max_conn, vine_dp_ssl *ssl)
 {
        websocket_op_s *op = (websocket_op_s *)calloc(1, sizeof(websocket_op_s));
        if (!op)
-               return -1;
+               return NULL;
 
        op->code = code;
        op->ws = ws;
@@ -258,7 +261,8 @@ static int _add_websocket_op_request(websocket_op_code_e code,
        }
 
        op_queue.push(op);
-       return 0;
+       return op;
+       //return 0;
 }
 
 static void _del_websocket_op_request(websocket_op_s *op)
@@ -284,8 +288,8 @@ static void _process_pending_destroy(struct lws_vhost *vh)
 
        lws_callback_all_protocol_vhost_args(vh, protocol, LWS_CALLBACK_USER, NULL, 0);
        listen_vh_list.erase(vh);
-       //lws_vhost_destroy(vh);
-       VINE_LOGI("vh[%p] is erased.", vh);
+       VINE_LOGI("vh[%p] will be destroyed.", vh);
+       lws_vhost_destroy(vh);
 
        while (!lws_service_adjust_timeout(g_context, 1, 0))
                lws_service_tsi(g_context, -1, 0);
@@ -343,13 +347,15 @@ static void _deinit(void)
        VINE_LOGI("-");
 }
 
-static void _process_websocket_op_request(void)
+static void _process_websocket_op_request(int fd)
 {
        RET_IF(op_queue.empty(), "operation queue is NULL");
 
        websocket_op_s *op = op_queue.front();
-       op_queue.pop();
        RET_IF(op == NULL, "op is NULL");
+       RET_IF(op->fd != fd, "Not matched event. op->fd[%d] fd[%d]", op->fd, fd);
+
+       op_queue.pop();
 
        if (!op->ws && op->code != WEBSOCKET_OP_DEINIT) {
                _del_websocket_op_request(op);
@@ -398,21 +404,17 @@ static void do_func(websocket_op_s *op)
                break;
        case WEBSOCKET_OP_WRITE:
                VINE_LOGI("Ignore.");
-               // Do!
+               // TODO: do write.
                break;
        case WEBSOCKET_OP_TERMINATE:
-               VINE_LOGI("Terminate.");
-               _process_pending_destroy(op->vh);
-               // Do!
-               break;
+               VINE_LOGI("Terminate. Do not handle here. ws[%p]", op->ws);
+               return;
        case WEBSOCKET_OP_DESTROY:
-               VINE_LOGI("destroy.");
-               //_destroy_websocket(op->ws);
-               // error
-               break;
+               VINE_LOGI("Destroy. Do not handle here. ws[%p]", op->ws);
+               return;
        case WEBSOCKET_OP_DEINIT:
-               VINE_LOGI("This cannot be called here.");
-               break;
+               VINE_LOGE("This cannot be called here.");
+               return;
        default:
                break;
        }
@@ -638,7 +640,7 @@ static int _websocket_protocol_cb(struct lws *wsi,
        case LWS_CALLBACK_RECEIVE:
        {
                VINE_LOGI("%d bytes is received from client.", len);
-               if (!ws)
+               if (!ws || len == 0)
                        return 0;
 
                size_t total_len = _save_data(ws, in, len,
@@ -721,7 +723,7 @@ static int _websocket_protocol_cb(struct lws *wsi,
        case LWS_CALLBACK_CLIENT_RECEIVE:
        {
                VINE_LOGI("%d bytes is received from server.", len);
-               if (!ws)
+               if (!ws || len == 0)
                        return 0;
 
                size_t total_len = _save_data(ws, in, len,
@@ -794,9 +796,13 @@ static void websocket_deinit(void)
 
        _flush_op_queue(NULL);
        // Run deinit in eventloop thread
-       _add_websocket_op_request(WEBSOCKET_OP_DEINIT,
+       websocket_op_s *op = _add_websocket_op_request(WEBSOCKET_OP_DEINIT,
                                NULL, NULL, 0, NULL, 0, NULL, 0, NULL);
-       _notify_websocket_op_request();
+       if (!op) {
+               VINE_LOGE("Failed to add deinit operation to queue");
+               return;
+       }
+       _notify_websocket_op_request(op);
 
        VINE_LOGD("lws is deinitialized.");
 }
@@ -898,12 +904,14 @@ static int websocket_open(vine_dp_plugin_h handle,
        websocket_s *ws = (websocket_s *)handle;
        ws->is_server = true;
        VINE_LOGD("ws[%p]", ws);
-       if (_add_websocket_op_request(WEBSOCKET_OP_OPEN,
+
+       websocket_op_s *op = _add_websocket_op_request(WEBSOCKET_OP_OPEN,
                                (websocket_s *)handle, NULL,
-                               addr_family, NULL, port, iface_name, max_conn, &ssl) < 0)
-               return VINE_DATA_PATH_ERROR_OPERATION_FAILED;
+                               addr_family, NULL, port, iface_name, max_conn, &ssl);
+       if (!op)
+               return VINE_DATA_PATH_ERROR_OUT_OF_MEMORY;
 
-       _notify_websocket_op_request();
+       _notify_websocket_op_request(op);
 
        return VINE_DATA_PATH_ERROR_NONE;
 }
@@ -944,12 +952,13 @@ static int websocket_connect(vine_dp_plugin_h handle,
        RET_VAL_IF(!handle, VINE_DATA_PATH_ERROR_INVALID_PARAMETER, "handle is NULL");
        RET_VAL_IF(!g_context, VINE_DATA_PATH_ERROR_INVALID_OPERATION, "g_context is NULL");
 
-       if (_add_websocket_op_request(WEBSOCKET_OP_CONNECT,
+       websocket_op_s *op = _add_websocket_op_request(WEBSOCKET_OP_CONNECT,
                                (websocket_s *)handle, NULL,
-                               addr_family, ip, port, iface_name, 0, &ssl) < 0)
-               return VINE_DATA_PATH_ERROR_OPERATION_FAILED;
+                               addr_family, ip, port, iface_name, 0, &ssl);
+       if (!op)
+               return VINE_DATA_PATH_ERROR_OUT_OF_MEMORY;
 
-       _notify_websocket_op_request();
+       _notify_websocket_op_request(op);
 
        return VINE_DATA_PATH_ERROR_NONE;
 }
@@ -1103,11 +1112,12 @@ static int websocket_write(vine_dp_plugin_h handle, unsigned char *buf, size_t l
        ws->write_buffer->push(wd);
        VINE_LOGD("websocket_data[%p] is pushed to write_buffer. len[%d]", wd, len);
 
-       if (_add_websocket_op_request(WEBSOCKET_OP_WRITE, ws, NULL,
-                               0, NULL, -1, NULL, 0, NULL) < 0)
-               return VINE_DATA_PATH_ERROR_OPERATION_FAILED;
+       websocket_op_s *op = _add_websocket_op_request(WEBSOCKET_OP_WRITE, ws, NULL,
+                               0, NULL, -1, NULL, 0, NULL);
+       if (!op)
+               return VINE_DATA_PATH_ERROR_OUT_OF_MEMORY;
 
-       _notify_websocket_op_request();
+       _notify_websocket_op_request(op);
 
        return VINE_DATA_PATH_ERROR_NONE;
 }
@@ -1125,10 +1135,11 @@ static int websocket_close(vine_dp_plugin_h handle)
        VINE_LOGD("ws[%p] will be closed.", ws);
        ws->close_requested = true;
        if (ws->is_server) {
-               if (_add_websocket_op_request(WEBSOCKET_OP_TERMINATE,
-                               ws, ws->vh, 0, NULL, 0, NULL, 0, NULL) < 0)
-                       return VINE_DATA_PATH_ERROR_OPERATION_FAILED;
-               _notify_websocket_op_request();
+               websocket_op_s *op = _add_websocket_op_request(WEBSOCKET_OP_TERMINATE,
+                               ws, ws->vh, 0, NULL, 0, NULL, 0, NULL);
+               if (!op)
+                       return VINE_DATA_PATH_ERROR_OUT_OF_MEMORY;
+               _notify_websocket_op_request(op);
                return VINE_DATA_PATH_ERROR_NONE;
        }
 
@@ -1142,7 +1153,7 @@ static void websocket_process_event(int fd, int events)
 {
        auto it = g_pollfds.find(fd);
        if (it == g_pollfds.end()) {
-               _process_websocket_op_request();
+               _process_websocket_op_request(fd);
                close(fd);
                return;
        }
@@ -1198,8 +1209,13 @@ static int websocket_destroy(vine_dp_plugin_h handle)
        }
 
        _flush_op_queue(ws);
-
        ws->destroy_requested = true;
+       websocket_op_s *op = _add_websocket_op_request(WEBSOCKET_OP_DESTROY,
+                               (websocket_s *)handle, NULL, 0, NULL, 0, NULL, 0, NULL);
+       if (!op)
+               return VINE_DATA_PATH_ERROR_OPERATION_FAILED;
+       _notify_websocket_op_request(op);
+
        return VINE_DATA_PATH_ERROR_NONE;
 }
 

diff --git a/src/vine-event-loop-epoll.cpp b/src/vine-event-loop-epoll.cpp
index 626da1c..5e7318a 100755 (executable)
--- a/src/vine-event-loop-epoll.cpp
+++ b/src/vine-event-loop-epoll.cpp
@@ -141,6 +141,7 @@ void vine_event_loop_epoll_stop()
        __cleanup = true;
        _vine_event_loop_epoll_wake_up();
        pthread_join(__vine_event_loop_epoll_tid, NULL);
+       VINE_LOGD("eventloop thread is terminated.");
        __vine_event_loop_epoll_tid = 0;
 }