evhiperfifo.c (12496B)
1 /*************************************************************************** 2 * _ _ ____ _ 3 * Project ___| | | | _ \| | 4 * / __| | | | |_) | | 5 * | (__| |_| | _ <| |___ 6 * \___|\___/|_| \_\_____| 7 * 8 * Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al. 9 * 10 * This software is licensed as described in the file COPYING, which 11 * you should have received as part of this distribution. The terms 12 * are also available at https://curl.se/docs/copyright.html. 13 * 14 * You may opt to use, copy, modify, merge, publish, distribute and/or sell 15 * copies of the Software, and permit persons to whom the Software is 16 * furnished to do so, under the terms of the COPYING file. 17 * 18 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY 19 * KIND, either express or implied. 20 * 21 * SPDX-License-Identifier: curl 22 * 23 ***************************************************************************/ 24 /* <DESC> 25 * multi socket interface together with libev 26 * </DESC> 27 */ 28 /* Example application source code using the multi socket interface to 29 * download many files at once. 30 * 31 * This example features the same basic functionality as hiperfifo.c does, 32 * but this uses libev instead of libevent. 33 * 34 * Written by Jeff Pohlmeyer, converted to use libev by Markus Koetter 35 36 Requires libev and a (POSIX?) system that has mkfifo(). 37 38 This is an adaptation of libcurl's "hipev.c" and libevent's "event-test.c" 39 sample programs. 40 41 When running, the program creates the named pipe "hiper.fifo" 42 43 Whenever there is input into the fifo, the program reads the input as a list 44 of URL's and creates some new easy handles to fetch each URL via the 45 curl_multi "hiper" API. 46 47 48 Thus, you can try a single URL: 49 % echo http://www.yahoo.com > hiper.fifo 50 51 Or a whole bunch of them: 52 % cat my-url-list > hiper.fifo 53 54 The fifo buffer is handled almost instantly, so you can even add more URL's 55 while the previous requests are still being downloaded. 56 57 Note: 58 For the sake of simplicity, URL length is limited to 1023 char's ! 59 60 This is purely a demo app, all retrieved data is simply discarded by the write 61 callback. 62 63 */ 64 65 #include <stdio.h> 66 #include <string.h> 67 #include <stdlib.h> 68 #include <sys/time.h> 69 #include <time.h> 70 #include <unistd.h> 71 #include <sys/poll.h> 72 #include <curl/curl.h> 73 #include <ev.h> 74 #include <fcntl.h> 75 #include <sys/stat.h> 76 #include <errno.h> 77 78 #define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */ 79 80 81 /* Global information, common to all connections */ 82 struct GlobalInfo { 83 struct ev_loop *loop; 84 struct ev_io fifo_event; 85 struct ev_timer timer_event; 86 CURLM *multi; 87 int still_running; 88 FILE *input; 89 }; 90 91 92 /* Information associated with a specific easy handle */ 93 struct ConnInfo { 94 CURL *easy; 95 char *url; 96 struct GlobalInfo *global; 97 char error[CURL_ERROR_SIZE]; 98 }; 99 100 101 /* Information associated with a specific socket */ 102 struct SockInfo { 103 curl_socket_t sockfd; 104 CURL *easy; 105 int action; 106 long timeout; 107 struct ev_io ev; 108 int evset; 109 struct GlobalInfo *global; 110 }; 111 112 static void timer_cb(EV_P_ struct ev_timer *w, int revents); 113 114 /* Update the event timer after curl_multi library calls */ 115 static int multi_timer_cb(CURLM *multi, long timeout_ms, struct GlobalInfo *g) 116 { 117 (void)multi; 118 printf("%s %li\n", __PRETTY_FUNCTION__, timeout_ms); 119 ev_timer_stop(g->loop, &g->timer_event); 120 if(timeout_ms >= 0) { 121 /* -1 means delete, other values are timeout times in milliseconds */ 122 double t = timeout_ms / 1000; 123 ev_timer_init(&g->timer_event, timer_cb, t, 0.); 124 ev_timer_start(g->loop, &g->timer_event); 125 } 126 return 0; 127 } 128 129 /* Die if we get a bad CURLMcode somewhere */ 130 static void mcode_or_die(const char *where, CURLMcode code) 131 { 132 if(CURLM_OK != code) { 133 const char *s; 134 switch(code) { 135 case CURLM_BAD_HANDLE: 136 s = "CURLM_BAD_HANDLE"; 137 break; 138 case CURLM_BAD_EASY_HANDLE: 139 s = "CURLM_BAD_EASY_HANDLE"; 140 break; 141 case CURLM_OUT_OF_MEMORY: 142 s = "CURLM_OUT_OF_MEMORY"; 143 break; 144 case CURLM_INTERNAL_ERROR: 145 s = "CURLM_INTERNAL_ERROR"; 146 break; 147 case CURLM_UNKNOWN_OPTION: 148 s = "CURLM_UNKNOWN_OPTION"; 149 break; 150 case CURLM_LAST: 151 s = "CURLM_LAST"; 152 break; 153 default: 154 s = "CURLM_unknown"; 155 break; 156 case CURLM_BAD_SOCKET: 157 s = "CURLM_BAD_SOCKET"; 158 fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s); 159 /* ignore this error */ 160 return; 161 } 162 fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s); 163 exit(code); 164 } 165 } 166 167 168 169 /* Check for completed transfers, and remove their easy handles */ 170 static void check_multi_info(struct GlobalInfo *g) 171 { 172 char *eff_url; 173 CURLMsg *msg; 174 int msgs_left; 175 struct ConnInfo *conn; 176 CURL *easy; 177 CURLcode res; 178 179 fprintf(MSG_OUT, "REMAINING: %d\n", g->still_running); 180 while((msg = curl_multi_info_read(g->multi, &msgs_left))) { 181 if(msg->msg == CURLMSG_DONE) { 182 easy = msg->easy_handle; 183 res = msg->data.result; 184 curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn); 185 curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url); 186 fprintf(MSG_OUT, "DONE: %s => (%d) %s\n", eff_url, res, conn->error); 187 curl_multi_remove_handle(g->multi, easy); 188 free(conn->url); 189 curl_easy_cleanup(easy); 190 free(conn); 191 } 192 } 193 } 194 195 196 197 /* Called by libevent when we get action on a multi socket */ 198 static void event_cb(EV_P_ struct ev_io *w, int revents) 199 { 200 struct GlobalInfo *g; 201 CURLMcode rc; 202 int action; 203 204 printf("%s w %p revents %i\n", __PRETTY_FUNCTION__, (void *)w, revents); 205 g = (struct GlobalInfo*) w->data; 206 207 action = ((revents & EV_READ) ? CURL_POLL_IN : 0) | 208 ((revents & EV_WRITE) ? CURL_POLL_OUT : 0); 209 rc = curl_multi_socket_action(g->multi, w->fd, action, &g->still_running); 210 mcode_or_die("event_cb: curl_multi_socket_action", rc); 211 check_multi_info(g); 212 if(g->still_running <= 0) { 213 fprintf(MSG_OUT, "last transfer done, kill timeout\n"); 214 ev_timer_stop(g->loop, &g->timer_event); 215 } 216 } 217 218 /* Called by libevent when our timeout expires */ 219 static void timer_cb(EV_P_ struct ev_timer *w, int revents) 220 { 221 struct GlobalInfo *g; 222 CURLMcode rc; 223 224 printf("%s w %p revents %i\n", __PRETTY_FUNCTION__, (void *)w, revents); 225 226 g = (struct GlobalInfo *)w->data; 227 228 rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0, 229 &g->still_running); 230 mcode_or_die("timer_cb: curl_multi_socket_action", rc); 231 check_multi_info(g); 232 } 233 234 /* Clean up the SockInfo structure */ 235 static void remsock(struct SockInfo *f, struct GlobalInfo *g) 236 { 237 printf("%s \n", __PRETTY_FUNCTION__); 238 if(f) { 239 if(f->evset) 240 ev_io_stop(g->loop, &f->ev); 241 free(f); 242 } 243 } 244 245 246 247 /* Assign information to a SockInfo structure */ 248 static void setsock(struct SockInfo *f, curl_socket_t s, CURL *e, int act, 249 struct GlobalInfo *g) 250 { 251 int kind = ((act & CURL_POLL_IN) ? EV_READ : 0) | 252 ((act & CURL_POLL_OUT) ? EV_WRITE : 0); 253 254 printf("%s \n", __PRETTY_FUNCTION__); 255 256 f->sockfd = s; 257 f->action = act; 258 f->easy = e; 259 if(f->evset) 260 ev_io_stop(g->loop, &f->ev); 261 ev_io_init(&f->ev, event_cb, f->sockfd, kind); 262 f->ev.data = g; 263 f->evset = 1; 264 ev_io_start(g->loop, &f->ev); 265 } 266 267 268 269 /* Initialize a new SockInfo structure */ 270 static void addsock(curl_socket_t s, CURL *easy, int action, 271 struct GlobalInfo *g) 272 { 273 struct SockInfo *fdp = calloc(1, sizeof(struct SockInfo)); 274 275 fdp->global = g; 276 setsock(fdp, s, easy, action, g); 277 curl_multi_assign(g->multi, s, fdp); 278 } 279 280 /* CURLMOPT_SOCKETFUNCTION */ 281 static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp) 282 { 283 struct GlobalInfo *g = (struct GlobalInfo*) cbp; 284 struct SockInfo *fdp = (struct SockInfo*) sockp; 285 const char *whatstr[]={ "none", "IN", "OUT", "INOUT", "REMOVE"}; 286 287 printf("%s e %p s %i what %i cbp %p sockp %p\n", 288 __PRETTY_FUNCTION__, e, s, what, cbp, sockp); 289 290 fprintf(MSG_OUT, 291 "socket callback: s=%d e=%p what=%s ", s, e, whatstr[what]); 292 if(what == CURL_POLL_REMOVE) { 293 fprintf(MSG_OUT, "\n"); 294 remsock(fdp, g); 295 } 296 else { 297 if(!fdp) { 298 fprintf(MSG_OUT, "Adding data: %s\n", whatstr[what]); 299 addsock(s, e, what, g); 300 } 301 else { 302 fprintf(MSG_OUT, 303 "Changing action from %s to %s\n", 304 whatstr[fdp->action], whatstr[what]); 305 setsock(fdp, s, e, what, g); 306 } 307 } 308 return 0; 309 } 310 311 312 /* CURLOPT_WRITEFUNCTION */ 313 static size_t write_cb(void *ptr, size_t size, size_t nmemb, void *data) 314 { 315 size_t realsize = size * nmemb; 316 struct ConnInfo *conn = (struct ConnInfo*) data; 317 (void)ptr; 318 (void)conn; 319 return realsize; 320 } 321 322 /* CURLOPT_XFERINFOFUNCTION */ 323 static int xferinfo_cb(void *p, curl_off_t dltotal, curl_off_t dlnow, 324 curl_off_t ult, curl_off_t uln) 325 { 326 struct ConnInfo *conn = (struct ConnInfo *)p; 327 (void)ult; 328 (void)uln; 329 330 fprintf(MSG_OUT, "Progress: %s (%" CURL_FORMAT_CURL_OFF_T 331 "/%" CURL_FORMAT_CURL_OFF_T ")\n", conn->url, dlnow, dltotal); 332 return 0; 333 } 334 335 336 /* Create a new easy handle, and add it to the global curl_multi */ 337 static void new_conn(const char *url, struct GlobalInfo *g) 338 { 339 struct ConnInfo *conn; 340 CURLMcode rc; 341 342 conn = calloc(1, sizeof(*conn)); 343 conn->error[0]='\0'; 344 345 conn->easy = curl_easy_init(); 346 if(!conn->easy) { 347 fprintf(MSG_OUT, "curl_easy_init() failed, exiting!\n"); 348 exit(2); 349 } 350 conn->global = g; 351 conn->url = strdup(url); 352 curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url); 353 curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb); 354 curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, conn); 355 curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L); 356 curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error); 357 curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn); 358 curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 0L); 359 curl_easy_setopt(conn->easy, CURLOPT_XFERINFOFUNCTION, xferinfo_cb); 360 curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn); 361 curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_TIME, 3L); 362 curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_LIMIT, 10L); 363 364 fprintf(MSG_OUT, 365 "Adding easy %p to multi %p (%s)\n", conn->easy, g->multi, url); 366 rc = curl_multi_add_handle(g->multi, conn->easy); 367 mcode_or_die("new_conn: curl_multi_add_handle", rc); 368 369 /* note that add_handle() sets a timeout to trigger soon so that the 370 necessary socket_action() gets called */ 371 } 372 373 /* This gets called whenever data is received from the fifo */ 374 static void fifo_cb(EV_P_ struct ev_io *w, int revents) 375 { 376 char s[1024]; 377 long int rv = 0; 378 int n = 0; 379 struct GlobalInfo *g = (struct GlobalInfo *)w->data; 380 381 (void)revents; 382 383 do { 384 s[0]='\0'; 385 rv = fscanf(g->input, "%1023s%n", s, &n); 386 s[n]='\0'; 387 if(n && s[0]) { 388 new_conn(s, g); /* if we read a URL, go get it! */ 389 } 390 else 391 break; 392 } while(rv != EOF); 393 } 394 395 /* Create a named pipe and tell libevent to monitor it */ 396 static int init_fifo(struct GlobalInfo *g) 397 { 398 struct stat st; 399 static const char *fifo = "hiper.fifo"; 400 curl_socket_t sockfd; 401 402 fprintf(MSG_OUT, "Creating named pipe \"%s\"\n", fifo); 403 if(lstat(fifo, &st) == 0) { 404 if((st.st_mode & S_IFMT) == S_IFREG) { 405 errno = EEXIST; 406 perror("lstat"); 407 return 1; 408 } 409 } 410 unlink(fifo); 411 if(mkfifo(fifo, 0600) == -1) { 412 perror("mkfifo"); 413 return 1; 414 } 415 sockfd = open(fifo, O_RDWR | O_NONBLOCK, 0); 416 if(sockfd == -1) { 417 perror("open"); 418 return 1; 419 } 420 g->input = fdopen(sockfd, "r"); 421 422 fprintf(MSG_OUT, "Now, pipe some URL's into > %s\n", fifo); 423 ev_io_init(&g->fifo_event, fifo_cb, sockfd, EV_READ); 424 ev_io_start(g->loop, &g->fifo_event); 425 return 0; 426 } 427 428 int main(int argc, char **argv) 429 { 430 struct GlobalInfo g; 431 (void)argc; 432 (void)argv; 433 434 memset(&g, 0, sizeof(g)); 435 g.loop = ev_default_loop(0); 436 437 if(init_fifo(&g)) 438 return 1; 439 g.multi = curl_multi_init(); 440 441 ev_timer_init(&g.timer_event, timer_cb, 0., 0.); 442 g.timer_event.data = &g; 443 g.fifo_event.data = &g; 444 curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb); 445 curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g); 446 curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb); 447 curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g); 448 449 /* we do not call any curl_multi_socket*() function yet as we have no handles 450 added! */ 451 452 ev_loop(g.loop, 0); 453 curl_multi_cleanup(g.multi); 454 return 0; 455 }