From 0f1f9ac088cfdb3fe089e6df44f6d3ca6801a95c Mon Sep 17 00:00:00 2001
From: Christos Houtouridis <houtouridis.ch@gmail.com>
Date: Sun, 6 Oct 2019 15:18:19 +0300
Subject: [PATCH] Seeker now is seperate thread

---
 src/client.c   | 45 ++++++++++++++++++++++++----------
 src/client.h   |  2 +-
 src/core.c     | 65 +++++++++++++++++++++++++++++++++++++++++---------
 src/core.h     |  4 ++++
 src/listener.c | 11 ++++++---
 src/main.c     | 26 ++++++++++----------
 src/msg_impl.h | 24 ++++++++++---------
 7 files changed, 127 insertions(+), 50 deletions(-)

diff --git a/src/client.c b/src/client.c
index bdb623e..993a15e 100644
--- a/src/client.c
+++ b/src/client.c
@@ -46,7 +46,11 @@ static bool_t ping (devAEM_t dev) {
  */
 static size_t seeker (void) {
    size_t cnt =0;                         // count devices on range
+
+   sleep (settings.seekerInterval);
+
    log_debug ("Debug: Pinging devices...\n");
+   devList_acquire ();
    for (int i=0 ; i<AEMLIST_SIZE ; ++i) {
       if (devList[i].dev == settings.me) {// Don't ping me, I'm not here, go away...
          devList[i].onRange = false;
@@ -59,11 +63,16 @@ static size_t seeker (void) {
          if (!devList[i].begin)
             devList[i].begin = time(NULL);   // first time only
          devList[i].end = time(NULL);        // every time
-         log_debug ("Debug: Device %u found\n", devList[i].dev);
+         devList_t cDev = devList[i];
+         devList_release ();
+         log_debug ("Debug: Device %u found\n", cDev.dev);
+         devList_acquire ();
       }
       else
          devList[i].onRange = false;      // Where is everybody?
    }
+   statsTimesPrint (devList);
+   devList_release ();
    log_debug ("Debug: %d devices found\n", cnt);
    return cnt;
 }
@@ -209,32 +218,31 @@ static void client (void) {
       msgList_release ();
       log_debug ("Debug: Message added to msgList at %d\n", at);
 
-      if (!seeker ())                           // If we are alone skip the rest
-         continue;
-
-      log_debug ("Debug: Devices found on range\n");
-
       msgList_acquire ();
       mIter_t it = msgList_begin (&msgList);    // get a message iterator
       // for each message -> for each recipient
       for (size_t i=0 ; i<msgList_size(&msgList) ; ++i, msgList_preInc (&it)) {
          for (size_t j=0 ; j<AEMLIST_SIZE ; ++j) {
+            // get current dev instance
+            devList_acquire ();
+            devList_t currentDev = devList[j];
+            devList_release ();
             // check when to send
-            if (devList[j].onRange              // is on range
+            if (currentDev.onRange              // is on range
              && !msgList.m[it].recipients[j]    // we haven't send the message to that device
              && msgList.m[it].cMsg.to != settings.me // the message it's not for me
             ) {
-               if (sendMsg (devList[j].dev, &msgList.m[it].cMsg)) {
+               if (sendMsg (currentDev.dev, &msgList.m[it].cMsg)) {
                   msgList.m[it].recipients[j] = true;
                   msgList_release ();
-                  statsUpdateOut (&msg, devList[j].dev);
-                  log_msg_out (&msg, devList[j].dev);
-                  log_debug ("Debug: Send message to device %u succeed\n", devList[j].dev);
+                  statsUpdateOut (&msg, currentDev.dev);
+                  log_msg_out (&msg, currentDev.dev);
+                  log_debug ("Debug: Send message to device %u succeed\n", currentDev.dev);
                   msgList_acquire ();
                }
                else {
                   msgList_release ();
-                  log_debug ("Debug: Send message to device %u failed\n", devList[j].dev);
+                  log_debug ("Debug: Send message to device %u failed\n", currentDev.dev);
                   msgList_acquire ();
                }
                //^ we try to send the message and mark the transmission on success
@@ -249,6 +257,19 @@ static void client (void) {
    return;
 }
 
+/*!
+ * pthread wrapper for \sa seeker()
+ * @param ptr
+ */
+void* pthSeeker (void* ptr) {
+   (void)&ptr;    // use parameter
+
+   while (true)
+      seeker ();
+   exit(1);       // we should not be here
+   return NULL;
+}
+
 /*!
  * pthread wrapper for \sa client()
  * @param ptr
diff --git a/src/client.h b/src/client.h
index 66e16b1..92c525e 100644
--- a/src/client.h
+++ b/src/client.h
@@ -10,7 +10,7 @@
 #include "core.h"
 #include "msg_impl.h"
 
-
+void* pthSeeker (void* ptr);
 void* pthClient (void* ptr);
 
 #endif /* __client_h__ */
diff --git a/src/core.c b/src/core.c
index 85730f1..1447e9f 100644
--- a/src/core.c
+++ b/src/core.c
@@ -23,6 +23,7 @@ msgList_t msgList;   //!< The message list for our application.
  * Local data types
  */
 static pthread_mutex_t lock_msgList;    //!< mutex for msgList locking
+static pthread_mutex_t lock_devList;    //!< mutex for devList locking
 static pthread_mutex_t lock_stderr;     //!< mutex for stderr locking
 static pthread_mutex_t lock_stdout;     //!< mutex for stderr locking
 static pthread_mutex_t lock_stats;      //!< mutex for stats locking
@@ -366,18 +367,26 @@ void msg_init (msg_t* msg) {
 //! @}
 
 
-//! msgList API
+//! devList API
 //! @{
 
-/*! Macro helper to saturate increased values */
-#define _top_saturate(test, apply, value) do { \
-   if (test >= value)   apply = value; \
-} while (0)
+/*!
+ * Initialize the devList
+ * @param msgList    Pointer to mesList t initialize
+ * @return     The status of the operation
+ */
+status_t devList_init (devList_t* devList) {
+   devAEM_t l[] = AEMLIST;
+   if (pthread_mutex_init(&lock_devList, NULL) != 0) {
+       log_error ("Error: mutex init has failed\n");
+       return MSG_ERROR;
+   }
+   memset ((void*)devList, 0, sizeof(devList_t));
+   for (size_t i =0 ; i<AEMLIST_SIZE ; ++i)
+      devList[i].dev = l[i];
 
-/*! Macro helper to saturate decreased values */
-#define _btm_saturate(test, apply, value) do { \
-   if (test < value)   apply = value; \
-while (0)
+   return MSG_OK;
+}
 
 /*!
  * Returns an iterator for \sa devList AND \sa msg_t.recipients
@@ -392,6 +401,26 @@ dIter_t devList_getIter (devAEM_t dev) {
    return -1;  // return end()
 }
 
+//! Acquires devList resources
+void devList_acquire (void) { pthread_mutex_lock(&lock_devList); }
+//! Releases devList resources
+void devList_release (void) { pthread_mutex_unlock(&lock_devList); }
+
+//! @}
+
+//! msgList API
+//! @{
+
+/*! Macro helper to saturate increased values */
+#define _top_saturate(test, apply, value) do { \
+   if (test >= value)   apply = value; \
+} while (0)
+
+/*! Macro helper to saturate decreased values */
+#define _btm_saturate(test, apply, value) do { \
+   if (test < value)   apply = value; \
+while (0)
+
 /*!
  * Initialize the msgList
  * @param msgList    Pointer to mesList t initialize
@@ -643,14 +672,28 @@ status_t statsPrint (stats_t* stats) {
    fprintf (fp, "Out direct messages: %d\n", stats->outDirectMsg);
    fprintf (fp, "Average message size: %g\n", stats->avMsgSize);
    fprintf (fp, "Average time to me: %g\n", stats->avTimeToMe);
+   fclose (fp);
+   return MSG_OK;
+}
 
+/*!
+ * Device online timing print functionality
+ * @param devList Pointer to devList to print
+ * @return        The status of the operation
+ */
+status_t statsTimesPrint (devList_t *devList) {
+   FILE* fp = fopen ("devices.txt", "w");
+   if (fp == NULL) {
+      fclose (fp);
+      return MSG_ERROR;
+   }
+   fprintf (fp, "\n Device timings\n================\n");
    for (size_t i =0 ; i<AEMLIST_SIZE ; ++i) {
-      fprintf (fp, "   Device %u found on %lld, last: %lld\n",
+      fprintf (fp, "Device %u found on %lld, last: %lld\n",
          devList[i].dev, devList[i].begin, devList[i].end);
    }
    fclose (fp);
    return MSG_OK;
 }
-
 //! @}
 
diff --git a/src/core.h b/src/core.h
index d455492..11a367e 100644
--- a/src/core.h
+++ b/src/core.h
@@ -29,7 +29,10 @@ bool_t    cMsg_equal (cMsg_t* m1, cMsg_t* m2);
 
 void      msg_init (msg_t* msg);
 
+status_t  devList_init (devList_t* devList);
 dIter_t   devList_getIter (devAEM_t dev);
+void      devList_acquire (void);
+void      devList_release (void);
 
 mIter_t   msgList_preInc (mIter_t* it);
 mIter_t   msgList_preDec (mIter_t* it);
@@ -47,6 +50,7 @@ void     statsUpdateCreate (msg_t* msg);
 void     statsUpdateIn (msg_t* msg, bool_t dup);
 void     statsUpdateOut (msg_t* msg, devAEM_t dev);
 status_t statsPrint (stats_t* stats);
+status_t statsTimesPrint (devList_t *devList);
 
 status_t log_init(void);
 void     log_msg_in (msg_t* msg);
diff --git a/src/listener.c b/src/listener.c
index edee376..af7693a 100755
--- a/src/listener.c
+++ b/src/listener.c
@@ -43,14 +43,19 @@ static void listen_handler (devAEM_t dev, char_t* buffer, size_t size) {
    // We have a copy
       msgList_release ();
       statsUpdateIn (&msg, true);   // message process
-      log_debug("Debug: Duplicate message from: %d\n", msg.sender);
+      log_debug ("Debug: Duplicate message from: %d\n", msg.sender);
    }
    // Processing...
+   devList_acquire();
+   dIter_t d = devList_getIter (dev);
+   dIter_t f = devList_getIter (msg.cMsg.from);
+   devList_release();
+
    msgList_acquire ();
    // Do not echo message to sender, he already has it
-   msgList.m[myCopy].recipients[devList_getIter (dev)] = true;
+   msgList.m[myCopy].recipients[d] = true;
    // don't push back message to creator, he already has it
-   msgList.m[myCopy].recipients[devList_getIter (msg.cMsg.from)] = true;
+   msgList.m[myCopy].recipients[f] = true;
    msgList_release ();
 }
 
diff --git a/src/main.c b/src/main.c
index fc13a2e..c277173 100644
--- a/src/main.c
+++ b/src/main.c
@@ -1,6 +1,6 @@
 /*!
  * \file main.c
- * This is the main file of the RTES final task
+ * This is the main file of the RTES final task.
  *
  * \author Christos Choutouridis AEM:8997 <cchoutou@ece.auth.gr>
  */
@@ -21,24 +21,23 @@
  */
 //! @{
 
-settings_t  settings_init (settings);
-devList_t   devList_init(devList[AEMLIST_SIZE]);
-stats_t     stats;
+settings_t  settings_init (settings);  //!< Application settings
+devList_t   devList[AEMLIST_SIZE];     //!< Device list
+stats_t     stats;                     //!< Statistical data
 
 //! @}
 
 /*!
  * CLI short options
  */
-const char *short_opt = "li:v:p:s:w:th";
+const char *short_opt = "v:i:p:s:w:th";
 
 /*!
  * CLI long options
  */
 const struct option long_opt[] = {
-   {"port",       required_argument,   NULL, 'l'},
-   {"interval",   required_argument,   NULL, 'i'},
    {"outlevel",   required_argument,   NULL, 'v'},
+   {"interval",   required_argument,   NULL, 'i'},
    {"pingtimeout",required_argument,   NULL, 'p'},
    {"sendtimeout",required_argument,   NULL, 's'},
    {"who",        required_argument,   NULL, 'w'},
@@ -50,7 +49,7 @@ const struct option long_opt[] = {
 /*!
  * \brief
  *    Parse input argument and fill the kcli_input_t struct
- * \param   in    Pointer to \ref kcli_input_t structure to fill
+ * \param   s     Pointer to settings_t data to fill
  * \param   argc  The argument count as passed to the main()
  * \param   argv  Argument array as passed to the main()
  * \return  The status of the operation
@@ -65,13 +64,12 @@ int parse_args (settings_t *s, int argc, char const *argv[]) {
          case -1:       /* no more arguments */
          case 0:        /* long options toggles */
             break;
-         case 'l':   s->port = atoi(optarg); break;
-         case 'i':   s->msgInterval = atoi (optarg); break;
          case 'v':
             s->outLevel = atoi (optarg);
             if (s->outLevel >= OUTLEVEL_2)   s->outLevel = OUTLEVEL_2;
             if (s->outLevel < OUTLEVEL_0)    s->outLevel = OUTLEVEL_0;
             break;
+         case 'i':   s->seekerInterval = atoi (optarg); break;
          case 'p':   s->pingTimeout = atoi (optarg); break;
          case 's':   s->sendTimeout.tv_sec = atoi (optarg); break;
          case 'w':   s->me = atoi (optarg); break;
@@ -89,22 +87,26 @@ int parse_args (settings_t *s, int argc, char const *argv[]) {
 }
 
 
-int main(int argc, char const *argv[])  {
+int main (int argc, char const *argv[])  {
 
+   // get command line arguments
    parse_args (&settings, argc, argv);
 
    // Initialize all subsystems
    log_init ();
    stats_init (&stats);
+   devList_init (devList);
    msgList_init (&msgList);
 
    // Create threads
-   pthread_t ptL, ptC;
+   pthread_t ptL, ptS, ptC;
    pthread_create (&ptL, NULL, pthListener, NULL);
+   pthread_create (&ptS, NULL, pthSeeker, NULL);
    pthread_create (&ptC, NULL, pthClient, NULL);
 
    // block here
    pthread_join (ptL, NULL);
+   pthread_join (ptS, NULL);
    pthread_join (ptC, NULL);
    return 0;
 }
diff --git a/src/msg_impl.h b/src/msg_impl.h
index c782a49..2db5677 100644
--- a/src/msg_impl.h
+++ b/src/msg_impl.h
@@ -21,17 +21,17 @@
  */
 #define AEMLIST_SIZE    (10)
 
-#define devList_init(l) l = { \
-   { 7700, 0, 0, 0}, \
-   { 8261, 0, 0, 0}, \
-   { 8765, 0, 0, 0}, \
-   { 8844, 0, 0, 0}, \
-   { 8880, 0, 0, 0}, \
-   { 8861, 0, 0, 0}, \
-   { 8877, 0, 0, 0}, \
-   { 8941, 0, 0, 0}, \
-   { 8934, 0, 0, 0}, \
-   { 8997, 0, 0, 0}  \
+#define AEMLIST { \
+   7700, \
+   8261, \
+   8765, \
+   8844, \
+   8880, \
+   8861, \
+   8877, \
+   8941, \
+   8934, \
+   8997  \
 }
 
 /*!
@@ -217,6 +217,7 @@ typedef enum {
 typedef struct {
    devAEM_t          me;
    uint16_t          port;
+   time_t            seekerInterval;
    time_t            msgInterval;
    time_t            msgRand;
    outLevel_en       outLevel;
@@ -230,6 +231,7 @@ extern settings_t settings;
 #define settings_init(s) s = {         \
    .me = 8997,                         \
    .port = 2288,                       \
+   .seekerInterval = 30,               \
    .msgInterval = 60,                  \
    .msgRand = 240,                     \
    .outLevel = OUTLEVEL_1,             \