curl/hiper/hipev.c

/*****************************************************************************
 *                                  _   _ ____  _
 *  Project                     ___| | | |  _ \| |
 *                             / __| | | | |_) | |
 *                            | (__| |_| |  _ <| |___
 *                             \___|\___/|_| \_\_____|
 *
 * $Id$
 *
 * Connect N connections. Z are idle, and X are active. Transfer as fast as
 * possible.
 *
 * Run for a specific amount of time (10 secs for now). Output detailed timing
 * information.
 *
 * Uses libevent.
 *
 */

/* The maximum number of simultanoues connections/transfers we support */
#define NCONNECTIONS 50000

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <sys/poll.h>

#include <curl/curl.h>

#include <event.h> /* for libevent */

#ifndef FALSE
#define FALSE 0
#endif

#ifndef TRUE
#define TRUE 1
#endif

#define MICROSEC 1000000 /* number of microseconds in one second */

/* The maximum time (in microseconds) we run the test */
#define RUN_FOR_THIS_LONG (5*MICROSEC)

/* Number of loops (seconds) we allow the total download amount and alive
   connections to remain the same until we bail out. Set this slightly higher
   when using asynch supported libcurl. */
#define IDLE_TIME 10

struct ourfdset {
  /* __fds_bits is what the Linux glibc headers use when they declare the
     fd_set struct so by using this we can actually avoid the typecase for the
     FD_SET() macro usage but it would hardly be portable */
  char __fds_bits[NCONNECTIONS/8];
};
#define FD2_ZERO(x) memset(x, 0, sizeof(struct ourfdset))

typedef struct ourfdset fd2_set;

struct globalinfo {
  size_t dlcounter;
};

struct connection {
  CURL *e;
  int id; /* just a counter for easy browsing */
  char *url;
  size_t dlcounter;
  struct globalinfo *global;
  char error[CURL_ERROR_SIZE];
  struct event ev[3]; /* maximum 3 events per handle NOTE: should this rather
                         be a define in a public curl header file or possibly
                         just documented somewhere or... ? */
};

struct fdinfo {
  /* create a link list of fdinfo structs */
  struct fdinfo *next;
  struct fdinfo *prev;
  curl_socket_t sockfd;
  CURL *easy;
  int action; /* as set by libcurl */
  long timeout; /* as set by libcurl */
};

static struct fdinfo *allsocks;

static struct fdinfo *findsock(curl_socket_t s)
{
  /* return the struct for the given socket */
  struct fdinfo *fdp = allsocks;

  while(fdp) {
    if(fdp->sockfd == s)
      break;
    fdp = fdp->next;
  }
  return fdp; /* a struct pointer or NULL */
}

static void remsock(curl_socket_t s)
{
  struct fdinfo *fdp = allsocks;

  while(fdp) {
    if(fdp->sockfd == s)
      break;
    fdp = fdp->next;
  }
  if(!fdp)
    /* did not find socket to remove! */
    return;

  if(fdp->prev)
    fdp->prev->next = fdp->next;
  if(fdp->next)
    fdp->next->prev = fdp->prev;
  else
    /* this was the last entry */
    allsocks = NULL;
}

static void setsock(struct fdinfo *fdp, curl_socket_t s, CURL *easy,
                    int action)
{
  fdp->sockfd = s;
  fdp->action = action;
  fdp->easy = easy;
}

static void addsock(curl_socket_t s, CURL *easy, int action)
{
  struct fdinfo *fdp = calloc(sizeof(struct fdinfo), 1);

  setsock(fdp, s, easy, action);

  if(allsocks) {
    fdp->next = allsocks;
    allsocks->prev = fdp;

    /* now set allsocks to point to the new struct */
    allsocks = fdp;
  }
  else
    allsocks = fdp;
}

static void fdinfo2fdset(fd2_set *fdread, fd2_set *fdwrite, int *maxfd)
{
  struct fdinfo *fdp = allsocks;
  int writable=0;

  FD2_ZERO(fdread);
  FD2_ZERO(fdwrite);

  *maxfd = 0;

#if 0
  printf("Wait for: ");
#endif

  while(fdp) {
    if(fdp->action & CURL_POLL_IN) {
      FD_SET(fdp->sockfd, (fd_set *)fdread);
    }
    if(fdp->action & CURL_POLL_OUT) {
      FD_SET(fdp->sockfd, (fd_set *)fdwrite);
      writable++;
    }

#if 0
    printf("%d (%s%s) ",
           fdp->sockfd,
           (fdp->action & CURL_POLL_IN)?"r":"",
           (fdp->action & CURL_POLL_OUT)?"w":"");
#endif

    if(fdp->sockfd > *maxfd)
      *maxfd = fdp->sockfd;

    fdp = fdp->next;
  }
#if 0
  if(writable)
    printf("Check for %d writable sockets\n", writable);
#endif
}

/* on port 8999 we run a fork enabled sws that supports 'idle' and 'stream' */
#define PORT "8999"

#define HOST "192.168.1.13"

#define URL_IDLE   "http://" HOST ":" PORT "/1000"
#define URL_ACTIVE "http://" HOST ":" PORT "/1001"


static int socket_callback(CURL *easy,      /* easy handle */
                           curl_socket_t s, /* socket */
                           int what,        /* see above */
                           void *userp)     /* "private" pointer */
{
  struct fdinfo *fdp;
  printf("socket %d easy %p what %d\n", s, easy, what);

  if(what == CURL_POLL_REMOVE)
    remsock(s);
  else {
    fdp = findsock(s);

    if(!fdp) {
      addsock(s, easy, what);
    }
    else {
      /* we already know about it, just change action/timeout */
      printf("Changing info for socket %d from %d to %d\n",
             s, fdp->action, what);
      setsock(fdp, s, easy, what);
    }
  }
  return 0; /* return code meaning? */
}


static size_t
writecallback(void *ptr, size_t size, size_t nmemb, void *data)
{
  size_t realsize = size * nmemb;
  struct connection *c = (struct connection *)data;

  c->dlcounter += realsize;
  c->global->dlcounter += realsize;

#if 0
  printf("%02d: %d, total %d\n",
         c->id, c->dlcounter, c->global->dlcounter);
#endif
  return realsize;
}

/* return the diff between two timevals, in us */
static long tvdiff(struct timeval *newer, struct timeval *older)
{
  return (newer->tv_sec-older->tv_sec)*1000000+
    (newer->tv_usec-older->tv_usec);
}


/* store the start time of the program in this variable */
static struct timeval timer;

static void timer_start(void)
{
  /* capture the time of the start moment */
  gettimeofday(&timer, NULL);
}

static struct timeval cont; /* at this moment we continued */

int still_running; /* keep number of running handles */

struct conncount {
  long time_us;
  long laps;
  long maxtime;
};

static struct timeval timerpause;
static void timer_pause(void)
{
  /* capture the time of the pause moment */
  gettimeofday(&timerpause, NULL);

  /* If we have a previous continue (all times except the first), we can now
     store the time for a whole "lap" */
  if(cont.tv_sec) {
    long lap;

    lap = tvdiff(&timerpause, &cont);
  }
}

static long paused; /* amount of us we have been pausing */

static void timer_continue(void)
{
  /* Capture the time of the restored operation moment, now calculate how long
     time we were paused and added that to the 'paused' variable.
   */
  gettimeofday(&cont, NULL);

  paused += tvdiff(&cont, &timerpause);
}

static long total; /* amount of us from start to stop */
static void timer_total(void)
{
  struct timeval stop;
  /* Capture the time of the operation stopped moment, now calculate how long
     time we were running and how much of that pausing.
   */
  gettimeofday(&stop, NULL);

  total = tvdiff(&stop, &timer);
}

struct globalinfo info;
struct connection *conns;

long selects;
long timeouts;

long multi_socket;
long performalive;
long performselect;
long topselect;

int num_total;
int num_idle;
int num_active;

static void report(void)
{
  int i;
  long active = total - paused;
  long numdl = 0;

  for(i=0; i < num_total; i++) {
    if(conns[i].dlcounter)
      numdl++;
  }

  printf("Summary from %d simultanoues transfers (%d active)\n",
         num_total, num_active);
  printf("%d out of %d connections provided data\n", numdl, num_total);

  printf("Total time: %ldus paused: %ldus curl_multi_socket(): %ldus\n",
         total, paused, active);

  printf("%d calls to select() "
         "Average time: %dus\n",
         selects, paused/selects);
  printf(" Average number of readable connections per select() return: %d\n",
         performselect/selects);

  printf(" Max number of readable connections for a single select() "
         "return: %d\n",
         topselect);

  printf("%ld calls to multi_socket(), "
         "Average time: %ldus\n",
         multi_socket, active/multi_socket);

  printf("%ld select() timeouts\n", timeouts);

  printf("Downloaded %ld bytes in %ld bytes/sec, %ld usec/byte\n",
         info.dlcounter,
         info.dlcounter/(total/1000000),
         total/info.dlcounter);

}

int main(int argc, char **argv)
{
  CURLM *multi_handle;
  CURLMsg *msg;
  CURLcode code = CURLE_OK;
  CURLMcode mcode = CURLM_OK;
  int rc;
  int i;
  fd2_set fdsizecheck;
  int selectmaxamount;
  struct fdinfo *fdp;
  char act;

  memset(&info, 0, sizeof(struct globalinfo));

  if(argc < 3) {
    printf("Usage: hiper-event [num idle] [num active]\n");
    return 1;
  }

  num_idle = atoi(argv[1]);
  num_active = atoi(argv[2]);

  num_total = num_idle + num_active;

  conns = calloc(num_total, sizeof(struct connection));
  if(!conns) {
    printf("Out of memory\n");
    return 3;
  }

  if(num_total >= NCONNECTIONS) {
    printf("Too many connections requested, increase NCONNECTIONS!\n");
    return 2;
  }

  event_init(); /* Initalize the event library */

  printf("About to do %d connections\n", num_total);

  /* init the multi stack */
  multi_handle = curl_multi_init();

  for(i=0; i< num_total; i++) {
    CURL *e;
    char *nl;

    memset(&conns[i], 0, sizeof(struct connection));

    if(i < num_idle)
      conns[i].url = URL_IDLE;
    else
      conns[i].url = URL_ACTIVE;

    e  = curl_easy_init();

    if(!e) {
      printf("curl_easy_init() for handle %d failed, exiting!\n", i);
      return 2;
    }

    conns[i].e = e;
    conns[i].id = i;
    conns[i].global = &info;

    curl_easy_setopt(e, CURLOPT_URL, conns[i].url);
    curl_easy_setopt(e, CURLOPT_WRITEFUNCTION, writecallback);
    curl_easy_setopt(e, CURLOPT_WRITEDATA, &conns[i]);
    curl_easy_setopt(e, CURLOPT_VERBOSE, 0);
    curl_easy_setopt(e, CURLOPT_ERRORBUFFER, conns[i].error);
    curl_easy_setopt(e, CURLOPT_PRIVATE, &conns[i]);

    /* add the easy to the multi */
    if(CURLM_OK != curl_multi_add_handle(multi_handle, e)) {
      printf("curl_multi_add_handle() returned error for %d\n", i);
      return 3;
    }
  }

  curl_multi_setopt(multi_handle, CURLMOPT_SOCKETFUNCTION, socket_callback);
  curl_multi_setopt(multi_handle, CURLMOPT_SOCKETDATA, NULL);

  /* we start the action by calling *socket() right away */
  while(CURLM_CALL_MULTI_PERFORM == curl_multi_socket_all(multi_handle));

  printf("Starting timer, expects to run for %ldus\n", RUN_FOR_THIS_LONG);
  timer_start();
  timer_pause();

  while(1) {
    struct timeval timeout;
    int rc; /* select() return code */
    long timeout_ms;

    fd2_set fdread;
    fd2_set fdwrite;
    int maxfd;

    curl_multi_timeout(multi_handle, &timeout_ms);

    /* set timeout to wait */
    timeout.tv_sec = timeout_ms/1000;
    timeout.tv_usec = (timeout_ms%1000)*1000;

    /* convert file descriptors from the transfers to fd_sets */
    fdinfo2fdset(&fdread, &fdwrite, &maxfd);

    selects++;
    rc = select(maxfd+1,
                (fd_set *)&fdread,
                (fd_set *)&fdwrite,
                NULL, &timeout);
    switch(rc) {
    case -1:
      /* select error */
      break;
    case 0:
      timeouts++;
      curl_multi_socket(multi_handle, CURL_SOCKET_TIMEOUT);
      break;

    default:
      /* timeout or readable/writable sockets */

      for(i=0, fdp = allsocks; fdp; fdp = fdp->next) {
        act = 0;
        if((fdp->action & CURL_POLL_IN) &&
           FD_ISSET(fdp->sockfd, &fdread)) {
          act |= CURL_POLL_IN;
          i++;
        }
        if((fdp->action & CURL_POLL_OUT) &&
           FD_ISSET(fdp->sockfd, &fdwrite)) {
          act |= CURL_POLL_OUT;
          i++;
        }

        if(act) {
          multi_socket++;
          timer_continue();
          if(act & CURL_POLL_OUT)
            act--;
          curl_multi_socket(multi_handle, fdp->sockfd);
          timer_pause();
        }
      }

      performselect += rc;
      if(rc > topselect)
        topselect = rc;
      break;
    }

    timer_total(); /* calculate the total time spent so far */

    if(total > RUN_FOR_THIS_LONG) {
      printf("Stopped after %ldus\n", total);
      break;
    }
  }

  if(still_running != num_total) {
    /* something made connections fail, extract the reason and tell */
    int msgs_left;
    struct connection *cptr;
    while ((msg = curl_multi_info_read(multi_handle, &msgs_left))) {
      if (msg->msg == CURLMSG_DONE) {
        curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &cptr);

        printf("%d => (%d) %s", cptr->id, msg->data.result, cptr->error);
      }
    }

  }

  curl_multi_cleanup(multi_handle);

  /* cleanup all the easy handles */
  for(i=0; i< num_total; i++)
    curl_easy_cleanup(conns[i].e);

  report();

  return code;
}