core.c

一个WEB服务器的性能测试工具

	}
      memcpy (&sin.sin_addr, he->h_addr_list[0], sizeof (sin.sin_addr));
    }
  else
    {
      if (!inet_aton (server, &sin.sin_addr))
	{
	  fprintf (stderr, "%s.core_addr_intern: invalid server address %s\n",
		   prog_name, server);
	  exit (1);
	}
    }
  h = hash_enter (server, server_len, port, &sin);
  if (!h)
    return 0;
  return &h->sin;
}

void
core_init (void)
{
  struct rlimit rlimit;

  memset (&hash_table, 0, sizeof (hash_table));
  memset (&rdfds, 0, sizeof (rdfds));
  memset (&wrfds, 0, sizeof (wrfds));
  memset (&myaddr, 0, sizeof (myaddr));
  memset (&port_free_map, 0xff, sizeof (port_free_map));

  /* Don't disturb just because a TCP connection closed on us... */
  signal (SIGPIPE, SIG_IGN);

#ifdef DONT_POLL
  /* This causes select() to take several milliseconds on both
     Linux/x86 and HP-UX 10.20.  */
  select_timeout.tv_sec  = (u_long) TIMER_INTERVAL;
  select_timeout.tv_usec = (u_long) (TIMER_INTERVAL * 1e6);
#else
  /* This causes httperf to become a CPU hog as it polls for
     filedescriptors to become readable/writable.  This is OK as long
     as httperf is the only (interesting) user-level process that
     executes on a machine.  */
  select_timeout.tv_sec  = 0;
  select_timeout.tv_usec = 0;
#endif

  /* boost open file limit to the max: */
  if (getrlimit (RLIMIT_NOFILE, &rlimit) < 0)
    {
      fprintf (stderr, "%s: failed to get number of open file limit: %s",
	       prog_name, strerror (errno));
      exit (1);
    }

  if (rlimit.rlim_max > FD_SETSIZE)
    {
      fprintf (stderr, "%s: warning: open file limit > FD_SETSIZE; "
	       "limiting max. # of open files to FD_SETSIZE\n", prog_name);
      rlimit.rlim_max = FD_SETSIZE;
    }

  rlimit.rlim_cur = rlimit.rlim_max;
  if (setrlimit (RLIMIT_NOFILE, &rlimit) < 0)
    {
      fprintf (stderr, "%s: failed to increase number of open file limit: %s",
	       prog_name, strerror (errno));
      exit (1);
    }

  if (verbose)
    printf ("%s: maximum number of open descriptors = %ld\n",
	    prog_name, rlimit.rlim_max);

  if (param.server)
    core_addr_intern (param.server, strlen (param.server), param.port);
}

#ifdef HAVE_SSL

void
core_ssl_connect (Conn *s)
{
  Any_Type arg;
  int ssl_err;

  if (DBG > 2)
    fprintf (stderr, "core_ssl_connect(conn=%p)\n", (void *) s);

  if (SSL_set_fd (s->ssl, s->sd) == 0)
    {
      ERR_print_errors_fp (stderr);
      exit (-1);
    }

  ssl_err = SSL_connect (s->ssl);
  if (ssl_err < 0)
    {
      int reason = SSL_get_error(s->ssl, ssl_err);

      if (reason == SSL_ERROR_WANT_READ || reason == SSL_ERROR_WANT_WRITE)
	{
	  if (DBG > 2)
	    fprintf (stderr, "core_ssl_connect: want to %s more...\n",
		     (reason == SSL_ERROR_WANT_READ) ? "read" : "write");
	  if (reason == SSL_ERROR_WANT_READ && !FD_ISSET (s->sd, &rdfds))
	    {
	      FD_CLR (s->sd, &wrfds);
	      set_active (s, &rdfds);
	    }
	  else if (reason == SSL_ERROR_WANT_WRITE && !FD_ISSET (s->sd, &wrfds))
	    {
	      FD_CLR (s->sd, &rdfds);
	      set_active (s, &wrfds);
	    }
	  return;
	}
      fprintf (stderr,
	       "%s: failed to connect to SSL server (err=%d, reason=%d)\n",
	       prog_name, ssl_err, reason);
      ERR_print_errors_fp (stderr);
      exit (-1);
    }

  s->state = S_CONNECTED;

  if (DBG > 0)
    fprintf (stderr, "core_ssl_connect: SSL is connected!\n");

  if (DBG > 1)
    {
      SSL_CIPHER *ssl_cipher;

      ssl_cipher = SSL_get_current_cipher (s->ssl);
      if (!ssl_cipher)
	fprintf (stderr, "core_ssl_connect: server refused all client cipher "
		 "suites!\n");
      else
	fprintf (stderr, "core_ssl_connect: cipher=%s, valid=%d, id=%lu\n",
		 ssl_cipher->name, ssl_cipher->valid, ssl_cipher->id);
    }

  arg.l = 0;
  event_signal (EV_CONN_CONNECTED, (Object *) s, arg);
}

#endif /* HAVE_SSL */

int
core_connect (Conn *s)
{
  int sd, result, len, async_errno;
  struct sockaddr_in *sin;
  struct linger linger;
  int myport, optval;
  Any_Type arg;
  static int prev_iteration = -1;
  static u_long burst_len;

  if (iteration == prev_iteration)
    ++burst_len;
  else
    {
      if (burst_len > max_burst_len)
	max_burst_len = burst_len;
      burst_len = 1;
      prev_iteration = iteration;
    }

  SYSCALL (SOCKET,
	   sd = socket (AF_INET, SOCK_STREAM, 0));
  if (sd < 0)
    {
      if (DBG > 0)
	fprintf (stderr, "%s.core_connect.socket: %s (max_sd=%d)\n",
		 prog_name, strerror (errno), max_sd);
      goto failure;
    }

  if (fcntl (sd, F_SETFL, O_NONBLOCK) < 0)
    {
      fprintf (stderr, "%s.core_connect.fcntl: %s\n",
	       prog_name, strerror (errno));
      goto failure;
    }

  if (param.close_with_reset)
    {
      linger.l_onoff = 1;
      linger.l_linger = 0;
      if (setsockopt (sd, SOL_SOCKET, SO_LINGER, &linger, sizeof (linger)) < 0)
	{
	  fprintf (stderr, "%s.core_connect.setsockopt(SO_LINGER): %s\n",
		   prog_name, strerror (errno));
	  goto failure;
	}
    }

  /* Disable Nagle algorithm so we don't delay needlessly when
     pipelining requests.  */
  optval = 1;
  if (setsockopt (sd, SOL_TCP, TCP_NODELAY, &optval, sizeof (optval)) < 0)
    {
      fprintf (stderr, "%s.core_connect.setsockopt(SO_SNDBUF): %s\n",
	       prog_name, strerror (errno));
      goto failure;
    }

  optval = param.send_buffer_size;
  if (setsockopt (sd, SOL_SOCKET, SO_SNDBUF, &optval, sizeof (optval)) < 0)
    {
      fprintf (stderr, "%s.core_connect.setsockopt(SO_SNDBUF): %s\n",
	       prog_name, strerror (errno));
      goto failure;
    }

  optval = param.recv_buffer_size;
  if (setsockopt (sd, SOL_SOCKET, SO_RCVBUF, &optval, sizeof (optval)) < 0)
    {
      fprintf (stderr, "%s.core_connect.setsockopt(SO_SNDBUF): %s\n",
	       prog_name, strerror (errno));
      goto failure;
    }

  s->sd = sd;
  if (sd >= alloced_sd_to_conn)
    {
      size_t size, old_size;

      old_size = alloced_sd_to_conn * sizeof (sd_to_conn[0]);
      alloced_sd_to_conn += 2048;
      size = alloced_sd_to_conn * sizeof (sd_to_conn[0]);
      if (sd_to_conn)
	sd_to_conn = realloc (sd_to_conn, size);
      else
	sd_to_conn = malloc (size);
      if (!sd_to_conn)
	{
	  if (DBG > 0)
	    fprintf (stderr, "%s.core_connect.realloc: %s\n",
		     prog_name, strerror (errno));
	  goto failure;
	}
      memset ((char *) sd_to_conn + old_size, 0, size - old_size);
    }
  assert (!sd_to_conn[sd]);
  sd_to_conn[sd] = s;

  sin = hash_lookup (s->hostname, s->hostname_len, s->port);
  if (!sin)
    {
      if (DBG > 0)
	fprintf (stderr, "%s.core_connect: unknown server/port %s:%d\n",
		 prog_name, s->hostname, s->port);
      goto failure;
    }

  arg.l = 0;
  event_signal (EV_CONN_CONNECTING, (Object *) s, arg);
  if (s->state >= S_CLOSING)
    goto failure;

  if (param.hog)
    {
      while (1)
	{
	  myport = port_get ();
	  if (myport < 0)
	    goto failure;

	  myaddr.sin_port = htons (myport);
	  SYSCALL (BIND,
		   result = bind (sd, &myaddr, sizeof (myaddr)));
	  if (result == 0)
	    break;

	  if (errno != EADDRINUSE && errno == EADDRNOTAVAIL)
	    {
	      if (DBG > 0)
		fprintf (stderr, "%s.core_connect.bind: %s\n",
			 prog_name, strerror (errno));
	      goto failure;
	    }
	}
      s->myport = myport;
    }

  SYSCALL (CONNECT,
	   result = connect (sd, sin, sizeof (*sin)));
  if (result == 0)
    {
#ifdef HAVE_SSL
      if (param.use_ssl)
	  core_ssl_connect (s);
      else
#endif
	{
	  s->state = S_CONNECTED;
	  arg.l = 0;
	  event_signal (EV_CONN_CONNECTED, (Object *) s, arg);
	}
    }
  else if (errno == EINPROGRESS)
    {
      /* The socket becomes writable only after the connection has
	 been established.  Hence we wait for writability to
	 detect connection establishment.  */
      s->state = S_CONNECTING;
      set_active (s, &wrfds);
      if (param.timeout > 0.0)
	{
	  arg.vp = s;
	  assert (!s->watchdog);
	  s->watchdog = timer_schedule (conn_timeout, arg, param.timeout);
	}
    }
  else
    {
      len = sizeof (async_errno);
      if (getsockopt (sd, SOL_SOCKET, SO_ERROR, &async_errno, &len) == 0
	  && async_errno != 0)
	errno = async_errno;

      if (DBG > 0)
	fprintf (stderr, "%s.core_connect.connect: %s (max_sd=%d)\n",
		 prog_name, strerror (errno), max_sd);
      goto failure;
    }
  return 0;

 failure:
  conn_failure (s, errno);
  return -1;
}

int
core_send (Conn *conn, Call *call)
{
  Any_Type arg;

  arg.l = 0;
  event_signal (EV_CALL_ISSUE, (Object *) call, arg);

  call->conn = conn;	/* NO refcounting here (see call.h).  */

  if (param.no_host_hdr)
    {
      call->req.iov[IE_HOST].iov_base = (caddr_t) "";
      call->req.iov[IE_HOST].iov_len = 0;
    }
  else if (!call->req.iov[IE_HOST].iov_base)
    {
      /* Default call's hostname to connection's hostname: */
      call->req.iov[IE_HOST].iov_base = (caddr_t) conn->hostname;
      call->req.iov[IE_HOST].iov_len = conn->hostname_len;
    }

  /* NOTE: the protocol version indicates what the _client_ can
     understand.  If we send HTTP/1.1, it doesn't mean that the server
     has to speak HTTP/1.1.  In other words, sending an HTTP/1.1
     header leaves it up to the server whether it wants to reply with
     a 1.0 or 1.1 reply.  */
  switch (call->req.version)
    {
    case 0x10000:
      if (param.no_host_hdr)
	{
	  call->req.iov[IE_PROTL].iov_base = (caddr_t) http10req_nohost;
	  call->req.iov[IE_PROTL].iov_len = sizeof (http10req_nohost) - 1;
	}
      else
	{
	  call->req.iov[IE_PROTL].iov_base = (caddr_t) http10req;
	  call->req.iov[IE_PROTL].iov_len = sizeof (http10req) - 1;
	}
      break;

    case 0x10001:
      if (param.no_host_hdr)
	{
	  call->req.iov[IE_PROTL].iov_base = http11req_nohost;
	  call->req.iov[IE_PROTL].iov_len = sizeof (http11req_nohost) - 1;
	}
      else
	{
	  call->req.iov[IE_PROTL].iov_base = http11req;
	  call->req.iov[IE_PROTL].iov_len = sizeof (http11req) - 1;
	}
      break;

    default:
      fprintf (stderr, "%s: unexpected version code %x\n",
	       prog_name, call->req.version);
      exit (1);
    }
  call->req.iov_index = 0;
  call->req.iov_saved = call->req.iov[0];

  /* insert call into connection's send queue: */
  call_inc_ref (call);
  call->sendq_next = 0;
  if (!conn->sendq)
    {
      conn->sendq = conn->sendq_tail = call;
      arg.l = 0;
      event_signal (EV_CALL_SEND_START, (Object *) call, arg);
      if (conn->state >= S_CLOSING)
	return -1;
      call->timeout = param.timeout ? timer_now () + param.timeout : 0.0;
      set_active (conn, &wrfds);
    }
  else
    {
      conn->sendq_tail->sendq_next = call;
      conn->sendq_tail = call;
    }
  return 0;
}

void
core_close (Conn *conn)
{
  Call *call, *call_next;
  Any_Type arg;
  int sd;

  if (conn->state >= S_CLOSING)
    return;			/* guard against recursive calls */
  conn->state = S_CLOSING;

  if (DBG >= 10)
    fprintf (stderr, "%s.core_close(conn=%p)\n", prog_name, conn);

  if (conn->watchdog)
    {
      timer_cancel (conn->watchdog);
      conn->watchdog = 0;
    }

  /* first, get rid of all pending calls: */
  for (call = conn->sendq; call; call = call_next)
    {
      call_next = call->sendq_next;
      call_dec_ref (call);
    }
  conn->sendq = 0;

  for (call = conn->recvq; call; call = call_next)
    {
      call_next = call->recvq_next;
      call_dec_ref (call);
    }
  conn->recvq = 0;

  sd = conn->sd;
  conn->sd = -1;

  arg.l = 0;
  event_signal (EV_CONN_CLOSE, (Object *) conn, arg);
  assert (conn->state == S_CLOSING);

#ifdef HAVE_SSL
  if (param.use_ssl)
    SSL_shutdown (conn->ssl);
#endif

  if (sd >= 0)
    {
      close (sd);
      sd_to_conn[sd] = 0;
      FD_CLR (sd, &wrfds);
      FD_CLR (sd, &rdfds);
    }
  if (conn->myport > 0)
    port_put (conn->myport);

  /* A connection that has been closed is not useful anymore, so we
     give up the reference obtained when creating the session.  This
     normally initiates destruction of the connection.  */
  conn_dec_ref (conn);
}

void
core_loop (void)
{
  int is_readable, is_writable, n, sd, bit, min_i, max_i, i = 0;
  fd_set readable, writable;
  fd_mask mask;
  Any_Type arg;
  Conn *conn;

  while (running)
    {
      struct timeval tv = select_timeout;

      timer_tick ();

      readable = rdfds;
      writable = wrfds;
      min_i = min_sd / NFDBITS;
      max_i = max_sd / NFDBITS;

      SYSCALL (SELECT,
	       n = select (max_sd + 1, &readable, &writable, 0, &tv));

      ++iteration;

      if (n <= 0)
	{
	  if (n < 0)
	    {
	      fprintf (stderr, "%s.core_loop: select failed: %s\n",
		       prog_name, strerror (errno));
	      exit (1);
	    }
	  continue;
	}

      while (n > 0)
	{
	  /* find the index of the fdmask that has something going on: */
	  do
	    {
	      ++i;
	      if (i > max_i)
		i = min_i;

	      assert (i <= max_i);
	      mask = readable.fds_bits[i] | writable.fds_bits[i];
	    }
	  while (!mask);
	  bit = 0;
	  sd = i*NFDBITS + bit;
	  do
	    {
	      if (mask & 1)
		{
		  --n;

		  is_readable =
		    (FD_ISSET (sd, &readable) && FD_ISSET (sd, &rdfds));
		  is_writable =
		    (FD_ISSET (sd, &writable) && FD_ISSET (sd, &wrfds));

		  if (is_readable || is_writable)
		    {
		      /* only handle sockets that haven't timed out yet */
		      conn = sd_to_conn[sd];

		      conn_inc_ref (conn);

		      if (conn->watchdog)
			{
			  timer_cancel (conn->watchdog);
			  conn->watchdog = 0;
			}
		      if (conn->state == S_CONNECTING)
			{
#ifdef HAVE_SSL
			  if (param.use_ssl)
			    core_ssl_connect (conn);
			  else
#endif
			    if (is_writable)
			      {
				FD_CLR (sd, &wrfds);
				conn->state = S_CONNECTED;
				arg.l = 0;
				event_signal (EV_CONN_CONNECTED,
					      (Object *) conn, arg);
			      }
			}
		      else
			{
			  if (is_writable && conn->sendq)
			    do_send (conn);
			  if (is_readable && conn->recvq)
			    do_recv (conn);
			}

		      conn_dec_ref (conn);

		      if (n > 0)
			timer_tick ();
		    }
		}
	      mask = ((u_long) mask) >> 1;
	      ++sd;
	    }
	  while (mask);
	}
    }
}

void
core_exit (void)
{
  running = 0;

  printf ("Maximum connect burst length: %lu\n", max_burst_len);

#ifdef TIME_SYSCALLS
  {
    u_int count;
    Time time;
    int i;

    printf ("Average syscall execution times:\n");
    for (i = 0; i < NELEMS (syscall_name); ++i)
      {
	count = syscall_count[i];
	time = syscall_time[i];
	printf ("\t%s:\t%.3f ms/call (%.3fs total, %u calls)\n",
		syscall_name[i], count > 0 ? 1e3*time/count : 0, time, count);
		
      }
    putchar ('\n');
  }
#endif
}