[GNUnet-SVN] r27737 - in msh: . src

[gnunet]

Author: harsha
Date: 2013-07-03 17:53:47 +0200 (Wed, 03 Jul 2013)
New Revision: 27737

Added:
   msh/src/scheduler.c
   msh/src/scheduler.h
   msh/src/test_scheduler.c
Modified:
   msh/
   msh/src/
   msh/src/Makefile.am
   msh/src/common.h
   msh/src/mshd.c
   msh/src/util.c
   msh/src/util.h
Log:
scheduler event loop based on libevent

Index: msh
===================================================================
--- msh 2013-07-03 15:20:12 UTC (rev 27736)
+++ msh 2013-07-03 15:53:47 UTC (rev 27737)

Property changes on: msh
___________________________________________________________________
Modified: svn:ignore
## -12,3 +12,7 ##
 aclocal.m4
 Makefile
 install-sh
+test-driver
+GRTAGS
+GPATH
+GTAGS
Index: msh/src
===================================================================
--- msh/src     2013-07-03 15:20:12 UTC (rev 27736)
+++ msh/src     2013-07-03 15:53:47 UTC (rev 27737)

Property changes on: msh/src
___________________________________________________________________
Modified: svn:ignore
## -3,4 +3,7 ##
 .deps
 mshd
 mping
-
+test-suite.log
+test-scheduler
+test-scheduler.log
+test-scheduler.trs
Modified: msh/src/Makefile.am
===================================================================
--- msh/src/Makefile.am 2013-07-03 15:20:12 UTC (rev 27736)
+++ msh/src/Makefile.am 2013-07-03 15:53:47 UTC (rev 27737)
@@ -3,3 +3,13 @@
 mping_SOURCES = mping.c
 
 mshd_SOURCES = mshd.c util.c util.h
+mshd_LDADD = -levent
+
+check_PROGRAMS = \
+  test-scheduler
+
+test_scheduler_SOURCES = test_scheduler.c scheduler.c scheduler.h common.h
+test_scheduler_LDADD = -levent
+
+TESTS = \
+  test-scheduler

Modified: msh/src/common.h
===================================================================
--- msh/src/common.h    2013-07-03 15:20:12 UTC (rev 27736)
+++ msh/src/common.h    2013-07-03 15:53:47 UTC (rev 27737)
@@ -20,10 +20,12 @@
 #include <net/if.h>
 #include <ifaddrs.h>
 #include <netdb.h>
-
+#include <signal.h>
 #include <unistd.h>
 
+#include <event2/event.h>
 
+
 #define MSH_OK 1
 
 #define MSH_SYSERR -1
@@ -43,6 +45,11 @@
 
 
 /**
+ * Always allocate and clear memory
+ */
+#define MSH_malloc(size) calloc (1, size);
+
+/**
  * Use this for internal assertion violations that are
  * not fatal (can be handled) but should not occur.
  */
@@ -63,3 +70,132 @@
  * Free if the given pointer is not NULL
  */
 #define MSH_free_non_null(ptr) do { if (NULL != (ptr)) free (ptr); } while(0)
+
+
+/* ******************** doubly-linked list *************** */
+/* To avoid mistakes: head->prev == tail->next == NULL     */
+
+/**
+ * Insert an element at the head of a DLL. Assumes that head, tail and
+ * element are structs with prev and next fields.
+ *
+ * @param head pointer to the head of the DLL
+ * @param tail pointer to the tail of the DLL
+ * @param element element to insert
+ */
+#define DLL_insert(head,tail,element) do { \
+  MSH_assert ( ( (element)->prev == NULL) && ((head) != (element))); \
+  MSH_assert ( ( (element)->next == NULL) && ((tail) != (element))); \
+  (element)->next = (head); \
+  (element)->prev = NULL; \
+  if ((tail) == NULL) \
+    (tail) = element; \
+  else \
+    (head)->prev = element; \
+  (head) = (element); } while (0)
+
+
+/**
+ * Insert an element at the tail of a DLL. Assumes that head, tail and
+ * element are structs with prev and next fields.
+ *
+ * @param head pointer to the head of the DLL
+ * @param tail pointer to the tail of the DLL
+ * @param element element to insert
+ */
+#define DLL_insert_tail(head,tail,element) do { \
+  MSH_assert ( ( (element)->prev == NULL) && ((head) != (element))); \
+  MSH_assert ( ( (element)->next == NULL) && ((tail) != (element))); \
+  (element)->prev = (tail); \
+  (element)->next = NULL; \
+  if ((head) == NULL) \
+    (head) = element; \
+  else \
+    (tail)->next = element; \
+  (tail) = (element); } while (0)
+
+
+/**
+ * Insert an element into a DLL after the given other element.  Insert
+ * at the head if the other element is NULL.
+ *
+ * @param head pointer to the head of the DLL
+ * @param tail pointer to the tail of the DLL
+ * @param other prior element, NULL for insertion at head of DLL
+ * @param element element to insert
+ */
+#define DLL_insert_after(head,tail,other,element) do { \
+  MSH_assert ( ( (element)->prev == NULL) && ((head) != (element))); \
+  MSH_assert ( ( (element)->next == NULL) && ((tail) != (element))); \
+  (element)->prev = (other); \
+  if (NULL == other) \
+    { \
+      (element)->next = (head); \
+      (head) = (element); \
+    } \
+  else \
+    { \
+      (element)->next = (other)->next; \
+      (other)->next = (element); \
+    } \
+  if (NULL == (element)->next) \
+    (tail) = (element); \
+  else \
+    (element)->next->prev = (element); } while (0)
+
+
+/**
+ * Insert an element into a DLL before the given other element.  Insert
+ * at the tail if the other element is NULL.
+ *
+ * @param head pointer to the head of the DLL
+ * @param tail pointer to the tail of the DLL
+ * @param other prior element, NULL for insertion at head of DLL
+ * @param element element to insert
+ */
+#define DLL_insert_before(head,tail,other,element) do { \
+  MSH_assert ( ( (element)->prev == NULL) && ((head) != (element))); \
+  MSH_assert ( ( (element)->next == NULL) && ((tail) != (element))); \
+  (element)->next = (other); \
+  if (NULL == other) \
+    { \
+      (element)->prev = (tail); \
+      (tail) = (element); \
+    } \
+  else \
+    { \
+      (element)->prev = (other)->prev; \
+      (other)->prev = (element); \
+    } \
+  if (NULL == (element)->prev) \
+    (head) = (element); \
+  else \
+    (element)->prev->next = (element); } while (0)
+
+
+/**
+ * Remove an element from a DLL. Assumes that head, tail and
+ * element point to structs with prev and next fields.
+ *
+ * Using the head or tail pointer as the element
+ * argument does NOT work with this macro.
+ * Make sure to store head/tail in another pointer
+ * and use it to remove the head or tail of the list.
+ *
+ * @param head pointer to the head of the DLL
+ * @param tail pointer to the tail of the DLL
+ * @param element element to remove
+ */
+#define DLL_remove(head,tail,element) do { \
+  MSH_assert ( ( (element)->prev != NULL) || ((head) == (element))); \
+  MSH_assert ( ( (element)->next != NULL) || ((tail) == (element))); \
+  if ((element)->prev == NULL) \
+    (head) = (element)->next;  \
+  else \
+    (element)->prev->next = (element)->next; \
+  if ((element)->next == NULL) \
+    (tail) = (element)->prev;  \
+  else \
+    (element)->next->prev = (element)->prev; \
+  (element)->next = NULL; \
+  (element)->prev = NULL; } while (0)

Modified: msh/src/mshd.c
===================================================================
--- msh/src/mshd.c      2013-07-03 15:20:12 UTC (rev 27736)
+++ msh/src/mshd.c      2013-07-03 15:53:47 UTC (rev 27737)
@@ -1,6 +1,7 @@
 #include "common.h"
 #include <mpi.h>
 #include "util.h"
+#include "scheduler.h"
 
 /**
  * The port number of our local socket
@@ -38,6 +39,10 @@
  */
 static unsigned int rwidth;
 
+/**
+ * event base for libevent
+ */
+static struct event_base *ebase;
 
 /**
  * Select loop for a socket
@@ -49,52 +54,7 @@
 static int
 sock_select (int sock, long timeout)
 {
-  struct sockaddr_in addr;
-  socklen_t addrlen;
-  struct timeval tv;
-  fd_set fdset;
-  int ret;
-  int nsock;
-  unsigned int cnt;
-  
-  tv.tv_sec = 0;
-  tv.tv_usec = timeout;
-  do {
-    FD_ZERO (&fdset);
-    FD_SET (sock, &fdset);
-    ret = select (sock+1, fdset, NULL, NULL, &tv);
-    if (-1 == ret)
-    {
-      switch (errno)
-      {
-      case EBADF:
-        MSH_assert (0);
-      case EINVAL:
-        MSH_assert (0);
-      case ENOMEM:
-        MSH_assert (0);
-      case EINTR:
-        return MSH_SYSERR;
-      default:
-        LOG_STRERROR ("select");
-        MSH_assert (0);
-      }
-    }
-    if (0 == ret)
-      break;
-    MSH_assert (FD_ISSET (sock, &fdset));
-    addrlen = sizeof (addr);
-    nsock = accept (sock, &addr, &addrlen);
-    if (nsock < 0)
-    {
-      LOG_STRERROR ("accept");
-      return MSH_SYSERR;
-    }    
-    MSH_assert (sizeof (addr) == addrlen);
-    (void) close (nsock);
-    
-  } while ()
-  
+  return MSH_SYSERR;
 }
 
 
@@ -190,8 +150,8 @@
     goto clo_ret;
   }
   for (cnt = 0; cnt < rwidth; cnt++)
-    send_addresses ((round * rwidth) + cnt);
-  sock_select (sock, timeout);
+    //send_addresses ((round * rwidth) + cnt);
+  sock_select (sock, 0);
   if (MPI_SUCCESS != MPI_Barrier (MPI_COMM_WORLD))
   {
     MSH_break (0);
@@ -205,9 +165,50 @@
 }
 
 
+/**
+ * Event callback for the first running task
+ *
+ * @param nosock we have no sockets associated with this callback
+ * @param flags EV_* flags
+ * @param cls NULL
+ */
+static void
+run (evutil_socket_t nosock, short flags, void *cls)
+{
+  LOG_DEBUG ("run task ran\n");
+  for (; round < nproc; round++)
+  {
+  }
+}
+
+#define N_SHUTDOWN_SIGNALS 2  /* SIGTERM, SIGINT */
+static struct event *ev_sigs[N_SHUTDOWN_SIGNALS];
+
+
+/**
+ * Event callback for the first running task
+ *
+ * @param signal the signal
+ * @param flags EV_* flags
+ * @param cls NULL
+ */
+static void
+sig_shutdown (evutil_socket_t signal, short flags, void *cls)
+{
+  struct event **ev_sig = cls;
+  unsigned int cnt;
+
+  *ev_sig = NULL;
+}
+
+
 int 
 main (int argc, char **argv)
 {
+  struct event *ev_run;
+  int ret;
+
+  ret = 1;
   if (MPI_SUCCESS != MPI_Init(&argc, &argv))
   {
     LOG_ERROR ("Failed to initialise MPI\n");
@@ -223,15 +224,26 @@
     LOG_ERROR ("Cannot determine our MPI rank\n");
     goto fail;
   }
-  GNUNET_OS_network_interfaces_list (&net_if_processor, NULL);
+  GNUNET_OS_network_interfaces_list (&net_if_processor, NULL);  
   if (0 == nips)
     LOG_ERROR ("No IP addresses found\n");
-  for (; rount < nproc; round++)
-    if (MSH_OK != verify_addresses ())
-      goto fail;
-
+  ebase = event_base_new ();
+  ev_run = evtimer_new (ebase, &run, NULL);
+  evtimer_add (ev_run, TV_IMMEDIATE);
+  if (0 != event_base_dispatch (ebase))
+  {
+    evtimer_del (ev_run);
+    LOG_ERROR ("Event loop dispatch error\n");
+    goto fail;
+  }
+  evtimer_del (ev_run);
+  ev_run = NULL;
+  ret = 0;
  fail:
   MSH_break (MPI_SUCCESS == MPI_Finalize());
   MSH_free_non_null (ip_addr_str);
-  return 1;
+  if (NULL != ebase) 
+    event_base_free (ebase);
+  //libevent_global_shutdown ();
+  return ret;
 }

Added: msh/src/scheduler.c
===================================================================
--- msh/src/scheduler.c                         (rev 0)
+++ msh/src/scheduler.c 2013-07-03 15:53:47 UTC (rev 27737)
@@ -0,0 +1,114 @@
+/**
+ * @file scheduler.c
+ * @brief task scheduler based on libevent
+ * @author Sree Harsha Totakura <address@hidden> 
+ */
+
+#include "scheduler.h"
+
+
+struct Task
+{
+  /**
+   * DLL next
+   */
+  struct Task *next;
+
+  /**
+   * DLL prev
+   */
+  struct Task *prev;
+  
+  struct event *ev;
+};
+
+/**
+ * Head for the DLL
+ */
+static struct Task *thead;
+
+/**
+ * Tail for the DLL
+ */
+static struct Task *ttail;
+
+
+static struct event_base *ebase;
+
+
+struct Task *
+scheduler_add_socket (evutil_socket_t sock, short flags, event_callback_fn cb, 
+                      void *cls, const struct timeval *tv)
+{
+  struct Task *task;
+
+  MSH_assert (NULL != ebase);
+  task = MSH_malloc (sizeof (struct Task));
+  task->ev = event_new (ebase, sock, flags, cb, cls);
+  if (0 != event_add (task->ev, tv))
+  {
+    free (task);
+    return NULL;
+  }
+  DLL_insert_tail (thead, ttail, task);
+  return task;
+}
+
+
+struct Task *
+scheduler_add (event_callback_fn cb, void *cls, const struct timeval *tv)
+{
+  return scheduler_add_socket (-1, 0, cb, cls, tv);
+}
+
+
+struct Task *
+scheduler_add_signal (int signal, event_callback_fn cb, void *cls, 
+                      const struct timeval *tv)
+{
+  return scheduler_add_socket (signal, EV_SIGNAL, cb, cls, tv);
+}
+
+
+void
+scheduler_remove (struct Task *task)
+{
+  DLL_remove (thead, ttail, task);
+  MSH_break (0 == event_del (task->ev));
+  event_free (task->ev);
+  free (task);
+}
+
+void
+scheduler_shutdown ()
+{
+  struct Task *task;
+
+  for (task = thead; NULL != task; task = task->next)
+  {
+    event_active (task->ev, EV_READ | EV_WRITE | EV_TIMEOUT, 0);
+  }
+}
+
+
+int
+scheduler_run (event_callback_fn cb, void *cls)
+{
+  struct Task *task;
+  struct event *sev;
+  int ret;
+
+  ebase = event_base_new ();
+  if (NULL == ebase)
+  {
+    LOG_ERROR ("Cannot allocate libevent event base\n");
+    return MSH_SYSERR;
+  }
+  sev = evtimer_new (ebase, cb, cls);
+  evtimer_add (sev, TV_IMMEDIATE);
+  ret = event_base_dispatch (ebase);
+  evtimer_del (sev);
+  event_free (sev);
+  event_base_free (ebase);  
+  return (1 == ret) ? MSH_OK : MSH_SYSERR;
+}

Added: msh/src/scheduler.h
===================================================================
--- msh/src/scheduler.h                         (rev 0)
+++ msh/src/scheduler.h 2013-07-03 15:53:47 UTC (rev 27737)
@@ -0,0 +1,31 @@
+#include "common.h"
+#include "event2/event.h"
+
+static struct timeval tv_immediate;
+
+#define TV_IMMEDIATE &tv_immediate
+
+#define is_shutdown_event(flags) ((flags & (EV_READ | EV_WRITE | EV_TIMEOUT)) 
== (EV_READ | EV_WRITE | EV_TIMEOUT))
+
+struct Task;
+
+struct Task *
+scheduler_add_socket (evutil_socket_t sock, short flags, event_callback_fn cb, 
+                      void *cls, const struct timeval *tv);
+
+struct Task *
+scheduler_add (event_callback_fn cb, void *cls, const struct timeval *tv);
+
+
+struct Task *
+scheduler_add_signal (int signal, event_callback_fn cb, void *cls, 
+                      const struct timeval *tv);
+
+void
+scheduler_remove (struct Task *task);
+
+void
+scheduler_shutdown ();
+
+int
+scheduler_run (event_callback_fn cb, void *cls);

Added: msh/src/test_scheduler.c
===================================================================
--- msh/src/test_scheduler.c                            (rev 0)
+++ msh/src/test_scheduler.c    2013-07-03 15:53:47 UTC (rev 27737)
@@ -0,0 +1,62 @@
+#include "common.h"
+#include "scheduler.h"
+
+struct Task *task;
+
+struct Task *sig_tasks[2];
+
+/**
+ * Event callback for a signal
+ *
+ * @param sig the signal
+ * @param flags EV_* flags
+ * @param cls NULL
+ */
+static void
+shutdown_signal (evutil_socket_t sig, short flags, void *cls)
+{
+  struct Task **sig_task = cls;
+  
+  scheduler_remove (*sig_task);
+  if (is_shutdown_event (flags))
+    return;
+  LOG_DEBUG ("Got signal %d.  Shutting down.\n", sig);
+  scheduler_shutdown ();
+}
+
+
+/**
+ * Event callback for the first running task
+ *
+ * @param nosock we have no sockets associated with this callback
+ * @param flags EV_* flags
+ * @param cls NULL
+ */
+static void
+run (evutil_socket_t nosock, short flags, void *cls)
+{
+  static unsigned int cnt;
+  struct timeval tv;
+
+  MSH_assert (-1 == nosock);
+  MSH_assert (NULL == cls);
+  MSH_assert (0 != (EV_TIMEOUT & flags));
+  LOG_DEBUG ("Run\n");
+  if (0 == cnt++)
+  {
+    tv.tv_sec = 10;
+    tv.tv_usec = 0;
+    task = scheduler_add (&run, NULL, &tv);
+    sig_tasks[0] = scheduler_add_signal (SIGTERM, &shutdown_signal, 
&sig_tasks[0], NULL);
+    sig_tasks[1] = scheduler_add_signal (SIGINT, &shutdown_signal, 
&sig_tasks[1], NULL);
+    return;
+  }
+  scheduler_remove (task);
+  //if (!is_shutdown_event (flags))
+  scheduler_shutdown ();
+}
+
+int main (int argc, char *argv[])
+{
+  MSH_assert (MSH_OK == scheduler_run (run, NULL));
+}

Modified: msh/src/util.c
===================================================================
--- msh/src/util.c      2013-07-03 15:20:12 UTC (rev 27736)
+++ msh/src/util.c      2013-07-03 15:53:47 UTC (rev 27737)
@@ -71,8 +71,7 @@
   }
   else
   {
-    tmp = malloc (size);
-    memset (tmp, 0, size);      /* client code should not rely on this, 
though... */
+    MSH_malloc (size);
     if (*oldCount > newCount)
       *oldCount = newCount;     /* shrink is also allowed! */
     memcpy (tmp, *old, elementSize * (*oldCount));

Modified: msh/src/util.h
===================================================================
--- msh/src/util.h      2013-07-03 15:20:12 UTC (rev 27736)
+++ msh/src/util.h      2013-07-03 15:53:47 UTC (rev 27737)
@@ -4,6 +4,7 @@
 #include <sys/types.h>
 #include <sys/socket.h>
 
+
 /**
  * Callback function invoked for each interface found.
  *