codemux.c

   1 #include <sys/types.h>
   2 #include <sys/socket.h>
   3 #include <sys/stat.h>
   4 #include <sys/wait.h>
   5 #include <netinet/in.h>
   6 #include <arpa/inet.h>
   7 #include <ctype.h>
   8 #include <errno.h>
   9 #include <fcntl.h>
  10 #include <signal.h>
  11 #include <stdio.h>
  12 #include <stdlib.h>
  13 #include <time.h>
  14 #include <unistd.h>
  15 #include <string.h>
  16 #include "codemuxlib.h"
  17
  18 #define CONF_FILE "/etc/codemux/codemux.conf"
  19 #define DEMUX_PORT 80
  20 #define PIDFILE "/var/run/codemux.pid"
  21 #define TARG_SETSIZE 4096
  22
  23 /* set aside some small number of fds for us, allow the rest for
  24    connections */
  25 #define MAX_CONNS ((TARG_SETSIZE-20)/2)
  26
  27 /* no single service can take more than half the connections */
  28 #define SERVICE_MAX (MAX_CONNS/2)
  29
  30 /* how many total connections before we get concerned about fairness
  31    among them */
  32 #define FAIRNESS_CUTOFF (MAX_CONNS * 0.85)
  33
  34
  35 typedef struct FlowBuf {
  36   int fb_refs;                  /* num refs */
  37   char *fb_buf;                 /* actual buffer */
  38   int fb_used;                  /* bytes used in buffer */
  39 } FlowBuf;
  40 #define FB_SIZE 3800            /* max usable size */
  41 #define FB_ALLOCSIZE 4000       /* extra to include IP address */
  42
  43 typedef struct SockInfo {
  44   int si_peerFd;                /* fd of peer */
  45   struct in_addr si_cliAddr;    /* address of client */
  46   int si_blocked;               /* are we blocked? */
  47   int si_needsHeaderSince;      /* since when are we waiting for a header */
  48   int si_whichService;          /* index of service */
  49   FlowBuf *si_readBuf;          /* read data into this buffer */
  50   FlowBuf *si_writeBuf;         /* drain this buffer for writing */
  51 } SockInfo;
  52
  53 static SockInfo sockInfo[TARG_SETSIZE]; /* fd number of peer socket */
  54
  55 typedef struct ServiceSig {
  56   char *ss_host;                /* suffix in host */
  57   char *ss_slice;
  58   short ss_port;
  59   int ss_slicePos;              /* position in slices array */
  60 } ServiceSig;
  61
  62 static ServiceSig *serviceSig;
  63 static int numServices;
  64 static int confFileReadTime;
  65 static int now;
  66
  67 typedef struct SliceInfo {
  68   char *si_sliceName;
  69   int si_inUse;                 /* do any services refer to this? */
  70   int si_numConns;
  71   int si_xid;
  72 } SliceInfo;
  73
  74 static SliceInfo *slices;
  75 static int numSlices;
  76 static int numActiveSlices;
  77 static int numTotalSliceConns;
  78 static int anySliceXidsNeeded;
  79
  80 typedef struct OurFDSet {
  81   long __fds_bits[TARG_SETSIZE/32];
  82 } OurFDSet;
  83 static OurFDSet masterReadSet, masterWriteSet;
  84 static int highestSetFd;
  85 static int numNeedingHeaders;   /* how many conns waiting on headers? */
  86
  87 static int numForks;
  88
  89 #ifndef SO_SETXID
  90 #define SO_SETXID SO_PEERCRED
  91 #endif
  92 /*-----------------------------------------------------------------*/
  93 static SliceInfo *
  94 ServiceToSlice(int whichService)
  95 {
  96   if (whichService < 0)
  97     return(NULL);
  98   return(&slices[serviceSig[whichService].ss_slicePos]);
  99 }
 100 /*-----------------------------------------------------------------*/
 101 static void
 102 DumpStatus(int fd)
 103 {
 104   char buf[65535];
 105   char *start = buf;
 106   int i;
 107   int len;
 108
 109   sprintf(start,
 110           "numForks %d, numActiveSlices %d, numTotalSliceConns %d\n"
 111           "numNeedingHeaders %d, anySliceXidsNeeded %d\n",
 112           numForks, numActiveSlices, numTotalSliceConns,
 113           numNeedingHeaders, anySliceXidsNeeded);
 114   start += strlen(start);
 115
 116   for (i = 0; i < numSlices; i++) {
 117     SliceInfo *si = &slices[i];
 118     sprintf(start, "Slice %d: %s xid %d, %d conns, inUse %d\n",
 119             i, si->si_sliceName, si->si_xid, si->si_numConns,
 120             si->si_inUse);
 121     start += strlen(start);
 122   }
 123
 124   for (i = 0; i < numServices; i++) {
 125     ServiceSig *ss = &serviceSig[i];
 126     sprintf(start, "Service %d: %s %s port %d, slice# %d\n", i, ss->ss_host,
 127             ss->ss_slice, (int) ss->ss_port, ss->ss_slicePos);
 128     start += strlen(start);
 129   }
 130
 131   len = start - buf;
 132   write(fd, buf, len);
 133 }
 134 /*-----------------------------------------------------------------*/
 135 static void
 136 GetSliceXids(void)
 137 {
 138   /* walks through /etc/passwd, and gets the uid for every slice we
 139      have */
 140   FILE *f;
 141   char *line;
 142   int i;
 143
 144   if (!anySliceXidsNeeded)
 145     return;
 146
 147   for (i = 0; i < numSlices; i++) {
 148     SliceInfo *si = &slices[i];
 149     si->si_inUse = 0;
 150   }
 151   for (i = 0; i < numServices; i++) {
 152     SliceInfo *si = ServiceToSlice(i);
 153     if (si != NULL)
 154       si->si_inUse++;
 155   }
 156
 157   if ((f = fopen("/etc/passwd", "r")) == NULL)
 158     return;
 159
 160   while ((line = GetNextLine(f)) != NULL) {
 161     char *temp;
 162     int xid;
 163
 164     if ((temp = strchr(line, ':')) == NULL)
 165       continue;                 /* weird line */
 166     *temp = '\0';               /* terminate slice name */
 167     temp++;
 168     if ((temp = strchr(temp+1, ':')) == NULL)
 169       continue;                 /* weird line */
 170     if ((xid = atoi(temp+1)) < 1)
 171       continue;                 /* weird xid */
 172
 173     /* we've got a slice name and xid, let's try to match */
 174     for (i = 0; i < numSlices; i++) {
 175       if (slices[i].si_xid == 0 &&
 176           strcasecmp(slices[i].si_sliceName, line) == 0) {
 177         slices[i].si_xid = xid;
 178         break;
 179       }
 180     }
 181   }
 182
 183   /* assume service 0 is the root service, and don't check it since
 184      it'll have xid zero */
 185   anySliceXidsNeeded = FALSE;
 186   for (i = 1; i < numSlices; i++) {
 187     if (slices[i].si_xid == 0 && slices[i].si_inUse > 0) {
 188       anySliceXidsNeeded = TRUE;
 189       break;
 190     }
 191   }
 192
 193   fclose(f);
 194 }
 195 /*-----------------------------------------------------------------*/
 196 static void
 197 SliceConnsInc(int whichService)
 198 {
 199   SliceInfo *si = ServiceToSlice(whichService);
 200
 201   if (si == NULL)
 202     return;
 203   numTotalSliceConns++;
 204   si->si_numConns++;
 205   if (si->si_numConns == 1)
 206     numActiveSlices++;
 207 }
 208 /*-----------------------------------------------------------------*/
 209 static void
 210 SliceConnsDec(int whichService)
 211 {
 212   SliceInfo *si = ServiceToSlice(whichService);
 213
 214   if (si == NULL)
 215     return;
 216   numTotalSliceConns--;
 217   si->si_numConns--;
 218   if (si->si_numConns == 0)
 219     numActiveSlices--;
 220 }
 221 /*-----------------------------------------------------------------*/
 222 static int
 223 WhichSlicePos(char *slice)
 224 {
 225   /* adds the new slice if necessary, returns the index into slice
 226      array. Never change the ordering of existing slices */
 227   int i;
 228   static int numSlicesAlloc;
 229
 230   for (i = 0; i < numSlices; i++) {
 231     if (strcasecmp(slice, slices[i].si_sliceName) == 0)
 232       return(i);
 233   }
 234
 235   if (numSlices >= numSlicesAlloc) {
 236     numSlicesAlloc = MAX(8, numSlicesAlloc * 2);
 237     slices = realloc(slices, numSlicesAlloc * sizeof(SliceInfo));
 238   }
 239
 240   memset(&slices[numSlices], 0, sizeof(SliceInfo));
 241   slices[numSlices].si_sliceName = strdup(slice);
 242   numSlices++;
 243   return(numSlices-1);
 244 }
 245 /*-----------------------------------------------------------------*/
 246 static void
 247 ReadConfFile(void)
 248 {
 249   int numAlloc = 0;
 250   int num = 0;
 251   ServiceSig *servs = NULL;
 252   FILE *f;
 253   char *line = NULL;
 254   struct stat statBuf;
 255   int i;
 256
 257   if (stat(CONF_FILE, &statBuf) != 0) {
 258     fprintf(stderr, "failed stat on codemux.conf\n");
 259     if (numServices)
 260       return;
 261     exit(-1);
 262   }
 263   if (statBuf.st_mtime == confFileReadTime)
 264     return;
 265
 266   if ((f = fopen(CONF_FILE, "r")) == NULL) {
 267     fprintf(stderr, "failed reading codemux.conf\n");
 268     if (numServices)
 269       return;
 270     exit(-1);
 271   }
 272
 273   /* conf file entries look like
 274      coblitz.codeen.org princeton_coblitz 3125
 275   */
 276
 277   while (1) {
 278     ServiceSig serv;
 279     int port;
 280     if (line != NULL)
 281       free(line);
 282
 283     if ((line = GetNextLine(f)) == NULL)
 284       break;
 285
 286     memset(&serv, 0, sizeof(serv));
 287     if (WordCount(line) != 3) {
 288       fprintf(stderr, "bad line: %s\n", line);
 289       continue;
 290     }
 291     serv.ss_port = port = atoi(GetField(line, 2));
 292     if (port < 1 || port > 65535 || port == DEMUX_PORT) {
 293       fprintf(stderr, "bad port: %s\n", line);
 294       continue;
 295     }
 296
 297     serv.ss_host = GetWord(line, 0);
 298     serv.ss_slice = GetWord(line, 1);
 299
 300     if (num == 0 && /* the first row must be an entry for apache */
 301         (strcmp(serv.ss_host, "*") != 0 ||
 302          strcmp(serv.ss_slice, "root") != 0)) {
 303       fprintf(stderr, "first row has to be for webserver\n");
 304       exit(-1);
 305     }
 306     if (num >= numAlloc) {
 307       numAlloc = MAX(numAlloc * 2, 8);
 308       servs = realloc(servs, numAlloc * sizeof(ServiceSig));
 309     }
 310     serv.ss_slicePos = WhichSlicePos(serv.ss_slice);
 311     if (slices[serv.ss_slicePos].si_inUse == 0 &&
 312         slices[serv.ss_slicePos].si_xid < 1)
 313       anySliceXidsNeeded = TRUE; /* if new/inactive, we need xid */
 314     servs[num] = serv;
 315     num++;
 316   }
 317
 318   fclose(f);
 319
 320   if (num == 1) {
 321     if (numServices == 0) {
 322       fprintf(stderr, "nothing found in codemux.conf\n");
 323       exit(-1);
 324     }
 325     return;
 326   }
 327
 328   for (i = 0; i < numServices; i++) {
 329     free(serviceSig[i].ss_host);
 330     free(serviceSig[i].ss_slice);
 331   }
 332   free(serviceSig);
 333   serviceSig = servs;
 334   numServices = num;
 335   confFileReadTime = statBuf.st_mtime;
 336 }
 337 /*-----------------------------------------------------------------*/
 338 static char *err400BadRequest =
 339 "HTTP/1.0 400 Bad Request\r\n"
 340 "Content-Type: text/html\r\n"
 341 "\r\n"
 342 "You are trying to access a PlanetLab node, and your\n"
 343 "request header exceeded the allowable size. Please\n"
 344 "try again if you believe this error is temporary.\n";
 345 /*-----------------------------------------------------------------*/
 346 static char *err503Unavailable =
 347 "HTTP/1.0 503 Service Unavailable\r\n"
 348 "Content-Type: text/html\r\n"
 349 "\r\n"
 350 "You are trying to access a PlanetLab node, but the service\n"
 351 "seems to be unavailable at the moment. Please try again.\n";
 352 /*-----------------------------------------------------------------*/
 353 static char *err503TooBusy =
 354 "HTTP/1.0 503 Service Unavailable\r\n"
 355 "Content-Type: text/html\r\n"
 356 "\r\n"
 357 "You are trying to access a PlanetLab node, but the service\n"
 358 "seems to be overloaded at the moment. Please try again.\n";
 359 /*-----------------------------------------------------------------*/
 360 static void
 361 SetFd(int fd, OurFDSet *set)
 362 {
 363   if (highestSetFd < fd)
 364     highestSetFd = fd;
 365   FD_SET(fd, set);
 366 }
 367 /*-----------------------------------------------------------------*/
 368 static void
 369 ClearFd(int fd, OurFDSet *set)
 370 {
 371   FD_CLR(fd, set);
 372 }
 373 /*-----------------------------------------------------------------*/
 374 static int
 375 RemoveHeader(char *lower, char *real, int totalSize, char *header)
 376 {
 377   /* returns number of characters removed */
 378   char h2[256];
 379   int start, end, len;
 380   char *temp, *conn;
 381
 382   sprintf(h2, "\n%s", header);
 383
 384   if ((conn = strstr(lower, h2)) == NULL)
 385     return(0);
 386
 387   conn++;
 388   /* determine how many characters to remove */
 389   if ((temp = strchr(conn, '\n')) != NULL)
 390     len = (temp - conn) + 1;
 391   else
 392     len = strlen(conn) + 1;
 393   start = conn - lower;
 394   end = start + len;
 395   memmove(&real[start], &real[end], totalSize - end);
 396   memmove(&lower[start], &lower[end], totalSize - end);
 397
 398   return(len);
 399 }
 400 /*-----------------------------------------------------------------*/
 401 static int
 402 InsertHeader(char *buf, int totalSize, char *header)
 403 {
 404   /* returns number of bytes inserted */
 405
 406   char h2[256];
 407   char *temp;
 408   int len;
 409
 410   sprintf(h2, "%s\r\n", header);
 411   len = strlen(h2);
 412
 413   /* if we don't encounter a \n, it means that we have only a single
 414      line, and we'd converted the \n to a \0 */
 415   if ((temp = strchr(buf, '\n')) == NULL)
 416     temp = strchr(buf, '\0');
 417   temp++;
 418
 419   memmove(temp + len, temp, totalSize - (temp - buf));
 420   memcpy(temp, h2, len);
 421
 422   return(len);
 423 }
 424 /*-----------------------------------------------------------------*/
 425 static int
 426 FindService(FlowBuf *fb, int *whichService, struct in_addr addr)
 427 {
 428   char *end;
 429   char *lowerBuf;
 430   char *hostVal;
 431   char *buf = fb->fb_buf;
 432   char orig[256];
 433 #if 0
 434   char *url;
 435   int i;
 436   int len;
 437 #endif
 438
 439   if (strstr(buf, "\n\r\n") == NULL && strstr(buf, "\n\n") == NULL)
 440     return(FAILURE);
 441
 442   /* insert client info after first line */
 443   sprintf(orig, "X-CoDemux-Client: %s", inet_ntoa(addr));
 444   fb->fb_used += InsertHeader(buf, fb->fb_used + 1, orig);
 445
 446   /* get just the header, so we can work on it */
 447   LOCAL_STR_DUP_LOWER(lowerBuf, buf);
 448   if ((end = strstr(lowerBuf, "\n\r\n")) == NULL)
 449     end = strstr(lowerBuf, "\n\n");
 450   *end = '\0';
 451
 452   /* remove any existing connection, keep-alive headers, add ours */
 453   fb->fb_used -= RemoveHeader(lowerBuf, buf, fb->fb_used + 1, "keep-alive:");
 454   fb->fb_used -= RemoveHeader(lowerBuf, buf, fb->fb_used + 1, "connection:");
 455   fb->fb_used += InsertHeader(buf, fb->fb_used + 1, "Connection: close");
 456   InsertHeader(lowerBuf, fb->fb_used + 1, "connection: close");
 457
 458   /* isolate host, see if it matches */
 459   if ((hostVal = strstr(lowerBuf, "\nhost:")) != NULL) {
 460     int i;
 461     hostVal += strlen("\nhost:");
 462     if ((end = strchr(hostVal, '\n')) != NULL)
 463       *end = '\0';
 464     if ((end = strchr(hostVal, ':')) != NULL)
 465       *end = '\0';
 466     while (isspace(*hostVal))
 467       hostVal++;
 468     if (strlen(hostVal) > 0) {
 469       hostVal = GetWord(hostVal, 0);
 470       for (i = 1; i < numServices; i++) {
 471         if (serviceSig[i].ss_host != NULL &&
 472             DoesDotlessSuffixMatch(hostVal, 0, serviceSig[i].ss_host)) {
 473           *whichService = i;
 474           free(hostVal);
 475           /* printf("%s", buf); */
 476           return(SUCCESS);
 477         }
 478       }
 479       free(hostVal);
 480     }
 481   }
 482
 483 #if 0
 484   /* see if URL prefix matches */
 485   if ((end = strchr(lowerBuf, '\n')) != NULL)
 486     *end = 0;
 487   if ((url = GetField(lowerBuf, 1)) == NULL ||
 488       url[0] != '/') {
 489     /* bad request - let apache handle it ? */
 490     *whichService = 0;
 491     return(SUCCESS);
 492   }
 493   url++;                        /* skip the leading slash */
 494   for (i = 1; i < numServices; i++) {
 495     if (serviceSig[i].ss_prefix != NULL &&
 496         (len = strlen(serviceSig[i].ss_prefix)) > 0 &&
 497         strncmp(url, serviceSig[i].ss_prefix, len) == 0 &&
 498         (url[len] == ' ' || url[len] == '/')) {
 499       int startPos = url - lowerBuf;
 500       int stripLen = len + ((url[len] == '/') ? 1 : 0);
 501       /* strip out prefix */
 502       fb->fb_used -= stripLen;
 503       memmove(&buf[startPos], &buf[startPos+stripLen],
 504               fb->fb_used + 1 - startPos);
 505       /* printf("%s", buf); */
 506       *whichService = i;
 507       return(SUCCESS);
 508     }
 509   }
 510 #endif
 511
 512   /* default to first service */
 513   *whichService = 0;
 514   return(SUCCESS);
 515 }
 516 /*-----------------------------------------------------------------*/
 517 static int
 518 StartConnect(int origFD, int whichService)
 519 {
 520   int sock;
 521   struct sockaddr_in dest;
 522   SockInfo *si;
 523
 524   /* create socket */
 525   if ((sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
 526     return(FAILURE);
 527   }
 528
 529   /* make socket non-blocking */
 530   if (fcntl(sock, F_SETFL, O_NONBLOCK) < 0) {
 531     close(sock);
 532     return(FAILURE);
 533   }
 534
 535   /* set addr structure */
 536   memset(&dest, 0, sizeof(dest));
 537   dest.sin_family = AF_INET;
 538   dest.sin_port = htons(serviceSig[whichService].ss_port);
 539   dest.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
 540
 541   /* start connection process - we should be told that it's in
 542      progress */
 543   if (connect(sock, (struct sockaddr *) &dest, sizeof(dest)) != -1 ||
 544       errno != EINPROGRESS) {
 545     close(sock);
 546     return(FAILURE);
 547   }
 548
 549   SetFd(sock, &masterWriteSet); /* determine when connect finishes */
 550   sockInfo[origFD].si_peerFd = sock;
 551   si = &sockInfo[sock];
 552   memset(si, 0, sizeof(SockInfo));
 553   si->si_peerFd = origFD;
 554   si->si_blocked = TRUE;        /* still connecting */
 555   si->si_whichService = whichService;
 556   si->si_writeBuf = sockInfo[origFD].si_readBuf;
 557   sockInfo[origFD].si_readBuf->fb_refs++;
 558   if (whichService >= 0)
 559     SliceConnsInc(whichService);
 560
 561   return(SUCCESS);
 562 }
 563 /*-----------------------------------------------------------------*/
 564 static int
 565 WriteAvailData(int fd)
 566 {
 567   SockInfo *si = &sockInfo[fd];
 568   FlowBuf *fb = si->si_writeBuf;
 569   int res;
 570
 571   /* printf("trying to write fd %d\n", fd); */
 572   if (fb->fb_used < 1 || si->si_blocked)
 573     return(SUCCESS);
 574
 575   /* printf("trying to write %d bytes\n", fb->fb_used); */
 576   /* write(STDOUT_FILENO, fb->fb_buf, fb->fb_used); */
 577   if ((res = write(fd, fb->fb_buf, fb->fb_used)) > 0) {
 578     fb->fb_used -= res;
 579     if (fb->fb_used > 0) {
 580       /* couldn't write all - assume blocked */
 581       memmove(fb->fb_buf, &fb->fb_buf[res], fb->fb_used);
 582       si->si_blocked = TRUE;
 583       SetFd(fd, &masterWriteSet);
 584     }
 585     /* printf("wrote %d\n", res); */
 586     return(SUCCESS);
 587   }
 588
 589   /* we might have been full but didn't realize it */
 590   if (res == -1 && errno == EAGAIN) {
 591     si->si_blocked = TRUE;
 592     SetFd(fd, &masterWriteSet);
 593     return(SUCCESS);
 594   }
 595
 596   /* otherwise, assume the worst */
 597   return(FAILURE);
 598 }
 599 /*-----------------------------------------------------------------*/
 600 static OurFDSet socksToCloseVec;
 601 static int numSocksToClose;
 602 static int whichSocksToClose[TARG_SETSIZE];
 603 /*-----------------------------------------------------------------*/
 604 static void
 605 CloseSock(int fd)
 606 {
 607   if (FD_ISSET(fd, &socksToCloseVec))
 608     return;
 609   SetFd(fd, &socksToCloseVec);
 610   whichSocksToClose[numSocksToClose] = fd;
 611   numSocksToClose++;
 612 }
 613 /*-----------------------------------------------------------------*/
 614 static void
 615 DecBuf(FlowBuf *buf)
 616 {
 617   if (buf == NULL)
 618     return;
 619   buf->fb_refs--;
 620   if (buf->fb_refs == 0) {
 621     free(buf->fb_buf);
 622     free(buf);
 623   }
 624 }
 625 /*-----------------------------------------------------------------*/
 626 static void
 627 ReallyCloseSocks(void)
 628 {
 629   int i;
 630
 631   memset(&socksToCloseVec, 0, sizeof(socksToCloseVec));
 632
 633   for (i = 0; i < numSocksToClose; i++) {
 634     int fd = whichSocksToClose[i];
 635     close(fd);
 636     DecBuf(sockInfo[fd].si_readBuf);
 637     DecBuf(sockInfo[fd].si_writeBuf);
 638     ClearFd(fd, &masterReadSet);
 639     ClearFd(fd, &masterWriteSet);
 640     if (sockInfo[fd].si_needsHeaderSince) {
 641       sockInfo[fd].si_needsHeaderSince = 0;
 642       numNeedingHeaders--;
 643     }
 644     if (sockInfo[fd].si_whichService >= 0) {
 645       SliceConnsDec(sockInfo[fd].si_whichService);
 646       sockInfo[fd].si_whichService = -1;
 647     }
 648   }
 649   numSocksToClose = 0;
 650 }
 651 /*-----------------------------------------------------------------*/
 652 static void
 653 SocketReadyToRead(int fd)
 654 {
 655   SockInfo *si = &sockInfo[fd];
 656   int spaceLeft;
 657   FlowBuf *fb;
 658   int res;
 659
 660   /* if peer is closed, close ourselves */
 661   if (si->si_peerFd < 0 && (!si->si_needsHeaderSince)) {
 662     CloseSock(fd);
 663     return;
 664   }
 665
 666   if ((fb = si->si_readBuf) == NULL) {
 667     fb = si->si_readBuf = calloc(1, sizeof(FlowBuf));
 668     fb->fb_refs = 1;
 669     if (si->si_peerFd >= 0) {
 670       sockInfo[si->si_peerFd].si_writeBuf = fb;
 671       fb->fb_refs = 2;
 672     }
 673   }
 674
 675   if (fb->fb_buf == NULL)
 676     fb->fb_buf = malloc(FB_ALLOCSIZE);
 677
 678   /* determine read buffer size - if 0, then block reads and return */
 679   if ((spaceLeft = FB_SIZE - fb->fb_used) < 0) {
 680     if (si->si_needsHeaderSince) {
 681       write(fd, err400BadRequest, strlen(err400BadRequest));
 682       CloseSock(fd);
 683       return;
 684     }
 685     else {
 686       ClearFd(fd, &masterReadSet);
 687       return;
 688     }
 689   }
 690
 691   /* read as much as allowed, and is available */
 692   if ((res = read(fd, &fb->fb_buf[fb->fb_used], spaceLeft)) == 0) {
 693     CloseSock(fd);
 694     if (fb->fb_used == 0 && si->si_peerFd >= 0) {
 695       CloseSock(si->si_peerFd);
 696       si->si_peerFd = -1;
 697     }
 698     return;
 699   }
 700   if (res == -1) {
 701     if (errno == EAGAIN)
 702       return;
 703     CloseSock(fd);
 704     if (fb->fb_used == 0 && si->si_peerFd >= 0) {
 705       CloseSock(si->si_peerFd);
 706       si->si_peerFd = -1;
 707     }
 708     return;
 709   }
 710   fb->fb_used += res;
 711   fb->fb_buf[fb->fb_used] = 0;  /* terminate it for convenience */
 712   printf("sock %d, read %d, total %d\n", fd, res, fb->fb_used);
 713
 714   /* if we need header, check if we've gotten it. if so, do
 715      modifications and continue. if not, check if we've read the
 716      maximum, and if so, fail */
 717   if (si->si_needsHeaderSince) {
 718     int whichService;
 719     SliceInfo *slice;
 720
 721 #define STATUS_REQ "GET /codemux/status.txt"
 722     if (strncasecmp(fb->fb_buf, STATUS_REQ, sizeof(STATUS_REQ)-1) == 0) {
 723       DumpStatus(fd);
 724       CloseSock(fd);
 725       return;
 726     }
 727
 728     //    printf("trying to find service\n");
 729     if (FindService(fb, &whichService, si->si_cliAddr) != SUCCESS)
 730       return;
 731     //    printf("found service %d\n", whichService);
 732     slice = ServiceToSlice(whichService);
 733
 734     /* no service can have more than some absolute max number of
 735        connections. Also, when we're too busy, start enforcing
 736        fairness across the servers */
 737     if (slice->si_numConns > SERVICE_MAX ||
 738         (numTotalSliceConns > FAIRNESS_CUTOFF &&
 739          slice->si_numConns > MAX_CONNS/numActiveSlices)) {
 740       write(fd, err503TooBusy, strlen(err503TooBusy));
 741       CloseSock(fd);
 742       return;
 743     }
 744
 745     if (slice->si_xid > 0) {
 746       static int first = 1;
 747       setsockopt(fd, SOL_SOCKET, SO_SETXID,
 748                  &slice->si_xid, sizeof(slice->si_xid));
 749       if (first) {
 750         /* just to log it for once */
 751         fprintf(stderr, "setsockopt() with XID = %d name = %s\n",
 752                 slice->si_xid, slice->si_sliceName);
 753         first = 0;
 754       }
 755     }
 756
 757     si->si_needsHeaderSince = 0;
 758     numNeedingHeaders--;
 759     if (StartConnect(fd, whichService) != SUCCESS) {
 760       write(fd, err503Unavailable, strlen(err503Unavailable));
 761       CloseSock(fd);
 762       return;
 763     }
 764     return;
 765   }
 766
 767   /* write anything possible */
 768   if (WriteAvailData(si->si_peerFd) != SUCCESS) {
 769     /* assume the worst and close */
 770     CloseSock(fd);
 771     CloseSock(si->si_peerFd);
 772     si->si_peerFd = -1;
 773   }
 774 }
 775 /*-----------------------------------------------------------------*/
 776 static void
 777 SocketReadyToWrite(int fd)
 778 {
 779   SockInfo *si = &sockInfo[fd];
 780
 781   /* unblock it and read what it has */
 782   si->si_blocked = FALSE;
 783   ClearFd(fd, &masterWriteSet);
 784   SetFd(fd, &masterReadSet);
 785
 786   /* enable reading on peer just in case it was off */
 787   if (si->si_peerFd >= 0)
 788     SetFd(si->si_peerFd, &masterReadSet);
 789
 790   /* if we have data, write it */
 791   if (WriteAvailData(fd) != SUCCESS) {
 792    /* assume the worst and close */
 793     CloseSock(fd);
 794     if (si->si_peerFd >= 0) {
 795       CloseSock(si->si_peerFd);
 796       si->si_peerFd = -1;
 797     }
 798     return;
 799   }
 800
 801   /* if peer is closed and we're done writing, we should close */
 802   if (si->si_peerFd < 0 && si->si_writeBuf->fb_used == 0)
 803     CloseSock(fd);
 804 }
 805 /*-----------------------------------------------------------------*/
 806 static void
 807 CloseReqlessConns(void)
 808 {
 809   static int lastSweep;
 810   int maxAge;
 811   int i;
 812
 813   if (lastSweep == now)
 814     return;
 815   lastSweep = now;
 816
 817   if (numTotalSliceConns + numNeedingHeaders > MAX_CONNS ||
 818       numNeedingHeaders > TARG_SETSIZE/20) {
 819     /* second condition is probably an attack - close aggressively */
 820     maxAge = 5;
 821   }
 822   else if (numTotalSliceConns + numNeedingHeaders > FAIRNESS_CUTOFF ||
 823            numNeedingHeaders > TARG_SETSIZE/40) {
 824     /* sweep a little aggressively */
 825     maxAge = 10;
 826   }
 827   else if (numNeedingHeaders > TARG_SETSIZE/80) {
 828     /* just sweep to close strays */
 829     maxAge = 30;
 830   }
 831   else {
 832     /* too little gained - not worth sweeping */
 833     return;
 834   }
 835
 836   /* if it's too old, close it */
 837   for (i = 0; i < highestSetFd+1; i++) {
 838     if (sockInfo[i].si_needsHeaderSince &&
 839         (now - sockInfo[i].si_needsHeaderSince) > maxAge)
 840       CloseSock(i);
 841   }
 842 }
 843 /*-----------------------------------------------------------------*/
 844 static void
 845 MainLoop(int lisSock)
 846 {
 847   int i;
 848   OurFDSet tempReadSet, tempWriteSet;
 849   int res;
 850   int lastConfCheck = 0;
 851
 852   signal(SIGPIPE, SIG_IGN);
 853
 854   while (1) {
 855     int newSock;
 856     int ceiling;
 857     struct timeval timeout;
 858
 859     now = time(NULL);
 860
 861     if (now - lastConfCheck > 300) {
 862       ReadConfFile();
 863       GetSliceXids();           /* always call - in case new slices created */
 864       lastConfCheck = now;
 865     }
 866
 867     /* see if there's any activity */
 868     tempReadSet = masterReadSet;
 869     tempWriteSet = masterWriteSet;
 870
 871     /* trim it down if needed */
 872     while (highestSetFd > 1 &&
 873            (!FD_ISSET(highestSetFd, &tempReadSet)) &&
 874            (!FD_ISSET(highestSetFd, &tempWriteSet)))
 875       highestSetFd--;
 876     timeout.tv_sec = 1;
 877     timeout.tv_usec = 0;
 878     res = select(highestSetFd+1, (fd_set *) &tempReadSet,
 879                  (fd_set *) &tempWriteSet, NULL, &timeout);
 880     if (res < 0 && errno != EINTR) {
 881       perror("select");
 882       exit(-1);
 883     }
 884
 885     now = time(NULL);
 886
 887     /* clear the bit for listen socket to avoid confusion */
 888     ClearFd(lisSock, &tempReadSet);
 889
 890     ceiling = highestSetFd+1;   /* copy it, since it changes during loop */
 891     /* pass data back and forth as needed */
 892     for (i = 0; i < ceiling; i++) {
 893       if (FD_ISSET(i, &tempWriteSet))
 894         SocketReadyToWrite(i);
 895     }
 896     for (i = 0; i < ceiling; i++) {
 897       if (FD_ISSET(i, &tempReadSet))
 898         SocketReadyToRead(i);
 899     }
 900
 901     /* see if we need to close conns w/o requests */
 902     CloseReqlessConns();
 903
 904     /* do all closes */
 905     ReallyCloseSocks();
 906
 907     /* try accepting new connections */
 908     do {
 909       struct sockaddr_in addr;
 910       socklen_t lenAddr = sizeof(addr);
 911       if ((newSock = accept(lisSock, (struct sockaddr *) &addr,
 912                             &lenAddr)) >= 0) {
 913         memset(&sockInfo[newSock], 0, sizeof(SockInfo));
 914         sockInfo[newSock].si_needsHeaderSince = now;
 915         numNeedingHeaders++;
 916         sockInfo[newSock].si_peerFd = -1;
 917         sockInfo[newSock].si_cliAddr = addr.sin_addr;
 918         sockInfo[newSock].si_whichService = -1;
 919         SetFd(newSock, &masterReadSet);
 920       }
 921     } while (newSock >= 0);
 922   }
 923 }
 924 /*-----------------------------------------------------------------*/
 925 static int
 926 InitDaemon(void)
 927 {
 928   pid_t pid;
 929   FILE *pidfile;
 930
 931   pidfile = fopen(PIDFILE, "w");
 932   if (pidfile == NULL) {
 933     fprintf(stderr, "%s creation failed\n", PIDFILE);
 934     return(-1);
 935   }
 936
 937   if ((pid = fork()) < 0) {
 938     fclose(pidfile);
 939     return(-1);
 940   }
 941   else if (pid != 0) {
 942     /* i'm the parent, writing down the child pid  */
 943     fprintf(pidfile, "%u\n", pid);
 944     fclose(pidfile);
 945     exit(0);
 946   }
 947
 948   /* close the pid file */
 949   fclose(pidfile);
 950
 951   /* routines for any daemon process
 952      1. create a new session
 953      2. change directory to the root
 954      3. change the file creation permission
 955   */
 956   setsid();
 957   chdir("/");
 958   umask(0);
 959
 960   return(0);
 961 }
 962 /*-----------------------------------------------------------------*/
 963 static int
 964 OpenLogFile(void)
 965 {
 966   static const char* logfile = "/var/log/codemux.log";
 967   static const char* oldlogfile = "/var/log/codemux.log.old";
 968   int logfd;
 969
 970   /* if the previous log file exists,
 971      rename it to the oldlogfile */
 972   if (access(logfile, F_OK) == 0) {
 973     if (rename(logfile, oldlogfile) < 0) {
 974       fprintf(stderr, "cannot rotate the logfile err=%s\n",
 975               strerror(errno));
 976       exit(-1);
 977     }
 978   }
 979
 980   logfd = open(logfile, O_WRONLY | O_APPEND | O_CREAT, 0600);
 981   if (logfd < 0) {
 982     fprintf(stderr, "cannot open the logfile err=%s\n",
 983             strerror(errno));
 984     exit(-1);
 985   }
 986
 987   /* duplicate logfile to stderr */
 988   if (dup2(logfd, STDERR_FILENO) != STDERR_FILENO) {
 989     fprintf(stderr, "cannot open the logfile err=%s\n",
 990             strerror(errno));
 991     exit(-1);
 992   }
 993
 994   /* set the close-on-exec flag */
 995   if (fcntl(STDERR_FILENO, F_SETFD, 1) != 0) {
 996     fprintf(stderr, "fcntl to set the close-on-exec flag failed err=%s\n",
 997             strerror(errno));
 998     exit(-1);
 999   }
1000
1001   return logfd;
1002 }
1003 /*-----------------------------------------------------------------*/
1004 int
1005 main(int argc, char *argv[])
1006 {
1007   int lisSock;
1008   int logFd;
1009
1010   /* do the daemon stuff */
1011   if (argc <= 1 || strcmp(argv[1], "-d") != 0) {
1012     if (InitDaemon() < 0) {
1013       fprintf(stderr, "codemux daemon_init() failed\n");
1014       exit(-1);
1015     }
1016   }
1017
1018   /* create the accept socket */
1019   if ((lisSock = CreatePrivateAcceptSocket(DEMUX_PORT, TRUE)) < 0) {
1020     fprintf(stderr, "failed creating accept socket\n");
1021     exit(-1);
1022   }
1023   SetFd(lisSock, &masterReadSet);
1024
1025   /* open the log file */
1026   logFd = OpenLogFile();
1027
1028   while (1) {
1029     numForks++;
1030     if (fork()) {
1031       /* this is the parent - just wait */
1032       while (wait3(NULL, 0, NULL) < 1)
1033         ;                       /* just keep waiting for a real pid */
1034     }
1035     else {
1036       /* child process */
1037       MainLoop(lisSock);
1038       exit(-1);
1039     }
1040   }
1041 }
1042 /*-----------------------------------------------------------------*/