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";
static char_t* _frm_msg_new = "new message: from=%d, to=%d, timestamp=%lld, text=%s";

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[16];
      strncpy (head, cMsg_getText(&msg->cMsg), sizeof(head));
      head[16] = 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[16];
      strncpy (head, cMsg_getText(&msg->cMsg), sizeof(head));
      head[16] = 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
   msg->to   = devList[rand()%AEMLIST_SIZE].dev;   // randomly select device
   msg->ts   = time(NULL);

   // stream the first fields and take the quote text iterator
   msg->text_it = sprintf (msg->text, "%d_%d_%lld_",
      msg->from,
      msg->to,
      msg->ts
   );
   // stream out the quote with a unique ID
   sprintf (&msg->text[msg->text_it], MESSAGE_BODY" #%d", msgID++);
}


/*!
 * 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;
   int d =0;
   for (size_t i =0; rawMsg[i] && i<size; ++i) {
      d += (rawMsg[i] == MSG_DELIMITER) ? 1:0;
   }
   if (d != 3)
      return MSG_ERROR;

   // Get message
   strcpy(cMsg->text, rawMsg);

   // Parse message
   char_t del[2] = {MSG_DELIMITER, '\0'};
   char_t* tok;

   tok = strtok (cMsg->text, del);
   cMsg->from = atoi (tok);

   tok = strtok(NULL, del);
   cMsg->to = atoi (tok);

   tok = strtok(NULL, del);
   cMsg->ts = atoll (tok);

   tok = strtok(NULL, del);
   cMsg->text_it = tok - cMsg->text;

   return MSG_OK;
}

/*! 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 (char_t*)& cMsg->text[cMsg->text_it]; }

/*!
 * 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 (cMsg_getText(m1), cMsg_getText(m2), sizeof(m1->text)))
      return false;
   return true;
}
//! @}

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

//! @}



/*!
 * 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
   uint32_t saved = stats.totalMsg - stats.duplicateMsg;
   stats.avMsgSize += strlen(cMsg_getText(&msg->cMsg)) / (saved -1);
   stats.avMsgSize *= (saved-1)/saved;

   pthread_mutex_unlock (&lock_stats);
}

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

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

   // averages
   uint32_t saved = stats.totalMsg - stats.duplicateMsg;
   stats.avMsgSize += strlen(cMsg_getText(&msg->cMsg)) / (saved -1);
   stats.avMsgSize *= (saved-1)/saved;

   if (settings.trackTime) {
      stats.avTimeToMe += ((tstamp_t)time(NULL) - msg->cMsg.ts) / (saved -1);
      stats.avTimeToMe *= (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);
}

//! @}