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

   1 #include "sockpool.h"
   2 #include <netinet/tcp.h>
   3
   4 #if defined(__i386__)
   5 inline static int test_and_set(volatile unsigned int *addr) {
   6     int oldval;
   7     /* Note: the "xchg" instruction does not need a "lock" prefix */
   8     __asm__ __volatile__("xchgl %0, %1"
   9         : "=r"(oldval), "=m"(*(addr))
  10         : "0"(1), "m"(*(addr)));
  11     return oldval;
  12 }
  13 inline static void UnLock(volatile unsigned int *addr) {
  14     int oldval;
  15     /* Note: the "xchg" instruction does not need a "lock" prefix */
  16     __asm__ __volatile__("xchgl %0, %1"
  17         : "=r"(oldval), "=m"(*(addr))
  18         : "0"(0), "m"(*(addr)));
  19 }
  20 #elif
  21 #   error need implementation of test_and_set
  22 #endif
  23
  24 #define MAXSPINS 1000
  25
  26 inline void Lock(volatile unsigned int *s) {
  27   while(test_and_set(s)) {
  28     int i=0;
  29     while(*s) {
  30       if (i++>MAXSPINS) {
  31         sched_yield();
  32         i=0;
  33       }
  34     }
  35   }
  36 }
  37
  38
  39 sockPoolHashTable_t *createSockPool(sockPoolHashTable_t * sockhash, unsigned int size) {
  40   if((sockhash = calloc(1, sizeof(sockPoolHashTable_t))) == NULL) {
  41     printf("Calloc error at %s line %d\n", __FILE__, __LINE__);
  42     return NULL;
  43   }
  44
  45   socknode_t **nodelist;
  46   if ((nodelist = calloc(size, sizeof(socknode_t *))) < 0) {
  47     printf("Calloc error at %s line %d\n", __FILE__, __LINE__);
  48     free(sockhash);
  49     return NULL;
  50   }
  51
  52   sockhash->table = nodelist;
  53   sockhash->inuse = NULL;
  54   sockhash->size = size;
  55   sockhash->mylock=0;
  56
  57   return sockhash;
  58 }
  59
  60 int createNewSocket(unsigned int mid) {
  61   int sd;
  62   int flag=1;
  63   if((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
  64     printf("%s() Error: In creating socket at %s, %d\n", __func__, __FILE__, __LINE__);
  65     return -1;
  66   }
  67   setsockopt(sd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(flag));
  68   struct sockaddr_in remoteAddr;
  69   bzero(&remoteAddr, sizeof(remoteAddr));
  70   remoteAddr.sin_family = AF_INET;
  71   remoteAddr.sin_port = htons(LISTEN_PORT);
  72   remoteAddr.sin_addr.s_addr = htonl(mid);
  73   if(connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
  74     printf("%s(): Error %d connecting to %s:%d\n", __func__, errno, inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
  75     close(sd);
  76     return -1;
  77   }
  78   return sd;
  79 }
  80
  81
  82 int getSockWithLock(sockPoolHashTable_t *sockhash, unsigned int mid) {
  83   socknode_t **ptr;
  84   int key = mid%(sockhash->size);
  85   int sd;
  86
  87   Lock(&sockhash->mylock);
  88   ptr=&sockhash->table[key];
  89
  90   while(*ptr!=NULL) {
  91     if (mid == (*ptr)->mid) {
  92       socknode_t *tmp=*ptr;
  93       sd = tmp->sd;
  94       *ptr=tmp->next;
  95       tmp->next=sockhash->inuse;
  96       sockhash->inuse=tmp;
  97       UnLock(&sockhash->mylock);
  98       return sd;
  99     }
 100     ptr=&((*ptr)->next);
 101   }
 102   UnLock(&sockhash->mylock);
 103   if((sd = createNewSocket(mid)) != -1) {
 104     socknode_t *inusenode = calloc(1, sizeof(socknode_t));
 105     insToListWithLock(sockhash, inusenode);
 106     return sd;
 107   } else {
 108     return -1;
 109   }
 110 }
 111
 112 int getSock(sockPoolHashTable_t *sockhash, unsigned int mid) {
 113   socknode_t **ptr;
 114   int key = mid%(sockhash->size);
 115   int sd;
 116
 117   ptr=&sockhash->table[key];
 118
 119   while(*ptr!=NULL) {
 120     if (mid == (*ptr)->mid) {
 121       socknode_t *tmp=*ptr;
 122       sd = tmp->sd;
 123       *ptr=tmp->next;
 124       tmp->next=sockhash->inuse;
 125       sockhash->inuse=tmp;
 126       return sd;
 127     }
 128     ptr=&((*ptr)->next);
 129   }
 130   if((sd = createNewSocket(mid)) != -1) {
 131     socknode_t *inusenode = calloc(1, sizeof(socknode_t));
 132     inusenode->next=sockhash->inuse;
 133     sockhash->inuse=inusenode;
 134     return sd;
 135   } else {
 136     return -1;
 137   }
 138 }
 139
 140 int getSock2(sockPoolHashTable_t *sockhash, unsigned int mid) {
 141   socknode_t **ptr;
 142   int key = mid%(sockhash->size);
 143   int sd;
 144
 145   ptr=&sockhash->table[key];
 146
 147   while(*ptr!=NULL) {
 148     if (mid == (*ptr)->mid) {
 149       return (*ptr)->sd;
 150     }
 151     ptr=&((*ptr)->next);
 152   }
 153   if((sd = createNewSocket(mid)) != -1) {
 154     *ptr=calloc(1, sizeof(socknode_t));
 155     (*ptr)->mid=mid;
 156     (*ptr)->sd=sd;
 157     return sd;
 158   } else {
 159     return -1;
 160   }
 161 }
 162
 163
 164 void insToListWithLock(sockPoolHashTable_t *sockhash, socknode_t *inusenode) {
 165     Lock(&sockhash->mylock);
 166     inusenode->next = sockhash->inuse;
 167     sockhash->inuse = inusenode;
 168     UnLock(&sockhash->mylock);
 169 }
 170
 171 void freeSock(sockPoolHashTable_t *sockhash, unsigned int mid, int sd) {
 172     int key = mid%(sockhash->size);
 173     socknode_t *ptr = sockhash->inuse;
 174     sockhash->inuse = ptr->next;
 175     ptr->mid = mid;
 176     ptr->sd = sd;
 177     ptr->next = sockhash->table[key];
 178     sockhash->table[key] = ptr;
 179 }
 180
 181 void freeSockWithLock(sockPoolHashTable_t *sockhash, unsigned int mid, int sd) {
 182   int key = mid%(sockhash->size);
 183   socknode_t *ptr;
 184   Lock(&sockhash->mylock);
 185   ptr = sockhash->inuse;
 186   sockhash->inuse = ptr->next;
 187   ptr->mid = mid;
 188   ptr->sd = sd;
 189   ptr->next = sockhash->table[key];
 190   sockhash->table[key] = ptr;
 191   UnLock(&sockhash->mylock);
 192 }
 193
 194 #if 0
 195 / ***************************************/
 196 * Array Implementation for socket resuse
 197 * ***************************************/
 198
 199 int num_machines;
 200
 201 sock_pool_t *initSockPool(unsigned int *mid, int machines) {
 202     sock_pool_t *sockpool;
 203     num_machines = machines;
 204     if ((sockpool = calloc(num_machines, sizeof(sock_pool_t))) < 0) {
 205         printf("%s(), Calloc error at %s, line %d\n", __func__, __FILE__, __LINE__);
 206         return NULL;
 207     }
 208     int i;
 209     for (i = 0; i < num_machines; i++) {
 210         if ((sockpool[i].sd = calloc(MAX_CONN_PER_MACHINE, sizeof(int))) < 0) {
 211             printf("%s(), Calloc error at %s, line %d\n", __func__, __FILE__, __LINE__);
 212             return NULL;
 213         }
 214         if ((sockpool[i].inuse = calloc(MAX_CONN_PER_MACHINE, sizeof(char))) < 0) {
 215             printf("%s(), Calloc error at %s, line %d\n", __func__, __FILE__, __LINE__);
 216             return NULL;
 217         }
 218         sockpool[i].mid = mid[i];
 219         int j;
 220         for(j = 0; j < MAX_CONN_PER_MACHINE; j++) {
 221             sockpool[i].sd[j] = -1;
 222         }
 223     }
 224
 225     return sockpool;
 226 }
 227
 228 int getSock(sock_pool_t *sockpool, unsigned int mid) {
 229     int i;
 230     for (i = 0; i < num_machines; i++) {
 231         if (sockpool[i].mid == mid) {
 232             int j;
 233             for (j = 0; j < MAX_CONN_PER_MACHINE; j++) {
 234                 if (sockpool[i].sd[j] != -1 && (sockpool[i].inuse[j] == 0)) {
 235                     sockpool[i].inuse[j] = 1;
 236                     return sockpool[i].sd[j];
 237                 }
 238                 if (sockpool[i].sd[j] == -1) {
 239                     //Open Connection
 240                     int sd;
 241                     if((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 242                         printf("%s() Error: In creating socket at %s, %d\n", __func__, __FILE__, __LINE__);
 243                         return -1;
 244                     }
 245                     struct sockaddr_in remoteAddr;
 246                     bzero(&remoteAddr, sizeof(remoteAddr));
 247                     remoteAddr.sin_family = AF_INET;
 248                     remoteAddr.sin_port = htons(LISTEN_PORT);
 249                     remoteAddr.sin_addr.s_addr = htonl(mid);
 250
 251                     if(connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
 252                         printf("%s(): Error %d connecting to %s:%d\n", __func__, errno, inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
 253                         close(sd);
 254                         return -1;
 255                     }
 256                     sockpool[i].sd[j] = sd;
 257                     sockpool[i].inuse[j] = 1;
 258                     return sockpool[i].sd[j];
 259                 }
 260             }
 261             printf("%s()->Error: Less number of MAX_CONN_PER_MACHINE\n", __func__);
 262             return -1;
 263         }
 264     }
 265     printf("%s()-> Error: Machine id not found\n", __func__);
 266
 267     return -1;
 268 }
 269
 270 int freeSock(sock_pool_t *sockpool, int sd) {
 271     int i;
 272     for (i = 0; i < num_machines; i++) {
 273         int j;
 274         for (j = 0; j < MAX_CONN_PER_MACHINE; j++) {
 275             if (sockpool[i].sd[j] == sd) {
 276                 sockpool[i].inuse[j] = 0;
 277                 return 0;
 278             }
 279         }
 280     }
 281     printf("%s() Error: Illegal socket descriptor %d\n", __func__, sd);
 282
 283     return -1;
 284 }
 285
 286 #endif