Created
March 16, 2018 10:47
-
-
Save bagder/5f724195dabf3b04c72e3ca8474eb488 to your computer and use it in GitHub Desktop.
attempt to repro pausing problem with multi_socket API
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#include <stdio.h> | |
#include <string.h> | |
#include <stdlib.h> | |
#include <sys/time.h> | |
#include <time.h> | |
#include <unistd.h> | |
#include <sys/poll.h> | |
#include <curl/curl.h> | |
#include <event2/event.h> | |
#include <event2/event_struct.h> | |
#include <fcntl.h> | |
#include <sys/stat.h> | |
#include <errno.h> | |
#include <sys/cdefs.h> | |
#ifdef __GNUC__ | |
#define _Unused __attribute__((unused)) | |
#else | |
#define _Unused | |
#endif | |
#define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */ | |
/* Global information, common to all connections */ | |
typedef struct _GlobalInfo | |
{ | |
struct event_base *evbase; | |
struct event timer_event; | |
CURLM *multi; | |
int still_running; | |
FILE *input; | |
} GlobalInfo; | |
/* Information associated with a specific easy handle */ | |
typedef struct _ConnInfo | |
{ | |
CURL *easy; | |
char *url; | |
GlobalInfo *global; | |
char error[CURL_ERROR_SIZE]; | |
struct event periodic; | |
struct { | |
char *mem; | |
unsigned size; | |
unsigned base; | |
unsigned avail; | |
} buffer; | |
int paused; | |
} ConnInfo; | |
/* Information associated with a specific socket */ | |
typedef struct _SockInfo | |
{ | |
curl_socket_t sockfd; | |
CURL *easy; | |
int action; | |
long timeout; | |
struct event ev; | |
GlobalInfo *global; | |
} SockInfo; | |
#define BUFFER_SIZE 2097152 | |
#define __case(code) \ | |
case code: s = __STRING(code) | |
/* Die if we get a bad CURLMcode somewhere */ | |
static void mcode_or_die(const char *where, CURLMcode code) | |
{ | |
if(CURLM_OK != code) { | |
const char *s; | |
switch(code) { | |
__case(CURLM_BAD_HANDLE); break; | |
__case(CURLM_BAD_EASY_HANDLE); break; | |
__case(CURLM_OUT_OF_MEMORY); break; | |
__case(CURLM_INTERNAL_ERROR); break; | |
__case(CURLM_UNKNOWN_OPTION); break; | |
__case(CURLM_LAST); break; | |
default: s = "CURLM_unknown"; break; | |
__case(CURLM_BAD_SOCKET); | |
fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s); | |
/* ignore this error */ | |
return; | |
} | |
fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s); | |
exit(code); | |
} | |
} | |
/* Update the event timer after curl_multi library calls */ | |
static int multi_timer_cb(CURLM *multi _Unused, long timeout_ms, GlobalInfo *g) | |
{ | |
struct timeval timeout; | |
CURLMcode rc; | |
timeout.tv_sec = timeout_ms/1000; | |
timeout.tv_usec = (timeout_ms%1000)*1000; | |
fprintf(MSG_OUT, "multi_timer_cb: Setting timeout to %ld ms\n", timeout_ms); | |
/* TODO | |
* | |
* if timeout_ms is 0, call curl_multi_socket_action() at once! | |
* | |
* if timeout_ms is -1, just delete the timer | |
* | |
* for all other values of timeout_ms, this should set or *update* | |
* the timer to the new value | |
*/ | |
if (timeout_ms == 0) { | |
rc = curl_multi_socket_action(g->multi, | |
CURL_SOCKET_TIMEOUT, 0, &g->still_running); | |
mcode_or_die("multi_timer_cb: curl_multi_socket_action", rc); | |
} else if (timeout_ms == -1) | |
evtimer_del(&g->timer_event); | |
else | |
evtimer_add(&g->timer_event, &timeout); | |
return 0; | |
} | |
/* Check for completed transfers, and remove their easy handles */ | |
static void check_multi_info(GlobalInfo *g) | |
{ | |
char *eff_url; | |
CURLMsg *msg; | |
int msgs_left; | |
ConnInfo *conn; | |
CURL *easy; | |
CURLcode res; | |
fprintf(MSG_OUT, "REMAINING: %d\n", g->still_running); | |
while((msg = curl_multi_info_read(g->multi, &msgs_left))) { | |
if(msg->msg == CURLMSG_DONE) { | |
easy = msg->easy_handle; | |
res = msg->data.result; | |
curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn); | |
curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url); | |
fprintf(MSG_OUT, "DONE: %s => (%d) %s\n", eff_url, res, conn->error); | |
curl_multi_remove_handle(g->multi, easy); | |
free(conn->url); | |
curl_easy_cleanup(easy); | |
free(conn->buffer.mem); | |
event_del(&conn->periodic); | |
free(conn); | |
} | |
} | |
} | |
/* Called by libevent when we get action on a multi socket */ | |
static void event_cb(int fd, short kind, void *userp) | |
{ | |
GlobalInfo *g = (GlobalInfo*) userp; | |
CURLMcode rc; | |
int action = | |
(kind & EV_READ ? CURL_CSELECT_IN : 0) | | |
(kind & EV_WRITE ? CURL_CSELECT_OUT : 0); | |
rc = curl_multi_socket_action(g->multi, fd, action, &g->still_running); | |
mcode_or_die("event_cb: curl_multi_socket_action", rc); | |
check_multi_info(g); | |
if(g->still_running <= 0) { | |
fprintf(MSG_OUT, "last transfer done, kill timeout\n"); | |
if(evtimer_pending(&g->timer_event, NULL)) { | |
evtimer_del(&g->timer_event); | |
} | |
} | |
} | |
/* Called by libevent when our timeout expires */ | |
static void timer_cb(int fd _Unused, short kind _Unused, void *userp) | |
{ | |
GlobalInfo *g = (GlobalInfo *)userp; | |
CURLMcode rc; | |
rc = curl_multi_socket_action(g->multi, | |
CURL_SOCKET_TIMEOUT, 0, &g->still_running); | |
mcode_or_die("timer_cb: curl_multi_socket_action", rc); | |
check_multi_info(g); | |
} | |
static int pausecounter = 0; | |
/* Called by libevent for our periodic timers */ | |
static void periodic_cb(int fd _Unused, short kind _Unused, void *userp) | |
{ | |
ConnInfo *conn = (ConnInfo *)userp; | |
fprintf(stderr, "tick: counter %d %s\n", pausecounter, | |
conn->paused?"[PAUSED]":""); | |
if (conn->paused) { | |
if (!--pausecounter) { | |
fprintf(stderr, "tick: resuming stream\n"); | |
conn->paused = 0; | |
curl_easy_pause(conn->easy, CURLPAUSE_CONT); | |
} | |
} | |
} | |
/* Clean up the SockInfo structure */ | |
static void remsock(SockInfo *f) | |
{ | |
if(f) { | |
event_del(&f->ev); | |
free(f); | |
} | |
} | |
/* Assign information to a SockInfo structure */ | |
static void setsock(SockInfo *f, curl_socket_t s, CURL *e, int act, | |
GlobalInfo *g) | |
{ | |
int kind = | |
(act&CURL_POLL_IN?EV_READ:0)|(act&CURL_POLL_OUT?EV_WRITE:0)|EV_PERSIST; | |
f->sockfd = s; | |
f->action = act; | |
f->easy = e; | |
event_del(&f->ev); | |
event_assign(&f->ev, g->evbase, f->sockfd, kind, event_cb, g); | |
event_add(&f->ev, NULL); | |
} | |
/* Initialize a new SockInfo structure */ | |
static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo *g) | |
{ | |
SockInfo *fdp = calloc(sizeof(SockInfo), 1); | |
fdp->global = g; | |
setsock(fdp, s, easy, action, g); | |
curl_multi_assign(g->multi, s, fdp); | |
} | |
/* CURLMOPT_SOCKETFUNCTION */ | |
static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp) | |
{ | |
GlobalInfo *g = (GlobalInfo*) cbp; | |
SockInfo *fdp = (SockInfo*) sockp; | |
const char *whatstr[]={ "none", "IN", "OUT", "INOUT", "REMOVE" }; | |
fprintf(MSG_OUT, | |
"socket callback: s=%d e=%p what=%s ", s, e, whatstr[what]); | |
if(what == CURL_POLL_REMOVE) { | |
fprintf(MSG_OUT, "\n"); | |
remsock(fdp); | |
} | |
else { | |
if(!fdp) { | |
fprintf(MSG_OUT, "Adding data: %s\n", whatstr[what]); | |
addsock(s, e, what, g); | |
} | |
else { | |
fprintf(MSG_OUT, | |
"Changing action from %s to %s\n", | |
whatstr[fdp->action], whatstr[what]); | |
setsock(fdp, s, e, what, g); | |
} | |
} | |
return 0; | |
} | |
/* CURLOPT_WRITEFUNCTION */ | |
static size_t write_cb(void *ptr _Unused, size_t size, size_t nmemb, void *data) | |
{ | |
size_t realsize = size * nmemb; | |
ConnInfo *conn _Unused = (ConnInfo*) data; | |
if (++pausecounter > 10) { | |
fprintf(MSG_OUT, "write_cb: PAUSE!\n"); | |
conn->paused = 1; | |
return CURL_WRITEFUNC_PAUSE; | |
} | |
fprintf(MSG_OUT, "write_cb: called %d\n", pausecounter); | |
return realsize; | |
} | |
/* CURLOPT_PROGRESSFUNCTION */ | |
static int prog_cb(void *p, double dltotal, double dlnow, double ult _Unused, | |
double uln _Unused) | |
{ | |
ConnInfo *conn _Unused = (ConnInfo *)p; | |
// fprintf(MSG_OUT, "Progress: %s (%g/%g)\n", conn->url, dlnow, dltotal); | |
return 0; | |
} | |
/* Create a new easy handle, and add it to the global curl_multi */ | |
static void new_conn(char *url, GlobalInfo *g) | |
{ | |
ConnInfo *conn; | |
CURLMcode rc; | |
struct timeval tv = (struct timeval) { 0, 100000 }; | |
conn = calloc(1, sizeof(ConnInfo)); | |
memset(conn, 0, sizeof(ConnInfo)); | |
conn->error[0]='\0'; | |
conn->easy = curl_easy_init(); | |
if(!conn->easy) { | |
fprintf(MSG_OUT, "curl_easy_init() failed, exiting!\n"); | |
exit(2); | |
} | |
conn->global = g; | |
conn->url = strdup(url); | |
curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url); | |
curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb); | |
curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, conn); | |
curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L); | |
curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error); | |
curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn); | |
curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 0L); | |
curl_easy_setopt(conn->easy, CURLOPT_PROGRESSFUNCTION, prog_cb); | |
curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn); | |
curl_easy_setopt(conn->easy, CURLOPT_FOLLOWLOCATION, 1L); | |
fprintf(MSG_OUT, | |
"Adding easy %p to multi %p (%s)\n", conn->easy, g->multi, url); | |
rc = curl_multi_add_handle(g->multi, conn->easy); | |
mcode_or_die("new_conn: curl_multi_add_handle", rc); | |
/* note that the add_handle() will set a time-out to trigger very soon so | |
that the necessary socket_action() call will be called by this app */ | |
event_assign(&conn->periodic, g->evbase, -1, EV_TIMEOUT | EV_PERSIST, periodic_cb, conn); | |
event_add(&conn->periodic, &tv); | |
conn->buffer.mem = malloc(BUFFER_SIZE); | |
conn->buffer.size = BUFFER_SIZE; | |
// conn->buffer.base = conn->buffer.avail = 0; | |
} | |
int main(int argc, char **argv) | |
{ | |
GlobalInfo g; | |
int n; | |
memset(&g, 0, sizeof(GlobalInfo)); | |
g.evbase = event_base_new(); | |
g.multi = curl_multi_init(); | |
evtimer_assign(&g.timer_event, g.evbase, timer_cb, &g); | |
/* setup the generic multi interface options we want */ | |
curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb); | |
curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g); | |
curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb); | |
curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g); | |
for (n = 1; n < argc; ++n) | |
new_conn(argv[n], &g); | |
event_base_dispatch(g.evbase); | |
/* this, of course, won't get called since only way to stop this program is | |
via ctrl-C, but it is here to show how cleanup /would/ be done. */ | |
event_del(&g.timer_event); | |
event_base_free(g.evbase); | |
curl_multi_cleanup(g.multi); | |
return 0; | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment