fixed vconn ref count and hashtables

3 files changed, 173 insertions, 109 deletions

diff --git a/ecp/src/ecp/core.h b/ecp/src/ecp/core.h
index 7ee7a9a..a4ae5c9 100644
--- a/ecp/src/ecp/core.h
+++ b/ecp/src/ecp/core.h
@@ -306,6 +306,19 @@ typedef struct ECPConnTable {
 #endif
 } ECPConnTable;
 
+#ifdef ECP_WITH_VCONN
+#ifdef ECP_WITH_HTABLE
+typedef struct ECPVConnTable {
+    ecp_ht_table_t *vconn_keys;
+    ecp_ht_table_t *vlink_keys;
+#ifdef ECP_WITH_PTHREAD
+    pthread_mutex_t vconn_keys_mutex;
+    pthread_mutex_t vlink_keys_mutex;
+#endif
+} ECPVConnTable;
+#endif
+#endif
+
 typedef struct ECPSocket {
     ECPContext *ctx;
     unsigned char running;
@@ -321,6 +334,11 @@ typedef struct ECPSocket {
     pthread_t rcvr_thd;
     pthread_mutex_t mutex;
 #endif
+#ifdef ECP_WITH_VCONN
+#ifdef ECP_WITH_HTABLE
+    ECPVConnTable vconn_table;
+#endif
+#endif
 } ECPSocket;
 
 typedef struct ECPConnection {
diff --git a/ecp/src/ecp/vconn/vconn.c b/ecp/src/ecp/vconn/vconn.c
index 28620a6..39da365 100644
--- a/ecp/src/ecp/vconn/vconn.c
+++ b/ecp/src/ecp/vconn/vconn.c
@@ -13,55 +13,47 @@
 
 #ifdef ECP_WITH_HTABLE
 
-static int insert_key_next(ECPVConnInb *vconn, unsigned char c_idx, ecp_ecdh_public_t *public) {
+static int insert_vconn_next(ECPVConnInb *vconn, unsigned char c_idx, ecp_ecdh_public_t *public) {
     ECPConnection *conn = &vconn->b;
     ECPSocket *sock = conn->sock;
     ECPDHPub *key = NULL;
     int rv = ECP_OK;
 
-    if (c_idx & ~ECP_ECDH_IDX_MASK) return ECP_ERR_ECDH_IDX;
-
 #ifdef ECP_WITH_PTHREAD
     pthread_mutex_lock(&conn->mutex);
 #endif
 
-    if (c_idx != vconn->key_next_curr) {
-        unsigned char _c_idx;
-
-        _c_idx = c_idx % ECP_MAX_NODE_KEY;
-        key = &vconn->key_next[_c_idx];
-
-        if (key->valid && (memcmp(public, &key->public, sizeof(key->public)) == 0)) {
-            key = NULL;
-        }
+    key = &vconn->vconn_next[c_idx % ECP_MAX_NODE_KEY];
+    if (key->valid && (memcmp(public, &key->public, sizeof(key->public)) == 0)) {
+        key = NULL;
     }
 
 #ifdef ECP_WITH_PTHREAD
     pthread_mutex_unlock(&conn->mutex);
 #endif
 
-    if (key) {
+    if (key == NULL) return ECP_OK;
+
 #ifdef ECP_WITH_PTHREAD
-        pthread_mutex_lock(&sock->conn_table.mutex);
-        pthread_mutex_lock(&conn->mutex);
+    pthread_mutex_lock(&sock->vconn_table.vconn_keys_mutex);
+    pthread_mutex_lock(&conn->mutex);
 #endif
 
-        if (key->valid) {
-            ecp_ht_remove(sock->conn_table.keys, &key->public);
-            key->valid = 0;
-        }
-        memcpy(&key->public, public, sizeof(key->public));
-        rv = ecp_ht_insert(sock->conn_table.keys, &key->public, conn);
-        if (!rv) {
-            key->valid = 1;
-            vconn->key_next_curr = c_idx;
-        }
+    if (key->valid) {
+        ecp_ht_remove(sock->vconn_table.vconn_keys, &key->public);
+        key->valid = 0;
+    }
+    memcpy(&key->public, public, sizeof(key->public));
+    rv = ecp_ht_insert(sock->vconn_table.vconn_keys, &key->public, conn);
+    if (!rv) {
+        key->valid = 1;
+        vconn->vconn_next_curr = c_idx;
+    }
 
 #ifdef ECP_WITH_PTHREAD
-        pthread_mutex_unlock(&conn->mutex);
-        pthread_mutex_unlock(&sock->conn_table.mutex);
+    pthread_mutex_unlock(&conn->mutex);
+    pthread_mutex_unlock(&sock->vconn_table.vconn_keys_mutex);
 #endif
-    }
 
     return rv;
 }
@@ -69,32 +61,21 @@ static int insert_key_next(ECPVConnInb *vconn, unsigned char c_idx, ecp_ecdh_pub
 static ssize_t handle_next(ECPConnection *conn, unsigned char *msg, size_t msg_size, ECP2Buffer *bufs) {
     ECPVConnInb *vconn = (ECPVConnInb *)conn;
     ECPSocket *sock = conn->sock;
-    int rv = ECP_OK;
 
     if (msg_size < ECP_SIZE_ECDH_PUB) return ECP_ERR_SIZE;
     if (ecp_conn_is_outb(conn)) return ECP_ERR;
 
 #ifdef ECP_WITH_PTHREAD
-    pthread_mutex_lock(&sock->conn_table.mutex);
+    pthread_mutex_lock(&conn->mutex);
 #endif
 
-    if (vconn->next == NULL) {
-        vconn->next = ecp_ht_search(sock->conn_table.keys, (ecp_ecdh_public_t *)msg);
-        if (vconn->next) {
-            ecp_conn_refcount_inc(vconn->next);
-        } else {
-            rv = ECP_ERR_NEXT;
-        }
-    } else {
-        rv = ECP_ERR_NEXT;
-    }
+    memcpy(&vconn->vlink_next.public, msg, sizeof(vconn->vlink_next.public));
+    vconn->vlink_next.valid = 1;
 
 #ifdef ECP_WITH_PTHREAD
-    pthread_mutex_unlock(&sock->conn_table.mutex);
+    pthread_mutex_unlock(&conn->mutex);
 #endif
 
-    if (rv) return rv;
-
     return ECP_SIZE_ECDH_PUB;
 }
 
@@ -116,14 +97,37 @@ static ssize_t handle_relay(ECPConnection *conn, unsigned char *msg, size_t msg_
         /* forward message */
         case ECP_CTYPE_VCONN: {
             ECPVConnInb *vconn = (ECPVConnInb *)conn;
+            int insert_next = 0;
 
             if (ecp_conn_is_outb(conn)) return ECP_ERR;
 
-            conn_next = vconn->next;
+            _idx = (idx & 0x0F);
+            conn_next = NULL;
+
+            if (_idx & ~ECP_ECDH_IDX_MASK) return ECP_ERR_ECDH_IDX;
+
+#ifdef ECP_WITH_PTHREAD
+            pthread_mutex_lock(&sock->vconn_table.vlink_keys_mutex);
+            pthread_mutex_lock(&conn->mutex);
+#endif
+
+            if (_idx != vconn->vconn_next_curr) insert_next = 1;
+
+            if (vconn->vlink_next.valid) conn_next = ecp_ht_search(sock->vconn_table.vlink_keys, &vconn->vlink_next.public);
             if (conn_next) {
-                _idx = (idx & 0x0F);
+                _rv = ecp_conn_refcount_inc(conn_next);
+                if (_rv) conn_next = NULL;
+            }
+
+#ifdef ECP_WITH_PTHREAD
+            pthread_mutex_unlock(&conn->mutex);
+            pthread_mutex_unlock(&sock->vconn_table.vlink_keys_mutex);
+#endif
 
-                _rv = insert_key_next(vconn, _idx, (ecp_ecdh_public_t *)(msg+ECP_SIZE_PROTO+1));
+            if (conn_next == NULL) return ECP_ERR_NEXT;
+
+            if (insert_next) {
+                _rv = insert_vconn_next(vconn, _idx, (ecp_ecdh_public_t *)(msg+ECP_SIZE_PROTO+1));
                 if (_rv) return _rv;
             }
 
@@ -133,18 +137,23 @@ static ssize_t handle_relay(ECPConnection *conn, unsigned char *msg, size_t msg_
         /* back message */
         case ECP_CTYPE_VLINK: {
 #ifdef ECP_WITH_PTHREAD
-            pthread_mutex_lock(&sock->conn_table.mutex);
+            pthread_mutex_lock(&sock->vconn_table.vconn_keys_mutex);
 #endif
 
-            conn_next = ecp_ht_search(sock->conn_table.keys, (ecp_ecdh_public_t *)(msg+ECP_SIZE_PROTO+1));
-            if (conn_next) ecp_conn_refcount_inc(conn_next);
+            conn_next = ecp_ht_search(sock->vconn_table.vconn_keys, (ecp_ecdh_public_t *)(msg+ECP_SIZE_PROTO+1));
+            if (conn_next) {
+                _rv = ecp_conn_refcount_inc(conn_next);
+                if (_rv) conn_next = NULL;
+            }
 
 #ifdef ECP_WITH_PTHREAD
-            pthread_mutex_unlock(&sock->conn_table.mutex);
+            pthread_mutex_unlock(&sock->vconn_table.vconn_keys_mutex);
 #endif
 
+            if (conn_next == NULL) return ECP_ERR_NEXT;
+
             _idx = (idx & 0xF0) >> 4;
-            if (conn_next && (_idx == ECP_ECDH_IDX_PERMA)) {
+            if (_idx == ECP_ECDH_IDX_PERMA) {
                 /* this is init reply */
                 msg[ECP_SIZE_PROTO] = (ECP_ECDH_IDX_PERMA << 4) | ECP_ECDH_IDX_NOKEY;
                 memmove(msg+ECP_SIZE_PROTO+1, msg+ECP_SIZE_PROTO+1+ECP_SIZE_ECDH_PUB, msg_size-(ECP_SIZE_PROTO+1+ECP_SIZE_ECDH_PUB));
@@ -158,17 +167,13 @@ static ssize_t handle_relay(ECPConnection *conn, unsigned char *msg, size_t msg_
             return ECP_ERR_CTYPE;
     }
 
-    if (conn_next == NULL) return ECP_ERR_NEXT;
-
     payload.buffer = msg - ECP_SIZE_MTYPE;
     payload.size = bufs->payload->size - (payload.buffer - bufs->payload->buffer);
 
     ecp_pld_set_type(payload.buffer, payload.size, ECP_MTYPE_EXEC);
     rv = ecp_pld_send(conn_next, bufs->packet, &payload, ECP_SIZE_MTYPE + _msg_size, ECP_SEND_FLAG_REPLY);
 
-    if (conn->type == ECP_CTYPE_VLINK) {
-        ecp_conn_refcount_dec(conn_next);
-    }
+    ecp_conn_refcount_dec(conn_next);
 
     if (rv < 0) return rv;
     return msg_size;
@@ -233,9 +238,9 @@ void ecp_vconn_init_inb(ECPVConnInb *vconn, ECPSocket *sock) {
     ECPConnection *conn = &vconn->b;
 
     ecp_conn_init(conn, sock, ECP_CTYPE_VCONN);
-    vconn->next = NULL;
-    memset(vconn->key_next, 0, sizeof(vconn->key_next));
-    vconn->key_next_curr = ECP_ECDH_IDX_INV;
+    memset(&vconn->vlink_next, 0, sizeof(vconn->vlink_next));
+    memset(&vconn->vconn_next, 0, sizeof(vconn->vconn_next));
+    vconn->vconn_next_curr = ECP_ECDH_IDX_INV;
 }
 
 #endif  /* ECP_WITH_HTABLE */
@@ -260,18 +265,16 @@ int ecp_vconn_open(ECPVConnOutb *vconn, ECPConnection *conn, ECPNode *node) {
 }
 
 int ecp_vconn_handle_open(ECPConnection *conn, ECP2Buffer *bufs) {
-    int rv = ECP_OK;
-
-    if (ecp_conn_is_outb(conn)) {
-        ECPVConnOutb *vconn = (ECPVConnOutb *)conn;
+    ECPVConnOutb *vconn = (ECPVConnOutb *)conn;
+    int rv;
 
-        if (vconn->next == NULL) return ECP_ERR;
+    if (ecp_conn_is_inb(conn)) return ECP_OK;
+    if (vconn->next == NULL) return ECP_ERR;
 
-        /* we should release incoming packet before sending next open packet */
-        ecp_tr_release(bufs->packet, 1);
-        rv = _ecp_conn_open(vconn->next, conn, NULL, 1);
-        if (rv) rv = ECP_ERR_NEXT;
-    }
+    /* we should release incoming packet before sending next open packet */
+    ecp_tr_release(bufs->packet, 1);
+    rv = _ecp_conn_open(vconn->next, conn, NULL, 1);
+    if (rv) rv = ECP_ERR_NEXT;
 
     return rv;
 }
@@ -286,23 +289,21 @@ void ecp_vconn_handle_close(ECPConnection *conn) {
     if (ecp_conn_is_outb(conn)) return;
 
 #ifdef ECP_WITH_PTHREAD
-    pthread_mutex_lock(&sock->conn_table.mutex);
+    pthread_mutex_lock(&sock->vconn_table.vconn_keys_mutex);
 #endif
 
     for (i=0; i<ECP_MAX_NODE_KEY; i++) {
         ECPDHPub *key;
 
-        key = &vconn->key_next[i];
+        key = &vconn->vconn_next[i];
         if (key->valid) {
-            ecp_ht_remove(sock->conn_table.keys, &key->public);
+            ecp_ht_remove(sock->vconn_table.vconn_keys, &key->public);
         }
     }
 
 #ifdef ECP_WITH_PTHREAD
-    pthread_mutex_unlock(&sock->conn_table.mutex);
+    pthread_mutex_unlock(&sock->vconn_table.vconn_keys_mutex);
 #endif
-
-    if (vconn->next) ecp_conn_refcount_dec(vconn->next);
 }
 
 #endif  /* ECP_WITH_HTABLE */
@@ -380,39 +381,22 @@ void ecp_vlink_init(ECPConnection *conn, ECPSocket *sock) {
 int ecp_vlink_handle_open(ECPConnection *conn, ECP2Buffer *bufs) {
     ECPSocket *sock = conn->sock;
     ECPDHPub *key;
-    int rv = ECP_OK;
-
-    key = &conn->remote.key_perma;
-
-#ifdef ECP_WITH_PTHREAD
-    pthread_mutex_lock(&sock->conn_table.mutex);
-    pthread_mutex_lock(&conn->mutex);
-#endif
+    int rv;
 
-    if (key->valid) {
-        ECPConnection *_conn;
+    if ((conn->parent == NULL) && ecp_conn_is_inb(conn)) return ECP_OK;
 
-        /* check for duplicates */
-        _conn = ecp_ht_search(sock->conn_table.keys, &key->public);
-        if (_conn) {
-#ifdef ECP_WITH_PTHREAD
-            pthread_mutex_lock(&_conn->mutex);
-#endif
-
-            _conn->remote.key_perma.valid = 0;
+    key = &conn->remote.key_perma;
+    if (!key->valid) return ECP_OK;
 
 #ifdef ECP_WITH_PTHREAD
-            pthread_mutex_unlock(&_conn->mutex);
+    pthread_mutex_lock(&sock->vconn_table.vlink_keys_mutex);
 #endif
 
-            ecp_ht_remove(sock->conn_table.keys, &key->public);
-        }
-        rv = ecp_ht_insert(sock->conn_table.keys, &key->public, conn);
-    }
+    ecp_ht_remove(sock->vconn_table.vlink_keys, &key->public);
+    rv = ecp_ht_insert(sock->vconn_table.vlink_keys, &key->public, conn);
 
 #ifdef ECP_WITH_PTHREAD
-    pthread_mutex_unlock(&conn->mutex);
-    pthread_mutex_unlock(&sock->conn_table.mutex);
+    pthread_mutex_unlock(&sock->vconn_table.vlink_keys_mutex);
 #endif
 
     return rv;
@@ -420,19 +404,22 @@ int ecp_vlink_handle_open(ECPConnection *conn, ECP2Buffer *bufs) {
 
 void ecp_vlink_handle_close(ECPConnection *conn) {
     ECPSocket *sock = conn->sock;
-    ECPDHPub *key = &conn->remote.key_perma;
+    ECPDHPub *key;
+
+    if ((conn->parent == NULL) && ecp_conn_is_inb(conn)) return;
+
+    key = &conn->remote.key_perma;
+    if (!key->valid) return;
 
-    if (key->valid) {
 #ifdef ECP_WITH_PTHREAD
-        pthread_mutex_lock(&sock->conn_table.mutex);
+    pthread_mutex_lock(&sock->vconn_table.vlink_keys_mutex);
 #endif
 
-        ecp_ht_remove(sock->conn_table.keys, &key->public);
+    ecp_ht_remove_kv(sock->vconn_table.vlink_keys, &key->public, conn);
 
 #ifdef ECP_WITH_PTHREAD
-        pthread_mutex_unlock(&sock->conn_table.mutex);
+    pthread_mutex_unlock(&sock->vconn_table.vlink_keys_mutex);
 #endif
-    }
 }
 
 #endif  /* ECP_WITH_HTABLE */
@@ -459,6 +446,63 @@ ssize_t ecp_vlink_handle_msg(ECPConnection *conn, ecp_seq_t seq, unsigned char m
     return rv;
 }
 
+int ecp_vconn_sock_create(ECPSocket *sock) {
+    int rv = ECP_OK;
+
+#ifdef ECP_WITH_HTABLE
+
+    memset(&sock->vconn_table, 0, sizeof(sock->vconn_table));
+
+#ifdef ECP_WITH_PTHREAD
+    rv = pthread_mutex_init(&sock->vconn_table.vconn_keys_mutex, NULL);
+    if (rv) {
+        return ECP_ERR;
+    }
+
+    rv = pthread_mutex_init(&sock->vconn_table.vlink_keys_mutex, NULL);
+    if (rv) {
+        pthread_mutex_destroy(&sock->vconn_table.vconn_keys_mutex);
+        return ECP_ERR;
+    }
+#endif
+
+    sock->vconn_table.vconn_keys = ecp_ht_create_keys();
+    if (sock->vconn_table.vconn_keys == NULL) rv = ECP_ERR_ALLOC;
+
+    if (!rv) {
+        sock->vconn_table.vlink_keys = ecp_ht_create_keys();
+        if (sock->vconn_table.vlink_keys == NULL) rv = ECP_ERR_ALLOC;
+    }
+
+    if (rv) {
+#ifdef ECP_WITH_PTHREAD
+        pthread_mutex_destroy(&sock->vconn_table.vlink_keys_mutex);
+        pthread_mutex_destroy(&sock->vconn_table.vconn_keys_mutex);
+#endif
+
+        if (sock->vconn_table.vlink_keys) ecp_ht_destroy(sock->vconn_table.vlink_keys);
+        if (sock->vconn_table.vconn_keys) ecp_ht_destroy(sock->vconn_table.vconn_keys);
+    }
+
+#endif  /* ECP_WITH_HTABLE */
+
+    return rv;
+}
+
+void ecp_vconn_sock_destroy(ECPSocket *sock) {
+#ifdef ECP_WITH_HTABLE
+
+    ecp_ht_destroy(sock->vconn_table.vlink_keys);
+    ecp_ht_destroy(sock->vconn_table.vconn_keys);
+
+#ifdef ECP_WITH_PTHREAD
+    pthread_mutex_destroy(&sock->vconn_table.vlink_keys_mutex);
+    pthread_mutex_destroy(&sock->vconn_table.vconn_keys_mutex);
+#endif
+
+#endif  /* ECP_WITH_HTABLE */
+}
+
 int ecp_vconn_handler_init(ECPContext *ctx, ECPConnHandler *vconn_handler) {
     int rv;
 
diff --git a/ecp/src/ecp/vconn/vconn.h b/ecp/src/ecp/vconn/vconn.h
index 9bdb3d3..f343846 100644
--- a/ecp/src/ecp/vconn/vconn.h
+++ b/ecp/src/ecp/vconn/vconn.h
@@ -13,9 +13,9 @@
 
 typedef struct ECPVConnInb {
     ECPConnection b;
-    ECPConnection *next;
-    ECPDHPub key_next[ECP_MAX_NODE_KEY];
-    unsigned char key_next_curr;
+    ECPDHPub vlink_next;
+    ECPDHPub vconn_next[ECP_MAX_NODE_KEY];
+    unsigned char vconn_next_curr;
 } ECPVConnInb;
 
 typedef struct ECPVConnOutb {
@@ -45,5 +45,7 @@ void ecp_vlink_handle_close(ECPConnection *conn);
 #endif
 ssize_t ecp_vlink_handle_msg(ECPConnection *conn, ecp_seq_t seq, unsigned char mtype, unsigned char *msg, size_t msg_size, ECP2Buffer *bufs);
 
+int ecp_vconn_sock_create(ECPSocket *sock);
+void ecp_vconn_sock_destroy(ECPSocket *sock);
 int ecp_vconn_handler_init(ECPContext *ctx, ECPConnHandler *vconn_handler);
 int ecp_vlink_handler_init(ECPContext *ctx, ECPConnHandler *vlink_handler);