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