hiperfifo.c 11 KB

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