Robust/src/Runtime/DSTM/interface/dstmserver.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <string.h>
   4 #include <pthread.h>
   5 #include <netdb.h>
   6 #include <fcntl.h>
   7 #include "dstm.h"
   8 #include "mlookup.h"
   9 #include "llookup.h"
  10
  11 #define LISTEN_PORT 2156
  12 #define BACKLOG 10 //max pending connections
  13 #define RECIEVE_BUFFER_SIZE 2048
  14
  15 extern int classsize[];
  16
  17 objstr_t *mainobjstore;
  18
  19 int dstmInit(void)
  20 {
  21         //todo:initialize main object store
  22         //do we want this to be a global variable, or provide
  23         //separate access funtions and hide the structure?
  24         mainobjstore = objstrCreate(DEFAULT_OBJ_STORE_SIZE);
  25         if (mhashCreate(HASH_SIZE, LOADFACTOR))
  26                 return 1; //failure
  27
  28         if (lhashCreate(HASH_SIZE, LOADFACTOR))
  29                 return 1; //failure
  30
  31         //pthread_t threadListen;
  32         //pthread_create(&threadListen, NULL, dstmListen, NULL);
  33
  34         return 0;
  35 }
  36
  37 void *dstmListen()
  38 {
  39         int listenfd, acceptfd;
  40         struct sockaddr_in my_addr;
  41         struct sockaddr_in client_addr;
  42         socklen_t addrlength = sizeof(struct sockaddr);
  43         pthread_t thread_dstm_accept;
  44         int i;
  45
  46         listenfd = socket(AF_INET, SOCK_STREAM, 0);
  47         if (listenfd == -1)
  48         {
  49                 perror("socket");
  50                 exit(1);
  51         }
  52
  53         my_addr.sin_family = AF_INET;
  54         my_addr.sin_port = htons(LISTEN_PORT);
  55         my_addr.sin_addr.s_addr = INADDR_ANY;
  56         memset(&(my_addr.sin_zero), '\0', 8);
  57
  58         if (bind(listenfd, (struct sockaddr *)&my_addr, addrlength) == -1)
  59         {
  60                 perror("bind");
  61                 exit(1);
  62         }
  63
  64         if (listen(listenfd, BACKLOG) == -1)
  65         {
  66                 perror("listen");
  67                 exit(1);
  68         }
  69
  70         printf("Listening on port %d, fd = %d\n", LISTEN_PORT, listenfd);
  71         while(1)
  72         {
  73                 acceptfd = accept(listenfd, (struct sockaddr *)&client_addr, &addrlength);
  74                 pthread_create(&thread_dstm_accept, NULL, dstmAccept, (void *)acceptfd);
  75         }
  76         pthread_exit(NULL);
  77 }
  78
  79 void *dstmAccept(void *acceptfd)
  80 {
  81         int numbytes,i,choice, oid;
  82         char buffer[RECIEVE_BUFFER_SIZE];
  83         char opcode[10];
  84         void *srcObj;
  85         objheader_t *h;
  86         int fd_flags = fcntl((int)acceptfd, F_GETFD), size;
  87
  88         printf("Recieved connection: fd = %d\n", (int)acceptfd);
  89         do
  90         {
  91                 numbytes = recv((int)acceptfd, (void *)buffer, sizeof(buffer), 0);
  92                 buffer[numbytes] = '\0';
  93                 if (numbytes == -1)
  94                 {
  95                         perror("recv");
  96                         pthread_exit(NULL);
  97                 }
  98                 else
  99                 {
 100                         sscanf(buffer, "%s %d\n", opcode, &oid);
 101
 102                         if (strcmp(opcode, "TRANS_RD") == 0) {
 103                                 printf("DEBUG -> Requesting:  %s %d\n", opcode, oid);
 104                                 srcObj = mhashSearch(oid);
 105                                 h = (objheader_t *) srcObj;
 106                                 printf("DEBUG -> Sending oid = %d, type %d\n", h->oid, h->type);
 107                                 size = sizeof(objheader_t) + sizeof(classsize[h->type]);
 108                                 if(send((int)acceptfd, srcObj, size, 0) < 0) {
 109                                         perror("");
 110                                 }
 111                                 //printf("DEBUG -> sent ...%d\n", write(acceptfd, srcObj, size));
 112                         }
 113                         else if (strcmp(opcode, "TRANS_COMMIT") == 0)
 114                                 printf(" This is a Commit\n");
 115                         else if (strcmp(opcode,"TRANS_ABORT") == 0)
 116                                 printf(" This is a Abort\n");
 117                         else
 118                                 printf(" This is a Broadcastt\n");
 119
 120                         //printf("Read %d bytes from %d\n", numbytes, (int)acceptfd);
 121                         //printf("%s", buffer);
 122                 }
 123
 124         //} while (numbytes != 0);
 125         } while (0);
 126         if (close((int)acceptfd) == -1)
 127         {
 128                 perror("close");
 129         }
 130         else
 131                 printf("Closed connection: fd = %d\n", (int)acceptfd);
 132         pthread_exit(NULL);
 133 }
 134