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