rtes_messenger/src/core.c

/*!
 * \file core.c
 *
 * \author Christos Choutouridis AEM:8997 <cchoutou@ece.auth.gr>
 */


#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>

#include <pthread.h>

#include "core.h"

pthread_mutex_t lock_msgList;
pthread_mutex_t lock_stderr;
pthread_mutex_t lock_stdout;
pthread_mutex_t lock_stats;

//! Helper API
//! @{

#define _ADHOC_SUBNET(A, B, C, D)   (((A)<<24) | ((B)<<16) | ((C)<<8) | (D))

devAEM_t addr2devAEM (uint32_t in_addr) {
   return (in_addr & 0x000000FF) + ((in_addr >> 8) & 0x000000FF) * 100;
}

in_addr_t devAEM2addr (devAEM_t dev) {
   uint32_t add  = _ADHOC_SUBNET (ADHOC_NET_A, ADHOC_NET_B, ADHOC_NET_C, ADHOC_NET_D);
            add |= (dev % 100) & 0x000000FF;
            add |= ((dev / 100) & 0x000000FF) << 8;
   return add;
}

devAEM_t ip2AEM (devIP_t* ip) {
   return ip->C*100 + ip->D;
}

devIP_t AEM2ip (devAEM_t dev) {
   devIP_t ip = {
      .A =ADHOC_NET_A, .B=ADHOC_NET_B, .C=dev/100, .D=dev%100
   };
   return ip;
}

devIP_t addr2ip (in_addr_t in_addr) {
   devIP_t ip = {
      .A = ADHOC_NET_A,
      .B = ADHOC_NET_B,
      .C = (in_addr >> 8) & 0x000000FF,
      .D = in_addr & 0x000000FF
   };
   return ip;
}
//! @}

//! log API
//! @{

static char_t* _frm_msg_io  = "dev=%d, message: from=%d, to=%d, timestamp=%lld, text=%s\n";
static char_t* _frm_msg_new = "new message: from=%d, to=%d, timestamp=%lld, text=%s\n";
#define  _HEAD_SIZE  25

status_t log_init (void) {
   if (pthread_mutex_init(&lock_stderr, NULL) != 0) {
       fprintf (stderr, "Error %s: mutex init has failed\n", __FUNCTION__ );
       return MSG_ERROR;
   }
   if (pthread_mutex_init(&lock_stdout, NULL) != 0) {
      fprintf (stderr, "Error %s: mutex init has failed\n", __FUNCTION__ );
      return MSG_ERROR;
   }
   return MSG_OK;
}

void log_msg_io (msg_t* msg) {
   if (settings.outLevel >= OUTLEVEL_1) {
      char_t head[_HEAD_SIZE];
      memcpy (head, msg->cMsg.text, (_HEAD_SIZE-1));
      head[_HEAD_SIZE-1] = 0;
      pthread_mutex_lock(&lock_stdout);
      fprintf (stdout, _frm_msg_io,
         msg->sender,
         cMsg_getFrom (&msg->cMsg),
         cMsg_getTo (&msg->cMsg),
         cMsg_getTs (&msg->cMsg),
         head
      );
      fflush(stdout);
      pthread_mutex_unlock(&lock_stdout);
   }
}

void log_msg_new (msg_t* msg) {
   if (settings.outLevel >= OUTLEVEL_1) {
      char_t head[_HEAD_SIZE];
      memcpy (head, msg->cMsg.text, (_HEAD_SIZE-1));
      head[_HEAD_SIZE-1] = 0;
      pthread_mutex_lock(&lock_stdout);
      fprintf (stdout, _frm_msg_new,
         cMsg_getFrom (&msg->cMsg),
         cMsg_getTo (&msg->cMsg),
         cMsg_getTs (&msg->cMsg),
         head
      );
      fflush(stdout);
      pthread_mutex_unlock(&lock_stdout);
   }
}

void log_debug (const char *fmt, ...) {
   if (settings.outLevel >= OUTLEVEL_2) {
      va_list ap;
      va_start(ap, fmt);
      pthread_mutex_lock(&lock_stdout);
      vfprintf (stdout, fmt, ap);
      fflush(stdout);
      pthread_mutex_unlock(&lock_stdout);
      va_end(ap);
   }
}

void log_error (const char *fmt, ...) {
   va_list ap;
   va_start(ap, fmt);
   pthread_mutex_lock(&lock_stderr);
   vfprintf (stderr, fmt, ap);
   fflush(stderr);
   pthread_mutex_unlock(&lock_stderr);
   va_end(ap);
}


//! @}

//! cMsg_t API
//! @{


/*!
 * Make a new message
 * @param msg     Pointer to message to create
 */
void cMsg_make (cMsg_t* msg) {
   static int msgID =0;                            // unique msg ID
   msg->from = settings.me;                        // from me
   do {
      // randomly select recipient device
      msg->to = devList[rand() % AEMLIST_SIZE].dev;
   } while (msg->to == settings.me);
   msg->ts = time(NULL);

   // stream the first fields and take the quote text iterator
   sprintf (msg->text, "%s #%d", MESSAGE_BODY, msgID++);
}

size_t cMsg_cat (cMsg_t* msg, char_t* buffer) {
   return sprintf (buffer, "%d_%d_%lld_%s",
      msg->from,
      msg->to,
      msg->ts,
      msg->text
   );
}

char_t* _strtok (char_t* str, char_t c, size_t max) {
   static char_t* last = NULL;
   char_t* ret = str;

   // init last
   if (str != NULL)  last = str;

   // loop
   for (size_t i=0 ; i<max ; ++i) {
      if (*last == c) {
         *last++ = 0;
         return ret;
      }
      ++last;
   }
   return NULL;
}

/*!
 * Parse an incoming message
 *
 * @param cMsg       Pointer to cMsg object to store the parsed data
 * @param rawMsg     Pointer to raw message
 * @param size       The size f raw message buffer
 * @return           The status of the operation
 *    @arg  MSG_OK      Success
 *    @arg  MSG_ERROR   Parse failure, incoming message format error
 */
status_t cMsg_parse (cMsg_t* cMsg, char_t* rawMsg, size_t size) {

   // Check message integrity
   if (size > MSG_TEXT_SIZE)
      return MSG_ERROR;

   // Parse message
   char_t* rest = rawMsg;
   char_t* tok[4];
   bool_t done = true;

   for (size_t i =0; i < 3; ++i) {
      tok[i] = _strtok (rest, MSG_DELIMITER, size);

      if (tok[i] == NULL) {
         done = false;
         break;
      }
      else {
         int l = strlen(rest);
         rest += l + 1;
         size -= l + 1;
      }
   }
   tok[3] = rest;

   if (done) {
      cMsg->from = atoi (tok[0]);
      cMsg->to = atoi (tok[1]);
      cMsg->ts = atoi (tok[2]);
      strcpy (cMsg->text, tok[3]);
      return MSG_OK;
   }

   return MSG_ERROR;
}

/*! getter for cMsg_t member fromAEM */
uint32_t cMsg_getFrom(cMsg_t* cMsg) { return cMsg->from; }
/*! getter for cMsg_t member toAEM */
uint32_t cMsg_getTo(cMsg_t* cMsg)   { return cMsg->to; }
/*! getter for cMsg_t member fromAEM */
uint64_t cMsg_getTs(cMsg_t* cMsg)   { return cMsg->ts; }
/*! getter for payload text member */
char_t*  cMsg_getText(cMsg_t* cMsg) { return cMsg->text; }

/*!
 * Predicate to check core message equality
 * @param m1   Pointer to message 1
 * @param m2   Pointer to message 2
 * @return     Equality result (true, false)
 */
bool_t cMsg_equal (cMsg_t* m1, cMsg_t* m2) {
   if (m1->from != m2->from)  return false;
   if (m1->to != m2->to)      return false;
   if (m1->ts != m2->ts)      return false;
   if (strncmp (m1->text, m2->text, sizeof(m1->text)))
      return false;
   return true;
}
//! @}

/*!
 * mgs_t API
 */
//! @{
void msg_init (msg_t* msg) {
   memset ((void*)msg, 0, sizeof(msg));
}

//! @}



/*!
 * Create a message list for our application.
 */
msgList_t msgList;


//! 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)

dIter_t devList_getIter (devAEM_t dev) {
   for (dIter_t i =0 ; i<AEMLIST_SIZE ; ++i) {
      if (devList[i].dev == dev)
         return i;
   }
   return -1;  // return end()
}


status_t msgList_init (msgList_t* msgList) {
   if (pthread_mutex_init(&lock_msgList, NULL) != 0) {
       log_error ("Error: mutex init has failed\n");
       return MSG_ERROR;
   }
   memset((void*)msgList, 0, sizeof(msgList_t));
   msgList->first =-1;
   msgList->last =-1;
   srand (time(NULL));
   return MSG_OK;
}

/*!
 * @brief   msgList iterator pre-increment in the msg_t direction
 *
 * This iterator force a ring buffer behavior. This function takes pointer
 * to iterator to alter but return the altered value so it can be directly
 * used in expressions
 *
 * @param it   Pointer to iterator to increase
 * @return     The iterator values
 */
mIter_t msgList_preInc (mIter_t* it) {
   if (++*it >= MSG_LIST_SIZE)  *it = 0;
   return *it;
}

/*!
 * @brief   msgList iterator pre-decrement in the msg_t direction
 *
 * This iterator force a ring buffer behavior. This function takes pointer
 * to iterator to alter but return the altered value so it can be directly
 * used in expressions
 *
 * @param it   Pointer to iterator to decrease
 * @return     The iterator values
 */
mIter_t msgList_preDec (mIter_t* it) {
   if (--*it < 0)  *it = MSG_LIST_SIZE;
   return *it;
}

mIter_t msgList_begin (msgList_t* this) {
   return this->first;
}

mIter_t msgList_last (msgList_t* this) {
   return this->last;
}

size_t msgList_size (msgList_t* this) {
   return this->size;
}

/*!
 * Searches for a message in the message list.
 *
 * @param this    The msgList object to work with
 * @param msg     Pointer to message to search
 * @return        Iterator to message if found, or -1 if not
 */
mIter_t msgList_find (msgList_t* this, msg_t* msg) {
   mIter_t it =this->last;  // get iterator
   // search from end to start to find msg, return on success
   for (size_t i=0 ; i < this->size ; ++i) {
      if (cMsg_equal (&this->m[it].cMsg, &msg->cMsg))
         return it;
      msgList_preDec(&it);
      // We start from the end as we think, its more possible
      // to find msg in the recent messages.
   }
   return (mIter_t)-1;     // fail to find
}



/*!
 * Add a new message in the message list
 *
 * @param this    The msgList object to work with
 * @param msg     Pointer to message
 * @return     Iterator to the added item (last)
 */
mIter_t msgList_add (msgList_t* this, msg_t* msg) {
   if (this->first == -1)                 // if its first time init "first" iterator
      this->first = 0;
   this->m[msgList_preInc(&this->last)] = *msg;             // store data *++it = *msg;
   _top_saturate(++this->size, this->size, MSG_LIST_SIZE);  // count the items with saturation

   // if we reacher full capacity, move along first also
   if ((this->first == this->last) && (this->size > 1))
      msgList_preInc(&this->first);
   return this->last;                     // return the iterator to newly created slot
}


void msgList_acquire () { pthread_mutex_lock (&lock_msgList); }
void msgList_release () { pthread_mutex_unlock (&lock_msgList); }

//! @}

//! Statistics API
//! @{

status_t stats_init (stats_t* s) {
   memset ((void*)s, 0, sizeof (stats_t));
   if (pthread_mutex_init(&lock_stats, NULL) != 0) {
       log_error ("Error: mutex init has failed\n");
       return MSG_ERROR;
   }
   return MSG_OK;
}

void statsUpdateCreate (msg_t* msg) {
   pthread_mutex_lock (&lock_stats);

   ++stats.totalMsg;
   ++stats.myMsg;

   // average message size
   int32_t saved = stats.totalMsg - stats.duplicateMsg;
   if ((saved-1) > 0) {
      int32_t l = strlen(msg->cMsg.text);
      stats.avMsgSize += l / (fpdata_t)(saved -1);
      stats.avMsgSize *= (fpdata_t)(saved-1)/saved;
   }
   pthread_mutex_unlock (&lock_stats);
}

void statsUpdateIn (msg_t* msg, bool_t dup) {
   pthread_mutex_lock (&lock_stats);

   bool_t forMe = msg->cMsg.to == settings.me;
   stats.totalMsg++;
   stats.duplicateMsg += (dup) ? 1:0;
   stats.forMeMsg += (forMe) ? 1:0;
   stats.inDirectMsg += (forMe && (msg->cMsg.from == msg->sender)) ? 1:0;

   // averages
   int32_t saved = stats.totalMsg - stats.duplicateMsg;
   if ((saved-1) > 0) {
      int32_t l = strlen(msg->cMsg.text);
      stats.avMsgSize += l / (fpdata_t)(saved -1);
      stats.avMsgSize *= (fpdata_t)(saved-1)/saved;

      if (settings.trackTime) {
         tstamp_t dt = (tstamp_t)time(NULL) - msg->cMsg.ts;
         if (dt < 0)
            dt = 0;
         stats.avTimeToMe += dt / (fpdata_t)(saved -1);
         stats.avTimeToMe *= (fpdata_t)(saved-1)/saved;
      }
   }
   pthread_mutex_unlock (&lock_stats);
}

void statsUpdateOut (msg_t* msg, devAEM_t dev) {
   pthread_mutex_lock (&lock_stats);
   stats.outDirectMsg += (msg->cMsg.to == dev) ? 1:0;
   pthread_mutex_unlock (&lock_stats);
}

status_t statsPrint (stats_t* stats) {
   FILE* fp = fopen ("statistics.txt", "w");
   if (fp == NULL) {
      fclose (fp);
      return MSG_ERROR;
   }
   fprintf (fp, "\nStatistics\n");
   fprintf (fp, "   total messages: %d\n", stats->totalMsg);
   fprintf (fp, "   duplicate messages: %d\n", stats->duplicateMsg);
   fprintf (fp, "   messages for me: %d\n", stats->forMeMsg);
   fprintf (fp, "   messages by me: %d\n",stats->myMsg);
   fprintf (fp, "   In messages direct for me: %d\n", stats->inDirectMsg);
   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);

   for (size_t i =0 ; i<AEMLIST_SIZE ; ++i) {
      fprintf (fp, "   Device %u found on %lld, last: %lld\n",
         devList[i].dev, devList[i].begin, devList[i].end);
   }
   fclose (fp);
   return MSG_OK;
}

//! @}