mirror
/
transmission
mirror of https://github.com/transmission/transmission
1
0
Fork 0
Code Issues Releases Wiki Activity

#915: Does too many wake-ups when idle

This commit is contained in:
Charles Kerr 2008-05-06 15:52:57 +00:00
parent 18871ac609
commit 6ba02efb58
4 changed files with 146 additions and 273 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
libtransmission/port-forwarding.c
View File

@ -188,7 +188,7 @@ tr_sharedInit( tr_handle * h, int isEnabled, int publicPort )
s->bindSocket = -1;
s->natpmp = tr_natpmpInit();
s->upnp = tr_upnpInit();
s->pulseTimer = tr_timerNew( h, sharedPulse, s, 500 );
s->pulseTimer = tr_timerNew( h, sharedPulse, s, 1000 );
s->isEnabled = isEnabled ? 1 : 0;
s->upnpStatus = TR_NAT_TRAVERSAL_UNMAPPED;
s->natpmpStatus = TR_NAT_TRAVERSAL_UNMAPPED;

240
libtransmission/trevent.c
View File

@ -11,58 +11,43 @@
*/
#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
@ -75,73 +60,64 @@ 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( ;; )
/* 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 );
switch( ch )
{
struct tr_event_command * cmd;
doDie = eh->die && !eh->timerCount;
if( doDie )
break;
/* get the next command */
tr_lockLock( eh->lock );
cmd = tr_list_pop_front( &eh->commands );
tr_lockUnlock( eh->lock );
if( cmd == NULL )
break;
/* process the command */
switch( cmd->mode )
case 'r': /* run in libevent thread */
{
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!" );
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;
}
/* cleanup */
tr_free( cmd );
}
if( !doDie )
evtimer_add( &eh->pulse, &eh->pulseInterval );
else {
assert( eh->timerCount == 0 );
event_del( &eh->pulse );
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;
}
case '\0': /* eof */
{
dbgmsg( "pipe eof reached... removing event listener" );
event_del( &eh->pipeEvent );
break;
}
default:
{
assert( 0 && "unhandled command type!" );
break;
}
}
}
@ -166,18 +142,18 @@ libeventThreadFunc( void * veh )
#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" );
}
@ -189,34 +165,23 @@ tr_eventInit( tr_handle * handle )
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 )
{
@ -227,37 +192,20 @@ 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
@ -271,14 +219,11 @@ tr_timerFree( tr_timer ** ptimer )
*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 );
}
}
@ -295,14 +240,20 @@ tr_timerNew( struct tr_handle * handle,
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;
@ -314,12 +265,21 @@ tr_runInEventThread( struct tr_handle * handle,
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 );
}
}

12
libtransmission/trevent.h
View File

@ -22,18 +22,6 @@ void tr_eventInit( struct tr_handle * tr_handle );
void tr_eventClose( struct tr_handle * tr_handle );
/**
**/
struct event;
enum evhttp_cmd_type;
struct evhttp_request;
struct evhttp_connection;
struct bufferevent;
/**
***
**/
typedef struct tr_timer tr_timer;

165
libtransmission/web.c
View File

@ -10,6 +10,7 @@
* $Id$
*/
#include <assert.h>
#include <stdlib.h> /* bsearch */
#include <event.h>
@ -26,19 +27,17 @@
(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;
};
@ -99,11 +98,7 @@ pump( tr_web * web )
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 );
@ -119,6 +114,18 @@ writeFunc( void * ptr, size_t size, size_t nmemb, void * task )
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 )
{
@ -129,6 +136,8 @@ 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 );
@ -146,8 +155,6 @@ addTask( void * vtask )
curl_easy_setopt( ch, CURLOPT_IPRESOLVE, CURL_IPRESOLVE_V4 );
curl_easy_setopt( ch, CURLOPT_ENCODING, "" );
curl_multi_add_handle( web->cm, ch );
pump( web );
}
}
@ -174,116 +181,41 @@ tr_webRun( tr_session * session,
}
}
#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 )
webDestroy( tr_web * web )
{
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 );
dbgmsg( "deleting web timer" );
assert( !web->running );
evtimer_del( &web->timer );
curl_multi_cleanup( web->cm );
tr_free( web );
}
/* 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 )
{
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 );
}
#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;
@ -300,16 +232,7 @@ tr_webInit( tr_session * session )
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
@ -319,13 +242,15 @@ tr_webInit( tr_session * session )
}
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;
}
/***