evhiperfifo.c 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450
  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. Requires libev and a (POSIX?) system that has mkfifo().
  36. This is an adaptation of libcurl's "hipev.c" and libevent's "event-test.c"
  37. sample programs.
  38. When running, the program creates the named pipe "hiper.fifo"
  39. Whenever there is input into the fifo, the program reads the input as a list
  40. of URL's and creates some new easy handles to fetch each URL via the
  41. curl_multi "hiper" API.
  42. Thus, you can try a single URL:
  43. % echo http://www.yahoo.com > hiper.fifo
  44. Or a whole bunch of them:
  45. % cat my-url-list > hiper.fifo
  46. The fifo buffer is handled almost instantly, so you can even add more URL's
  47. while the previous requests are still being downloaded.
  48. Note:
  49. For the sake of simplicity, URL length is limited to 1023 char's !
  50. This is purely a demo app, all retrieved data is simply discarded by the write
  51. callback.
  52. */
  53. #include <stdio.h>
  54. #include <string.h>
  55. #include <stdlib.h>
  56. #include <sys/time.h>
  57. #include <time.h>
  58. #include <unistd.h>
  59. #include <sys/poll.h>
  60. #include <curl/curl.h>
  61. #include <ev.h>
  62. #include <fcntl.h>
  63. #include <sys/stat.h>
  64. #include <errno.h>
  65. #define DPRINT(x...) printf(x)
  66. #define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */
  67. /* Global information, common to all connections */
  68. typedef struct _GlobalInfo
  69. {
  70. struct ev_loop *loop;
  71. struct ev_io fifo_event;
  72. struct ev_timer timer_event;
  73. CURLM *multi;
  74. int still_running;
  75. FILE *input;
  76. } GlobalInfo;
  77. /* Information associated with a specific easy handle */
  78. typedef struct _ConnInfo
  79. {
  80. CURL *easy;
  81. char *url;
  82. GlobalInfo *global;
  83. char error[CURL_ERROR_SIZE];
  84. } ConnInfo;
  85. /* Information associated with a specific socket */
  86. typedef struct _SockInfo
  87. {
  88. curl_socket_t sockfd;
  89. CURL *easy;
  90. int action;
  91. long timeout;
  92. struct ev_io ev;
  93. int evset;
  94. GlobalInfo *global;
  95. } SockInfo;
  96. static void timer_cb(EV_P_ struct ev_timer *w, int revents);
  97. /* Update the event timer after curl_multi library calls */
  98. static int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo *g)
  99. {
  100. DPRINT("%s %li\n", __PRETTY_FUNCTION__, timeout_ms);
  101. ev_timer_stop(g->loop, &g->timer_event);
  102. if(timeout_ms >= 0) {
  103. /* -1 means delete, other values are timeout times in milliseconds */
  104. double t = timeout_ms / 1000;
  105. ev_timer_init(&g->timer_event, timer_cb, t, 0.);
  106. ev_timer_start(g->loop, &g->timer_event);
  107. }
  108. return 0;
  109. }
  110. /* Die if we get a bad CURLMcode somewhere */
  111. static void mcode_or_die(const char *where, CURLMcode code)
  112. {
  113. if(CURLM_OK != code) {
  114. const char *s;
  115. switch(code) {
  116. case CURLM_BAD_HANDLE:
  117. s = "CURLM_BAD_HANDLE";
  118. break;
  119. case CURLM_BAD_EASY_HANDLE:
  120. s = "CURLM_BAD_EASY_HANDLE";
  121. break;
  122. case CURLM_OUT_OF_MEMORY:
  123. s = "CURLM_OUT_OF_MEMORY";
  124. break;
  125. case CURLM_INTERNAL_ERROR:
  126. s = "CURLM_INTERNAL_ERROR";
  127. break;
  128. case CURLM_UNKNOWN_OPTION:
  129. s = "CURLM_UNKNOWN_OPTION";
  130. break;
  131. case CURLM_LAST:
  132. s = "CURLM_LAST";
  133. break;
  134. default:
  135. s = "CURLM_unknown";
  136. break;
  137. case CURLM_BAD_SOCKET:
  138. s = "CURLM_BAD_SOCKET";
  139. fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s);
  140. /* ignore this error */
  141. return;
  142. }
  143. fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s);
  144. exit(code);
  145. }
  146. }
  147. /* Check for completed transfers, and remove their easy handles */
  148. static void check_multi_info(GlobalInfo *g)
  149. {
  150. char *eff_url;
  151. CURLMsg *msg;
  152. int msgs_left;
  153. ConnInfo *conn;
  154. CURL *easy;
  155. CURLcode res;
  156. fprintf(MSG_OUT, "REMAINING: %d\n", g->still_running);
  157. while((msg = curl_multi_info_read(g->multi, &msgs_left))) {
  158. if(msg->msg == CURLMSG_DONE) {
  159. easy = msg->easy_handle;
  160. res = msg->data.result;
  161. curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
  162. curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url);
  163. fprintf(MSG_OUT, "DONE: %s => (%d) %s\n", eff_url, res, conn->error);
  164. curl_multi_remove_handle(g->multi, easy);
  165. free(conn->url);
  166. curl_easy_cleanup(easy);
  167. free(conn);
  168. }
  169. }
  170. }
  171. /* Called by libevent when we get action on a multi socket */
  172. static void event_cb(EV_P_ struct ev_io *w, int revents)
  173. {
  174. DPRINT("%s w %p revents %i\n", __PRETTY_FUNCTION__, w, revents);
  175. GlobalInfo *g = (GlobalInfo*) w->data;
  176. CURLMcode rc;
  177. int action = ((revents & EV_READ) ? CURL_POLL_IN : 0) |
  178. ((revents & EV_WRITE) ? CURL_POLL_OUT : 0);
  179. rc = curl_multi_socket_action(g->multi, w->fd, action, &g->still_running);
  180. mcode_or_die("event_cb: curl_multi_socket_action", rc);
  181. check_multi_info(g);
  182. if(g->still_running <= 0) {
  183. fprintf(MSG_OUT, "last transfer done, kill timeout\n");
  184. ev_timer_stop(g->loop, &g->timer_event);
  185. }
  186. }
  187. /* Called by libevent when our timeout expires */
  188. static void timer_cb(EV_P_ struct ev_timer *w, int revents)
  189. {
  190. DPRINT("%s w %p revents %i\n", __PRETTY_FUNCTION__, w, revents);
  191. GlobalInfo *g = (GlobalInfo *)w->data;
  192. CURLMcode rc;
  193. rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0,
  194. &g->still_running);
  195. mcode_or_die("timer_cb: curl_multi_socket_action", rc);
  196. check_multi_info(g);
  197. }
  198. /* Clean up the SockInfo structure */
  199. static void remsock(SockInfo *f, GlobalInfo *g)
  200. {
  201. printf("%s \n", __PRETTY_FUNCTION__);
  202. if(f) {
  203. if(f->evset)
  204. ev_io_stop(g->loop, &f->ev);
  205. free(f);
  206. }
  207. }
  208. /* Assign information to a SockInfo structure */
  209. static void setsock(SockInfo *f, curl_socket_t s, CURL *e, int act,
  210. GlobalInfo *g)
  211. {
  212. printf("%s \n", __PRETTY_FUNCTION__);
  213. int kind = ((act & CURL_POLL_IN) ? EV_READ : 0) |
  214. ((act & CURL_POLL_OUT) ? EV_WRITE : 0);
  215. f->sockfd = s;
  216. f->action = act;
  217. f->easy = e;
  218. if(f->evset)
  219. ev_io_stop(g->loop, &f->ev);
  220. ev_io_init(&f->ev, event_cb, f->sockfd, kind);
  221. f->ev.data = g;
  222. f->evset = 1;
  223. ev_io_start(g->loop, &f->ev);
  224. }
  225. /* Initialize a new SockInfo structure */
  226. static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo *g)
  227. {
  228. SockInfo *fdp = calloc(1, sizeof(SockInfo));
  229. fdp->global = g;
  230. setsock(fdp, s, easy, action, g);
  231. curl_multi_assign(g->multi, s, fdp);
  232. }
  233. /* CURLMOPT_SOCKETFUNCTION */
  234. static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp)
  235. {
  236. DPRINT("%s e %p s %i what %i cbp %p sockp %p\n",
  237. __PRETTY_FUNCTION__, e, s, what, cbp, sockp);
  238. GlobalInfo *g = (GlobalInfo*) cbp;
  239. SockInfo *fdp = (SockInfo*) sockp;
  240. const char *whatstr[]={ "none", "IN", "OUT", "INOUT", "REMOVE"};
  241. fprintf(MSG_OUT,
  242. "socket callback: s=%d e=%p what=%s ", s, e, whatstr[what]);
  243. if(what == CURL_POLL_REMOVE) {
  244. fprintf(MSG_OUT, "\n");
  245. remsock(fdp, g);
  246. }
  247. else {
  248. if(!fdp) {
  249. fprintf(MSG_OUT, "Adding data: %s\n", whatstr[what]);
  250. addsock(s, e, what, g);
  251. }
  252. else {
  253. fprintf(MSG_OUT,
  254. "Changing action from %s to %s\n",
  255. whatstr[fdp->action], whatstr[what]);
  256. setsock(fdp, s, e, what, g);
  257. }
  258. }
  259. return 0;
  260. }
  261. /* CURLOPT_WRITEFUNCTION */
  262. static size_t write_cb(void *ptr, size_t size, size_t nmemb, void *data)
  263. {
  264. size_t realsize = size * nmemb;
  265. ConnInfo *conn = (ConnInfo*) data;
  266. (void)ptr;
  267. (void)conn;
  268. return realsize;
  269. }
  270. /* CURLOPT_PROGRESSFUNCTION */
  271. static int prog_cb(void *p, double dltotal, double dlnow, double ult,
  272. double uln)
  273. {
  274. ConnInfo *conn = (ConnInfo *)p;
  275. (void)ult;
  276. (void)uln;
  277. fprintf(MSG_OUT, "Progress: %s (%g/%g)\n", conn->url, dlnow, dltotal);
  278. return 0;
  279. }
  280. /* Create a new easy handle, and add it to the global curl_multi */
  281. static void new_conn(char *url, GlobalInfo *g)
  282. {
  283. ConnInfo *conn;
  284. CURLMcode rc;
  285. conn = calloc(1, sizeof(ConnInfo));
  286. conn->error[0]='\0';
  287. conn->easy = curl_easy_init();
  288. if(!conn->easy) {
  289. fprintf(MSG_OUT, "curl_easy_init() failed, exiting!\n");
  290. exit(2);
  291. }
  292. conn->global = g;
  293. conn->url = strdup(url);
  294. curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url);
  295. curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb);
  296. curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, conn);
  297. curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L);
  298. curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error);
  299. curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn);
  300. curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 0L);
  301. curl_easy_setopt(conn->easy, CURLOPT_PROGRESSFUNCTION, prog_cb);
  302. curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn);
  303. curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_TIME, 3L);
  304. curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_LIMIT, 10L);
  305. fprintf(MSG_OUT,
  306. "Adding easy %p to multi %p (%s)\n", conn->easy, g->multi, url);
  307. rc = curl_multi_add_handle(g->multi, conn->easy);
  308. mcode_or_die("new_conn: curl_multi_add_handle", rc);
  309. /* note that the add_handle() will set a time-out to trigger very soon so
  310. that the necessary socket_action() call will be called by this app */
  311. }
  312. /* This gets called whenever data is received from the fifo */
  313. static void fifo_cb(EV_P_ struct ev_io *w, int revents)
  314. {
  315. char s[1024];
  316. long int rv = 0;
  317. int n = 0;
  318. GlobalInfo *g = (GlobalInfo *)w->data;
  319. do {
  320. s[0]='\0';
  321. rv = fscanf(g->input, "%1023s%n", s, &n);
  322. s[n]='\0';
  323. if(n && s[0]) {
  324. new_conn(s, g); /* if we read a URL, go get it! */
  325. }
  326. else
  327. break;
  328. } while(rv != EOF);
  329. }
  330. /* Create a named pipe and tell libevent to monitor it */
  331. static int init_fifo(GlobalInfo *g)
  332. {
  333. struct stat st;
  334. static const char *fifo = "hiper.fifo";
  335. curl_socket_t sockfd;
  336. fprintf(MSG_OUT, "Creating named pipe \"%s\"\n", fifo);
  337. if(lstat (fifo, &st) == 0) {
  338. if((st.st_mode & S_IFMT) == S_IFREG) {
  339. errno = EEXIST;
  340. perror("lstat");
  341. exit(1);
  342. }
  343. }
  344. unlink(fifo);
  345. if(mkfifo (fifo, 0600) == -1) {
  346. perror("mkfifo");
  347. exit(1);
  348. }
  349. sockfd = open(fifo, O_RDWR | O_NONBLOCK, 0);
  350. if(sockfd == -1) {
  351. perror("open");
  352. exit(1);
  353. }
  354. g->input = fdopen(sockfd, "r");
  355. fprintf(MSG_OUT, "Now, pipe some URL's into > %s\n", fifo);
  356. ev_io_init(&g->fifo_event, fifo_cb, sockfd, EV_READ);
  357. ev_io_start(g->loop, &g->fifo_event);
  358. return (0);
  359. }
  360. int main(int argc, char **argv)
  361. {
  362. GlobalInfo g;
  363. (void)argc;
  364. (void)argv;
  365. memset(&g, 0, sizeof(GlobalInfo));
  366. g.loop = ev_default_loop(0);
  367. init_fifo(&g);
  368. g.multi = curl_multi_init();
  369. ev_timer_init(&g.timer_event, timer_cb, 0., 0.);
  370. g.timer_event.data = &g;
  371. g.fifo_event.data = &g;
  372. curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
  373. curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g);
  374. curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
  375. curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g);
  376. /* we do not call any curl_multi_socket*() function yet as we have no handles
  377. added! */
  378. ev_loop(g.loop, 0);
  379. curl_multi_cleanup(g.multi);
  380. return 0;
  381. }