Changeset 97

Timestamp:

02/16/09 16:54:59 (15 years ago)

Author:

anonymous

Message:

cleaning up code

Location:

vtcross/trunk/CR_shell/src

Files:

• main_client.c (deleted)
• main_cognitive_radio.cpp (modified) (16 diffs)

vtcross/trunk/CR_shell/src/main_cognitive_radio.cpp

@@ -8 +8 @@
 #include "tinystr.h"
 #include <arpa/inet.h>
+#include <sys/mman.h>
+#include <fcntl.h>
+#include <sys/ioctl.h>
+
 //#include "socket/ServerSocket.h"
 //#include "socket/SocketException.h"
@@ -13 +17 @@
 using namespace std;
 
-#define SERVER_PORT 30000
+#define CE_SERVER_PORT 30001
+#define POLICY_SERVER_PORT 30003
 
 struct CE_Info {
@@ -19 +24 @@
         int numParameters;
         int numObservables;
+    int policy_engine;
+    int policy_socket;
+    int ce_socket;
 };
 
@@ -50 +58 @@
 };
 
+void DieWithError(char *errorMessage)
+{
+        perror(errorMessage);
+            exit(1);
+}
+
 void print_current_config(Utility * uList[], Parameter * pList[], Observable * oList[], CE_Info * ce_info) {
         int i = 0;
@@ -63 +77 @@
         for(i = 0; i < ce_info->numParameters; i++) {
                 cout << "Parameter:  " << pList[i]->name << endl;
-        printf("Radio Operation Profile has been sucessfully sent.\n");
+        printf("Cognitive Radio:: Radio Operation Profile has been sucessfully sent.\n");
                 cout << "       Units:   " << pList[i]->units << endl;
                 cout << "       Min:     " << pList[i]->min << endl;
@@ -230 +244 @@
 {
     int i,n;
-    
+   
     n = recv(socket,buffer,256,MSG_PEEK);
+    
     for(i=0;i<256;i++){
         if(strcmp(&buffer[i],"\0") == 0) break;
@@ -249 +264 @@
         message.append("\0");   
         // Write message back to client
-    n = write(socketfd,message.c_str(),(message.size()+1));
+    n = send(socketfd,message.c_str(),(message.size()+1),0);
     if (n<0)
         error("Error sending to client\n");
@@ -274 +289 @@
         //int total_bytes;   
 
-        printf("Sending Radio Operating Profile to Cognitive Engine.\n\n");
+        printf("Cognitive Radio:: Sending Radio Operating Profile to Cognitive Engine.\n\n");
  
         // utilities
@@ -345 +360 @@
         float * past_exp[]) 
 {
-    char buffer[256];
     int i, j;
         char counter[55];
@@ -384 +398 @@
 }
 
+
+int RequestPolicyValidation(Parameter * pList[], CE_Info *ce_info)
+{
+        char counter[55];
+        char var[50];
+    int i;
+    string control_msg;
+    
+    int socketfd = ce_info->policy_socket;
+
+    // Control message that validation request is coming
+    control_msg = "val";
+        SendMessage(socketfd,control_msg);
+
+    printf("Cognitive Radio:: Here. %i\n\n", socketfd);
+
+        // Send parameter information 
+    sprintf(counter,"%i",ce_info->numParameters);
+        SendMessage(socketfd,counter);
+        for(i = 0; i < ce_info->numParameters; i++) {
+                SendMessage(socketfd,pList[i]->name);
+                SendMessage(socketfd,pList[i]->units);
+                sprintf(var,"%f",pList[i]->min);
+                SendMessage(socketfd,var);
+                sprintf(var,"%f",pList[i]->max);
+                SendMessage(socketfd,var);
+                sprintf(var,"%f",pList[i]->step);
+                SendMessage(socketfd,var);
+                sprintf(var,"%f",pList[i]->value);
+                SendMessage(socketfd,var);
+                
+        }
+    return 1;
+
+}
+
+
 int RequestCEOptimization(int sockfd, Utility *uList[], 
         Parameter *pList[], Observable *oList[],
@@ -409 +460 @@
 
 
+    // If policy engine is connect, validate new values
+    if(ce_info->policy_engine == 1) {
+
+        printf("Cognitive Radio:: Found Policy Engine!\n");
+        printf("Cognitive Radio:: Validating parameters with Policy Engine\n\n");
+        RequestPolicyValidation(pList,ce_info); 
+        printf("Cognitive Radio:: Done\n\n");
+
+    }
+
+
     return 1;
 }
@@ -430 +492 @@
 }
 
-void StartMessaging(int socketfd, Utility * uList[], Parameter * pList[], Observable * oList[], CE_Info * ce_info) 
+void InitializePE(int socket, CE_Info * ce_info) 
+{
+    // Policy Engine is connected
+    // Set global policy engine value to 1
+    ce_info->policy_engine = 1;
+    ce_info->policy_socket = socket;
+
+    return;
+}
+
+void InitializeCE(int socketfd, Utility * uList[], Parameter * pList[], Observable * oList[], CE_Info * ce_info) 
 {
         LoadCEConfiguration(socketfd, uList, pList, oList, ce_info);
@@ -464 +536 @@
     past_exp[1][6] = 0.462; // overall utility
 
-
         // update ce with experience
-    printf("Sending Previous Experience to New Cognitive Engine.\n\n");
+    printf("Cognitive Radio:: Sending Previous Experience to New Cognitive Engine.\n\n");
     UpdateCEExperience(socketfd, num_rows, num_cols, past_exp); 
 
@@ -472 +543 @@
 }
 
-int StartShell(int port,Utility * uList[], Parameter * pList[], Observable * oList[], CE_Info * ce_info) {
-        // Start socket server
-        int sockfd, newsockfd, clilen;
-        struct sockaddr_in serv_addr, cli_addr;
-        char str[20];
-
-        // Setup server socket connection
-        sockfd = socket(AF_INET, SOCK_STREAM, 0);
-        if (sockfd < 0) 
-                error("ERROR opening socket");
-        bzero((char *) &serv_addr, sizeof(serv_addr));
-        serv_addr.sin_family = AF_INET;
-        serv_addr.sin_port = htons(port);
-        serv_addr.sin_addr.s_addr = INADDR_ANY;
-        if (bind(sockfd, (struct sockaddr *) &serv_addr,
-                sizeof(serv_addr)) < 0) 
-                error("ERROR on binding");
-        listen(sockfd,5);
-        clilen = sizeof(cli_addr);
-
-        //while(1) {
-                newsockfd = accept(sockfd, 
-                        (struct sockaddr *) &cli_addr, 
-                        (socklen_t*)&clilen);
-                if (newsockfd < 0) 
-                        error("ERROR on accept");
-
-        inet_ntop(AF_INET,&(cli_addr.sin_addr),str,20); 
-        printf("Received a connection from a Cognitive Engine with IP address: %s\n\n",str);
-
-                // Begin parsing the messages
-                StartMessaging(newsockfd, uList, pList, oList, ce_info);
-        //}
+int AcceptTCPConnection(int servSock)
+{
+    int clntSock;                    /* Socket descriptor for client */
+    struct sockaddr_in echoClntAddr;
+    unsigned int clntLen;
+
+    /* Set the size of the in-out parameter */
+    clntLen = sizeof(echoClntAddr);
+
+    /* Wait for a client to connect */
+    //if ((clntSock = accept(servSock, (struct sockaddr *) &echoClntAddr, &clntLen)) < 0) {
+    if ((clntSock = accept(servSock, NULL, NULL)) < 0) {
+        return -1;
+    } 
+   
+    /* clntSock is connected to a client! */
+    
+    printf("Handling client %s\n", inet_ntoa(echoClntAddr.sin_addr));
+
+    return clntSock;
+}
+
+int CreateTCPServerSocket(unsigned short port)
+{
+    int sock;                        /* socket to create */
+    struct sockaddr_in echoServAddr; /* Local address */
+
+    /* Create socket for incoming connections */
+    if ((sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0)
+        DieWithError("socket() failed");
+      
+    /* Construct local address structure */
+    memset(&echoServAddr, 0, sizeof(echoServAddr));   /* Zero out structure */
+    echoServAddr.sin_family = AF_INET;                /* Internet address family */
+    echoServAddr.sin_addr.s_addr = htonl(INADDR_ANY); /* Any incoming interface */
+    echoServAddr.sin_port = htons(port);              /* Local port */
+
+    /* Bind to the local address */
+    if (bind(sock, (struct sockaddr *) &echoServAddr, sizeof(echoServAddr)) < 0)
+        DieWithError("bind() failed");
+
+    /* Mark the socket so it will listen for incoming connections */
+    if (listen(sock, 5) < 0) {
+        printf("listen() failed\n");
+        return 0;
+    }
+
+    return sock;
+}
+
+void HandleTCPClient(int socketfd, Utility * uList[], Parameter * pList[], Observable * oList[], CE_Info * ce_info) 
+{
+    char buffer[256];        /* Buffer for echo string */
+
+    /* Receive message from client */
+    bzero(buffer,256);
+    ReceiveMessage(socketfd,buffer);
+
+    printf("Cognitive Radio:: Message Received - %s.\n\n", buffer);
+
+    if(strcmp(buffer,"c_register") == 0) 
+            InitializeCE(socketfd, uList, pList, oList, ce_info);
+
+    if(strcmp(buffer,"p_register") == 0)
+            InitializePE(socketfd, ce_info);
+
+    if(strcmp(buffer,"optimize") == 0)
+            RunSimulator(socketfd, uList, pList, oList, ce_info);
+        
+    //close(socketfd);    /* Close client socket */
+}
+
+int StartServers(Utility * uList[], Parameter * pList[], Observable * oList[], CE_Info * ce_info) {
+    int * servSock; 
+    int running = 1;
+    struct timeval selTimeout;
+    int timeout = 10;
+    int cognitive_engine = 0; 
+    int policy_engine = 1; 
+    int port, rc, on = 1;
+    int new_sd;
+    int desc_ready = 0;
+    fd_set sockSet;
+    
+    servSock = (int *) malloc(2 * sizeof(int));
+
+    servSock[cognitive_engine] = CreateTCPServerSocket(CE_SERVER_PORT);
+    servSock[policy_engine] = CreateTCPServerSocket(POLICY_SERVER_PORT);
+
+
+    int maxDescriptor = servSock[cognitive_engine];
+
+    if(servSock[cognitive_engine] < servSock[policy_engine])
+        maxDescriptor = servSock[policy_engine];
+
+    rc = setsockopt(servSock[cognitive_engine], SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof(on));
+    if(rc < 0)
+    {
+        perror("setsockopt() failed");
+        close(servSock[cognitive_engine]);
+        exit(-1);
+    }
+    
+    rc = setsockopt(servSock[policy_engine], SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof(on));
+    if(rc < 0)
+    {
+        perror("setsockopt() failed");
+        close(servSock[policy_engine]);
+        exit(-1);
+    }
+    
+    rc = ioctl(servSock[cognitive_engine], FIONBIO, (char*)&on);
+    if(rc < 0)
+    {
+        perror("ioctl() failed");
+        close(servSock[cognitive_engine]);
+        exit(-1);
+    }
+    
+    rc = ioctl(servSock[policy_engine], FIONBIO, (char*)&on);
+    if(rc < 0)
+    {
+        perror("ioctl() failed");
+        close(servSock[policy_engine]);
+        exit(-1);
+    }
+    
+    printf("Starting server:  Hit return to shutdown\n");
+    while (running)
+    {
+        /* Zero socket descriptor vector and set for server sockets */
+        /* This must be reset every time select() is called */
+        FD_ZERO(&sockSet);
+        /* Add keyboard to descriptor vector */
+        FD_SET(STDIN_FILENO, &sockSet);
+        FD_SET(servSock[cognitive_engine], &sockSet);
+        FD_SET(servSock[policy_engine], &sockSet);
+
+        /* Timeout specification */
+        /* This must be reset every time select() is called */
+        selTimeout.tv_sec = timeout;       /* timeout (secs.) */
+        selTimeout.tv_usec = 0;            /* 0 microseconds */
+
+        /* Suspend program until descriptor is ready or timeout */
+        rc = select(maxDescriptor + 1, &sockSet, NULL, NULL, &selTimeout); 
+        if (rc == 0)
+            printf("No echo requests for %i secs...Server still alive\n", timeout);
+        else 
+        {
+            if (FD_ISSET(0, &sockSet)) /* Check keyboard */
+            {
+                printf("Shutting down server\n");
+                getchar();
+                running = 0;
+            }
+
+            desc_ready = rc;
+
+            for (port = 0; port <= maxDescriptor && desc_ready > 0; port++) {
+                if (FD_ISSET(port, &sockSet))
+                {
+                    printf("Request on port %d:  ", port);
+                        desc_ready -= 1;
+
+
+                    if( (port == servSock[cognitive_engine]) || (port == servSock[policy_engine])) { 
+                       
+                        do
+                        { 
+                            new_sd = AcceptTCPConnection(port);
+                            if(new_sd < 0) 
+                            {
+                                break;
+                            }
+                            
+                            HandleTCPClient(new_sd, uList, pList, oList, ce_info);
+                            FD_SET(new_sd,&sockSet);
+                            if(new_sd > maxDescriptor) 
+                                maxDescriptor = new_sd;
+                            printf("New incoming connection - %i\n\n",new_sd);
+                        } while(new_sd != -1);
+                    } else {
+                        
+                        printf("Request on already open descriptor.\n\n");
+                        HandleTCPClient(port, uList, pList, oList, ce_info);
+
+                    }
+
+                }
+            }
+        }
+    }
+
+    /* Close sockets */
+    for (port = 0; port < 2; port++)
+        close(servSock[port]);
+
+    /* Free list of sockets */
+    free(servSock);        
+        
         return 0;
 }
@@ -514 +752 @@
         // CognitiveEngineShell Shell;
         string pFilename; 
-        int pid;
+    int fd;
 
         Utility * uList[10];
         Parameter * pList[10];
         Observable * oList[10];
-        CE_Info ce_info;
+        struct CE_Info *ce_info;
+
+    if((fd = open("/dev/zero", O_RDWR)) == -1)
+            return 1;
+
+    ce_info = (struct CE_Info *)mmap(0,sizeof(CE_Info),PROT_READ|PROT_WRITE,MAP_SHARED,fd,0);
+
+    close(fd);
 
         if(argc < 2) { 
@@ -537 +782 @@
 
         cout << "\n\nInitialize:: Attemping to parse " << pFilename << "." << endl;
-        parse_ce_config( &doc , uList, pList, oList, &ce_info);
+        parse_ce_config( &doc , uList, pList, oList, ce_info);
         cout << "Initialize:: Configuration file parsing completed.\n" << endl;
 
     //print_current_config(uList, pList, oList, &ce_info);
         
-    pid = fork();
-        if(pid == 0) {
-                // In child process - open policy engine port.
-        //      StartShell(30000,uList, pList, oList, &ce_info);
-        } else {
-                // In parent process - open cognitive engine port.
-                StartShell(30001,uList, pList, oList, &ce_info);
-        } 
-        
-
+   StartServers(uList, pList, oList, ce_info);
    return 1;
 }