Pluggable distribution socket implementation for EI

3 files changed, 99 insertions, 55 deletions

diff --git a/lib/erl_interface/src/epmd/epmd_port.c b/lib/erl_interface/src/epmd/epmd_port.c
index 2ec418b24a..492c3fb3aa 100644
--- a/lib/erl_interface/src/epmd/epmd_port.c
+++ b/lib/erl_interface/src/epmd/epmd_port.c
@@ -62,31 +62,38 @@
 int ei_epmd_connect_tmo(struct in_addr *inaddr, unsigned ms)
 {
   static unsigned int epmd_port = 0;
-  struct sockaddr_in saddr;
-  int sd;
-  int res;
+  int port, sd, err;
+  struct in_addr ip_addr;
+  struct sockaddr_in addr;
+  unsigned tmo = ms == 0 ? EI_SCLBK_INF_TMO : ms;
+
+  err = ei_socket__(&sd);
+  if (err) {
+      erl_errno = err;
+      return -1;
+  }
 
   if (epmd_port == 0) {
       char* port_str = getenv("ERL_EPMD_PORT");
       epmd_port = (port_str != NULL) ? atoi(port_str) : EPMD_PORT;
   }
-  memset(&saddr, 0, sizeof(saddr)); 
-  saddr.sin_port = htons(epmd_port);
-  saddr.sin_family = AF_INET;
 
-  if (!inaddr) saddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
-  else memmove(&saddr.sin_addr,inaddr,sizeof(saddr.sin_addr));
+  port = (int) epmd_port;
 
-  if (((sd = socket(PF_INET, SOCK_STREAM, 0)) < 0))
-  {
-      erl_errno = errno;
-      return -1;
+  if (!inaddr) {      
+      ip_addr.s_addr = htonl(INADDR_LOOPBACK);
+      inaddr = &ip_addr;
   }
+  
+  memset((void *) &addr, 0, sizeof(struct sockaddr_in));
+  memcpy((void *) &addr.sin_addr, (void *) inaddr, sizeof(addr.sin_addr));
+  addr.sin_family = AF_INET;
+  addr.sin_port = htons(port);
 
-  if ((res = ei_connect_t(sd,(struct sockaddr *)&saddr,sizeof(saddr),ms)) < 0) 
-  {
-      erl_errno = (res == -2) ? ETIMEDOUT : errno;
-      closesocket(sd);
+  err = ei_connect_t__(sd, (void *) &addr, sizeof(addr), tmo);
+  if (err) {
+      erl_errno = err;
+      ei_close__(sd);
       return -1;
   }
 
@@ -104,6 +111,9 @@ static int ei_epmd_r4_port (struct in_addr *addr, const char *alive,
   int port;
   int dist_high, dist_low, proto;
   int res;
+  int err;
+  ssize_t dlen;
+  unsigned tmo = ms == 0 ? EI_SCLBK_INF_TMO : ms;
 #if defined(VXWORKS)
   char ntoabuf[32];
 #endif
@@ -124,10 +134,14 @@ static int ei_epmd_r4_port (struct in_addr *addr, const char *alive,
       return -1;
   }
 
-  if ((res = ei_write_fill_t(fd, buf, len+2, ms)) != len+2) {
-    closesocket(fd);
-    erl_errno = (res == -2) ? ETIMEDOUT : EIO;
-    return -1;
+  dlen = len + 2;
+  err = ei_write_fill_t__(fd, buf, &dlen, tmo);
+  if (!err && dlen != (ssize_t) len + 2)
+      erl_errno = EIO;
+  if (err) {
+      ei_close__(fd);
+      EI_CONN_SAVE_ERRNO__(err);
+      return -1;
   }
 
 #ifdef VXWORKS
@@ -142,12 +156,15 @@ static int ei_epmd_r4_port (struct in_addr *addr, const char *alive,
 		 "-> PORT2_REQ alive=%s ip=%s",alive,inet_ntoa(*addr));
 #endif
 
-  /* read first two bytes (response type, response) */
-  if ((res = ei_read_fill_t(fd, buf, 2, ms)) != 2) {
-    EI_TRACE_ERR0("ei_epmd_r4_port","<- CLOSE");
-    erl_errno = (res == -2) ? ETIMEDOUT : EIO;
-    closesocket(fd);
-    return -2;			/* version mismatch */
+  dlen = (ssize_t) 2;
+  err = ei_read_fill_t__(fd, buf, &dlen, tmo);
+  if (!err && dlen != (ssize_t) 2)
+      erl_errno = EIO;
+  if (err) {
+      EI_TRACE_ERR0("ei_epmd_r4_port","<- CLOSE");
+      ei_close__(fd);
+      EI_CONN_SAVE_ERRNO__(err);
+      return -2;
   }
 
   s = buf;
@@ -156,7 +173,7 @@ static int ei_epmd_r4_port (struct in_addr *addr, const char *alive,
   if (res != EI_EPMD_PORT2_RESP) { /* response type */
     EI_TRACE_ERR1("ei_epmd_r4_port","<- unknown (%d)",res);
     EI_TRACE_ERR0("ei_epmd_r4_port","-> CLOSE");
-    closesocket(fd);
+    ei_close__(fd);
     erl_errno = EIO;
     return -1;
   }
@@ -167,7 +184,7 @@ static int ei_epmd_r4_port (struct in_addr *addr, const char *alive,
   if ((res = get8(s))) {
     /* got negative response */
     EI_TRACE_ERR1("ei_epmd_r4_port","<- PORT2_RESP result=%d (failure)",res);
-    closesocket(fd);
+    ei_close__(fd);
     erl_errno = EIO;
     return -1;
   }
@@ -175,14 +192,18 @@ static int ei_epmd_r4_port (struct in_addr *addr, const char *alive,
   EI_TRACE_CONN1("ei_epmd_r4_port","<- PORT2_RESP result=%d (ok)",res);
 
   /* expecting remaining 8 bytes */
-  if ((res = ei_read_fill_t(fd,buf,8,ms)) != 8) {
+  dlen = (ssize_t) 8;
+  err = ei_read_fill_t__(fd, buf, &dlen, tmo);
+  if (!err && dlen != (ssize_t) 8)
+      err = EIO;
+  if (err) {
     EI_TRACE_ERR0("ei_epmd_r4_port","<- CLOSE");
-    erl_errno = (res == -2) ? ETIMEDOUT : EIO;
-    closesocket(fd);
+    ei_close__(fd);
+    EI_CONN_SAVE_ERRNO__(err);
     return -1;
   }
   
-  closesocket(fd);
+  ei_close__(fd);
   s = buf;
 
   port = get16be(s);
diff --git a/lib/erl_interface/src/epmd/epmd_publish.c b/lib/erl_interface/src/epmd/epmd_publish.c
index 47d68a6db0..20b8e867e8 100644
--- a/lib/erl_interface/src/epmd/epmd_publish.c
+++ b/lib/erl_interface/src/epmd/epmd_publish.c
@@ -68,8 +68,10 @@ static int ei_epmd_r4_publish (int port, const char *alive, unsigned ms)
   int nlen = strlen(alive);
   int len = elen + nlen + 13; /* hard coded: be careful! */
   int n;
-  int res, creation;
-  
+  int err, res, creation;
+  ssize_t dlen;
+  unsigned tmo = ms == 0 ? EI_SCLBK_INF_TMO : ms;
+
   if (len > sizeof(buf)-2)
   {
     erl_errno = ERANGE;
@@ -93,29 +95,39 @@ static int ei_epmd_r4_publish (int port, const char *alive, unsigned ms)
 
   if ((fd = ei_epmd_connect_tmo(NULL,ms)) < 0) return fd;
 
-  if ((res = ei_write_fill_t(fd, buf, len+2, ms)) != len+2) {
-    closesocket(fd);
-    erl_errno = (res == -2) ? ETIMEDOUT : EIO;
-    return -1;
+  dlen = (ssize_t) len+2;
+  err = ei_write_fill_t__(fd, buf, &dlen, tmo);
+  if (!err && dlen != (ssize_t) len + 2)
+      erl_errno = EIO;
+  if (err) {
+      ei_close__(fd);
+      EI_CONN_SAVE_ERRNO__(err);
+      return -1;
   }
 
   EI_TRACE_CONN6("ei_epmd_r4_publish",
 		 "-> ALIVE2_REQ alive=%s port=%d ntype=%d "
 		 "proto=%d dist-high=%d dist-low=%d",
 		 alive,port,'H',EI_MYPROTO,EI_DIST_HIGH,EI_DIST_LOW);
-  
-  if ((n = ei_read_fill_t(fd, buf, 4, ms)) != 4) {
+
+  dlen = (ssize_t) 4;
+  err = ei_read_fill_t__(fd, buf, &dlen, tmo);
+  n = (int) dlen;
+  if (!err && n != 4)
+      err = EIO;
+  if (err) {
     EI_TRACE_ERR0("ei_epmd_r4_publish","<- CLOSE");
-    closesocket(fd);
-    erl_errno = (n == -2) ? ETIMEDOUT : EIO;
+    ei_close__(fd);
+    EI_CONN_SAVE_ERRNO__(err);
     return -2;			/* version mismatch */
   }
+
   /* Don't close fd here! It keeps us registered with epmd */
   s = buf;
   if (((res=get8(s)) != EI_EPMD_ALIVE2_RESP)) {  /* response */
     EI_TRACE_ERR1("ei_epmd_r4_publish","<- unknown (%d)",res);
     EI_TRACE_ERR0("ei_epmd_r4_publish","-> CLOSE");
-    closesocket(fd);
+    ei_close__(fd);
     erl_errno = EIO;
     return -1;
   }
@@ -124,7 +136,7 @@ static int ei_epmd_r4_publish (int port, const char *alive, unsigned ms)
 
   if (((res=get8(s)) != 0)) {           /* 0 == success */
       EI_TRACE_ERR1("ei_epmd_r4_publish"," result=%d (fail)",res);
-    closesocket(fd);
+    ei_close__(fd);
     erl_errno = EIO;
     return -1;
   }
diff --git a/lib/erl_interface/src/epmd/epmd_unpublish.c b/lib/erl_interface/src/epmd/epmd_unpublish.c
index 255d0ffb59..c112f74147 100644
--- a/lib/erl_interface/src/epmd/epmd_unpublish.c
+++ b/lib/erl_interface/src/epmd/epmd_unpublish.c
@@ -58,7 +58,9 @@ int ei_unpublish_tmo(const char *alive, unsigned ms)
     char buf[EPMDBUF];
     char *s = (char*)buf;
     int len = 1 + strlen(alive);
-    int fd, res;
+    int fd, err;
+    ssize_t dlen;
+    unsigned tmo = ms == 0 ? EI_SCLBK_INF_TMO : ms;
 
     if (len > sizeof(buf)-3) {
 	erl_errno = ERANGE;
@@ -72,20 +74,29 @@ int ei_unpublish_tmo(const char *alive, unsigned ms)
     /* FIXME can't connect, return success?! At least commen whats up */
     if ((fd = ei_epmd_connect_tmo(NULL,ms)) < 0) return fd;
 
-    if ((res = ei_write_fill_t(fd, buf, len+2,ms)) != len+2) {
-	closesocket(fd);
-	erl_errno = (res == -2) ? ETIMEDOUT : EIO;
-	return -1;
+    dlen = (ssize_t) len+2;
+    err = ei_write_fill_t__(fd, buf, &dlen, tmo);
+    if (!err && dlen != (ssize_t) len + 2)
+        erl_errno = EIO;
+    if (err) {
+        ei_close__(fd);
+        EI_CONN_SAVE_ERRNO__(err);
+        return -1;
     }
 
     EI_TRACE_CONN1("ei_unpublish_tmo","-> STOP %s",alive);
-  
-    if ((res = ei_read_fill_t(fd, buf, 7, ms)) != 7) {
-	closesocket(fd);
-	erl_errno = (res == -2) ? ETIMEDOUT : EIO;
-	return -1; 
+
+    dlen = (ssize_t) 7;
+    err = ei_read_fill_t__(fd, buf, &dlen, tmo);
+    if (!err && dlen != (ssize_t) 7)
+        erl_errno = EIO;
+    if (err) {
+        ei_close__(fd);
+        EI_CONN_SAVE_ERRNO__(err);
+        return -1;
     }
-    closesocket(fd);
+    
+    ei_close__(fd);
     buf[7]=(char)0;		/* terminate the string */
   
     if (!strcmp("STOPPED",(char *)buf)) {