[GNUnet-SVN] r27953 - msh/src

[gnunet]

Author: harsha
Date: 2013-07-12 14:59:22 +0200 (Fri, 12 Jul 2013)
New Revision: 27953

Added:
   msh/src/addressmap.c
   msh/src/addressmap.h
Modified:
   msh/src/Makefile.am
   msh/src/mshd.c
   msh/src/mtypes.h
Log:
- address map


Modified: msh/src/Makefile.am
===================================================================
--- msh/src/Makefile.am 2013-07-12 11:22:50 UTC (rev 27952)
+++ msh/src/Makefile.am 2013-07-12 12:59:22 UTC (rev 27953)
@@ -3,7 +3,7 @@
 mping_SOURCES = mping.c
 
 mshd_SOURCES = mshd.c util.c util.h scheduler.c scheduler.h \
-  common.h bitmap.c bitmap.h
+  common.h bitmap.c bitmap.h addressmap.c addressmap.h
 mshd_LDADD = -levent
 mshd_CPPFLAGS = $(LIBEVENT_CPPFLAGS)
 mshd_LDFLAGS =  $(LIBEVENT_LDFLAGS)
@@ -29,4 +29,5 @@
 
 TESTS = \
   test-scheduler \
-  test-scheduler-socket
+  test-scheduler-socket \
+  test-bitmap

Added: msh/src/addressmap.c
===================================================================
--- msh/src/addressmap.c                                (rev 0)
+++ msh/src/addressmap.c        2013-07-12 12:59:22 UTC (rev 27953)
@@ -0,0 +1,255 @@
+/**
+ * @file addressmap.c
+ * @brief implementation of address maps
+ * @author Sree Harsha Totakura <address@hidden> 
+ */
+
+#include "common.h"
+#include "util.h"
+#include "addressmap.h"
+
+
+/**
+ * An address of an instance
+ */
+struct InstanceAddr
+{
+  struct InstanceAddr *next;
+
+  struct InstanceAddr *prev;
+
+  /**
+   * The length of the instance address
+   */
+  socklen_t addrlen;
+
+  /**
+   * The instance address to be followed here
+   */
+  struct sockaddr saddr[0];
+};
+
+
+/**
+ * Instance address information
+ */
+struct InstanceAddrInfo
+{
+  struct InstanceAddr *addr_head;
+
+  struct InstanceAddr *addr_tail;
+
+  /**
+   * Number of addresses in the above array
+   */
+  unsigned int naddrs;
+  
+  /**
+   * The MPI id of the instance to whom these addresses belong to
+   */
+  unsigned int rank;
+};
+
+
+#define instance_address_ip(iaddr) \
+  ((uint32_t) ((struct sockaddr_in *)iaddr->saddr)->sin_addr.s_addr)
+
+
+/**
+ * Create an instance address structure from its port number and ip address
+ *
+ * @param port the number in network byte format
+ * @param ip 32-bit IP address in network byte format
+ * @return instace address structure
+ */
+struct InstanceAddr *
+instance_address_create_sockaddr_in (uint16_t port, in_addr_t ip)
+{
+  struct InstanceAddr *iaddr;
+  struct sockaddr_in *sinaddr;
+  socklen_t addrlen;
+
+  addrlen = sizeof (struct sockaddr_in);
+  iaddr = MSH_malloc (sizeof (struct InstanceAddr) + addrlen);
+  iaddr->addrlen = addrlen;
+  sinaddr = (struct sockaddr_in *) iaddr->saddr;
+  sinaddr->sin_family = AF_INET;
+  sinaddr->sin_port = port;
+  sinaddr->sin_addr.s_addr = ip;
+  return iaddr;
+}
+
+
+/**
+ * Create an instance address information object
+ *
+ * @param rank the MPI rank of the instance
+ * @return the instance address information object
+ */
+struct InstanceAddrInfo *
+instance_address_info_create (unsigned int rank)
+{
+  struct InstanceAddrInfo *iainfo;
+
+  iainfo = MSH_malloc (sizeof (struct InstanceAddrInfo));
+  iainfo->rank = rank;
+  return iainfo;
+}
+
+
+/**
+ * Add an address to the address information object
+ *
+ * @param iainfo the instance information object
+ * @param iaddr the address to add
+ */
+void
+instance_address_info_add_address (struct InstanceAddrInfo *iainfo,
+                                   struct InstanceAddr *iaddr)
+{
+  struct InstanceAddr *addr;
+  
+  addr = iainfo->addr_head;  
+  while (NULL != addr)
+  {
+    if (instance_address_ip (iaddr) == instance_address_ip (addr))
+    {
+      MSH_break (0);
+      return;
+    }
+    if (instance_address_ip (iaddr) < instance_address_ip (addr))
+      break;
+    addr = addr->next;
+  }
+  if (NULL == addr)
+  {
+    DLL_insert_tail (iainfo->addr_head, iainfo->addr_tail, iaddr);
+    return;
+  }
+  DLL_insert_before (iainfo->addr_head, iainfo->addr_tail, addr, iaddr);  
+}
+
+
+/**
+ * Iterate over all addresses in the given instance address info object
+ *
+ * @param iainfo the instance address info object
+ * @param cb the callback to call for each iterated address
+ * @param cls the closure for the above callback
+ * @return MSH_OK if the iteration completed successfully; MSH_SYSERR if the
+ *           iteration was terminated
+ */
+int
+instance_address_info_iterate_addresses (struct InstanceAddrInfo *iainfo,
+                                         
instance_address_info_address_iterate_cb
+                                         cb,
+                                         void *cls)
+{
+  struct InstanceAddr *iaddr;
+  
+  MSH_assert (NULL != cb);
+  iaddr = iainfo->addr_head;
+  while (NULL != iaddr)
+  {
+    if (MSH_OK != cb (cls, iaddr->saddr, iaddr->addrlen))
+      return MSH_SYSERR;
+    iaddr = iaddr->next;
+  }
+  return MSH_OK;
+}
+                                         
+
+
+/**
+ * Retrieve the rank of the instance from its address information object 
+ *
+ * @param iainfo the address information object
+ * @return the rank of the instance
+ */
+unsigned int
+instance_address_get_rank (const struct InstanceAddrInfo *iainfo)
+{
+  return iainfo->rank;
+}
+
+
+/**
+ * Free an instance's address information object along with the instance 
address
+ * objects
+ *
+ * @param iainfos the instance address information
+ */
+void
+instance_address_info_destroy (struct InstanceAddrInfo *iainfo)
+{
+  struct InstanceAddr *addr;
+
+  while (NULL != (addr = iainfo->addr_head))
+  {
+    DLL_remove (iainfo->addr_head, iainfo->addr_tail, addr);
+    free (addr);
+  }
+  free (iainfo);
+}
+
+
+/**
+ * Handle for address map
+ */
+struct AddressMap
+{
+  struct InstanceAddrInfo **map;
+
+  unsigned int size;
+};
+
+
+struct AddressMap *
+addressmap_create (unsigned int nproc)
+{
+  struct AddressMap *m;
+
+  m = MSH_malloc (sizeof (struct InstanceAddrInfo *) * nproc);
+  m->size = nproc;
+  return m;
+}
+
+
+void
+addressmap_add (struct AddressMap *m, struct InstanceAddrInfo *new)
+{
+  struct InstanceAddrInfo *old;
+  struct InstanceAddr *old_ia;
+  struct InstanceAddr *new_ia;
+  unsigned int rank;
+
+  rank = new->rank;
+  MSH_assert (rank < m->size);
+  old = m->map[rank];
+  if (NULL == old)
+  {
+    m->map[rank] = new;
+    return;
+  }
+  old_ia = old->addr_head;
+  new_ia = new->addr_head;
+  while (NULL != new_ia)
+  {
+    while ((NULL != old_ia) && (instance_address_ip (new_ia) >=
+                                instance_address_ip (old_ia)))
+    {
+      old_ia = old_ia->next;
+    }
+    if (NULL == old_ia)
+    {
+      DLL_insert_tail (old->addr_head, old->addr_tail, new_ia);
+      goto next;
+    }
+    DLL_insert_before (old->addr_head, old->addr_tail, old_ia, new_ia);
+
+  next:    
+    new_ia = new_ia->next;
+  }
+}
+
+

Added: msh/src/addressmap.h
===================================================================
--- msh/src/addressmap.h                                (rev 0)
+++ msh/src/addressmap.h        2013-07-12 12:59:22 UTC (rev 27953)
@@ -0,0 +1,96 @@
+/**
+ * @file addressmap.h
+ * @brief interface for address maps
+ * @author Sree Harsha Totakura <address@hidden> 
+ */
+
+#ifndef ADDRESSMAP_H_
+#define ADDRESSMAP_H_
+
+/**
+ * An address of an instance
+ */
+struct InstanceAddr;
+
+
+/**
+ * Instance address information
+ */
+struct InstanceAddrInfo;
+
+
+/**
+ * Create an instance address structure from its port number and ip address
+ *
+ * @param port the number in network byte format
+ * @param ip 32-bit IP address in network byte format
+ * @return instace address structure
+ */
+struct InstanceAddr *
+instance_address_create_sockaddr_in (uint16_t port, in_addr_t ip);
+
+
+/**
+ * Create an instance address information object
+ *
+ * @param rank the MPI rank of the instance
+ * @return the instance address information object
+ */
+struct InstanceAddrInfo *
+instance_address_info_create (unsigned int rank);
+
+
+/**
+ * Add an address to the address information object
+ *
+ * @param iainfo the instance information object
+ * @param iaddr the address to add
+ */
+void
+instance_address_info_add_address (struct InstanceAddrInfo *iainfo,
+                                   struct InstanceAddr *iaddr);
+
+
+/**
+ * Retrieve the rank of the instance from its address information object 
+ *
+ * @param iainfo the address information object
+ * @return the rank of the instance
+ */
+unsigned int
+instance_address_get_rank (const struct InstanceAddrInfo *iainfo);
+
+
+/**
+ * Free an instance's address information object along with the instance 
address
+ * objects
+ *
+ * @param iainfos the instance address information
+ */
+void
+instance_address_info_destroy (struct InstanceAddrInfo *iainfos);
+
+
+/**
+ * Callback for iterating all the instance addresses present in an instance
+ * address info object
+ *
+ * @param cls the closure passed to instance_address_info_iterate_addresses()
+ * @param addr the address of the instance
+ * @param addrlen the address length
+ * @return MSH_OK to continue iteration; MSH_SYSERR to terminate
+ */
+typedef int (*instance_address_info_address_iterate_cb) (void *cls,
+                                                         struct sockaddr *addr,
+                                                         socklen_t addrlen);
+
+
+/**
+ * Opaque handle for address map
+ */
+typedef struct AddressMap AddressMap;
+
+
+#endif  /* ADDRESSMAP_H_ */
+
+/* End of addressmap.h */

Modified: msh/src/mshd.c
===================================================================
--- msh/src/mshd.c      2013-07-12 11:22:50 UTC (rev 27952)
+++ msh/src/mshd.c      2013-07-12 12:59:22 UTC (rev 27953)
@@ -4,47 +4,9 @@
 #include "scheduler.h"
 #include "mtypes.h"
 #include "bitmap.h"
+#include "addressmap.h"
 
-
 /**
- * An address of an instance
- */
-struct InstanceAddr
-{
-  /**
-   * The length of the instance address
-   */
-  socklen_t addrlen;
-
-  /**
-   * The instance address to be followed here
-   */
-  struct sockaddr saddr[0];
-};
-
-/**
- * Instance address information
- */
-struct InstanceAddrInfo
-{
-  /**
-   * Array of addresses
-   */
-  struct InstanceAddr **addrs;
-
-  /**
-   * Number of addresses in the above array
-   */
-  unsigned int naddrs;
-  
-  /**
-   * The MPI id of the instance to whom these addresses belong to
-   */
-  unsigned int source;
-};
-
-
-/**
  * Context for verifying addresses
  */
 struct VerifyAddressesCtx
@@ -75,11 +37,16 @@
   struct Task *close_task;
 
   /**
-   * The index of the address being verified in association with this context
+   * the instance's address being verified in association with this context
    */
-  unsigned int naddr;
+  struct sockaddr *saddr;
 
   /**
+   * The length of the above address
+   */
+  socklen_t addrlen;
+
+  /**
    * The socket file descriptor associated with the connection used to verify
    * the address 
    */
@@ -357,23 +324,6 @@
 
 
 /**
- * Free an instance's address information
- *
- * @param iainfos the instance address information
- */
-static void
-free_instance_addresses (struct InstanceAddrInfo *iainfos)
-{
-  unsigned int cnt;
-  
-  for (cnt = 0; cnt < iainfos->naddrs; cnt++)
-    free (iainfos->addrs[cnt]);
-  free (iainfos->addrs);
-  free (iainfos);
-}
-
-
-/**
  * Callback trigger to finalise a round
  *
  * @param sock -1 do not use this
@@ -401,7 +351,7 @@
     free (ctx);
   }
   for (cnt = 0; cnt < rwidth; cnt++)
-    free_instance_addresses (riainfos[cnt]);
+    instance_address_info_destroy (riainfos[cnt]);
   if (IS_SHUTDOWN_EVENT (flags))
     return;
   MSH_close (listen_sock);
@@ -451,7 +401,7 @@
   lb = rank - round * rwidth - rwidth + nproc;
   MSH_assert (0 <= lb);
   lb %= nproc;
-  source = ctx->iainfo->source;
+  source = instance_address_get_rank (ctx->iainfo);
   if (lb <= source)
     off = source - lb;
   else
@@ -472,7 +422,6 @@
 socket_open_cb (int sockfd, void *cls)
 {
   struct VerifyAddressesCtx *ctx = cls;
-  struct InstanceAddr *iaddr;
   uint32_t id;
 
   ctx->soh = NULL;
@@ -482,9 +431,8 @@
     /* FIXME: Check if we already got a mapping for the instance */
     goto err_ret;
   }
-  iaddr = ctx->iainfo->addrs[ctx->naddr];
   LOG_DEBUG ("%d: Opened a connection to %s\n", rank, 
-             saddr2str (iaddr->saddr, iaddr->addrlen));
+             saddr2str (ctx->saddr, ctx->addrlen));
   ctx->sock = sockfd;
   id = htonl ((uint32_t) rank);
   if (sizeof (uint32_t) != write (sockfd, &id, sizeof (uint32_t)))
@@ -503,6 +451,28 @@
 }
 
 
+static int
+address_iterator_cb (void *cls, struct sockaddr *saddr, socklen_t addrlen)
+{
+  struct VerifyAddressesCtx *ctx;
+  struct InstanceAddrInfo *iainfo = cls;
+
+  ctx = MSH_malloc (sizeof (struct VerifyAddressesCtx));
+  ctx->soh = scheduler_open_socket (saddr, addrlen, &socket_open_cb, ctx);
+  ctx->iainfo = iainfo;
+  ctx->saddr = saddr;
+  ctx->addrlen = addrlen;
+  ctx->sock = -1;
+  if (NULL == ctx->soh)
+  {
+    MSH_break (0);
+    free (ctx);
+    return MSH_SYSERR;
+  }
+  DLL_insert_tail (vactx_head, vactx_tail, ctx);
+}
+
+
 /**
  * Verify the addresses of an instance by connecting to the instance's listen
  * socket
@@ -515,27 +485,13 @@
 static int
 verify_addresses (struct InstanceAddrInfo *iainfo)
 {
-  struct VerifyAddressesCtx *ctx;
+  
   struct InstanceAddr *iaddr;
-  unsigned int cnt;
-
-  for (cnt = 0; cnt < iainfo->naddrs; cnt++)
-  {
-    iaddr = iainfo->addrs[cnt];
-    ctx = MSH_malloc (sizeof (struct VerifyAddressesCtx));
-    ctx->naddr = cnt;
-    ctx->soh = scheduler_open_socket (iaddr->saddr, iaddr->addrlen,
-                                      &socket_open_cb, ctx);
-    ctx->iainfo = iainfo;
-    ctx->sock = -1;
-    if (NULL == ctx->soh)
-    {
-      MSH_break (0);
-      free (ctx);
-      return MSH_SYSERR;
-    }
-    DLL_insert_tail (vactx_head, vactx_tail, ctx);
-  }
+  
+  if (MSH_OK != instance_address_info_iterate_addresses (iainfo,
+                                                         &address_iterator_cb,
+                                                         iainfo))
+    return MSH_SYSERR;
   return MSH_OK;
 }
 
@@ -550,12 +506,9 @@
 static struct InstanceAddrInfo *
 parse_verify_address_msg (struct MSH_MSG_VerifyAddress *msg, int source)
 {
-  struct InstanceAddr *addr;
-  struct sockaddr_in *inaddr;
+  struct InstanceAddr *iaddr;
   struct InstanceAddrInfo *iainfo;
   size_t size;
-  socklen_t addrlen;
-  uint16_t port;  
   uint16_t nips;
   uint16_t cnt;
 
@@ -567,19 +520,12 @@
     LOG_ERROR ("Parsing failed\n");
     return NULL;
   }
-  iainfo = MSH_malloc (sizeof (struct InstanceAddrInfo));
-  iainfo->source = source;
+  iainfo = instance_address_info_create (source);
   for (cnt = 0; cnt < nips; cnt++)
   {
-    addrlen = sizeof (struct sockaddr_in); /* IPv4 */
-    addr = MSH_malloc (sizeof (struct InstanceAddr) + addrlen);
-    addr->addrlen = addrlen;
-    inaddr = (struct sockaddr_in *) addr->saddr;
-    inaddr->sin_family = AF_INET;
-    /* assign directly as address and port already in NB format */
-    inaddr->sin_port = msg->port;
-    inaddr->sin_addr.s_addr = (in_addr_t) msg->ipaddrs[cnt];
-    MSH_array_append (iainfo->addrs, iainfo->naddrs, addr);
+    iaddr = instance_address_create_sockaddr_in (msg->port,
+                                                (in_addr_t) msg->ipaddrs[cnt]);
+    instance_address_info_add_address (iainfo, iaddr);
   }
   return iainfo;
 }
@@ -653,7 +599,7 @@
   for (cnt=0; cnt < rwidth; cnt++)
   {
     if (NULL != iainfos[cnt])
-      free_instance_addresses (iainfos[cnt]);
+      instance_address_info_destroy (iainfos[cnt]);
   }
   free (iainfos);
   return NULL;

Modified: msh/src/mtypes.h
===================================================================
--- msh/src/mtypes.h    2013-07-12 11:22:50 UTC (rev 27952)
+++ msh/src/mtypes.h    2013-07-12 12:59:22 UTC (rev 27953)
@@ -18,6 +18,10 @@
   uint16_t size MSH_PACKED;
 };
 
+
+/**
+ * Message for sending addresses for verification
+ */
 struct MSH_MSG_VerifyAddress
 {
   /**
@@ -64,6 +68,49 @@
 };
 
 
+/**
+ * Message for signifying transmission of an address map
+ */
+struct MSH_MSG_AddressMap
+{
+  /**
+   * Header for this message
+   */
+  struct MSH_MessageHeader header;
+
+  /**
+   * Number of instances in this address map
+   */
+  uint16_t num MSH_PACKED;
+};
+
+
+/**
+ * Structure for representing verified addresses of an instance.  This does not
+ * denote a message but is used in @see MSH_MSG_AddressMap
+ */
+struct MSH_MSG_InstanceAdresses
+{  
+  /**
+   * The rank of the instance
+   */
+  uint16_t rank MSH_PACKED;
+
+  /**
+   * The number of addresses
+   */
+  uint16_t nips MSH_PACKED;
+
+  /**
+   * IPv4 addresses to follow as 32 bit unsigned integers
+   */
+  uint32_t ipaddrs[0];
+
+};
+
+
+
+
 /*********************************************************************
  * MPI tag numbers for each message type
  *********************************************************************/