/home/dko/projects/mobilec/trunk/src/acc.c

Go to the documentation of this file.

/*[
 * Copyright (c) 2007 Integration Engineering Laboratory
                      University of California, Davis
 *
 * Permission to use, copy, and distribute this software and its
 * documentation for any purpose with or without fee is hereby granted,
 * provided that the above copyright notice appear in all copies and
 * that both that copyright notice and this permission notice appear
 * in supporting documentation.
 *
 * Permission to modify the software is granted, but not the right to
 * distribute the complete modified source code.  Modifications are to
 * be distributed as patches to the released version.  Permission to
 * distribute binaries produced by compiling modified sources is granted,
 * provided you
 *   1. distribute the corresponding source modifications from the
 *    released version in the form of a patch file along with the binaries,
 *   2. add special version identification to distinguish your version
 *    in addition to the base release version number,
 *   3. provide your name and address as the primary contact for the
 *    support of your modified version, and
 *   4. retain our contact information in regard to use of the base
 *    software.
 * Permission to distribute the released version of the source code along
 * with corresponding source modifications in the form of a patch file is
 * granted with same provisions 2 through 4 for binary distributions.
 *
 * This software is provided "as is" without express or implied warranty
 * to the extent permitted by applicable law.
]*/

#ifndef _WIN32
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <netdb.h>
#include <sys/un.h>
#include <unistd.h>
#include <sys/time.h>
#include <netdb.h>
#include <pthread.h>
#else
#include <winsock.h>
#include <windows.h>
#include <time.h>
#endif

#include <stdlib.h>
#include "include/acc.h"
#include "include/connection.h"
#include "include/data_structures.h"
#include "include/macros.h"
#include "include/mc_error.h"
#include "include/mc_platform.h"
#include "include/message.h"
#include "include/mtp_http.h"
#include "include/xml_parser.h"
#include "include/fipa_acl_envelope.h"

#define BACKLOG 10

acc_p
acc_Initialize(struct mc_platform_s* mc_platform)
{
  acc_p acc;
  acc = (acc_p)malloc(sizeof(acc_t));
  acc->mc_platform = mc_platform;

  acc->waiting = 0;
  acc->waiting_lock = (MUTEX_T*)malloc(sizeof(MUTEX_T));
  MUTEX_INIT(acc->waiting_lock);
  acc->waiting_cond = (COND_T*)malloc(sizeof(COND_T));
  COND_INIT(acc->waiting_cond);

  return acc;
}

  int 
acc_Destroy(acc_p acc)
{
  if(acc == NULL) {
    return MC_SUCCESS;
  }
  free(acc);
  acc = NULL;
  return MC_SUCCESS;
}

#ifndef _WIN32
  void*
acc_MessageHandlerThread(void* arg)
#else
  DWORD WINAPI
acc_MessageHandlerThread( LPVOID arg )
#endif
{
  mc_platform_p mc_platform = (mc_platform_p)arg;
  message_p message;
  agent_p agent;
  int mobile_agent_counter = 1;
  char* tmpstr;
  char* origname;
  int i;

  while(1) 
  {
    MUTEX_LOCK(mc_platform->message_queue->lock);
    MUTEX_LOCK(mc_platform->quit_lock);
    while(mc_platform->message_queue->size == 0 && !mc_platform->quit) {
      MUTEX_UNLOCK(mc_platform->quit_lock);
      COND_WAIT
        (
         mc_platform->message_queue->cond,
         mc_platform->message_queue->lock
        );
      MUTEX_LOCK(mc_platform->quit_lock);
    }
    if ( mc_platform->message_queue->size == 0 && mc_platform->quit )
    {
      MUTEX_UNLOCK(mc_platform->quit_lock);
      MUTEX_UNLOCK(mc_platform->message_queue->lock);
      return 0;
    }

    MUTEX_UNLOCK(mc_platform->quit_lock);
    MUTEX_UNLOCK(mc_platform->message_queue->lock);
    message = message_queue_Pop(mc_platform->message_queue);
    if (message == NULL) {
      printf("POP ERROR\n");
      continue;
    }
    /* Send MC Signal */
    MUTEX_LOCK(mc_platform->MC_signal_lock);
    mc_platform->MC_signal = MC_RECV_MESSAGE;
    COND_BROADCAST(mc_platform->MC_signal_cond);
    MUTEX_UNLOCK(mc_platform->MC_signal_lock);
    MUTEX_LOCK(mc_platform->giant_lock);
    while(mc_platform->giant == 0) {
      COND_WAIT
        (
         mc_platform->giant_cond,
         mc_platform->giant_lock
        );
    }
    MUTEX_UNLOCK(mc_platform->giant_lock);
    /* See if the destination is here. If it is, process the message.
     * Otherwise, send the message to it's destination */
    if( message->to_address == NULL) {
      /* Process message. Create the agent if it is an agent, or
       * process the ACL if it's an ACL message. */
      switch(message->message_type) {
        case MOBILE_AGENT:
          agent = agent_Initialize
            (
             mc_platform,
             message,
             mobile_agent_counter
            );
          if (agent != NULL) {
            /* We need to make sure there are no agent-name collisions. */
            i = 1;
            if( agent_queue_SearchName( mc_platform->agent_queue, agent->name ) ) {
              origname = agent->name;
              while( agent_queue_SearchName( mc_platform->agent_queue, agent->name ) ) {
                /* An agent with a matching name was found! For now, let us just
                 * rename the agent. */
                tmpstr = (char*)malloc(sizeof(char)* strlen(origname)+5);
                sprintf(tmpstr, "%s_%04d", origname, i);
                agent->name = tmpstr;
                i++;
              }
              fprintf(stderr, "Warning: Agent '%s' has been renamed to '%s'.\n",
                  origname, agent->name);
              free(origname);
            }
            mobile_agent_counter++;
            agent_queue_Add
              (
               mc_platform->agent_queue,
               agent
              );
          }
          message_Destroy(message);
          /* Send MC_Signal */
          MUTEX_LOCK(mc_platform->MC_signal_lock);
          mc_platform->MC_signal = MC_RECV_AGENT;
          COND_BROADCAST(mc_platform->MC_signal_cond);
          MUTEX_UNLOCK(mc_platform->MC_signal_lock);
          MUTEX_LOCK(mc_platform->giant_lock);
          while(mc_platform->giant == 0) {
            COND_WAIT(
                mc_platform->giant_cond,
                mc_platform->giant_lock
                );
          }
          MUTEX_UNLOCK(mc_platform->giant_lock);
          /* Set the ams to run */
          MUTEX_LOCK(mc_platform->ams->runflag_lock);
          mc_platform->ams->run = 1;
          COND_BROADCAST(mc_platform->ams->runflag_cond);
          MUTEX_UNLOCK(mc_platform->ams->runflag_lock);
          break;
        case FIPA_ACL:
          /* This case should never happen now, due to a reorganization
           * of MobileC. This enum will be deprecated in a future release. */
          break;
        case RETURN_MSG:
          /* Add the persistent neutral agent. */
          agent = agent_Initialize
            (
             mc_platform,
             message,
             mobile_agent_counter
            );
          if (agent != NULL) {
            MUTEX_LOCK(agent->lock);
            agent->datastate->persistent = 1;
            agent->agent_status = MC_AGENT_NEUTRAL;
            MUTEX_UNLOCK(agent->lock);
            mobile_agent_counter++;
            agent_queue_Add
              (
               mc_platform->agent_queue,
               agent
              );
          }
          message_Destroy(message);
          /* Send MC_Signal */
          MUTEX_LOCK(mc_platform->MC_signal_lock);
          mc_platform->MC_signal = MC_RECV_RETURN;
          COND_BROADCAST(mc_platform->MC_signal_cond);
          MUTEX_UNLOCK(mc_platform->MC_signal_lock);
          MUTEX_LOCK(mc_platform->giant_lock);
          while(mc_platform->giant == 0) {
            COND_WAIT(
                mc_platform->giant_cond,
                mc_platform->giant_lock
                );
          }
          MUTEX_UNLOCK(mc_platform->giant_lock);
          /* Set the ams to run */
          MUTEX_LOCK(mc_platform->ams->runflag_lock);
          mc_platform->ams->run = 1;
          COND_BROADCAST(mc_platform->ams->runflag_cond);
          MUTEX_UNLOCK(mc_platform->ams->runflag_lock);
          break;
/* MC_SECURITY */
#ifdef MC_SECURITY
        case ENCRYPTED_DATA:
          if (mc_platform->enable_security) {
            message_queue_Add
              (
               mc_platform->asm_message_queue,
               message
              );
          } else {
            WARN("mc_security not enabled. Discarding ENCRYPTED_DATA message.");
            message_Destroy(message);
          }
          break;
        case ENCRYPTION_INITIALIZE: 
          if (mc_platform->enable_security) {
          asm_queue_Add
            (
             mc_platform->asm_queue,
             asm_node_Initialize(message, mc_platform->security_manager)
            );
          /* Manually wake up the asm thread */
          MUTEX_LOCK(mc_platform->asm_message_queue->lock);
          COND_SIGNAL(mc_platform->asm_message_queue->cond);
          MUTEX_UNLOCK(mc_platform->asm_message_queue->lock);
          message_Destroy(message);
          } else {
            WARN("mc_security not enabled. Discarding ENCRYPTION_INITIALIZE message.");
            message_Destroy(message);
          }
          break;
        case REQUEST_ENCRYPTION_INITIALIZE:
          if (mc_platform->enable_security) {
          asm_SendEncryptionData(
              mc_platform->security_manager,
              message->from_address
              );
          message_Destroy(message);
          } else {
            WARN("mc_security not enabled. Discarding REQUEST_ENCRYPTION_INITIALIZE message.");
            message_Destroy(message);
          }
          break;
#endif /* END MC_SECURITY */
        case RELAY:
        case REQUEST:
        case SUBSCRIBE:
        case CANCEL:
        case N_UNDRSTD:
        case QUER_IF:
        case QUER_REF:
        case AGENT_UPDATE:
          fprintf(stderr, "FIXME: Message type %d not processable.%s:%d\n",
              message->message_type, __FILE__, __LINE__ );
          message_Destroy(message);
          break;
        default:
          fprintf(stderr, "Unknown message type:%d %s:%d\n",
              message->message_type, __FILE__, __LINE__);
          message_Destroy(message);
      }
    } else {
#ifdef MC_SECURITY
      if (mc_platform->enable_security) {
        if (
            (message->message_type != ENCRYPTED_DATA) &&
            (message->message_type != ENCRYPTION_INITIALIZE) &&
            (message->message_type != REQUEST_ENCRYPTION_INITIALIZE)
           )
        {
          message_queue_Add
            (
             mc_platform->asm_message_queue,
             message
            );
          continue;
        }
      }
#endif
      message_Send
        (
         message
        );
      message_Destroy
        (
         message
        );
    }
  }
  return 0;
}

#ifndef _WIN32
  void*
acc_Thread(void* arg)
#else
  DWORD WINAPI
acc_Thread( LPVOID arg )
#endif
{
  connection_p connection;
  message_p message;
  mtp_http_p mtp_http;
  mc_platform_p mc_platform = (mc_platform_p)arg;
  fipa_acl_envelope_p fipa_envelope;
  fipa_acl_message_p fipa_message;
  fipa_message_string_p fipa_message_string;
  int err;
  int i, j;
  agent_t* agent;
  /* We want to watch the connectlist for incoming connections, initialize 
   * them into messages, and add the messages to the appropriate queue. */
  while(1) {
    connection = NULL;
    message = NULL;
    mtp_http = NULL;
    MUTEX_LOCK(mc_platform->connection_queue->lock);
    MUTEX_LOCK(mc_platform->quit_lock);
    while (mc_platform->connection_queue->size == 0 && !mc_platform->quit) {
      MUTEX_UNLOCK(mc_platform->quit_lock);
      COND_WAIT(
          mc_platform->connection_queue->cond,
          mc_platform->connection_queue->lock
          );
      MUTEX_LOCK(mc_platform->quit_lock);
    }
    if 
      (
       mc_platform->connection_queue->size == 0 &&
       mc_platform->quit
      )
      {
        MUTEX_UNLOCK(mc_platform->quit_lock);
        MUTEX_UNLOCK(mc_platform->connection_queue->lock);
        return 0;
      }
    MUTEX_UNLOCK(mc_platform->quit_lock);
    MUTEX_UNLOCK(mc_platform->connection_queue->lock);
    /* Send MC Signal */
    MUTEX_LOCK(mc_platform->MC_signal_lock);
    mc_platform->MC_signal = MC_RECV_CONNECTION;
    COND_BROADCAST(mc_platform->MC_signal_cond);
    MUTEX_UNLOCK(mc_platform->MC_signal_lock);

    /* Check the giant lock to make sure we can continue */
    MUTEX_LOCK(mc_platform->giant_lock);
    while (mc_platform->giant == 0) {
      COND_WAIT(
          mc_platform->giant_cond,
          mc_platform->giant_lock
          );
    }
    MUTEX_UNLOCK(mc_platform->giant_lock);

    /* Continue with normal operation */
    connection = connection_queue_Pop(mc_platform->connection_queue);
    mtp_http = mtp_http_New();
    if ( mtp_http_InitializeFromConnection(mtp_http, connection ) )
    {
      connection_Destroy(connection);
      mtp_http_Destroy(mtp_http);
      continue;
    }

    switch(mtp_http->http_performative)
    {
      case HTTP_POST:
      case HTTP_PUT:
        /* See if the incoming message is going to any of our services. */
        /* See if it is a message for the AMS */
        if(
            !strcmp(mtp_http->target, "/ams") ||
            !strcmp( strrchr(mtp_http->target, (int)'/'), "/ams" )
            ) {
          message = message_New();
          /*message->message_parts = mtp_http->message_parts;*/
          message->message_body = (char*)malloc
            (
             sizeof(char) * 
             (strlen((char*)mtp_http->content->data)+1)
            );
          strcpy(message->message_body, (char*)mtp_http->content->data);
          message->xml_root = mxmlLoadString
            (
             NULL,
             message->message_body,
             MXML_NO_CALLBACK
            );
          if(message_xml_parse(message)) {
            fprintf(stderr, "Error parsing message. %s:%d\n",
                __FILE__,__LINE__);
            message_Destroy(message);
            mtp_http_Destroy(mtp_http);
            continue;
          }
          mtp_http_Destroy(mtp_http);
          break;
        } else if 
          ( 
           !strcmp(mtp_http->target, "/acc") ||
           !strcmp( strrchr(mtp_http->target, (int)'/'), "/acc")
          ) {
          /* Make sure there are two parts to the html message.
           * We expect to receive an xml envelope with a message attached. */
          if (mtp_http->message_parts != 2) {
            fprintf(stderr, "Error parsing message. %s:%d\n",
                __FILE__,__LINE__);
            mtp_http_Destroy(mtp_http);
            continue;
          }
          /* Now we need to parse the envelope */
          fipa_envelope = fipa_acl_envelope_New();
          err = fipa_envelope_Parse(fipa_envelope, (char*)mtp_http->content[0].data);
          if (err) {
            fprintf(stderr, "Error parsing message. %s:%d\n",
                __FILE__, __LINE__);
            fipa_acl_envelope_Destroy(fipa_envelope);
            mtp_http_Destroy(mtp_http);
            continue;
          }
          /* Now we need to check the envelope receivers and put copies of the message into
           * each receiving agent's mailbox. */
          for(i = 0; i < fipa_envelope->num_params; i++) {
            for(j = 0; j < fipa_envelope->params[i]->to->num; j++) {
              agent = agent_queue_SearchName(
                  mc_platform->agent_queue,
                  fipa_envelope->params[i]->to->fipa_agent_identifiers[j]->name
                  );
              if (agent != NULL) {
                /* Parse, copy, and deliver the message to the agent. */
                fipa_message_string = fipa_message_string_New();
                fipa_message_string->message = strdup((char*)mtp_http->content[1].data);
                fipa_message_string->parse = fipa_message_string->message;
                fipa_message = fipa_acl_message_New();
                err = fipa_acl_Parse(fipa_message, fipa_message_string);
                if (err) {
                  fipa_message_string_Destroy(fipa_message_string);
                  fipa_acl_message_Destroy(fipa_message);
                  fipa_acl_envelope_Destroy(fipa_envelope);
                  mtp_http_Destroy(mtp_http);
                  continue;
                }
                agent_mailbox_Post( agent->mailbox, fipa_message);
                fipa_message_string_Destroy(fipa_message_string);
              }
            }
          }
          fipa_acl_envelope_Destroy(fipa_envelope);
          mtp_http_Destroy(mtp_http);
          continue;
        }
        else {
          /* FIXME: We don't support other recipients yet */
          fprintf(stderr, "Unsupported. %s:%d\n", __FILE__, __LINE__);
          mtp_http_Destroy(mtp_http);
        }
      default:
        fprintf(stderr, "unsupported http performative. %s:%d\n",
            __FILE__, __LINE__);
    }

    /* If we get here, then we have an incoming message for the ams */
    connection_Destroy(connection);
    switch(message->message_type) {
      case RELAY:
      case REQUEST:
      case SUBSCRIBE:
      case CANCEL:
      case N_UNDRSTD:
      case MOBILE_AGENT:
      case QUER_IF:
      case QUER_REF:
      case AGENT_UPDATE:
      case RETURN_MSG:
      case FIPA_ACL:
        message_queue_Add(mc_platform->message_queue, message);
        break;
#ifdef MC_SECURITY
      case ENCRYPTED_DATA:
      case ENCRYPTION_INITIALIZE:
        if (mc_platform->enable_security) {
          message_queue_Add(mc_platform->asm_message_queue, message);
        } else {
          WARN("MC Security not enabled. Discarding message...");
          message_Destroy(message);
        }
        break;
      case REQUEST_ENCRYPTION_INITIALIZE:
        if (mc_platform->enable_security) {
          message_queue_Add(mc_platform->message_queue, message);
        } else {
          WARN("MC Security not enabled. Discarding message...");
          message_Destroy(message);
        }
        break;
#endif
      default:
        fprintf(stderr, "Unknown message type:%d. Rejecting message.%s:%d\n",
            message->message_type,
            __FILE__, __LINE__ );
        free(message);
        break;
    }
  }
}

  void
acc_Start(mc_platform_p mc_platform)
{
  acc_p acc = mc_platform->acc;
#ifndef _WIN32
  pthread_attr_t attr;
  pthread_attr_init(&attr);
  if(mc_platform->stack_size[MC_THREAD_ACC] != -1) {
  pthread_attr_setstacksize
    (
     &attr,
     mc_platform->stack_size[MC_THREAD_ACC]
    );
  }
#else
  int stack_size;
  if (mc_platform->stack_size[MC_THREAD_ACC] < 1) {
    /* In windows, 0 is default, not min */
    stack_size = mc_platform->stack_size[MC_THREAD_ACC]+1; 
  } else {
    stack_size = mc_platform->stack_size[MC_THREAD_ACC];
  }
#endif
  THREAD_CREATE
    ( 
     &acc->thread,
     acc_Thread,
     mc_platform 
    );
  THREAD_CREATE
    ( 
     &acc->message_handler_thread,
     acc_MessageHandlerThread,
     mc_platform
    );
  THREAD_CREATE
    (
     &acc->listen_thread,
     listen_Thread,
     mc_platform 
    );
}

#ifndef _WIN32
  void*
listen_Thread(void* arg)
#else
  DWORD WINAPI
listen_Thread( LPVOID arg )
#endif
{
#ifndef _WIN32
  int connectionsockfd; 
  int sockfd;
  struct sockaddr_in sktin;
  struct sockaddr_in peer_addr;
#else
  SOCKET connectionsockfd;
  SOCKET sockfd;
  struct sockaddr_in sktin;
  struct sockaddr_in peer_addr;
#endif

  connection_p connection;
  u_long connection_number;
  int connectionlen;
  mc_platform_p mc_platform = (mc_platform_p)arg;

  /* basic initialization */
  connection_number = 0;

  connectionlen = sizeof(struct sockaddr_in);

  /* Set up the socket */
  sockfd = socket(PF_INET, SOCK_STREAM, 0);
  sktin.sin_family = AF_INET;
  sktin.sin_port = htons(mc_platform->port);
  sktin.sin_addr.s_addr = INADDR_ANY;
  memset(sktin.sin_zero, '\0', sizeof sktin.sin_zero);
  if (bind(sockfd, (struct sockaddr *)&sktin, sizeof(struct sockaddr))
      == -1) {
    fprintf(stderr, "bind() error. %s:%d\n",
        __FILE__, __LINE__ );
    exit(1);
  }
  listen(sockfd, BACKLOG);

  /* this while loop runs continuously creating connections */
  while(1)
  {
    /* Set waiting flag */
    MUTEX_LOCK(mc_platform->acc->waiting_lock);
    mc_platform->acc->waiting = 1;
    COND_BROADCAST(mc_platform->acc->waiting_cond);
    MUTEX_UNLOCK(mc_platform->acc->waiting_lock);
#ifndef _WIN32
    if((connectionsockfd = accept(sockfd, 
            (struct sockaddr *)&peer_addr, 
            (socklen_t *)&connectionlen)) < 0)
#else
      if((connectionsockfd = accept(sockfd,
              (struct sockaddr *)&peer_addr,
              (int*)&connectionlen)) == INVALID_SOCKET)
#endif
      {
        fprintf(stderr, "ListenThread: accept error \n");
#ifdef _WIN32
        printf("Error number: %d\n", WSAGetLastError() );
        continue;
#endif
        continue;
      }
      else 
      {
        /* Set waiting flag */
        MUTEX_LOCK(mc_platform->acc->waiting_lock);
        mc_platform->acc->waiting = 0;
        COND_BROADCAST(mc_platform->acc->waiting_cond);
        MUTEX_UNLOCK(mc_platform->acc->waiting_lock);
        /* create a new connection and insert it into the connection linked 
         * list */
        connection = (connection_p)malloc(sizeof(connection_t));
        CHECK_NULL(connection, exit(0););
        connection->connect_id = rand();
        connection->remote_hostname = NULL;
        connection->addr = peer_addr;
        connection->serverfd = sockfd;
        connection->clientfd = connectionsockfd;

        /* add the connection to list and increment the number of connections */
        connection_queue_Add(mc_platform->connection_queue, connection);
      }
  }

  /* Free the current thread */
#ifndef _WIN32
  pthread_exit(0);
#else
  ExitThread(0);
#endif

  return 0;
}

---