2
0

hiperfifo.c 12 KB

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