ephiperfifo.c (15033B)
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 API usage with epoll and timerfd 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 epoll and timerfd instead of libevent. 33 * 34 * Written by Jeff Pohlmeyer, converted to use epoll by Josh Bialkowski 35 36 Requires a Linux system with epoll 37 38 When running, the program creates the named pipe "hiper.fifo" 39 40 Whenever there is input into the fifo, the program reads the input as a list 41 of URL's and creates some new easy handles to fetch each URL via the 42 curl_multi "hiper" API. 43 44 45 Thus, you can try a single URL: 46 % echo http://www.yahoo.com > hiper.fifo 47 48 Or a whole bunch of them: 49 % cat my-url-list > hiper.fifo 50 51 The fifo buffer is handled almost instantly, so you can even add more URL's 52 while the previous requests are still being downloaded. 53 54 Note: 55 For the sake of simplicity, URL length is limited to 1023 char's ! 56 57 This is purely a demo app, all retrieved data is simply discarded by the write 58 callback. 59 60 */ 61 62 #include <errno.h> 63 #include <fcntl.h> 64 #include <signal.h> 65 #include <stdio.h> 66 #include <stdlib.h> 67 #include <string.h> 68 #include <sys/epoll.h> 69 #include <sys/stat.h> 70 #include <sys/time.h> 71 #include <sys/timerfd.h> 72 #include <sys/types.h> 73 #include <time.h> 74 #include <unistd.h> 75 76 #include <curl/curl.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 int epfd; /* epoll filedescriptor */ 84 int tfd; /* timer filedescriptor */ 85 int fifofd; /* fifo filedescriptor */ 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 GlobalInfo *global; 108 }; 109 110 #define mycase(code) \ 111 case code: s = __STRING(code) 112 113 /* Die if we get a bad CURLMcode somewhere */ 114 static void mcode_or_die(const char *where, CURLMcode code) 115 { 116 if(CURLM_OK != code) { 117 const char *s; 118 switch(code) { 119 mycase(CURLM_BAD_HANDLE); break; 120 mycase(CURLM_BAD_EASY_HANDLE); break; 121 mycase(CURLM_OUT_OF_MEMORY); break; 122 mycase(CURLM_INTERNAL_ERROR); break; 123 mycase(CURLM_UNKNOWN_OPTION); break; 124 mycase(CURLM_LAST); break; 125 default: s = "CURLM_unknown"; break; 126 mycase(CURLM_BAD_SOCKET); 127 fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s); 128 /* ignore this error */ 129 return; 130 } 131 fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s); 132 exit(code); 133 } 134 } 135 136 static void timer_cb(struct GlobalInfo *g, int revents); 137 138 /* Update the timer after curl_multi library does its thing. Curl informs the 139 * application through this callback what it wants the new timeout to be, 140 * after it does some work. */ 141 static int multi_timer_cb(CURLM *multi, long timeout_ms, struct GlobalInfo *g) 142 { 143 struct itimerspec its; 144 145 fprintf(MSG_OUT, "multi_timer_cb: Setting timeout to %ld ms\n", timeout_ms); 146 147 if(timeout_ms > 0) { 148 its.it_interval.tv_sec = 0; 149 its.it_interval.tv_nsec = 0; 150 its.it_value.tv_sec = timeout_ms / 1000; 151 its.it_value.tv_nsec = (timeout_ms % 1000) * 1000 * 1000; 152 } 153 else if(timeout_ms == 0) { 154 /* libcurl wants us to timeout now, however setting both fields of 155 * new_value.it_value to zero disarms the timer. The closest we can 156 * do is to schedule the timer to fire in 1 ns. */ 157 its.it_interval.tv_sec = 0; 158 its.it_interval.tv_nsec = 0; 159 its.it_value.tv_sec = 0; 160 its.it_value.tv_nsec = 1; 161 } 162 else { 163 memset(&its, 0, sizeof(its)); 164 } 165 166 timerfd_settime(g->tfd, /* flags= */0, &its, NULL); 167 return 0; 168 } 169 170 171 /* Check for completed transfers, and remove their easy handles */ 172 static void check_multi_info(struct GlobalInfo *g) 173 { 174 char *eff_url; 175 CURLMsg *msg; 176 int msgs_left; 177 struct ConnInfo *conn; 178 CURL *easy; 179 CURLcode res; 180 181 fprintf(MSG_OUT, "REMAINING: %d\n", g->still_running); 182 while((msg = curl_multi_info_read(g->multi, &msgs_left))) { 183 if(msg->msg == CURLMSG_DONE) { 184 easy = msg->easy_handle; 185 res = msg->data.result; 186 curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn); 187 curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url); 188 fprintf(MSG_OUT, "DONE: %s => (%d) %s\n", eff_url, res, conn->error); 189 curl_multi_remove_handle(g->multi, easy); 190 free(conn->url); 191 curl_easy_cleanup(easy); 192 free(conn); 193 } 194 } 195 } 196 197 /* Called by libevent when we get action on a multi socket filedescriptor */ 198 static void event_cb(struct GlobalInfo *g, int fd, int revents) 199 { 200 CURLMcode rc; 201 struct itimerspec its; 202 203 int action = ((revents & EPOLLIN) ? CURL_CSELECT_IN : 0) | 204 ((revents & EPOLLOUT) ? CURL_CSELECT_OUT : 0); 205 206 rc = curl_multi_socket_action(g->multi, fd, action, &g->still_running); 207 mcode_or_die("event_cb: curl_multi_socket_action", rc); 208 209 check_multi_info(g); 210 if(g->still_running <= 0) { 211 fprintf(MSG_OUT, "last transfer done, kill timeout\n"); 212 memset(&its, 0, sizeof(its)); 213 timerfd_settime(g->tfd, 0, &its, NULL); 214 } 215 } 216 217 /* Called by main loop when our timeout expires */ 218 static void timer_cb(struct GlobalInfo *g, int revents) 219 { 220 CURLMcode rc; 221 uint64_t count = 0; 222 ssize_t err = 0; 223 224 err = read(g->tfd, &count, sizeof(uint64_t)); 225 if(err == -1) { 226 /* Note that we may call the timer callback even if the timerfd is not 227 * readable. It's possible that there are multiple events stored in the 228 * epoll buffer (i.e. the timer may have fired multiple times). The event 229 * count is cleared after the first call so future events in the epoll 230 * buffer fails to read from the timer. */ 231 if(errno == EAGAIN) { 232 fprintf(MSG_OUT, "EAGAIN on tfd %d\n", g->tfd); 233 return; 234 } 235 } 236 if(err != sizeof(uint64_t)) { 237 fprintf(stderr, "read(tfd) == %ld", err); 238 perror("read(tfd)"); 239 } 240 241 rc = curl_multi_socket_action(g->multi, 242 CURL_SOCKET_TIMEOUT, 0, &g->still_running); 243 mcode_or_die("timer_cb: curl_multi_socket_action", rc); 244 check_multi_info(g); 245 } 246 247 248 249 /* Clean up the SockInfo structure */ 250 static void remsock(struct SockInfo *f, struct GlobalInfo *g) 251 { 252 if(f) { 253 if(f->sockfd) { 254 if(epoll_ctl(g->epfd, EPOLL_CTL_DEL, f->sockfd, NULL)) 255 fprintf(stderr, "EPOLL_CTL_DEL failed for fd: %d : %s\n", 256 f->sockfd, strerror(errno)); 257 } 258 free(f); 259 } 260 } 261 262 263 264 /* Assign information to a SockInfo structure */ 265 static void setsock(struct SockInfo *f, curl_socket_t s, CURL *e, int act, 266 struct GlobalInfo *g) 267 { 268 struct epoll_event ev; 269 int kind = ((act & CURL_POLL_IN) ? EPOLLIN : 0) | 270 ((act & CURL_POLL_OUT) ? EPOLLOUT : 0); 271 272 if(f->sockfd) { 273 if(epoll_ctl(g->epfd, EPOLL_CTL_DEL, f->sockfd, NULL)) 274 fprintf(stderr, "EPOLL_CTL_DEL failed for fd: %d : %s\n", 275 f->sockfd, strerror(errno)); 276 } 277 278 f->sockfd = s; 279 f->action = act; 280 f->easy = e; 281 282 ev.events = kind; 283 ev.data.fd = s; 284 if(epoll_ctl(g->epfd, EPOLL_CTL_ADD, s, &ev)) 285 fprintf(stderr, "EPOLL_CTL_ADD failed for fd: %d : %s\n", 286 s, strerror(errno)); 287 } 288 289 290 291 /* Initialize a new SockInfo structure */ 292 static void addsock(curl_socket_t s, CURL *easy, int action, 293 struct GlobalInfo *g) 294 { 295 struct SockInfo *fdp = (struct SockInfo*)calloc(1, sizeof(struct SockInfo)); 296 297 fdp->global = g; 298 setsock(fdp, s, easy, action, g); 299 curl_multi_assign(g->multi, s, fdp); 300 } 301 302 /* CURLMOPT_SOCKETFUNCTION */ 303 static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp) 304 { 305 struct GlobalInfo *g = (struct GlobalInfo*) cbp; 306 struct SockInfo *fdp = (struct SockInfo*) sockp; 307 const char *whatstr[]={ "none", "IN", "OUT", "INOUT", "REMOVE" }; 308 309 fprintf(MSG_OUT, 310 "socket callback: s=%d e=%p what=%s ", s, e, whatstr[what]); 311 if(what == CURL_POLL_REMOVE) { 312 fprintf(MSG_OUT, "\n"); 313 remsock(fdp, g); 314 } 315 else { 316 if(!fdp) { 317 fprintf(MSG_OUT, "Adding data: %s\n", whatstr[what]); 318 addsock(s, e, what, g); 319 } 320 else { 321 fprintf(MSG_OUT, 322 "Changing action from %s to %s\n", 323 whatstr[fdp->action], whatstr[what]); 324 setsock(fdp, s, e, what, g); 325 } 326 } 327 return 0; 328 } 329 330 331 332 /* CURLOPT_WRITEFUNCTION */ 333 static size_t write_cb(void *ptr, size_t size, size_t nmemb, void *data) 334 { 335 (void)ptr; 336 (void)data; 337 return size * nmemb; 338 } 339 340 341 /* CURLOPT_PROGRESSFUNCTION */ 342 static int prog_cb(void *p, double dltotal, double dlnow, double ult, 343 double uln) 344 { 345 struct ConnInfo *conn = (struct ConnInfo *)p; 346 (void)ult; 347 (void)uln; 348 349 fprintf(MSG_OUT, "Progress: %s (%g/%g)\n", conn->url, dlnow, dltotal); 350 return 0; 351 } 352 353 354 /* Create a new easy handle, and add it to the global curl_multi */ 355 static void new_conn(const char *url, struct GlobalInfo *g) 356 { 357 struct ConnInfo *conn; 358 CURLMcode rc; 359 360 conn = (struct ConnInfo*)calloc(1, sizeof(*conn)); 361 conn->error[0] = '\0'; 362 363 conn->easy = curl_easy_init(); 364 if(!conn->easy) { 365 fprintf(MSG_OUT, "curl_easy_init() failed, exiting!\n"); 366 exit(2); 367 } 368 conn->global = g; 369 conn->url = strdup(url); 370 curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url); 371 curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb); 372 curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, conn); 373 curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L); 374 curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error); 375 curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn); 376 curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 0L); 377 curl_easy_setopt(conn->easy, CURLOPT_PROGRESSFUNCTION, prog_cb); 378 curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn); 379 curl_easy_setopt(conn->easy, CURLOPT_FOLLOWLOCATION, 1L); 380 curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_TIME, 3L); 381 curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_LIMIT, 10L); 382 fprintf(MSG_OUT, 383 "Adding easy %p to multi %p (%s)\n", conn->easy, g->multi, url); 384 rc = curl_multi_add_handle(g->multi, conn->easy); 385 mcode_or_die("new_conn: curl_multi_add_handle", rc); 386 387 /* note that the add_handle() sets a timeout to trigger soon so that the 388 * necessary socket_action() call gets called by this app */ 389 } 390 391 /* This gets called whenever data is received from the fifo */ 392 static void fifo_cb(struct GlobalInfo *g, int revents) 393 { 394 char s[1024]; 395 long int rv = 0; 396 int n = 0; 397 398 do { 399 s[0]='\0'; 400 rv = fscanf(g->input, "%1023s%n", s, &n); 401 s[n]='\0'; 402 if(n && s[0]) { 403 new_conn(s, g); /* if we read a URL, go get it! */ 404 } 405 else 406 break; 407 } while(rv != EOF); 408 } 409 410 /* Create a named pipe and tell libevent to monitor it */ 411 static const char *fifo = "hiper.fifo"; 412 static int init_fifo(struct GlobalInfo *g) 413 { 414 struct stat st; 415 curl_socket_t sockfd; 416 struct epoll_event epev; 417 418 fprintf(MSG_OUT, "Creating named pipe \"%s\"\n", fifo); 419 if(lstat(fifo, &st) == 0) { 420 if((st.st_mode & S_IFMT) == S_IFREG) { 421 errno = EEXIST; 422 perror("lstat"); 423 return 1; 424 } 425 } 426 unlink(fifo); 427 if(mkfifo(fifo, 0600) == -1) { 428 perror("mkfifo"); 429 return 1; 430 } 431 sockfd = open(fifo, O_RDWR | O_NONBLOCK, 0); 432 if(sockfd == -1) { 433 perror("open"); 434 return 1; 435 } 436 437 g->fifofd = sockfd; 438 g->input = fdopen(sockfd, "r"); 439 440 epev.events = EPOLLIN; 441 epev.data.fd = sockfd; 442 epoll_ctl(g->epfd, EPOLL_CTL_ADD, sockfd, &epev); 443 444 fprintf(MSG_OUT, "Now, pipe some URL's into > %s\n", fifo); 445 return 0; 446 } 447 448 static void clean_fifo(struct GlobalInfo *g) 449 { 450 epoll_ctl(g->epfd, EPOLL_CTL_DEL, g->fifofd, NULL); 451 fclose(g->input); 452 unlink(fifo); 453 } 454 455 456 int g_should_exit_ = 0; 457 458 void sigint_handler(int signo) 459 { 460 g_should_exit_ = 1; 461 } 462 463 int main(int argc, char **argv) 464 { 465 struct GlobalInfo g; 466 struct itimerspec its; 467 struct epoll_event ev; 468 struct epoll_event events[10]; 469 (void)argc; 470 (void)argv; 471 472 g_should_exit_ = 0; 473 signal(SIGINT, sigint_handler); 474 475 memset(&g, 0, sizeof(g)); 476 g.epfd = epoll_create1(EPOLL_CLOEXEC); 477 if(g.epfd == -1) { 478 perror("epoll_create1 failed"); 479 return 1; 480 } 481 482 g.tfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); 483 if(g.tfd == -1) { 484 perror("timerfd_create failed"); 485 return 1; 486 } 487 488 memset(&its, 0, sizeof(its)); 489 its.it_interval.tv_sec = 0; 490 its.it_value.tv_sec = 1; 491 timerfd_settime(g.tfd, 0, &its, NULL); 492 493 ev.events = EPOLLIN; 494 ev.data.fd = g.tfd; 495 epoll_ctl(g.epfd, EPOLL_CTL_ADD, g.tfd, &ev); 496 497 if(init_fifo(&g)) 498 return 1; 499 g.multi = curl_multi_init(); 500 501 /* setup the generic multi interface options we want */ 502 curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb); 503 curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g); 504 curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb); 505 curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g); 506 507 /* we do not call any curl_multi_socket*() function yet as we have no handles 508 added! */ 509 510 fprintf(MSG_OUT, "Entering wait loop\n"); 511 fflush(MSG_OUT); 512 while(!g_should_exit_) { 513 int idx; 514 int err = epoll_wait(g.epfd, events, 515 sizeof(events)/sizeof(struct epoll_event), 10000); 516 if(err == -1) { 517 /* !checksrc! disable ERRNOVAR 1 */ 518 if(errno == EINTR) { 519 fprintf(MSG_OUT, "note: wait interrupted\n"); 520 continue; 521 } 522 else { 523 perror("epoll_wait"); 524 return 1; 525 } 526 } 527 528 for(idx = 0; idx < err; ++idx) { 529 if(events[idx].data.fd == g.fifofd) { 530 fifo_cb(&g, events[idx].events); 531 } 532 else if(events[idx].data.fd == g.tfd) { 533 timer_cb(&g, events[idx].events); 534 } 535 else { 536 event_cb(&g, events[idx].data.fd, events[idx].events); 537 } 538 } 539 } 540 541 fprintf(MSG_OUT, "Exiting normally.\n"); 542 fflush(MSG_OUT); 543 544 curl_multi_cleanup(g.multi); 545 clean_fifo(&g); 546 return 0; 547 }