*/
#include <assert.h>
+#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <signal.h>
+#ifdef WIN32
+ #include <fcntl.h>
+ #define pipe(f) _pipe(f, 1000, _O_BINARY)
+#else
+ #include <unistd.h>
+#endif
+
#include <event.h>
-#include <evhttp.h>
#include "transmission.h"
-#include "list.h"
#include "platform.h"
#include "trevent.h"
#include "utils.h"
-/* #define DEBUG */
-#ifdef DEBUG
-#include <stdio.h>
-#undef tr_dbg
-#define tr_dbg( a, b... ) fprintf(stderr, a "\n", ##b )
-#endif
-
/***
****
***/
typedef struct tr_event_handle
{
+ uint8_t die;
+ int fds[2];
tr_lock * lock;
tr_handle * h;
tr_thread * thread;
- tr_list * commands;
struct event_base * base;
- struct event pulse;
- struct timeval pulseInterval;
- uint8_t die;
-
- int timerCount;
+ struct event pipeEvent;
}
tr_event_handle;
-#ifdef DEBUG
-static int reads = 0;
-static int writes = 0;
-#endif
-
-enum mode
-{
- TR_EV_TIMER_ADD,
- TR_EV_EXEC
-};
-
typedef int timer_func(void*);
struct tr_timer
uint8_t inCallback;
};
-struct tr_event_command
+struct tr_run_data
{
- int mode;
-
- struct tr_timer * timer;
-
- struct evhttp_connection * evcon;
- struct evhttp_request * req;
- enum evhttp_cmd_type evtype;
- char * uri;
-
- struct bufferevent * bufev;
- short enable;
- short disable;
- char * buf;
- size_t buflen;
-
- void (*func)( void* );
+ void (*func)( void * );
void * user_data;
};
+#define dbgmsg(fmt...) tr_deepLog( __FILE__, __LINE__, "event", ##fmt )
+
static void
-pumpList( int i UNUSED, short s UNUSED, void * veh )
+readFromPipe( int fd, short eventType, void * veh )
{
+ char ch;
+ int ret;
tr_event_handle * eh = veh;
- int doDie;
+ dbgmsg( "readFromPipe: eventType is %hd", eventType );
- for( ;; )
- {
- struct tr_event_command * cmd;
+ /* read the command type */
+ ch = '\0';
+ do {
+ ret = read( fd, &ch, 1 );
+ } while( !eh->die && ret<0 && errno==EAGAIN );
+ dbgmsg( "command is [%c], ret is %d, errno is %d", ch, ret, (int)errno );
- doDie = eh->die && !eh->timerCount;
- if( doDie )
+ switch( ch )
+ {
+ case 'r': /* run in libevent thread */
+ {
+ struct tr_run_data data;
+ const size_t nwant = sizeof( data );
+ const ssize_t ngot = read( fd, &data, nwant );
+ if( !eh->die && ( ngot == (ssize_t)nwant ) ) {
+ dbgmsg( "invoking function in libevent thread" );
+ (data.func)( data.user_data );
+ }
break;
-
- /* get the next command */
- tr_lockLock( eh->lock );
- cmd = tr_list_pop_front( &eh->commands );
- tr_lockUnlock( eh->lock );
- if( cmd == NULL )
+ }
+ case 't': /* create timer */
+ {
+ tr_timer * timer;
+ const size_t nwant = sizeof( timer );
+ const ssize_t ngot = read( fd, &timer, nwant );
+ if( !eh->die && ( ngot == (ssize_t)nwant ) ) {
+ dbgmsg( "adding timer in libevent thread" );
+ evtimer_add( &timer->event, &timer->tv );
+ }
break;
-
- /* process the command */
- switch( cmd->mode )
+ }
+ case '\0': /* eof */
{
- case TR_EV_TIMER_ADD:
- evtimer_add( &cmd->timer->event, &cmd->timer->tv );
- ++eh->timerCount;
- break;
-
- case TR_EV_EXEC:
- (cmd->func)( cmd->user_data );
- break;
-
- default:
- assert( 0 && "unhandled command type!" );
+ dbgmsg( "pipe eof reached... removing event listener" );
+ event_del( &eh->pipeEvent );
+ break;
}
-
- /* cleanup */
- tr_free( cmd );
- }
-
- if( !doDie )
- evtimer_add( &eh->pulse, &eh->pulseInterval );
- else {
- assert( eh->timerCount == 0 );
- event_del( &eh->pulse );
+ default:
+ {
+ assert( 0 && "unhandled command type!" );
+ break;
+ }
}
}
#endif
eh->base = event_init( );
- event_set_log_callback( logFunc );
- evtimer_set( &eh->pulse, pumpList, veh );
- evtimer_add( &eh->pulse, &eh->pulseInterval );
eh->h->events = eh;
+ event_set_log_callback( logFunc );
+
+ /* listen to the pipe's read fd */
+ event_set( &eh->pipeEvent, eh->fds[0], EV_READ|EV_PERSIST, readFromPipe, veh );
+ event_add( &eh->pipeEvent, NULL );
event_dispatch( );
tr_lockFree( eh->lock );
event_base_free( eh->base );
-
eh->h->events = NULL;
-
tr_free( eh );
tr_dbg( "Closing libevent thread" );
}
eh = tr_new0( tr_event_handle, 1 );
eh->lock = tr_lockNew( );
+ pipe( eh->fds );
eh->h = handle;
- eh->pulseInterval = tr_timevalMsec( 100 );
eh->thread = tr_threadNew( libeventThreadFunc, eh, "libeventThreadFunc" );
}
void
tr_eventClose( tr_handle * handle )
{
- tr_event_handle * eh = handle->events;
-
- tr_lockLock( eh->lock );
- tr_list_free( &eh->commands, tr_free );
- eh->die = TRUE;
- tr_lockUnlock( eh->lock );
+ handle->events->die = TRUE;
+ tr_deepLog( __FILE__, __LINE__, NULL, "closing trevent pipe" );
+ close( handle->events->fds[1] );
}
/**
***
**/
-static void
-pushList( struct tr_event_handle * eh, struct tr_event_command * command )
-{
- tr_lockLock( eh->lock );
- tr_list_append( &eh->commands, command );
- tr_lockUnlock( eh->lock );
-}
-
int
tr_amInEventThread( struct tr_handle * handle )
{
***
**/
-static int
-timerCompareFunc( const void * va, const void * vb )
-{
- const struct tr_event_command * a = va;
- const struct tr_timer * b = vb;
- return a->timer == b ? 0 : 1;
-}
-
static void
timerCallback( int fd UNUSED, short event UNUSED, void * vtimer )
{
int more;
struct tr_timer * timer = vtimer;
- void * del;
- del = tr_list_remove( &timer->eh->commands, timer, timerCompareFunc );
-
- if( del != NULL ) /* there's a TIMER_DEL command queued for this timer... */
- more = FALSE;
- else {
- timer->inCallback = 1;
- more = (*timer->func)( timer->user_data );
- timer->inCallback = 0;
- }
+ timer->inCallback = 1;
+ more = (*timer->func)( timer->user_data );
+ timer->inCallback = 0;
if( more )
evtimer_add( &timer->event, &timer->tv );
else
tr_timerFree( &timer );
-
- tr_free( del );
}
void
*ptimer = NULL;
/* destroy the timer directly or via the command queue */
- if( timer!=NULL && !timer->inCallback ) {
- void * del;
+ if( timer && !timer->inCallback )
+ {
assert( tr_amInEventThread( timer->eh->h ) );
- del = tr_list_remove( &timer->eh->commands, timer, timerCompareFunc );
- --timer->eh->timerCount;
event_del( &timer->event );
tr_free( timer );
- tr_free( del );
}
}
timer->eh = handle->events;
evtimer_set( &timer->event, timerCallback, timer );
- if( tr_amInThread( handle->events->thread ) ) {
+ if( tr_amInThread( handle->events->thread ) )
+ {
evtimer_add( &timer->event, &timer->tv );
- ++handle->events->timerCount;
- } else {
- struct tr_event_command * cmd = tr_new0( struct tr_event_command, 1 );
- cmd->mode = TR_EV_TIMER_ADD;
- cmd->timer = timer;
- pushList( handle->events, cmd );
+ }
+ else
+ {
+ const char ch = 't';
+ int fd = handle->events->fds[1];
+ tr_lock * lock = handle->events->lock;
+
+ tr_lockLock( lock );
+ write( fd, &ch, 1 );
+ write( fd, &timer, sizeof(timer) );
+ tr_lockUnlock( lock );
}
return timer;
void * user_data )
{
if( tr_amInThread( handle->events->thread ) )
+ {
(func)( user_data );
- else {
- struct tr_event_command * cmd = tr_new0( struct tr_event_command, 1 );
- cmd->mode = TR_EV_EXEC;
- cmd->func = func;
- cmd->user_data = user_data;
- pushList( handle->events, cmd );
+ }
+ else
+ {
+ const char ch = 'r';
+ int fd = handle->events->fds[1];
+ tr_lock * lock = handle->events->lock;
+ struct tr_run_data data;
+
+ tr_lockLock( lock );
+ write( fd, &ch, 1 );
+ data.func = func;
+ data.user_data = user_data;
+ write( fd, &data, sizeof(data) );
+ tr_lockUnlock( lock );
}
}
* $Id$
*/
+#include <assert.h>
#include <stdlib.h> /* bsearch */
#include <event.h>
(LIBCURL_VERSION_MAJOR == (major) && LIBCURL_VERSION_MINOR == (minor) && \
LIBCURL_VERSION_PATCH >= (micro)))
-//#if CURL_CHECK_VERSION(7,16,0)
-//#define USE_CURL_MULTI_SOCKET
-//#else
-#define PULSE_MSEC 150
-//#endif
+#define PULSE_MSEC 500
#define dbgmsg(fmt...) tr_deepLog( __FILE__, __LINE__, "web", ##fmt )
struct tr_web
{
+ unsigned int dying : 1;
+ unsigned int running : 1;
+ int remain;
CURLM * cm;
tr_session * session;
- int remain;
struct event timer;
};
int unused;
CURLMcode rc;
do {
-#ifdef USE_CURL_MULTI_SOCKET
- rc = curl_multi_socket_all( web->cm, &unused );
-#else
rc = curl_multi_perform( web->cm, &unused );
-#endif
} while( rc == CURLM_CALL_MULTI_PERFORM );
if ( rc == CURLM_OK )
processCompletedTasks( web );
return byteCount;
}
+static void
+ensureTimerIsRunning( tr_web * web )
+{
+ if( !web->running )
+ {
+ struct timeval tv = tr_timevalMsec( PULSE_MSEC );
+ dbgmsg( "starting web timer" );
+ web->running = 1;
+ evtimer_add( &web->timer, &tv );
+ }
+}
+
static void
addTask( void * vtask )
{
struct tr_web * web = task->session->web;
CURL * ch;
+ ensureTimerIsRunning( web );
+
++web->remain;
dbgmsg( "adding task #%lu [%s] (%d remain)", task->tag, task->url, web->remain );
curl_easy_setopt( ch, CURLOPT_IPRESOLVE, CURL_IPRESOLVE_V4 );
curl_easy_setopt( ch, CURLOPT_ENCODING, "" );
curl_multi_add_handle( web->cm, ch );
-
- pump( web );
}
}
}
}
-#ifdef USE_CURL_MULTI_SOCKET
-
-/* libevent says that sock is ready to be processed, so tell libcurl */
-static void
-ev_sock_cb( int sock, short action, void * vweb )
-{
- tr_web * web = vweb;
- CURLMcode rc;
- int mask, unused;
-
- switch (action & (EV_READ|EV_WRITE)) {
- case EV_READ: mask = CURL_CSELECT_IN; break;
- case EV_WRITE: mask = CURL_CSELECT_OUT; break;
- case EV_READ|EV_WRITE: mask = CURL_CSELECT_IN|CURL_CSELECT_OUT; break;
- default: tr_err( "Unknown event %hd\n", action ); return;
- }
-
- do {
- rc = curl_multi_socket_action( web->cm, sock, mask, &unused );
- } while( rc == CURLM_CALL_MULTI_PERFORM );
- if ( rc == CURLM_OK )
- processCompletedTasks( web );
- else
- tr_err( "%s (%d)", curl_multi_strerror(rc), (int)sock );
-
-}
-
-/* CURLMPOPT_SOCKETFUNCTION */
-/* libcurl wants us to tell it when sock is ready to be processed */
static void
-multi_sock_cb( CURL * easy UNUSED,
- curl_socket_t sock,
- int action,
- void * vweb,
- void * assigndata )
+webDestroy( tr_web * web )
{
- tr_web * web = vweb;
- struct event * ev = assigndata;
-
- if( action == CURL_POLL_REMOVE ) {
- if( ev ) {
- dbgmsg( "deleting libevent socket polling" );
- event_del( ev );
- tr_free( ev );
- curl_multi_assign( web->cm, sock, NULL );
- }
- } else {
- int kind;
- if( ev ) {
- event_del( ev );
- } else {
- ev = tr_new0( struct event, 1 );
- curl_multi_assign( web->cm, sock, ev );
- }
- kind = EV_PERSIST;
- if( action & CURL_POLL_IN ) kind |= EV_READ;
- if( action & CURL_POLL_OUT ) kind |= EV_WRITE;
- event_set( ev, sock, kind, ev_sock_cb, web );
- event_add( ev, NULL );
- }
-}
-
-/* libevent says that timeout_ms have passed, so tell libcurl */
-static void
-event_timer_cb( int socket UNUSED, short action UNUSED, void * vweb )
-{
- int unused;
- CURLMcode rc;
- tr_web * web = vweb;
-
- do {
- rc = curl_multi_socket( web->cm, CURL_SOCKET_TIMEOUT, &unused );
- } while( rc == CURLM_CALL_MULTI_PERFORM );
- if ( rc == CURLM_OK )
- processCompletedTasks( web );
- else
- tr_err( "%s", curl_multi_strerror(rc) );
-}
-
-/* CURLMPOPT_TIMERFUNCTION */
-static void
-multi_timer_cb( CURLM *multi UNUSED, long timeout_ms, void * vweb )
-{
- tr_web * web = vweb;
- struct timeval tv = tr_timevalMsec( timeout_ms );
- evtimer_add( &web->timer, &tv );
+ dbgmsg( "deleting web timer" );
+ assert( !web->running );
+ evtimer_del( &web->timer );
+ curl_multi_cleanup( web->cm );
+ tr_free( web );
}
-#else
-
static void
pulse( int socket UNUSED, short action UNUSED, void * vweb )
{
tr_web * web = vweb;
- struct timeval tv = tr_timevalMsec( PULSE_MSEC );
+ assert( web->running );
pump( web );
evtimer_del( &web->timer );
- evtimer_add( &web->timer, &tv );
-}
-#endif
+ web->running = web->remain > 0;
+
+ if( web->running ) {
+ struct timeval tv = tr_timevalMsec( PULSE_MSEC );
+ evtimer_add( &web->timer, &tv );
+ } else if( web->dying ) {
+ webDestroy( web );
+ } else {
+ dbgmsg( "stopping web timer" );
+ }
+}
tr_web*
tr_webInit( tr_session * session )
{
-#ifndef USE_CURL_MULTI_SOCKET
- struct timeval tv = tr_timevalMsec( PULSE_MSEC );
-#endif
static int curlInited = FALSE;
tr_web * web;
web->cm = curl_multi_init( );
web->session = session;
-#ifdef USE_CURL_MULTI_SOCKET
- evtimer_set( &web->timer, event_timer_cb, web );
- curl_multi_setopt( web->cm, CURLMOPT_SOCKETDATA, web );
- curl_multi_setopt( web->cm, CURLMOPT_SOCKETFUNCTION, multi_sock_cb );
- curl_multi_setopt( web->cm, CURLMOPT_TIMERDATA, web );
- curl_multi_setopt( web->cm, CURLMOPT_TIMERFUNCTION, multi_timer_cb );
-#else
evtimer_set( &web->timer, pulse, web );
- evtimer_add( &web->timer, &tv );
-#endif
#if CURL_CHECK_VERSION(7,16,3)
curl_multi_setopt( web->cm, CURLMOPT_MAXCONNECTS, 10 );
#endif
}
void
-tr_webClose( tr_web ** web )
+tr_webClose( tr_web ** web_in )
{
- dbgmsg( "deleting web->timer" );
- evtimer_del( &(*web)->timer );
- curl_multi_cleanup( (*web)->cm );
- tr_free( *web );
- *web = NULL;
+ tr_web * web = *web_in;
+ *web_in = NULL;
+
+ if( !web->running )
+ webDestroy( web );
+ else
+ web->dying = 1;
}
/***
projects / transmission / commitdiff