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 resuse
 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