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

   1 #include "sockpool.h"
   2 #include <netinet/tcp.h>
   3
   4 #if defined(__i386__)
   5 inline 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 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 4
  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 sockPoolHashTable_t *createSockPool(sockPoolHashTable_t * sockhash, unsigned int size) {
  39   if((sockhash = calloc(1, sizeof(sockPoolHashTable_t))) == NULL) {
  40     printf("Calloc error at %s line %d\n", __FILE__, __LINE__);
  41     return NULL;
  42   }
  43
  44   socknode_t **nodelist;
  45   if ((nodelist = calloc(size, sizeof(socknode_t *))) < 0) {
  46     printf("Calloc error at %s line %d\n", __FILE__, __LINE__);
  47     free(sockhash);
  48     return NULL;
  49   }
  50
  51   sockhash->table = nodelist;
  52   sockhash->inuse = NULL;
  53   sockhash->size = size;
  54   sockhash->mask = size - 1;
  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     perror("socket connect: ");
  75     printf("%s(): Error %d connecting to %s:%d\n", __func__, errno, inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
  76     close(sd);
  77     return -1;
  78   }
  79   return sd;
  80 }
  81
  82 int getSockWithLock(sockPoolHashTable_t *sockhash, unsigned int mid) {
  83   socknode_t **ptr;
  84   int key = mid&(sockhash->mask);
  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     inusenode->sd = sd;
 106     inusenode->mid = mid;
 107     insToListWithLock(sockhash, inusenode);
 108     return sd;
 109   } else {
 110     return -1;
 111   }
 112 }
 113
 114 int getSock(sockPoolHashTable_t *sockhash, unsigned int mid) {
 115   socknode_t **ptr;
 116   int key = mid&(sockhash->mask);
 117   int sd;
 118
 119   ptr=&(sockhash->table[key]);
 120
 121   while(*ptr!=NULL) {
 122     if (mid == (*ptr)->mid) {
 123       socknode_t *tmp=*ptr;
 124       sd = tmp->sd;
 125       *ptr=tmp->next;
 126       tmp->next=sockhash->inuse;
 127       sockhash->inuse=tmp;
 128       return sd;
 129     }
 130     ptr=&((*ptr)->next);
 131   }
 132   if((sd = createNewSocket(mid)) != -1) {
 133     socknode_t *inusenode = calloc(1, sizeof(socknode_t));
 134     inusenode->next=sockhash->inuse;
 135     sockhash->inuse=inusenode;
 136     return sd;
 137   } else {
 138     return -1;
 139   }
 140 }
 141
 142 int getSock2(sockPoolHashTable_t *sockhash, unsigned int mid) {
 143   socknode_t **ptr;
 144   int key = mid&(sockhash->mask);
 145   int sd;
 146
 147   ptr=&(sockhash->table[key]);
 148
 149   while(*ptr!=NULL) {
 150     if (mid == (*ptr)->mid) {
 151       return (*ptr)->sd;
 152     }
 153     ptr=&((*ptr)->next);
 154   }
 155   if((sd = createNewSocket(mid)) != -1) {
 156     *ptr=calloc(1, sizeof(socknode_t));
 157     (*ptr)->mid=mid;
 158     (*ptr)->sd=sd;
 159     return sd;
 160   } else {
 161     return -1;
 162   }
 163 }
 164
 165 /*socket pool with multiple TR threads asking to connect to same machine  */
 166 int getSock2WithLock(sockPoolHashTable_t *sockhash, unsigned int mid) {
 167   socknode_t **ptr;
 168   int key = mid&(sockhash->mask);
 169   int sd;
 170
 171   Lock(&sockhash->mylock);
 172   ptr=&(sockhash->table[key]);
 173   while(*ptr!=NULL) {
 174     if (mid == (*ptr)->mid) {
 175       UnLock(&sockhash->mylock);
 176       return (*ptr)->sd;
 177     }
 178     ptr=&((*ptr)->next);
 179   }
 180   UnLock(&sockhash->mylock);
 181   if((sd = createNewSocket(mid)) != -1) {
 182     socknode_t *inusenode = calloc(1, sizeof(socknode_t));
 183     inusenode->sd = sd;
 184     inusenode->mid = mid;
 185     addSockWithLock(sockhash, inusenode);
 186     return sd;
 187   } else {
 188     return -1;
 189   }
 190 }
 191
 192 void addSockWithLock(sockPoolHashTable_t *sockhash, socknode_t *ptr) {
 193   int key = ptr->mid&(sockhash->mask);
 194   Lock(&sockhash->mylock);
 195   ptr->next = sockhash->table[key];
 196   sockhash->table[key] = ptr;
 197   UnLock(&sockhash->mylock);
 198 }
 199
 200 void insToListWithLock(sockPoolHashTable_t *sockhash, socknode_t *inusenode) {
 201   Lock(&sockhash->mylock);
 202   inusenode->next = sockhash->inuse;
 203   sockhash->inuse = inusenode;
 204   UnLock(&sockhash->mylock);
 205 }
 206
 207 void freeSock(sockPoolHashTable_t *sockhash, unsigned int mid, int sd) {
 208   int key = mid&(sockhash->mask);
 209   socknode_t *ptr = sockhash->inuse;
 210   sockhash->inuse = ptr->next;
 211   ptr->mid = mid;
 212   ptr->sd = sd;
 213   ptr->next = sockhash->table[key];
 214   sockhash->table[key] = ptr;
 215 }
 216
 217 void freeSockWithLock(sockPoolHashTable_t *sockhash, unsigned int mid, int sd) {
 218   int key = mid&(sockhash->mask);
 219   socknode_t *ptr;
 220   Lock(&sockhash->mylock);
 221   ptr = sockhash->inuse;
 222   sockhash->inuse = ptr->next;
 223   ptr->mid = mid;
 224   ptr->sd = sd;
 225   ptr->next = sockhash->table[key];
 226   sockhash->table[key] = ptr;
 227   UnLock(&sockhash->mylock);
 228 }
 229
 230 #if 0
 231 /***************************************
 232  * Array Implementation for socket reuse
 233  ****************************************/
 234
 235 int num_machines;
 236
 237 sock_pool_t *initSockPool(unsigned int *mid, int machines) {
 238   sock_pool_t *sockpool;
 239   num_machines = machines;
 240   if ((sockpool = calloc(num_machines, sizeof(sock_pool_t))) < 0) {
 241     printf("%s(), Calloc error at %s, line %d\n", __func__, __FILE__, __LINE__);
 242     return NULL;
 243   }
 244   int i;
 245   for (i = 0; i < num_machines; i++) {
 246     if ((sockpool[i].sd = calloc(MAX_CONN_PER_MACHINE, sizeof(int))) < 0) {
 247       printf("%s(), Calloc error at %s, line %d\n", __func__, __FILE__, __LINE__);
 248       return NULL;
 249     }
 250     if ((sockpool[i].inuse = calloc(MAX_CONN_PER_MACHINE, sizeof(char))) < 0) {
 251       printf("%s(), Calloc error at %s, line %d\n", __func__, __FILE__, __LINE__);
 252       return NULL;
 253     }
 254     sockpool[i].mid = mid[i];
 255     int j;
 256     for(j = 0; j < MAX_CONN_PER_MACHINE; j++) {
 257       sockpool[i].sd[j] = -1;
 258     }
 259   }
 260
 261   return sockpool;
 262 }
 263
 264 int getSock(sock_pool_t *sockpool, unsigned int mid) {
 265   int i;
 266   for (i = 0; i < num_machines; i++) {
 267     if (sockpool[i].mid == mid) {
 268       int j;
 269       for (j = 0; j < MAX_CONN_PER_MACHINE; j++) {
 270         if (sockpool[i].sd[j] != -1 && (sockpool[i].inuse[j] == 0)) {
 271           sockpool[i].inuse[j] = 1;
 272           return sockpool[i].sd[j];
 273         }
 274         if (sockpool[i].sd[j] == -1) {
 275           //Open Connection
 276           int sd;
 277           if((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 278             printf("%s() Error: In creating socket at %s, %d\n", __func__, __FILE__, __LINE__);
 279             return -1;
 280           }
 281           struct sockaddr_in remoteAddr;
 282           bzero(&remoteAddr, sizeof(remoteAddr));
 283           remoteAddr.sin_family = AF_INET;
 284           remoteAddr.sin_port = htons(LISTEN_PORT);
 285           remoteAddr.sin_addr.s_addr = htonl(mid);
 286
 287           if(connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
 288             printf("%s(): Error %d connecting to %s:%d\n", __func__, errno, inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
 289             close(sd);
 290             return -1;
 291           }
 292           sockpool[i].sd[j] = sd;
 293           sockpool[i].inuse[j] = 1;
 294           return sockpool[i].sd[j];
 295         }
 296       }
 297       printf("%s()->Error: Less number of MAX_CONN_PER_MACHINE\n", __func__);
 298       return -1;
 299     }
 300   }
 301   printf("%s()-> Error: Machine id not found\n", __func__);
 302
 303   return -1;
 304 }
 305
 306 int freeSock(sock_pool_t *sockpool, int sd) {
 307   int i;
 308   for (i = 0; i < num_machines; i++) {
 309     int j;
 310     for (j = 0; j < MAX_CONN_PER_MACHINE; j++) {
 311       if (sockpool[i].sd[j] == sd) {
 312         sockpool[i].inuse[j] = 0;
 313         return 0;
 314       }
 315     }
 316   }
 317   printf("%s() Error: Illegal socket descriptor %d\n", __func__, sd);
 318
 319   return -1;
 320 }
 321
 322 #endif