improved (v)conn init / open / init vlink / open vlink API; implemented randezvous hashing

4 files changed, 32 insertions, 36 deletions

diff --git a/ecp/server/dir.c b/ecp/server/dir.c
index b941050..8064d0c 100644
--- a/ecp/server/dir.c
+++ b/ecp/server/dir.c
@@ -29,7 +29,9 @@ static int dir_process_enable;
 
 static SRVConfig *srv_config;
 
-#define MAX(X, Y)   (((X) > (Y)) ? (X) : (Y))
+/* other directory servers */
+#define _dir_item_is_dir(dir_item)      (((dir_item)->roles & ECP_ROLE_DIR) && (memcmp(&(dir_item)->node.key_perma.public, &srv_config->key_perma.public, sizeof(srv_config->key_perma.public)) != 0))
+#define MAX(X, Y)                       (((X) > (Y)) ? (X) : (Y))
 
 /* messages from clients */
 ssize_t dir_send_online(ECPConnection *conn, uint8_t region) {
@@ -414,14 +416,19 @@ void dir_process_item(ECPDirItem *dir_item, ECPSocket *sock, ecp_ecdh_public_t *
             if (rv) pthread_rwlock_unlock(&dir_shadow_rwlock);
         }
         if (rv) {
-            if (node) {
-                dir_destroy_node(node);
-                node = NULL;
-            }
-            LOG(LOG_ERR, "dir_process_item: err:%d\n", rv);
+            if (node) dir_destroy_node(node);
+            /* ignore duplicate hashtable inserts */
+            if (rv != ECP_ERR_DUP) LOG(LOG_ERR, "dir_process_item: create node err:%d\n", rv);
+
             return;
         }
 
+        /* try to open connection to other directory servers */
+        if (_dir_item_is_dir(dir_item)) {
+            rv = dir_open_conn(node, sock);
+            if (rv) LOG(LOG_ERR, "dir_process_item: conn open err:%d\n", rv);
+        }
+
         LOG(LOG_DEBUG, "dir_process_item: new node\n");
     }
 
@@ -496,7 +503,7 @@ int dir_open_conn(DIRNode *node, ECPSocket *sock) {
     ecp_conn_init(node->conn, sock, CTYPE_DIR);
     ecp_conn_set_flags(node->conn, ECP_CONN_FLAG_VBOX);
 
-    rv = ecp_conn_open(node->conn, &node->dir_item.node);
+    rv = ecp_conn_open(node->conn, &node->dir_item.node.key_perma.public, &node->dir_item.node.addr);
     if (rv) {
         free(node->conn);
         node->conn = NULL;
@@ -522,16 +529,7 @@ int dir_create_node(ECPDirItem *dir_item, ECPSocket *sock, DIRNode **node) {
 
     _node->dir_item = *dir_item;
     _node->is_new = 1;
-
-    /* open connection to other directory servers */
-    if ((dir_item->roles & ECP_ROLE_DIR) && (memcmp(&dir_item->node.key_perma.public, &srv_config->key_perma.public, sizeof(srv_config->key_perma.public)) != 0)) {
-        rv = dir_open_conn(_node, sock);
-        if (rv) {
-            pthread_mutex_destroy(&_node->mutex);
-            free(_node);
-            return rv;
-        }
-    }
+    _node->conn = NULL;
 
     *node = _node;
     return ECP_OK;
@@ -802,7 +800,8 @@ void dir_announce(ECPSocket *sock, int ann_period) {
             do {
                 node = ecp_ht_itr_value(&itr);
 
-                if ((node->dir_item.roles & ECP_ROLE_DIR) && (memcmp(&node->dir_item.node.key_perma.public, &srv_config->key_perma.public, sizeof(srv_config->key_perma.public)) != 0)) {
+                /* announce to other directory servers */
+                if (_dir_item_is_dir(&node->dir_item)) {
                     if (announce_cnt < MAX_NODE_ANNOUNCE) {
                         announce_node[announce_cnt] = node;
                         announce_cnt++;
@@ -922,7 +921,7 @@ void dir_init_switch(ECPSocket *sock, int init_ann) {
     sleep(init_ann);
 }
 
-int dir_init_ann(ECPSocket *sock, ECPNode *node) {
+int dir_init_ann(ECPSocket *sock, ecp_ecdh_public_t *public, ecp_tr_addr_t *addr) {
     ECPConnection *conn;
     int i, is_open;
     int rv;
@@ -933,7 +932,7 @@ int dir_init_ann(ECPSocket *sock, ECPNode *node) {
 
     ecp_conn_init(conn, sock, CTYPE_DIR);
     ecp_conn_set_flags(conn, ECP_CONN_FLAG_VBOX);
-    rv = ecp_conn_open(conn, node);
+    rv = ecp_conn_open(conn, public, addr);
     if (rv) {
         free(conn);
         LOG(LOG_ERR, "dir_init_ann: conn open err:%d\n", rv);
diff --git a/ecp/server/dir.h b/ecp/server/dir.h
index dbcbc2e..e5eba75 100644
--- a/ecp/server/dir.h
+++ b/ecp/server/dir.h
@@ -73,5 +73,5 @@ int dir_init_dir_cnt(unsigned int dir_cnt, uint16_t serial);
 void dir_announce(ECPSocket *sock, int ann_period);
 int dir_start_announce(ECPSocket *sock);
 void dir_init_switch(ECPSocket *sock, int init_ann);
-int dir_init_ann(ECPSocket *sock, ECPNode *node);
+int dir_init_ann(ECPSocket *sock, ecp_ecdh_public_t *public, ecp_tr_addr_t *addr);
 int dir_init(ECPSocket *sock);
diff --git a/ecp/server/server.c b/ecp/server/server.c
index a7140a1..fd8b317 100644
--- a/ecp/server/server.c
+++ b/ecp/server/server.c
@@ -42,7 +42,7 @@ SRVConfig *srv_get_config(void) {
 }
 
 static void handle_err(ECPConnection *conn, unsigned char mtype, int err) {
-    LOG(LOG_ERR, "handle_err: ctype:%d mtype:%d err:%d\n", conn->type, mtype, err);
+    LOG(LOG_ERR, "handle_err: ctype:0x%.2x mtype:0x%.2x err:%d\n", conn->type, mtype, err);
 }
 
 static ssize_t conn_auth(ECPSocket *sock, ECPConnection *parent, unsigned char ctype, ecp_ecdh_public_t *public, unsigned char *msg, size_t msg_size) {
@@ -307,20 +307,19 @@ static void dir_result(ECPSocket *sock, ECPDirList *list, int rv) {
 
 int main(int argc, char *argv[]) {
     char *endptr;
-    unsigned int rcvr_num;
     char *init_switch;
-    ECPNode node;
+    ecp_ecdh_public_t remote_pub;
+    ecp_tr_addr_t remote_addr;
     ECPContext ctx;
     ECPSocket sock;
     ECPConnection dir_conn;
     ECPConnHandler vconn_handler;
     ECPConnHandler vlink_handler;
     ECPConnHandler dir_handler;
-    int _argi, _argc, fd;
+    int _argi, _argc, remote, fd;
     int rv;
 
     memset(&srv_config, 0, sizeof(srv_config));
-    memset(&node, 0, sizeof(node));
 
     if (argc < 1) fail("Bad argument count\n");
 
@@ -443,21 +442,19 @@ int main(int argc, char *argv[]) {
         if (rv) fail("ACL load failed\n");
     }
 
+    remote = 0;
     if (_argc == 2) {
-        ecp_ecdh_public_t node_pub;
-
+        remote = 1;
         fd = open(argv[_argi], O_RDONLY);
         if (fd < 0) fail("Unable to open public key file: %s\n", argv[_argi]);
 
-        rv = ecp_util_read_key(fd, &node_pub, NULL, NULL);
+        rv = ecp_util_read_key(fd, &remote_pub, NULL, NULL);
         close(fd);
         if (rv) fail("Bad public key file: %s\n", argv[_argi]);
         _argi++;
         _argc--;
 
-        ecp_node_init(&node, &node_pub, NULL);
-
-        rv = ecp_node_set_addr(&node, argv[_argi]);
+        rv = ecp_addr_init(&remote_addr, argv[_argi]);
         if (rv) fail("Bad remote address: %s\n", argv[_argi]);
         _argi++;
         _argc--;
@@ -512,12 +509,12 @@ int main(int argc, char *argv[]) {
     if (rv) fail("start_receivers err:%d\n", rv);
 
     init_switch = getenv("ECP_INITSW");
-    if (node.key_perma.valid) {
+    if (remote) {
         if (init_switch == NULL) {
             rv = ecp_dir_set_handler(&ctx, &dir_handler, dir_result);
             if (rv) fail("ecp_dir_set_handler err:%d\n", rv);
 
-            rv = ecp_dir_get(&dir_conn, &sock, &node, 0);
+            rv = ecp_dir_get(&dir_conn, &sock, &remote_pub, &remote_addr, 0);
             if (rv) fail("ecp_dir_get err:%d\n", rv);
 
             pthread_mutex_lock(&dir_sync_mutex);
@@ -527,7 +524,7 @@ int main(int argc, char *argv[]) {
             pthread_mutex_unlock(&dir_sync_mutex);
             LOG(LOG_DEBUG, "ecp_dir_get: done\n");
         } else {
-            rv = dir_init_ann(&sock, &node);
+            rv = dir_init_ann(&sock, &remote_pub, &remote_addr);
             if (rv) fail("dir_init_ann err:%d\n", rv);
         }
     }
diff --git a/ecp/server/vlink.c b/ecp/server/vlink.c
index 445734c..eb2484e 100644
--- a/ecp/server/vlink.c
+++ b/ecp/server/vlink.c
@@ -32,7 +32,7 @@ int vlink_open_conn(ECPSocket *sock, ECPNode *node, ECPConnection **_conn) {
     if (conn == NULL) return ECP_ERR_ALLOC;
 
     ecp_vlink_init(conn, sock);
-    rv = ecp_conn_open(conn, node);
+    rv = ecp_conn_open(conn, &node->key_perma.public, &node->addr);
     if (rv) {
         free(conn);
         return rv;