Changeset 255 for vtcross/branches

Timestamp:

05/15/09 11:47:36 (15 years ago)

Author:

wrodgers

Message:

Implemented service DB, transfer methods

Files:

• vtcross/branches/wrodgers/ServiceManagementLayer.cpp (modified) (12 diffs)

vtcross/branches/wrodgers/ServiceManagementLayer.cpp

@@ -5 +5 @@
  */
 
-/* DESCRIPTION OF FILE.
+/* 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...
  */
 
@@ -39 +55 @@
 #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;
 
 
@@ -70 +99 @@
 
     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);
 }
 
@@ -137 +206 @@
 ServiceManagementLayer::CESignalHandler(int32_t ID)
 {
-   char buffer[256];    
-   memset(buffer, 0, 256);     
-   //
-   ReadMessage(cogEngSrv, buffer);
-   if(strcmp(buffer, "register_engine_cognitive") == 0) {
-        RegisterCognitiveEngine(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);
+    }
 
 
@@ -185 +257 @@
 }
 
-
+/* 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)
 {
-}
-
-
-void 
-ServiceManagementLayer::ReceiveServices()
-{
+    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);
+
 }
 
@@ -213 +356 @@
 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);
 }
 
@@ -230 +383 @@
 
 void
-ServiceManagementLayer::RegisterCognitiveEngine(int32_t socketFD)
+ServiceManagementLayer::RegisterCognitiveEngine(int32_t ID)
 {
     SendMessage(shellSocketFD, "register_engine_cognitive");
-    LOG("Cognitive Engine:: Registration message sent to shell.\n");
-
-    TransferRadioConfiguration(socketFD);
-    TransferExperience(socketFD);
+    LOG("ServiceManagementLayer:: CE registration message forwarded to shell.\n");
+
+    TransferRadioConfiguration(ID);
+    TransferExperience(ID);
 
     numberOfCognitiveEngines++;
@@ -242 +395 @@
 }
 
-
-void 
-ServiceManagementLayer::DeregisterCognitiveEngine(int32_t socketFD)
-{
-    LOG("Cognitive Radio Shell:: Received deregistration message from Cognitive Engine.\n");
+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--;
@@ -252 +421 @@
         CE_Present = false;
 
-    SendMessage(socketFD, "deregister_ack");
-    shutdown(socketFD, 2);
-    close(socketFD);
+    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()
@@ -282 +463 @@
         /* Timeout specification */
         /* This must be reset every time select() is called */
-        selTimeout.tv_sec = timeout;       /* timeout (secs.) */
+        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, log that fact and check instead for messages from the shell
-            LOG("No echo requests for %i secs...Server still alive\n", timeout);
+            //If not, check instead for messages from the shell
             ShellSignalHandler();       
         }
@@ -298 +480 @@
             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, NULL);
+            recvfrom(cogEngSrv, buffer, 256, MSG_PEEK, (struct sockaddr*) &sock_addr, &sock_len);
             bool found = false;
             //Is it from a previously logged address?
@@ -313 +496 @@
                 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++;