src/network/TCPServer.cxx

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by Prabakaran Thirumalai   *
   3  *   praba_tuty@yahoo.com   *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
  19  ***************************************************************************/
  20 #include <CSql.h>
  21 #include <Network.h>
  22 #include <SqlNetworkHandler.h>
  23
  24 DbRetVal TCPServer::start()
  25 {
  26     DbRetVal rv = OK;
  27     if (port == 0 )
  28     {
  29         printError(ErrBadArg, "Set the port first before starting\n");
  30         return ErrBadArg;
  31     }
  32     struct sockaddr_in my_addr;
  33     if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
  34        printError(ErrOS, "Unable to create socket\n");
  35        return ErrOS;
  36     }
  37     my_addr.sin_family = AF_INET;
  38     my_addr.sin_port = htons(port);
  39     my_addr.sin_addr.s_addr = INADDR_ANY;
  40     memset(&(my_addr.sin_zero), '\0', 8);
  41     if (bind(sockfd, (struct sockaddr *)&my_addr,
  42                      sizeof(struct sockaddr)) == -1) {
  43         printError(ErrOS, "bind failed");
  44         return ErrOS;
  45     }
  46     if (listen(sockfd, 10) == -1) {
  47         printError(ErrOS, "listen failed");
  48         return ErrOS;
  49     }
  50     return rv;
  51
  52 }
  53 DbRetVal TCPServer::stop()
  54 {
  55     DbRetVal rv = OK;
  56     close (sockfd);
  57     return rv;
  58 }
  59 DbRetVal TCPServer::handleClient()
  60 {
  61 //   printf("PRABA::handling client \n");
  62    DbRetVal rv = OK;
  63    socklen_t addressLen = sizeof(struct sockaddr);
  64    clientfd = accept(sockfd, (struct sockaddr*) &clientAddress, &addressLen);
  65    int ret = os::fork();
  66    if (ret) {
  67        //parent
  68        os::signal(SIGCHLD, SIG_IGN);
  69        close(clientfd);
  70        return OK;
  71    }else if (ret == 0) {
  72        //child
  73        int response = 1;
  74        int ret = os::send(clientfd, &response, 4, 0);
  75        if (ret == -1)
  76        {
  77            printError(ErrOS, "Unable to communicate to peer\n");
  78            return ErrOS;
  79        }
  80 //       printf("sent connect ack packet to client\n");
  81        fd_set fdset;
  82        struct timeval timeout;
  83        SqlNetworkHandler handler;
  84        PacketHeader header;
  85        while(true) {
  86            FD_ZERO(&fdset);
  87            FD_SET(clientfd, &fdset);
  88            timeout.tv_sec = 5;
  89            timeout.tv_usec = 0;
  90            int ret = os::select(clientfd+1, &fdset, 0, 0, &timeout);
  91            if (ret > 0) {
  92 //                printf("something in fd\n");
  93                int numbytes = os::recv(clientfd,&header,sizeof(PacketHeader),0);
  94                if (numbytes == -1)
  95                {
  96                    printError(ErrOS, "Error reading from socket\n");
  97                    return ErrOS;
  98                }
  99 //               printf("HEADER says packet type is %d\n", header.packetType);
 100                char *buffer = NULL;
 101                if (header.packetType != SQL_NW_PKT_DISCONNECT &&
 102                    header.packetType != SQL_NW_PKT_COMMIT     &&
 103                    header.packetType != SQL_NW_PKT_ROLLBACK)
 104                {
 105                    buffer = (char*) malloc(header.packetLength);
 106                    numbytes = os::recv(clientfd,buffer,header.packetLength,0);
 107                    if (numbytes == -1)
 108                    {
 109                        printError(ErrOS, "Error reading from socket\n");
 110                        return ErrOS;
 111                    }
 112                }
 113                ResponsePacket *rpkt = (ResponsePacket *) handler.process(header, buffer);
 114                numbytes = os::send(clientfd, rpkt, sizeof(ResponsePacket), 0);
 115                if (numbytes == -1)
 116                {
 117                    printError(ErrOS, "Error writing to socket\n");
 118                    return ErrOS;
 119                }
 120                char *ptr = (char *)&rpkt->retVal;
 121                if (*ptr==0) continue;
 122                if (*(ptr + 1)  == 1) continue; // for end of fetch
 123                NetworkStmt *stmt=NULL;
 124                int params =  *(ptr + 2);
 125                int proj = *(ptr + 3);
 126                if ((header.packetType == SQL_NW_PKT_PREPARE && params != 0) ||
 127                    (header.packetType == SQL_NW_PKT_PREPARE && proj != 0)) {
 128                    if (params) {
 129                        SqlPacketParamMetadata *prmpkt = new SqlPacketParamMetadata();
 130                        prmpkt->stmtID = rpkt->stmtID;
 131                        ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 132                        while (stmtIter.hasElement()) {
 133                            stmt = (NetworkStmt*) stmtIter.nextElement();
 134                            if (stmt->stmtID == prmpkt->stmtID) break;
 135                        }
 136                        prmpkt->noParams = stmt->paramList.size();
 137                        rv = prmpkt->marshall();
 138                        if (rv != OK) {
 139                            printf("marshall failed\n");
 140                        }
 141                        rv = send(SQL_NW_PKT_PARAM_METADATA, prmpkt->getMarshalledBuffer(), prmpkt->getBufferSize());
 142                        if (rv != OK) {
 143                            printf("Error in sending the metadata to the client\n");
 144                            exit(1);
 145                        }
 146                    }
 147                    if (proj) {
 148                        //fill projection list and send it to client
 149                        SqlPacketProjMetadata *prjpkt = new SqlPacketProjMetadata();
 150                        prjpkt->stmtID = rpkt->stmtID;
 151                        ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 152                        while (stmtIter.hasElement()) {
 153                            stmt = (NetworkStmt*) stmtIter.nextElement();
 154                            if (stmt->stmtID == prjpkt->stmtID) break;
 155                        }
 156                        prjpkt->noProjs = stmt->projList.size();
 157                        rv = prjpkt->marshall();
 158                        if (rv != OK) {
 159                            printf("marshall failed\n");
 160                        }
 161                        rv = send(SQL_NW_PKT_PROJ_METADATA, prjpkt->getMarshalledBuffer(), prjpkt->getBufferSize());
 162                        if (rv != OK) {
 163                            printf("Error in sending the metadata to the client\n");
 164                            exit(1);
 165                        }
 166                    }
 167                }
 168                if (header.packetType == SQL_NW_PKT_FETCH) {
 169                    SqlPacketResultSet *rspkt = new SqlPacketResultSet();
 170                    rspkt->stmtID = rpkt->stmtID;
 171                    ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 172                    while (stmtIter.hasElement()) {
 173                        stmt = (NetworkStmt*) stmtIter.nextElement();
 174                        if (stmt->stmtID == rspkt->stmtID) break;
 175                    }
 176                    rspkt->noProjs = stmt->projList.size();
 177                    rspkt->setProjList(stmt->projList);
 178                    rspkt->marshall();
 179                    if (rv != OK) { printf("marshall failed\n"); }
 180                    rv = send(SQL_NW_PKT_RESULT_SET, rspkt->getMarshalledBuffer(), rspkt->getBufferSize());
 181                    if (rv != OK) {
 182                        printf("Error in sending the metadata to the client\n");
 183                        exit(1);
 184                    }
 185                }
 186                if (header.packetType == SQL_NW_PKT_DISCONNECT) {
 187                    exit(0);
 188                }
 189            } else printf("Nothing in fd %d\n", ret);
 190        }
 191    }else {
 192         printError(ErrOS, "Unable to fork new process");
 193         return ErrOS;
 194    }
 195    return OK;
 196 }
 197
 198 DbRetVal TCPServer::send(NetworkPacketType type, char *buf, int len)
 199 {
 200     DbRetVal rv = OK;
 201     void* totalBuffer = malloc(sizeof(PacketHeader)+ len);
 202     PacketHeader *hdr=  new PacketHeader();
 203     hdr->packetType = type;
 204     hdr->packetLength = len;
 205     hdr->srcNetworkID = 0;//networkid;
 206     hdr->version = 1;
 207     memcpy(((char*)totalBuffer) + sizeof(PacketHeader) , buf, len);
 208     int numbytes=0;
 209     if ((numbytes=os::send(clientfd, hdr, sizeof(PacketHeader), 0)) == -1) {
 210         printError(ErrOS, "Unable to send the packet\n");
 211         return ErrOS;
 212     }
 213 //    printf("Sent bytes %d\n", numbytes);
 214     if ((numbytes=os::send(clientfd, buf, len, 0)) == -1) {
 215         printError(ErrOS, "Unable to send the packet\n");
 216         return ErrOS;
 217     }
 218 //    printf("Sent bytes %d\n", numbytes);
 219     free(totalBuffer);
 220     return rv;
 221 }