You need to call curl_multi_perform after each curl_multi_add_handle. It
will perform much better. By comparison, I can completely saturate a t3
from a single cpu linux box. Choose min/max connections you want to keep
active ( and your bandwidth can handle ), and then adjust a number for
usleep to try to keep this number alive. I choose say, between 50 and
100 connections, and then if it is less then 50 decrease the sleep
interval, if it is more then 100, increase it. Be sure to call curl
multi cleanup on each easy thread.

Jason

Martin Vogt wrote:

>Hello,
>
>this is a new program, which demonstrates the problem.
>(Although I dont know if libcurl should be used this way)
>
>What does the program do:
>
>My idea is to not start 1000 requests in parallel, but use a set
>of requests "on the fly".
>Evertime a request has finished, it should add a new one,
>until 1000 requests are done.
>
>On my tests I dont get a predictible behaviour. Sometimes
>it works without timeouts, then the first call directly timeouts.
>
>It looks like, that if I return the size of the copied bytes in dev_null
>The request is not set to the state CURLE_OK and I cannot remove
>it from the msgqueue.
>
>regards,
>
>Martin
>
>
>
>
>
>------------------------------------------------------------------------
>
>/*
> testing curl with parallel request
> Author: Martin Vogt
>
>*/
>
>#include <stdio.h>
>#include <stdlib.h>
>#include <string.h>
>#include <getopt.h>
>#include <curl/curl.h>
>#include <iostream>
>
>
>static int bytecounter;
>static int still_running;
>static int removed=0;
>static int concurreny=0;
>static int on_the_fly=0;
>static int requests=0;
>static char* url=NULL;
>
>using namespace std;
>
>/**
>
> build:
>
> g++ -o curltest curltest.cpp -lcurl
>
>
>*/
>
>
>void usage() {
> printf("curltest is frontend for testing libdbaccess\n");
> printf("Usage : curltest [-u url] [-n requests] [-c concurreny] [-h]\n\n");
>
> printf("-u : url to fetch\n");
> printf("-n : requests to do\n");
> printf("-c : number of concurrent requests \n");
>
> printf("-h : this message\n\n");
>
> printf("THIS SOFTWARE COMES WITH ABSOLUTELY NO WARRANTY!");
> printf(" USE AT YOUR OWN RISK!\n\n");
> printf("example:\n");
> printf("./curltest -u http://media3.cluster/a.jpg -n 1000 -c 50\n");
>
>}
>
>#define TRANSFER_SIZE 16384
>extern "C" size_t dev_null(void *ptr,size_t size,size_t nmemb,void *stream) {
> char buf[TRANSFER_SIZE];
> int back=TRANSFER_SIZE;
> int bytes=size*nmemb;
> if (bytes > TRANSFER_SIZE) bytes=TRANSFER_SIZE;
> memcpy(buf,ptr,bytes);
> bytecounter+=bytes;
> cout << "devnull called read:"<<bytes<<endl;
> return bytes;
>}
>
>/**
> This works as follows. We know how many request we want to do.
> requests=1000
> then we know how many requests we want to do in parallel
> concurreny=50
>
> As a start we fill concurreny requests in the multiqueue,
> everytime we remove a curl context, and we have still requests
> to do we add one again.
>
> This means that there are always a good number of requests "on the fly"
>
>*/
>
>void insert_more_request(CURLM *multi_handle) {
> // do we still have something to do?
> if (removed >= requests) {
> // NO
> return;
> }
> // how much requests are running
> if (on_the_fly >= concurreny) {
> // NO need to insert mor
> return;
> }
> // how much requests do we still need to do
> int left_request=requests-on_the_fly-removed;
> cout << "still to do:"<<left_request<<endl;
> // how much request can we possibly insert
> int can_insert=concurreny-on_the_fly;
> // do not insert more the the fixed number of concurrent requests
> if (left_request > can_insert) {
> left_request=can_insert;
> }
> cout << "url is:"<<url<<endl;
> cout << "can_insert:"<<can_insert<<endl;
> cout << "doing insert:"<<left_request<<endl;
> int i;
> CURLMcode mstatus;
> for(i=0;i<left_request;i++) {
> CURL* ctx = curl_easy_init() ;
> curl_easy_setopt(ctx, CURLOPT_URL, url);
> curl_easy_setopt(ctx, CURLOPT_WRITEFUNCTION, dev_null);
> mstatus=curl_multi_add_handle(multi_handle, ctx);
> cout << "status:"<<curl_multi_strerror(mstatus)<<endl;
> on_the_fly++;
> }
>}
>
>
>int remove_eof_handles(CURLM *multi_handle) {
>
> CURLMsg* curlmsg;
> int msgs_in_queue=0;
> curlmsg=curl_multi_info_read(multi_handle,&msgs_in_queue);
> cout << "msgs_in_queue:"<<msgs_in_queue<<endl;
> cout << "curlmsg:"<<curlmsg<<endl;
> while (curlmsg != NULL) {
> if (curlmsg->msg == CURLMSG_DONE) {
> CURLcode status;
> long val;
> status=curl_easy_getinfo(curlmsg->easy_handle,
> CURLINFO_HTTP_CODE,&val);
> if (status != CURLE_OK) {
> cout << "error CURL_OK:"<<status<<endl;
> }
> curl_multi_remove_handle(multi_handle,curlmsg->easy_handle);
> curl_easy_cleanup(curlmsg->easy_handle);
> removed++;
> on_the_fly--;
> cout << "removed one"<<endl;
> }
> curlmsg=curl_multi_info_read(multi_handle,&msgs_in_queue);
> }
> return 0;
>}
>
>
>int doloop(CURLM *multi_handle) {
>
> while(removed < requests) {
>
> CURLMcode mstatus=CURLM_CALL_MULTI_PERFORM;
> while(mstatus == CURLM_CALL_MULTI_PERFORM) {
> mstatus=curl_multi_perform(multi_handle, &still_running);
> remove_eof_handles(multi_handle);
> insert_more_request(multi_handle);
> mstatus=curl_multi_perform(multi_handle, &still_running);
> }
>
> struct timeval timeout;
> int rc; /* select() return code */
>
> fd_set fdread;
> fd_set fdwrite;
> fd_set fdexcep;
> int maxfd;
>
> FD_ZERO(&fdread);
> FD_ZERO(&fdwrite);
> FD_ZERO(&fdexcep);
>
> /* set a suitable timeout to play around with */
> timeout.tv_sec = 1;
> timeout.tv_usec = 0;
>
> /* get file descriptors from the transfers */
> curl_multi_fdset(multi_handle, &fdread, &fdwrite, &fdexcep, &maxfd);
>
> cout << "maxfd:"<<maxfd<<endl;
> rc = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);
>
> switch(rc) {
> case -1:
> /* select error */
> printf("select error\n");
> break;
> case 0:
> printf("timeout!\n");
> default:
> /* timeout or readable/writable sockets */
> break;
> }
> }
>}
>
>
>
>
>int main(int argc, char **argv) {
>
> int c;
>
>
> bytecounter=0;
>
>
> while(1) {
> c = getopt (argc, argv, "u:hn:c:");
> if (c == -1) break;
> switch(c) {
> case 'u': {
> url=strdup(optarg);
> break;
> }
> case 'h': {
> usage();
> exit(0);
> }
> case 'n': {
> requests=atoi(optarg);
> break;
> }
> case 'c': {
> concurreny=atoi(optarg);
> break;
> }
> default:
> printf ("?? getopt returned character code 0%o ??\n", c);
> usage();
> exit(-1);
> }
> }
>
> if ( (url == NULL) ) {
> usage();
> exit(-1);
> }
>
> CURLcode res;
> CURLM *multi_handle;
> curl_global_init( CURL_GLOBAL_ALL ) ;
>
> multi_handle = curl_multi_init();
> if (multi_handle == NULL) {
> cout << "error setting up curl"<<endl;
> exit(-1);
> }
>
>
>
> doloop(multi_handle);
>
>
> cout << "bytecounter:"<<bytecounter<<" still:"<<still_running<<endl;
> curl_multi_cleanup(multi_handle);
> exit(0);
>}
>
>
>
>
>
>
Received on 2005-09-01