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

   1 #include "dstm.h"
   2 #include "clookup.h"
   3 #include "mlookup.h"
   4 #include "llookup.h"
   5 #include<sys/types.h>
   6 #include<sys/socket.h>
   7 #include<netdb.h>
   8 #include<netinet/in.h>
   9
  10 #define LISTEN_PORT 2156
  11 #define MACHINE_IP "127.0.0.1"
  12 #define RECEIVE_BUFFER_SIZE 2048
  13
  14 extern int classsize[];
  15
  16 transrecord_t *transStart()
  17 {
  18         transrecord_t *tmp = malloc(sizeof(transrecord_t));
  19         tmp->cache = objstrCreate(1048576);
  20         tmp->lookupTable = chashCreate(HASH_SIZE, LOADFACTOR);
  21         return tmp;
  22 }
  23
  24 objheader_t *transRead(transrecord_t *record, unsigned int oid)
  25 {
  26         unsigned int machinenumber;
  27         objheader_t *tmp, *objheader;
  28         void *objcopy;
  29         int size;
  30         void *buf;
  31                 //check cache
  32         if((objheader =(objheader_t *)chashSearch(record->lookupTable, oid)) != NULL){
  33                 return(objheader);
  34         } else if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
  35                 //Look up in Machine lookup table and found
  36                 printf("oid not found in local cache\n");
  37                 tmp = mhashSearch(oid);
  38                 size = sizeof(objheader_t)+classsize[tmp->type];
  39                 //Copy into cache
  40                 objcopy = objstrAlloc(record->cache, size);
  41                 memcpy(objcopy, (void *)tmp, size);
  42                 //Insert into cache's lookup table
  43                 chashInsert(record->lookupTable, objheader->oid, objcopy);
  44                 return(objcopy);
  45         } else {
  46                 printf("oid not found in local machine lookup\n");
  47                 machinenumber = lhashSearch(oid);
  48                 objcopy = getRemoteObj(record, machinenumber, oid);
  49                 if(objcopy == NULL)
  50                         printf("Object not found in Machine %d\n", machinenumber);
  51                 else
  52                         return(objcopy);
  53         }
  54 }
  55
  56 objheader_t *transCreateObj(transrecord_t *record, unsigned short type)
  57 {
  58         objheader_t *tmp = (objheader_t *) objstrAlloc(record->cache, (sizeof(objheader_t) + classsize[type]));
  59         tmp->oid = getNewOID();
  60         tmp->type = type;
  61         tmp->version = 1;
  62         tmp->rcount = 0; //? not sure how to handle this yet
  63         tmp->status |= NEW;
  64         chashInsert(record->lookupTable, tmp->oid, tmp);
  65         return tmp;
  66 }
  67
  68 int transCommit(transrecord_t *record){
  69         //Move objects to machine that hosts it
  70
  71 }
  72
  73 int transAbort(transrecord_t *record){
  74
  75 }
  76
  77 //mnun will be used to represent the machine IP address later
  78 void *getRemoteObj(transrecord_t *record, unsigned int mnum, unsigned int oid) {
  79         int sd, size;
  80         struct sockaddr_in serv_addr;
  81         struct hostent *server;
  82         char buffer[RECEIVE_BUFFER_SIZE],control;
  83         objheader_t *h;
  84         void *objcopy;
  85
  86         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
  87                 perror("Error in socket");
  88                 return NULL;
  89         }
  90         bzero((char*) &serv_addr, sizeof(serv_addr));
  91         serv_addr.sin_family = AF_INET;
  92         serv_addr.sin_port = htons(LISTEN_PORT);
  93         serv_addr.sin_addr.s_addr = inet_addr(MACHINE_IP);
  94
  95         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
  96                 perror("Error in connect");
  97                 return NULL;
  98         }
  99         bzero((char *)buffer,sizeof(buffer));
 100         control = READ_REQUEST;
 101         buffer[0] = control;
 102         memcpy(buffer+1, &oid, sizeof(int));
 103         if (write(sd, buffer, sizeof(buffer)) < 0) {
 104                 perror("Error sending message");
 105                 return NULL;
 106         }
 107
 108 #ifdef DEBUG1
 109         printf("DEBUG -> ready to rcv ...\n");
 110 #endif
 111         read(sd, buffer, sizeof(buffer));
 112         close(sd);
 113         if (buffer[0] == OBJECT_NOT_FOUND) {
 114                 return NULL;
 115         } else {
 116
 117                 h = (objheader_t *) buffer+1;
 118                 size = sizeof(objheader_t) + sizeof(classsize[h->type]);
 119 #ifdef DEBUG1
 120         printf("DEBUG -> Received: oid = %d, type = %d\n", h->oid, h->type);
 121 #endif
 122         }
 123         objcopy = objstrAlloc(record->cache, size);
 124         memcpy(objcopy, (void *)buffer+1, size);
 125         //Insert into cache's lookup table
 126         chashInsert(record->lookupTable, oid, objcopy);
 127         return objcopy;
 128 }