root/vtcross/branches/wrodgers/ServiceManagementLayer.cpp @ 255

/* Virginia Tech Cognitive Radio Open Source Systems
 * Virginia Tech, 2009
 *
 * LICENSE INFORMATION GOES HERE
 */

/* When a CE connects, the address of it's local socket is stored and it is assigned an ID number.
 * Until it sends a message ending communication, any communications from it are processed with
 * that ID number.  The "sendto" command uses the address data to send messages directly back to
 * it's local socket, allowing us to get away with a single-socket implementation on the SML-side.
 *
 * Services are stored in a SQLite DB by the ID of the CE that registered them.
 *
 * Mission support has not been implemented yet, but will be soon.
 */

/* KNOWN ISSUE: Note that this implementation would be unstable for CE's on any sort of network
 * since it relies on asyncronous communication occuring linearly and with one partner at a time.
 * The breakdown occurs specifically when the CE has to send ack messages back to the SML;
 * since there may be other messages queued on the socket since communication started,
 * there is no guarentee that the message would be the first up and I have yet to discover a method
 * for peaking more than 1 item into the message queue without starting to pop messages off first.
 * Still a work in progress though...
 */

#include <cstdlib>
#include <cstring>
#include <stdint.h>

#include "../../trunk/src/include/vtcross/common.h"
#include "../../trunk/src/include/vtcross/components.h"
#include "../../trunk/src/include/vtcross/containers.h"
#include "../../trunk/src/include/vtcross/debug.h"
#include "../../trunk/src/include/vtcross/error.h"
#include "../../trunk/src/include/vtcross/socketcomm.h"


#include <cstdlib>
#include <cstring>
#include <stdint.h>
#include <math.h>


#include <arpa/inet.h>
#include <iostream>
#include <netinet/in.h>
#include <netdb.h>
#include <fcntl.h>
#include <sys/ioctl.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/wait.h>

#include "../../trunk/src/include/tinyxml/tinyxml.h"
#include "../../trunk/src/include/tinyxml/tinystr.h"
#include "sqlite3.h"

typedef struct services_s * services_DB;

struct services_s {
    char filename[64];
    char tablename[64];
    char command[2048];
    sqlite3 *db; 
    unsigned int num_columns;
};

services_DB _services_DB;




ServiceManagementLayer::ServiceManagementLayer()
{
    LOG("Creating Service Management Layer.\n");
    shellSocketFD = -1;
    numberOfCognitiveEngines = 0;
    CE_Present = false;
    cogEngSrv = 1;

    Current_ID = 0;
    LoadConfiguration();
}


ServiceManagementLayer::~ServiceManagementLayer()
{
}


ServiceManagementLayer::ServiceManagementLayer(const char* serverName, \
        const char* serverPort)
{
    LOG("Creating Service Management Layer.\n");

    ConnectToShell(serverName, serverPort);
    CE_List = new CE_Reg[10];

    LoadConfiguration();
    CreateServicesDB();
}

void
CreateServicesDB()
{
    _services_DB = (services_DB) malloc(sizeof(struct services_s));

    // create database

    // copy filename
    unsigned int i=0;
    strcpy(_services_DB->filename, "Services_Table");

    // execute create database command
    // database handle
    //_services_DB->db = NULL;
    sqlite3_open(_services_DB->filename, &(_services_DB->db));
    char* cols[] = {"ID_Num", "Service_Name"};

    // create table

    // copy tablename
    strcpy(_services_DB->tablename, "Services");

    // number of columns in the table
    _services_DB->num_columns = 2;

    // generate command
    strcpy(_services_DB->command, "CREATE TABLE ");
    strcat(_services_DB->command, _services_DB->tablename);
    strcat(_services_DB->command, "(");
    strcat(_services_DB->command, cols[0]);
    strcat(_services_DB->command, " INT, ");
    strcat(_services_DB->command, cols[1]);
    strcat(_services_DB->command, " TEXT");
    strcat(_services_DB->command, ");");

    // execute create table command
    sqlite3_exec(_services_DB->db, _services_DB->command, NULL, 0, NULL);
}


void 
ServiceManagementLayer::SendComponentType()
{
    SendMessage(shellSocketFD, "response_sml");
    LOG("SML responded to GetRemoteComponentType query.\n");
}


void 
ServiceManagementLayer::ConnectToShell(const char* serverName, \
        const char* serverPort)
{
    shellSocketFD = ClientSocket(serverName, serverPort);

    RegisterComponent();
}


void 
ServiceManagementLayer::ShellSignalHandler()
{
    char buffer[256];

    memset(buffer, 0, 256); 
    ReadMessage(shellSocketFD, buffer);

    // TODO
    // If we send integer op codes rather than strings, this process will be
    // MUCH faster since instead of donig string compares we can simply
    // switch on the integer value...
    if(strcmp(buffer, "register_service") == 0) {
        if(strcmp(buffer, "policy_geo") == 0) {
        }
        else if(strcmp(buffer, "policy_time") == 0) {
        }
        else if(strcmp(buffer, "policy_spectrum") == 0) {
        }
        else if(strcmp(buffer, "policy_spacial") == 0) {
        }
    }
    else if(strcmp(buffer, "deregister_service") == 0) {
        if(strcmp(buffer, "policy_geo") == 0) {
        }
        else if(strcmp(buffer, "policy_time") == 0) {
        }
        else if(strcmp(buffer, "policy_spectrum") == 0) {
        }
        else if(strcmp(buffer, "policy_spacial") == 0) {
        }
    }
    else if(strcmp(buffer, "query_component_type") == 0) {
        SendComponentType();
    }
    else if(strcmp(buffer, "reset_sml") == 0) {
        Reset();
    }
    else if(strcmp(buffer, "shutdown_sml") == 0) {
        Shutdown();
    }
}

void
ServiceManagementLayer::CESignalHandler(int32_t ID)
{
    char buffer[256];   
    memset(buffer, 0, 256);     
    //
    ReadMessage(cogEngSrv, buffer);
    if(strcmp(buffer, "register_engine_cognitive") == 0) {
         RegisterCognitiveEngine(ID);
    }
    else if(strcmp(buffer, "register_service") == 0) {
         ReceiveServices(ID);
    }


}



void 
ServiceManagementLayer::Shutdown()
{
    DeregisterComponent();
}


void 
ServiceManagementLayer::Reset()
{
    DeregisterComponent();
    LoadConfiguration();
}


void 
ServiceManagementLayer::RegisterComponent()
{
    SendMessage(shellSocketFD, "register_sml");
    LOG("ServiceManagementLayer:: Registration message sent.\n");
}


void 
ServiceManagementLayer::DeregisterComponent()
{
    SendMessage(shellSocketFD, "deregister_sml");
    LOG("ServiceManagementLayer:: Deregistration message sent.\n");

    shutdown(shellSocketFD, 2);
    close(shellSocketFD);
    shellSocketFD = -1;
    LOG("ServiceManagementLayer:: Shell socket closed.\n");
}

/* Streams config data directly from the shell to the CE, and checks
 * for an "ack" message from the CE after every sent message 
 * to know when to stop communication.
 */
void 
ServiceManagementLayer::TransferRadioConfiguration(int32_t ID)
{
    struct timeval selTimeout;
    fd_set sockSet;
    int32_t rc = 0;
    char buffer[256];
    //Send data until the CE sends an ACK message back
    while(rc == 0){
        memset(buffer, 0, 256);
        //Receive data from Shell
        ReadMessage(shellSocketFD, buffer);
        //Send data to CE
        sendto(cogEngSrv, buffer, 256, NULL, (struct sockaddr*) CE_List[ID].sock_ptr, CE_List[ID].sock_len); 
        FD_ZERO(&sockSet);
        FD_SET(cogEngSrv, &sockSet);
        selTimeout.tv_sec = 0;
        selTimeout.tv_usec = 0;
        //Check if there is a message on the CE socket ready to be processed
        rc = select(cogEngSrv + 1, &sockSet, NULL, NULL, &selTimeout);
    }
    memset(buffer, 0, 256);
    //Once there is a message on the CE socket, read it and forward it on to the shell
    ReadMessage(cogEngSrv, buffer);
    SendMessage(shellSocketFD, buffer);
}

//Simmilar to TransferRadioConfig, just with Experience data
void 
ServiceManagementLayer::TransferExperience(int32_t ID)
{
    struct timeval selTimeout;
    fd_set sockSet;
    int32_t rc = 0;
    char buffer[256];
    //Send data until the CE sends an ACK message back
    while(rc == 0){
        memset(buffer, 0, 256);
        //Receive data from Shell
        ReadMessage(shellSocketFD, buffer);
        //Send data to CE
        sendto(cogEngSrv, buffer, 256, NULL, (struct sockaddr*) CE_List[ID].sock_ptr, CE_List[ID].sock_len); 
        FD_ZERO(&sockSet);
        FD_SET(cogEngSrv, &sockSet);
        selTimeout.tv_sec = 0;
        selTimeout.tv_usec = 0;
        //Check if there is a message on the CE socket ready to be processed
        rc = select(cogEngSrv + 1, &sockSet, NULL, NULL, &selTimeout);
    }
    memset(buffer, 0, 256);
    //Once there is a message on the CE socket, read it and forward it on to the shell
    ReadMessage(cogEngSrv, buffer);
    SendMessage(shellSocketFD, buffer);
}


void 
ServiceManagementLayer::ReceiveServices(int32_t ID)
{
    char buffer[256];
    memset(buffer, 0, 256);
    ReadMessage(cogEngSrv, buffer);
    char* cols[] = {"ID_Num", "Service_Name"};
    
    // generate command
    //printf("%s\n", _services_DB->command);
    strcpy(_services_DB->command, "insert into ");
    strcat(_services_DB->command, _services_DB->tablename); 
    strcat(_services_DB->command, " (");
    strcat(_services_DB->command, cols[0]);
    strcat(_services_DB->command, ", ");
    strcat(_services_DB->command, cols[1]);
    strcat(_services_DB->command, ") ");
    strcat(_services_DB->command, " values(");
    sprintf(_services_DB->command, "%s%d", _services_DB->command, ID);
    strcat(_services_DB->command, ", ");
    strcat(_services_DB->command, buffer);
    strcat(_services_DB->command, ");");
    
    //printf("search command: %s\n", _services_DB->command);
    // execute add command
    sqlite3_exec(_services_DB->db, _services_DB->command, NULL, 0, NULL);

}


void 
ServiceManagementLayer::SetActiveMission()
{
}


void 
ServiceManagementLayer::ListServices()
{
    // generate commandi
    strcpy(_services_DB->command, "select ");
    strcat(_services_DB->command, _services_DB->tablename);
    strcat(_services_DB->command, ".* from ");
    strcat(_services_DB->command, _services_DB->tablename);
    strcat(_services_DB->command, ";");

    // execute print (select all)  command
    sqlite3_exec(_services_DB->db, _services_DB->command, NULL, 0, NULL);
    printf("database %s, table %s:\n", _services_DB->filename, _services_DB->tablename);
}


void 
ServiceManagementLayer::ReloadConfiguration()
{
    LOG("ServiceManagementLayer:: Reloading Configuration.\n");
}


void 
ServiceManagementLayer::LoadConfiguration()
{
    LOG("ServiceManagementLayer:: Loading Configuration.\n");
}

void
ServiceManagementLayer::RegisterCognitiveEngine(int32_t ID)
{
    SendMessage(shellSocketFD, "register_engine_cognitive");
    LOG("ServiceManagementLayer:: CE registration message forwarded to shell.\n");

    TransferRadioConfiguration(ID);
    TransferExperience(ID);

    numberOfCognitiveEngines++;
    CE_Present = true;
}

void 
ServiceManagementLayer::DeregisterServices(int32_t ID)
{
    char str_buffer[64];
    strcpy(_services_DB->command, "delete ");
    strcat(_services_DB->command, _services_DB->tablename);
    strcat(_services_DB->command, ".* from ");
    strcat(_services_DB->command, _services_DB->tablename);
    strcat(_services_DB->command, " where ");
    strcat(_services_DB->command, "ID_Num");
    strcat(_services_DB->command, "==");
    sprintf(str_buffer, "%d;", ID);
    strcat(_services_DB->command, str_buffer);
    sqlite3_exec(_services_DB->db, _services_DB->command, NULL, 0, NULL);
}


void 
ServiceManagementLayer::DeregisterCognitiveEngine(int32_t ID)
{
    LOG("ServiceManagementLayer:: CE deregistration message forwarded to shell.\n");

    numberOfCognitiveEngines--;
    if(numberOfCognitiveEngines == 0)
        CE_Present = false;

    SendMessage(shellSocketFD, "deregister_engine_cognitive");
    char buffer[256];
    memset(buffer, 0, 256);
    ReadMessage(shellSocketFD, buffer);
    sendto(cogEngSrv, buffer, 256, NULL, (struct sockaddr*) CE_List[ID].sock_ptr, CE_List[ID].sock_len);
    if(strcmp("deregister_ack", buffer) != 0) {
        ERROR(1, "SML:: Failed to close CE socket\n");
    }
    CE_List[ID].sock_ptr = NULL;
    CE_List[ID].ID_num = -1;

    LOG("Cognitive Radio Shell:: Socket closed.\n");
}

/* The basic idea here is that when any time a message is sent over the socket, the sender's address is sent too.
 * We take advantage of this in order to uniquely identify the component for future communications since
 * there is only one socket on the SML-side.
 */
void
ServiceManagementLayer::StartSMLServer()
{
    struct timeval selTimeout;
    int32_t running = 1;
    int32_t port, rc, new_sd = 1;
    int32_t desc_ready = 1;
    int32_t timeout = 10;
    fd_set sockSet;

    cogEngSrv = CreateTCPServerSocket(CEPort);
    int32_t maxDescriptor = cogEngSrv;

    if(InitializeTCPServerPort(cogEngSrv) == -1)
        ERROR(1,"Error initializing primary port\n");

    while (running) {
        /* Zero socket descriptor vector and set for server sockets */
        /* This must be reset every time select() is called */
        FD_ZERO(&sockSet);
        FD_SET(cogEngSrv, &sockSet);

        /* Timeout specification */
        /* This must be reset every time select() is called */
        selTimeout.tv_sec = 0;       /* timeout (secs.) */
        selTimeout.tv_usec = 0;            /* 0 microseconds */
        //Changed both to zero so that select will check messages from the shell instead of blocking
        //when there is no command from the CE's to be processed

        //Check if there is a message on the socket waiting to be read
        rc = select(cogEngSrv + 1, &sockSet, NULL, NULL, &selTimeout);
        if(rc == 0){
            //If not, check instead for messages from the shell
            ShellSignalHandler();       
        }
        else {
            //If so, process the address information and pass the component ID on to the handler
            char buffer[256];
            memset(buffer, 0, 256);
            struct sockaddr_in sock_addr;
            memset((void *) &sock_addr, 0, sizeof(sock_addr));
            socklen_t sock_len;
            //Peak at the next message on the socket to determine its address of orgin
            recvfrom(cogEngSrv, buffer, 256, MSG_PEEK, (struct sockaddr*) &sock_addr, &sock_len);
            bool found = false;
            //Is it from a previously logged address?
            for(int i = 0; i < Current_ID; i++){
                if((*CE_List[i].sock_ptr).sin_addr.s_addr == sock_addr.sin_addr.s_addr){
                        //If so, pass the ID number of that component into the signal handler to process the message
                        found=true;
                        CESignalHandler(i);
                }
            }
            //If not, log the address and pass the ID number of that component into the signal handler
            if(!found){
                CE_List[Current_ID].sock_ptr = &sock_addr;
                CE_List[Current_ID].ID_num = Current_ID;
                CE_List[Current_ID].sock_len = sock_len;
                CESignalHandler(Current_ID);
                Current_ID++;
            }
        }
    }        

    /* Close sockets */
    close(cogEngSrv);

    return;
}

/*



rc = select(cogEngSrv + 1, &sockSet, NULL, NULL, &selTimeout); 
        if(rc == 0){
            LOG("No echo requests for %i secs...Server still alive\n", timeout);
            ShellSignalHandler();       
        }
        else {
            desc_ready = rc;

            for(port = 0; port <= maxDescriptor && desc_ready > 0; port++) {
                if(FD_ISSET(port, &sockSet)) {
                    desc_ready -= 1;

                    //Check if request is new or on an existing open descriptor
                    if(port == cogEngSrv) { 
                        do {
                            new_sd = AcceptTCPConnection(port);
                            if(new_sd < 0) 
                                break;
                           
                            CE_List[Current_ID].FD = new_sd;
                            CE_List[Current_ID].IDNum = Current_ID;
                            HandleMessage(CE_List[Current_ID]);
                            Current_ID++;
        
                            FD_SET(new_sd,&sockSet);
                            if(new_sd > maxDescriptor) 
                                maxDescriptor = new_sd;
                            //LOG("New incoming connection - %i\n\n",new_sd);
                        } while(new_sd != -1);
                    } 
                    else {
                        //LOG("Request on already open descriptor.\n\n");
                        for(int16_t i = 0; i < Current_ID; i++)
                        {
                                if(CE_List[i].FD == port)
                                        HandleMessage(CE_List[i]);
                        }
                    }
                }
            }
    }
*/