Pluggable distribution socket implementation for EI

4 files changed, 70 insertions, 108 deletions

diff --git a/lib/erl_interface/test/ei_accept_SUITE.erl b/lib/erl_interface/test/ei_accept_SUITE.erl
index 78a433d21b..9c9c3f86b6 100644
--- a/lib/erl_interface/test/ei_accept_SUITE.erl
+++ b/lib/erl_interface/test/ei_accept_SUITE.erl
@@ -81,12 +81,10 @@ ei_accept(Config) when is_list(Config) ->
 
 ei_threaded_accept(Config) when is_list(Config) ->
     Einode = filename:join(proplists:get_value(data_dir, Config), "eiaccnode"),
-    N = 1, % 3,
+    N = 3,
     Host = atom_to_list(node()),
-    Port = 6767,
-    start_einode(Einode, N, Host, Port),
+    start_einode(Einode, N, Host),
     io:format("started eiaccnode"),
-    %%spawn_link(fun() -> start_einode(Einode, N, Host, Port) end),
     TestServerPid = self(),
     [spawn_link(fun() -> send_rec_einode(I, TestServerPid) end) || I <- lists:seq(0, N-1)],
     [receive I -> ok end || I <- lists:seq(0, N-1) ],
@@ -159,10 +157,9 @@ send_rec_einode(N, TestServerPid) ->
               ct:fail(EINode)
     end.
 
-start_einode(Einode, N, Host, Port) ->
+start_einode(Einode, N, Host) ->
     Einodecmd = Einode ++ " " ++ atom_to_list(erlang:get_cookie())
-    ++ " " ++ integer_to_list(N) ++ " " ++ Host ++ " "
-    ++ integer_to_list(Port) ++ " nothreads",
+    ++ " " ++ integer_to_list(N) ++ " " ++ Host,
     io:format("Einodecmd  ~p ~n", [Einodecmd]),      
     open_port({spawn, Einodecmd}, []),
     ok.
diff --git a/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c b/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c
index 50df848b69..f41d741609 100644
--- a/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c
+++ b/lib/erl_interface/test/ei_accept_SUITE_data/ei_accept_test.c
@@ -125,45 +125,26 @@ static void cmd_ei_connect_init(char* buf, int len)
     ei_x_free(&res);
 }
 
-static int my_listen(int port)
-{
-    int listen_fd;
-    struct sockaddr_in addr;
-    const char *on = "1";
-    
-    if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
-	return -1;
-    
-    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, on, sizeof(on));
-    
-    memset((void*) &addr, 0, (size_t) sizeof(addr));
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(port);
-    addr.sin_addr.s_addr = htonl(INADDR_ANY);
-    
-    if (bind(listen_fd, (struct sockaddr*) &addr, sizeof(addr)) < 0)
-	return -1;
-
-    listen(listen_fd, 5);
-    return listen_fd;
-}
-
 static void cmd_ei_publish(char* buf, int len)
 {
     int index = 0;
-    int listen, r;
-    long port;
+    int iport, lfd, r;
+    long lport;
     ei_x_buff x;
     int i;
 
     /* get port */
-    if (ei_decode_long(buf, &index, &port) < 0)
+    if (ei_decode_long(buf, &index, &lport) < 0)
 	fail("expected int (port)");
     /* Make a listen socket */
-    if ((listen = my_listen(port)) <= 0)
+
+    iport = (int) lport;
+    lfd = ei_listen(&ec, &iport, 5);
+    if (lfd < 0)
 	fail("listen");
+    lport = (long) iport;
     
-    if ((i = ei_publish(&ec, port)) == -1)
+    if ((i = ei_publish(&ec, lport)) == -1)
 	fail("ei_publish");
 #ifdef VXWORKS
     save_fd(i);
@@ -171,7 +152,7 @@ static void cmd_ei_publish(char* buf, int len)
     /* send listen-fd, result and errno */
     ei_x_new_with_version(&x);
     ei_x_encode_tuple_header(&x, 3);
-    ei_x_encode_long(&x, listen);
+    ei_x_encode_long(&x, (long) lfd);
     ei_x_encode_long(&x, i);
     ei_x_encode_long(&x, erl_errno);
     send_bin_term(&x);
diff --git a/lib/erl_interface/test/ei_accept_SUITE_data/eiaccnode.c b/lib/erl_interface/test/ei_accept_SUITE_data/eiaccnode.c
index 308f843530..c850d20f3c 100644
--- a/lib/erl_interface/test/ei_accept_SUITE_data/eiaccnode.c
+++ b/lib/erl_interface/test/ei_accept_SUITE_data/eiaccnode.c
@@ -47,8 +47,6 @@
 #define MAIN main
 #endif
 
-static int my_listen(int port);
-
 /*
    A small einode.
    To be called from the test case ei_accept_SUITE:multi_thread
@@ -64,7 +62,6 @@ static int my_listen(int port);
 */
 
 static const char* cookie, * desthost;
-static int port; /* actually base port */
 
 #ifndef SD_SEND
 #ifdef SHUTWR
@@ -74,10 +71,6 @@ static int port; /* actually base port */
 #endif
 #endif
 
-#ifndef __WIN32__
-#define closesocket(fd) close(fd)
-#endif
-
 #ifdef __WIN32__
 static DWORD WINAPI
 #else
@@ -86,26 +79,32 @@ static void*
     einode_thread(void* num)
 {
     int n = (int)num;
+    int port;
     ei_cnode ec;
-    char myname[100], destname[100];
+    char myname[100], destname[100], filename[100];
     int r, fd, listen;
     ErlConnect conn;
     erlang_msg msg;
-/*    FILE* f;*/
+    FILE* file;
 
-    sprintf(myname, "eiacc%d", n);
-    printf("thread %d (%s) listening\n", n, myname, destname);
+    sprintf(filename, "eiacc%d_trace.txt", n);
+    file = fopen(filename, "w");
+
+    sprintf(myname, "eiacc%d", n); fflush(file);
     r = ei_connect_init(&ec, myname, cookie, 0);
-    if ((listen = my_listen(port+n)) <= 0) {
-	printf("listen err\n");
+    port = 0;
+    listen = ei_listen(&ec, &port, 5);
+    if (listen <= 0) {
+	fprintf(file, "listen err\n"); fflush(file);
 	exit(7);
     }
-    if (ei_publish(&ec, port + n) == -1) {
-	printf("ei_publish port %d\n", port+n);
+    fprintf(file, "thread %d (%s:%s) listening on port %d\n", n, myname, destname, port);
+    if (ei_publish(&ec, port) == -1) {
+	fprintf(file, "ei_publish port %d\n", port+n); fflush(file);
 	exit(8);
     }
     fd = ei_accept(&ec, listen, &conn);
-    printf("ei_accept %d\n", fd);
+    fprintf(file, "ei_accept %d\n", fd); fflush(file);
     if (fd >= 0) {
 	ei_x_buff x, xs;
 	int index, version;
@@ -117,37 +116,38 @@ static void*
 	    if (got == ERL_TICK)
 		continue;
 	    if (got == ERL_ERROR) {
-		printf("receive error %d\n", n);
+		fprintf(file, "receive error %d\n", n); fflush(file);
 		return 0;
 	    }
-	    printf("received %d\n", got);
+	    fprintf(file, "received %d\n", got); fflush(file);
 	    break;
 	}
 	index = 0;
 	if (ei_decode_version(x.buff, &index, &version) != 0) {
-	    printf("ei_decode_version %d\n", n);
+	    fprintf(file, "ei_decode_version %d\n", n); fflush(file);
 	    return 0;
 	}
 	if (ei_decode_pid(x.buff, &index, &pid) != 0) {
-	    printf("ei_decode_pid %d\n", n);
+	    fprintf(file, "ei_decode_pid %d\n", n); fflush(file);
 	    return 0;
 	}
-/*	fprintf(f, "got pid from %s \n", pid.node);*/
+	fprintf(file, "got pid from %s \n", pid.node); fflush(file);
 	ei_x_new_with_version(&xs);
 	ei_x_encode_tuple_header(&xs, 2);
 	ei_x_encode_long(&xs, n);
 	ei_x_encode_pid(&xs, &pid);
 	r = ei_send(fd, &pid, xs.buff, xs.index);
-/*	fprintf(f, "sent %d bytes %d\n", xs.index, r);*/
+	fprintf(file, "sent %d bytes %d\n", xs.index, r); fflush(file);
 	shutdown(fd, SD_SEND);
-	closesocket(fd);
+        ei_close_connection(fd);
 	ei_x_free(&x);
 	ei_x_free(&xs);
     } else {
-	printf("coudn't connect fd %d r %d\n", fd, r);
+	fprintf(file, "coudn't connect fd %d r %d\n", fd, r); fflush(file);
     }
-    printf("done thread %d\n", n);
-/*    fclose(f);*/
+    ei_close_connection(listen);
+    fprintf(file, "done thread %d\n", n);
+    fclose(file);
     return 0;
 }
 
@@ -170,12 +170,14 @@ MAIN(int argc, char *argv[])
     if (n > 100)
 	exit(2);
     desthost = argv[3];
-    port = atoi(argv[4]);
-#ifndef VXWORKS	   
-    no_threads = argv[5] != NULL && strcmp(argv[5], "nothreads") == 0;
-#else
+    if (argc == 3)
+        no_threads = 0;
+    else
+        no_threads = argv[4] != NULL && strcmp(argv[4], "nothreads") == 0;
+#ifdef VXWORKS
     no_threads = 1;
 #endif
+
     for (i = 0; i < n; ++i) {
 	if (!no_threads) {
 #ifndef VXWORKS
@@ -209,27 +211,3 @@ MAIN(int argc, char *argv[])
     printf("ok\n");
     return 0;
 }
-
-static int my_listen(int port)
-{
-    int listen_fd;
-    struct sockaddr_in addr;
-    const char *on = "1";
-    
-    if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
-	return -1;
-    
-    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, on, sizeof(on));
-    
-    memset((void*) &addr, 0, (size_t) sizeof(addr));
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(port);
-    addr.sin_addr.s_addr = htonl(INADDR_ANY);
-    
-    if (bind(listen_fd, (struct sockaddr*) &addr, sizeof(addr)) < 0)
-	return -1;
-
-    listen(listen_fd, 5);
-    return listen_fd;
-}
-
diff --git a/lib/erl_interface/test/erl_eterm_SUITE_data/cnode.c b/lib/erl_interface/test/erl_eterm_SUITE_data/cnode.c
index bead0f8413..b87feb9dfc 100644
--- a/lib/erl_interface/test/erl_eterm_SUITE_data/cnode.c
+++ b/lib/erl_interface/test/erl_eterm_SUITE_data/cnode.c
@@ -20,7 +20,7 @@
 
 #include <stdlib.h>
 #include <stdio.h>
-
+#include <string.h>
 #include "ei.h"
 #include "erl_interface.h"
 
@@ -68,6 +68,7 @@ MAIN(int argc, char **argv)
     char host[80];
     int number;
     ETERM *ref, *ref1, *ref2;
+    FILE *dfile = fopen("cnode_debug_printout", "w");
 
     erl_init(NULL, 0);
 
@@ -80,28 +81,30 @@ MAIN(int argc, char **argv)
     gethostname(host, sizeof(host));
     sprintf(node, "c%d@%s", number, host);
 
-    printf("s = %d\n", s);
+    fprintf(dfile, "s = %d\n", s); fflush(dfile);
 
     sprintf(server, "test_server@%s", host);
     fd = erl_connect(server);
-    printf("fd = %d\n", fd);
+    fprintf(dfile, "fd = %d\n", fd);
 
-/*    printf("dist = %d\n", erl_distversion(fd)); */
+/*    fprintf(dfile, "dist = %d\n", erl_distversion(fd)); */
 
 #if 1
     ref = erl_mk_long_ref(node, 4711, 113, 98, 0);
 #else
     ref = erl_mk_ref(node, 4711, 0);
 #endif
-    printf("ref = %d\n", ref);
+    fprintf(dfile, "ref = %p\n", ref); fflush(dfile);
 
     s = erl_reg_send(fd, "mip", ref);
-    printf("s = %d\n", s);
+    fprintf(dfile, "s = %d\n", s); fflush(dfile);
 
     {
       ETERM* emsg;
       emsg = SELF(fd);
-      erl_reg_send(fd,"mip",emsg);
+      fprintf(dfile, "pid = %p\n", emsg); fflush(dfile);
+      s = erl_reg_send(fd,"mip",emsg);
+      fprintf(dfile, "s2 = %d\n", s); fflush(dfile);
       erl_free_term(emsg);
     }
 
@@ -116,28 +119,29 @@ MAIN(int argc, char **argv)
 #endif
 	switch (s) {
 	  case ERL_TICK:
-	    printf("tick\n");
+	    fprintf(dfile, "tick\n");
 	    break;
 	  case ERL_ERROR:
-	    printf("error\n");
+            fprintf(dfile, "error: %s (%d)\n", strerror(erl_errno), erl_errno);
 	    break;
 	  case ERL_MSG:
-	    printf("msg %d\n", msgsize);
+	    fprintf(dfile, "msg %d\n", msgsize);
 	    break;
 	  default:
-	    printf("unknown result %d\n", s);
+	    fprintf(dfile, "unknown result %d\n", s);
 	    break;
 	}
+        fflush(dfile);
     } while (s == ERL_TICK);
 
     s = erl_reg_send(fd, "mip", msg.msg);
-    printf("s = %d\n", s);
+    fprintf(dfile, "s = %d\n", s); fflush(dfile);
     s = erl_reg_send(fd, "mip", msg.to);
-    printf("s = %d\n", s);
+    fprintf(dfile, "s = %d\n", s); fflush(dfile);
 #if 0
     /* from = NULL! */
     s = erl_reg_send(fd, "mip", msg.from);
-    printf("s = %d\n", s);
+    fprintf(dfile, "s = %d\n", s); fflush(dfile);
 #endif
 
 #if 0
@@ -150,17 +154,19 @@ MAIN(int argc, char **argv)
     ref1 = erl_mk_long_ref(node, 4711, 113, 98, 0);
     ref2 = erl_mk_ref(node, 4711, 0);
     s = erl_encode(ref1, buf1);
-    printf("enc1 s = %d\n", s);
+    fprintf(dfile, "enc1 s = %d\n", s); fflush(dfile);
     s = erl_encode(ref2, buf2);
-    printf("enc2 s = %d\n", s);
+    fprintf(dfile, "enc2 s = %d\n", s); fflush(dfile);
     s = erl_compare_ext(buf1, buf2);
-    printf("comp s = %d\n", s);
+    fprintf(dfile, "comp s = %d\n", s); fflush(dfile);
     
     /* Compare, in another way */
     s = erl_match(ref1, ref2);
-    printf("match s = %d\n", s);
+    fprintf(dfile, "match s = %d\n", s); fflush(dfile);
 #endif
 
+    fclose(dfile);
+
     erl_close_connection(fd);
 
     return 0;