rtes_messenger/src/client.c

/*!
 * \file client.c
 * This file contains all the client and device search functionalities.
 *
 * \author Christos Choutouridis AEM:8997 <cchoutou@ece.auth.gr>
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <sys/select.h>
#include <fcntl.h>
#include <errno.h>

#include "client.h"
/*!
 * Transmit buffer
 */
static char_t tx_buffer [MSG_TEXT_SIZE];

/*!
 * Ping functionality. We use system and pass a test command in order to
 * find out if a particular device is on range.
 * @param dev  The device to ping
 * @return     The status of the operation
 *    @arg  true  The device was on range and replyed to our ping
 *    @arg  false The device wasn't on range
 */
static bool_t ping (devAEM_t dev) {
   char_t cmd[72];
   devIP_t ip  = AEM2ip (dev);

   // ask host to ping and make a little pre-process before take the answer
   sprintf (cmd,
      "test $(ping -c1 -w%d %u.%u.%u.%u| grep received |cut -d' ' -f4) = 1",
      (int)settings.pingTimeout, ip.A, ip.B, ip.C, ip.D
   );
   return (system(cmd) == 0) ? true:false;
}

/*!
 * Search for devices on range and mark the time of first and last sight
 * @return  The number of devices on range
 */
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;
         continue;
      }
      if (ping (devList[i].dev)) {        // Noc noc....
         devList[i].onRange = true;       // Who's there?
         ++cnt;                           // No one, bye bye!
         // Mark the time for first and last time we saw this device
         if (!devList[i].begin)
            devList[i].begin = time(NULL);   // first time only
         devList[i].end = time(NULL);        // every time
         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;
}

/*!
 * @brief
 *    Send message functionality
 * TCP socket communication for sending a message to device. We use non-blocking
 * connect/select in order to be able to timeout from select. We also use send with
 * timeout.
 *
 * @param dev     The device to send the message
 * @param msg     Pointer to message to send
 * @return        The status of the operation
 * @note
 *    We treat all error of this function as "non-fatal".
 */
static bool_t sendMsg (devAEM_t dev, cMsg_t* msg) {
   int sock;
   sockaddr_in_t srvAddr;
   timeval_t timeout = settings.sendTimeout;
   long arg;
   bool_t ret = true;      // have faith

   // Make address
   memset(&srvAddr, 0, sizeof (srvAddr));
   srvAddr.sin_family= AF_INET;
   srvAddr.sin_port= htons (settings.port);
   srvAddr.sin_addr.s_addr = htonl (devAEM2addr (dev));
   devIP_t ip  = AEM2ip (dev);

   // Create socket for sending
   if ((sock = socket (PF_INET, SOCK_STREAM, 0)) == -1)
      return false;
   log_debug ("Debug: Socket for sending to %u.%u.%u.%u created\n", ip.A, ip.B, ip.C, ip.D);
   do {
      // Set non-blocking connect
      if((arg = fcntl(sock, F_GETFL, NULL)) < 0) {
         ret = false;
         log_debug ("Debug: Reading socket's file status failed\n");
         break;
      }
      if(fcntl(sock, F_SETFL, arg | O_NONBLOCK) < 0) {
         ret = false;
         log_debug ("Debug: Set socket to non-blocking mode failed\n");
         break;
      }
      log_debug ("Debug: Socket switched to non-blocking mode\n");

      // Trying to connect with timeout
      log_debug ("Debug: Try to connect with timeout\n");
      if (connect (sock, (sockaddr_t*)&srvAddr, sizeof(srvAddr)) == -1) {
         if (errno == EINPROGRESS) {
            fd_set myset;
            FD_ZERO (&myset);
            FD_SET (sock, &myset);
            if (select (sock+1, NULL, &myset, NULL, &timeout) <= 0) {
               ret = false;
               log_debug ("Debug: Connection failed in select()\n");
               break;
            }
            // Socket selected for write
            int valopt;
            socklen_t lon = sizeof(int);
            if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (void*)(&valopt), &lon) < 0) {
               ret = false;
               log_debug ("Debug: Reading socket's options failed\n");
               break;
            }
            // Check the value returned...
            if (valopt) {
               ret = false;
               log_debug ("Debug: Delayed connection failed\n");
               break;
            }
         }
         else {
            ret = false;
            log_debug ("Debug: Connection failed in select()\n");
            break;
         }
      }
      log_debug ("Debug: Connection to %u.%u.%u.%u succeed\n", ip.A, ip.B, ip.C, ip.D);

      // Set to blocking mode again...
      if( (arg = fcntl(sock, F_GETFL, NULL)) < 0) {
         ret = false;
         log_debug ("Debug: Reading socket's file status failed\n");
         break;
      }
      arg &= (~O_NONBLOCK);
      if( fcntl(sock, F_SETFL, arg) < 0) {
         ret = false;
         log_debug ("Debug: Set socket back to blocking mode failed\n");
         break;
      }
      log_debug ("Debug: Socket switched to blocking mode\n");

      // send the data with timeout
      if (setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO, (void*)&timeout, sizeof(timeout)) == -1) {
         ret = false;
         log_debug ("Debug: Setting send timeout failed\n");
         break;
      }
      log_debug ("Debug: Setting send timeout succeed\n");
      size_t len = cMsg_serialize (msg, tx_buffer);
      if (send (sock, tx_buffer, len, MSG_CONFIRM) == -1) {
         ret =  false;
         log_debug ("Debug: Sending failed\n");
         break;
      }
      log_debug ("Debug: Sending succeed\n");
   } while (0);

   close (sock);
   log_debug ("Debug: Closing socket\n");
   return ret;
}

/*!
 * @brief
 *    Client functionality
 *
 * Creates a message in a random interval, seeks for devices on range
 * and loop in entire msgList in order to send all messages
 * @note
 *    This function never returns
 */
static void client (void) {
   msg_t msg;                                   // new message buffer
   while (true) {
      // Idle until the time comes
      sleep (
         settings.msgIntervalMin +
         (rand() % (settings.msgIntervalMax - settings.msgIntervalMin))
      );
      memset ((void*)&msg, 0, sizeof (msg));    // create a new message
      cMsg_make (&msg.cMsg);
      msg.sender = settings.me;
      log_msg_new (&msg);

      statsUpdateCreate (&msg);

      msgList_acquire ();                       // try to lock resources
      mIter_t at = msgList_add (&msgList, &msg); // Add message to msgList
      msgList_release ();
      log_debug ("Debug: Message added to msgList at %d\n", at);

      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 (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 (currentDev.dev, &msgList.m[it].cMsg)) {
                  msgList.m[it].recipients[j] = true;
                  msgList_release ();
                  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", currentDev.dev);
                  msgList_acquire ();
               }
               //^ we try to send the message and mark the transmission on success
               // if send fail, don't worry it may succeed the next time.
            }
         }
      }
      msgList_release ();                       // Unlock resources
      if (statsPrint (&stats) == MSG_ERROR)
         log_error ("Error: Writing to statistics file failed\n");
   }
   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
 */
void* pthClient (void* ptr) {
   (void)&ptr;    // use parameter

   client ();
   exit(1);       // we should not be here, client never returns
   return NULL;
}