diff options
Diffstat (limited to 'hiper')
| -rw-r--r-- | hiper/hipev.c | 550 |
1 files changed, 550 insertions, 0 deletions
diff --git a/hiper/hipev.c b/hiper/hipev.c new file mode 100644 index 000000000..d82d7349f --- /dev/null +++ b/hiper/hipev.c @@ -0,0 +1,550 @@ +/***************************************************************************** + * _ _ ____ _ + * 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; +} |