ephiperfifo.c 15 KB

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