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