From: Kurt Zeilenga Date: Wed, 12 Oct 2005 23:17:28 +0000 (+0000) Subject: Align Operation buffers using LBER_ALIGNED_BUFFER X-Git-Tag: OPENLDAP_REL_ENG_2_2_MP~283 X-Git-Url: https://git.sur5r.net/?a=commitdiff_plain;h=6db7e605ee06b033f260d0dc02dcd4ed189a12a2;p=openldap Align Operation buffers using LBER_ALIGNED_BUFFER --- diff --git a/libraries/libldap_r/tpool.c b/libraries/libldap_r/tpool.c index ba87eaa989..bf8f908e56 100644 --- a/libraries/libldap_r/tpool.c +++ b/libraries/libldap_r/tpool.c @@ -29,6 +29,10 @@ #define LDAP_THREAD_POOL_IMPLEMENTATION #include "ldap_thr_debug.h" /* May rename symbols defined below */ +#ifdef LDAP_DEVEL +#define SLAP_SEM_LOAD_CONTROL +#endif + #ifndef LDAP_THREAD_HAVE_TPOOL typedef enum ldap_int_thread_pool_state_e { @@ -118,6 +122,123 @@ ldap_int_thread_pool_shutdown ( void ) return(0); } +typedef struct ldap_lazy_sem_t { + ldap_pvt_thread_mutex_t ls_mutex; + ldap_pvt_thread_cond_t ls_cond; + int ls_sem_value; + /* + * when more than ls_lazy_count number of resources + * becmoes available, the thread wating for the resources will + * be waken up in order to prevent frequent blocking/waking-up + */ + unsigned int ls_lazy_count; + /* + * only one thread(listener) will wait on this semaphore + * using a flag instead of a list + */ + int ls_wait; +} ldap_lazy_sem_t; + +ldap_lazy_sem_t* thread_pool_sem = NULL; + +int +ldap_lazy_sem_init( unsigned int value, unsigned int lazyness ) { + thread_pool_sem = (ldap_lazy_sem_t*) LDAP_CALLOC(1, + sizeof( ldap_lazy_sem_t )); + + if( thread_pool_sem == NULL ) return -1; + + ldap_pvt_thread_mutex_init( &thread_pool_sem->ls_mutex ); + ldap_pvt_thread_cond_init( &thread_pool_sem->ls_cond ); + thread_pool_sem->ls_sem_value = value; + thread_pool_sem->ls_lazy_count = lazyness; + thread_pool_sem->ls_wait = 0; + + return 0; +} + +/* + * ldap_lazy_sem_wait is used if a caller is blockable(listener). + * Otherwise use ldap_lazy_sem_dec (worker) + */ +int +ldap_lazy_sem_op_submit( ldap_lazy_sem_t* ls ) { + if ( ls == NULL ) return -1; + + /* only worker thread has its thread ctx */ + if ( ldap_pvt_thread_pool_context() ) { + /* worker thread */ + return ldap_lazy_sem_dec( ls ); + } else { + /* listener */ + return ldap_lazy_sem_wait( ls ); + } +} + +/* + * test if given semaphore's count is zero. + * If 0, the caller is blocked + * If not, the count is decremented. + */ +int +ldap_lazy_sem_wait ( ldap_lazy_sem_t* ls ) { + ldap_pvt_thread_mutex_lock( &ls->ls_mutex ); + +lazy_sem_retry: + if ( ls->ls_sem_value <= 0 ) { + /* no more avaliable resources */ + ls->ls_wait = 1; + ldap_pvt_thread_cond_wait( &ls->ls_cond, &ls->ls_mutex ); + goto lazy_sem_retry; + } else { + /* avaliable resources */ + ls->ls_sem_value--; + } + + ldap_pvt_thread_mutex_unlock( &ls->ls_mutex ); + + return 0; +} + +/* + * decrement the count without blocking + * even when the count becomes less than or equal to 0 + */ +int +ldap_lazy_sem_dec ( ldap_lazy_sem_t* ls ) { + ldap_pvt_thread_mutex_lock( &ls->ls_mutex ); + + ls->ls_sem_value--; + + ldap_pvt_thread_mutex_unlock( &ls->ls_mutex ); + + return 0; +} + +/* + * Increment the count by one and test if it is greater or + * equal to lazyness. If it is, wake up a blocked thread. + */ +int +ldap_lazy_sem_post( ldap_lazy_sem_t* ls ) { + + if( ls == NULL ) return (-1); + + ldap_pvt_thread_mutex_lock( &ls->ls_mutex ); + + ls->ls_sem_value++; + if ( ls->ls_wait ) { + if ( ls->ls_sem_value >= ls->ls_lazy_count ) { + ls->ls_wait = 0; + ldap_pvt_thread_cond_signal( &ls->ls_cond ); + } + } + + ldap_pvt_thread_mutex_unlock( &ls->ls_mutex ); + + return 0; +} + int ldap_pvt_thread_pool_init ( ldap_pvt_thread_pool_t *tpool, @@ -257,6 +378,10 @@ ldap_pvt_thread_pool_submit ( } ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex); +#ifdef SLAP_SEM_LOAD_CONTROL + ldap_lazy_sem_op_submit( thread_pool_sem ); +#endif + if (need_thread) { int rc; @@ -408,6 +533,11 @@ ldap_pvt_thread_pool_destroy ( ldap_pvt_thread_pool_t *tpool, int run_pending ) ldap_pvt_thread_cond_destroy(&pool->ltp_cond); ldap_pvt_thread_mutex_destroy(&pool->ltp_mutex); LDAP_FREE(pool); +#ifdef SLAP_SEM_LOAD_CONTROL + if ( thread_pool_sem ) { + LDAP_FREE( thread_pool_sem ); + } +#endif return(0); } @@ -484,6 +614,9 @@ ldap_int_thread_pool_wrapper ( ctx->ltc_start_routine(ltc_key, ctx->ltc_arg); +#ifdef SLAP_SEM_LOAD_CONTROL + ldap_lazy_sem_post( thread_pool_sem ); +#endif ldap_pvt_thread_mutex_lock(&pool->ltp_mutex); LDAP_SLIST_REMOVE(&pool->ltp_active_list, ctx, ldap_int_thread_ctx_s, ltc_next.al); diff --git a/servers/slapd/back-bdb/config.c b/servers/slapd/back-bdb/config.c index 5c5ff8428e..fa2bf98de5 100644 --- a/servers/slapd/back-bdb/config.c +++ b/servers/slapd/back-bdb/config.c @@ -172,8 +172,8 @@ bdb_online_index( void *ctx, void *arg ) struct bdb_info *bdb = be->be_private; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; - Operation *op = (Operation *)opbuf; + OperationBuffer opbuf; + Operation *op = (Operation *) &opbuf; DBC *curs; DBT key, data; diff --git a/servers/slapd/back-monitor/init.c b/servers/slapd/back-monitor/init.c index cccdec1531..50cb2b44c1 100644 --- a/servers/slapd/back-monitor/init.c +++ b/servers/slapd/back-monitor/init.c @@ -642,7 +642,7 @@ monitor_filter2ndn( struct berval *ndn ) { Connection conn = { 0 }; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; SlapReply rs = { 0 }; slap_callback cb = { NULL, monitor_filter2ndn_cb, NULL, NULL }; @@ -654,7 +654,7 @@ monitor_filter2ndn( return -1; } - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, &conn ); op->o_tag = LDAP_REQ_SEARCH; diff --git a/servers/slapd/back-sql/init.c b/servers/slapd/back-sql/init.c index 1c9913febd..40a4478c1f 100644 --- a/servers/slapd/back-sql/init.c +++ b/servers/slapd/back-sql/init.c @@ -224,8 +224,8 @@ backsql_db_open( SQLHDBC dbh = SQL_NULL_HDBC; struct berbuf bb = BB_NULL; - char opbuf[ OPERATION_BUFFER_SIZE ]; - Operation* op = (Operation *)opbuf; + OperationBuffer opbuf; + Operation* op = (Operation *) &opbuf; Debug( LDAP_DEBUG_TRACE, "==>backsql_db_open(): " "testing RDBMS connection\n", 0, 0, 0 ); @@ -581,8 +581,8 @@ backsql_db_close( int backsql_connection_destroy( Backend *bd, Connection *c ) { - char opbuf[ OPERATION_BUFFER_SIZE ]; - Operation* op = (Operation *)opbuf; + OperationBuffer opbuf; + Operation* op = (Operation *) &opbuf; op->o_hdr = (Opheader *)&op[ 1 ]; op->o_connid = c->c_connid; diff --git a/servers/slapd/bconfig.c b/servers/slapd/bconfig.c index d7d4daa773..2b827f30cc 100644 --- a/servers/slapd/bconfig.c +++ b/servers/slapd/bconfig.c @@ -2623,7 +2623,7 @@ config_setup_ldif( BackendDB *be, const char *dir, int readit ) { setup_cookie sc; slap_callback cb = { NULL, config_ldif_resp, NULL, NULL }; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; SlapReply rs = {REP_RESULT}; Filter filter = { LDAP_FILTER_PRESENT }; @@ -2675,7 +2675,7 @@ config_setup_ldif( BackendDB *be, const char *dir, int readit ) { if ( readit ) { void *thrctx = ldap_pvt_thread_pool_context(); - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, thrctx ); filter.f_desc = slap_schema.si_ad_objectClass; @@ -4063,7 +4063,7 @@ config_back_db_open( BackendDB *be ) BackendInfo *bi; ConfigArgs c; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; slap_callback cb = { NULL, slap_null_cb, NULL, NULL }; SlapReply rs = {REP_RESULT}; @@ -4075,7 +4075,7 @@ config_back_db_open( BackendDB *be ) if ( cfb->cb_use_ldif ) { thrctx = ldap_pvt_thread_pool_context(); - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, thrctx ); op->o_dn = be->be_rootdn; diff --git a/servers/slapd/connection.c b/servers/slapd/connection.c index 06918c2a3d..d4f77c908e 100644 --- a/servers/slapd/connection.c +++ b/servers/slapd/connection.c @@ -43,9 +43,26 @@ #include "slapi/slapi.h" #endif + +#ifdef SLAP_MULTI_CONN_ARRAY +/* for Multiple Connection Arrary (MCA) Support */ +static ldap_pvt_thread_mutex_t* connections_mutex; +static Connection **connections = NULL; + +/* set to the number of processors */ +#define NUM_CONNECTION_ARRAY 2 + +/* partition the array in a modulo manner */ +#define MCA_conn_array_id( fd ) ((int)fd%NUM_CONNECTION_ARRAY) +#define MCA_conn_array_element_id( fd ) ((int)fd/NUM_CONNECTION_ARRAY) + +#define MCA_GET_CONNECTION(fd) &(connections[MCA_conn_array_id(fd)])[MCA_conn_array_element_id( fd )] +#define MCA_GET_CONN_MUTEX(fd) &connections_mutex[MCA_conn_array_id(fd)] +#else /* protected by connections_mutex */ static ldap_pvt_thread_mutex_t connections_mutex; static Connection *connections = NULL; +#endif static ldap_pvt_thread_mutex_t conn_nextid_mutex; static unsigned long conn_nextid = 0; @@ -82,10 +99,17 @@ connection_state2str( int state ) static Connection* connection_get( ber_socket_t s ); +#ifdef SLAP_LIGHTWEIGHT_LISTENER +static int connection_input( Connection *c, Operation** op ); +#else static int connection_input( Connection *c ); +#endif static void connection_close( Connection *c ); static int connection_op_activate( Operation *op ); +#ifdef SLAP_LIGHTWEIGHT_LISTENER +static void connection_op_queue( Operation *op ); +#endif static int connection_resched( Connection *conn ); static void connection_abandon( Connection *conn ); static void connection_destroy( Connection *c ); @@ -96,6 +120,65 @@ static ldap_pvt_thread_start_t connection_operation; * Initialize connection management infrastructure. */ int connections_init(void) +#ifdef SLAP_MULTI_CONN_ARRAY +{ + int i, j; + Connection* conn; + + assert( connections == NULL ); + + if( connections != NULL) { + Debug( LDAP_DEBUG_ANY, "connections_init: already initialized.\n", + 0, 0, 0 ); + return -1; + } + + connections_mutex = (ldap_pvt_thread_mutex_t*) ch_calloc( NUM_CONNECTION_ARRAY, sizeof(ldap_pvt_thread_mutex_t) ); + if( connections_mutex == NULL ) { + Debug( LDAP_DEBUG_ANY, + "connections_init: allocation of connection mutex[%d] failed\n", i, 0, 0 ); + return -1; + } + + connections = (Connection**) ch_calloc( NUM_CONNECTION_ARRAY, sizeof(Connection*)); + if( connections == NULL ) { + Debug( LDAP_DEBUG_ANY, + "connections_init: allocation of connection[%d] failed\n", 0, 0, 0 ); + return -1; + } + + for ( i = 0; i < NUM_CONNECTION_ARRAY; i++ ) { + ldap_pvt_thread_mutex_init( connections_mutex+i ); + connections[i] = (Connection*) ch_calloc( (dtblsize/NUM_CONNECTION_ARRAY), sizeof(Connection) ); + if( connections[i] == NULL ) { + Debug( LDAP_DEBUG_ANY, + "connections_init: allocation (%d*%ld) of connection array[%d] failed\n", + dtblsize, (long) sizeof(Connection), i ); + return -1; + } + } + + /* should check return of every call */ + ldap_pvt_thread_mutex_init( &conn_nextid_mutex ); + + assert( connections[0]->c_struct_state == SLAP_C_UNINITIALIZED ); + assert( connections[NUM_CONNECTION_ARRAY-1]->c_struct_state == SLAP_C_UNINITIALIZED ); + + for ( i = 0; i < NUM_CONNECTION_ARRAY; i++ ) { + conn = connections[i]; + for ( j = 0; j < (dtblsize/NUM_CONNECTION_ARRAY); j++ ) { + conn[j].c_conn_idx = j; + } + } + + /* + * per entry initialization of the Connection array initialization + * will be done by connection_init() + */ + + return 0; +} +#else { int i; @@ -132,11 +215,55 @@ int connections_init(void) return 0; } +#endif /* * Destroy connection management infrastructure. */ + int connections_destroy(void) +#ifdef SLAP_MULTI_CONN_ARRAY +{ + int i; + ber_socket_t j; + + if( connections == NULL) { + Debug( LDAP_DEBUG_ANY, "connections_destroy: nothing to destroy.\n", + 0, 0, 0 ); + return -1; + } + + for ( i = 0; i < NUM_CONNECTION_ARRAY; i++ ) { + Connection* conn = connections[i]; + for ( j = 0; j < (dtblsize/NUM_CONNECTION_ARRAY); j++ ) { + if( conn[j].c_struct_state != SLAP_C_UNINITIALIZED ) { + ber_sockbuf_free( conn[j].c_sb ); + ldap_pvt_thread_mutex_destroy( &conn[j].c_mutex ); + ldap_pvt_thread_mutex_destroy( &conn[j].c_write_mutex ); + ldap_pvt_thread_cond_destroy( &conn[j].c_write_cv ); +#ifdef LDAP_SLAPI + /* FIX ME!! */ + if ( slapi_plugins_used ) { + slapi_int_free_object_extensions( SLAPI_X_EXT_CONNECTION, + &connections[i] ); + } +#endif + } + } + } + + for ( i = 0; i < NUM_CONNECTION_ARRAY; i++ ) { + free( connections[i] ); + connections[i] = NULL; + ldap_pvt_thread_mutex_destroy( &connections_mutex[i] ); + } + + ldap_pvt_thread_mutex_destroy( &conn_nextid_mutex ); + + return 0; + +} +#else { ber_socket_t i; @@ -156,7 +283,8 @@ int connections_destroy(void) ldap_pvt_thread_cond_destroy( &connections[i].c_write_cv ); #ifdef LDAP_SLAPI if ( slapi_plugins_used ) { - slapi_int_free_object_extensions( SLAPI_X_EXT_CONNECTION, &connections[i] ); + slapi_int_free_object_extensions( SLAPI_X_EXT_CONNECTION, + &connections[i] ); } #endif } @@ -169,11 +297,45 @@ int connections_destroy(void) ldap_pvt_thread_mutex_destroy( &conn_nextid_mutex ); return 0; } +#endif /* * shutdown all connections */ int connections_shutdown(void) +#ifdef SLAP_MULTI_CONN_ARRAY +{ + int i; + ber_socket_t j; + + for ( i = 0; i < NUM_CONNECTION_ARRAY; i++ ) { + Connection* conn = connections[i]; + ldap_pvt_thread_mutex_lock( &connections_mutex[i] ); + for ( j = 0; j < (dtblsize/NUM_CONNECTION_ARRAY); j++ ) { + if( conn[j].c_struct_state != SLAP_C_USED ) { + continue; + } + /* give persistent clients a chance to cleanup */ + if( conn[j].c_conn_state == SLAP_C_CLIENT ) { + ldap_pvt_thread_pool_submit( &connection_pool, + conn[j].c_clientfunc, conn[j].c_clientarg ); + continue; + } + + ldap_pvt_thread_mutex_lock( &conn[j].c_mutex ); + /* connections_mutex and c_mutex are locked */ + connection_closing( &conn[j], "connection shutdown" ); + connection_close( &conn[j] ); + ldap_pvt_thread_mutex_unlock( &conn[j].c_mutex ); + } + + ldap_pvt_thread_mutex_unlock( &connections_mutex[i] ); + } + + return 0; + +} +#else { ber_socket_t i; @@ -203,6 +365,7 @@ int connections_shutdown(void) return 0; } +#endif /* * Timeout idle connections. @@ -251,7 +414,11 @@ static Connection* connection_get( ber_socket_t s ) } #ifndef HAVE_WINSOCK +#ifdef SLAP_MULTI_CONN_ARRAY + c = MCA_GET_CONNECTION(s); +#else c = &connections[s]; +#endif assert( c->c_struct_state != SLAP_C_UNINITIALIZED ); @@ -367,10 +534,18 @@ long connection_init( assert( s < dtblsize ); #endif +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_lock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_lock( &connections_mutex ); +#endif #ifndef HAVE_WINSOCK +#ifdef SLAP_MULTI_CONN_ARRAY + c = MCA_GET_CONNECTION(s); +#else c = &connections[s]; +#endif #else { @@ -497,7 +672,11 @@ long connection_init( c->c_close_reason = "?"; /* should never be needed */ ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_SET_FD, &s ); ldap_pvt_thread_mutex_unlock( &c->c_mutex ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return 0; } @@ -585,7 +764,11 @@ long connection_init( slap_sasl_external( c, ssf, authid ); ldap_pvt_thread_mutex_unlock( &c->c_mutex ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif backend_connection_init(c); @@ -611,10 +794,12 @@ void connection2anonymous( Connection *c ) ch_free(c->c_dn.bv_val); } BER_BVZERO( &c->c_dn ); + if ( !BER_BVISNULL( &c->c_ndn ) ) { ch_free(c->c_ndn.bv_val); } BER_BVZERO( &c->c_ndn ); + if ( !BER_BVISNULL( &c->c_sasl_authz_dn ) ) { ber_memfree_x( c->c_sasl_authz_dn.bv_val, NULL ); } @@ -817,21 +1002,34 @@ unsigned long connections_nextid(void) unsigned long id; assert( connections != NULL ); - ldap_pvt_thread_mutex_lock( &connections_mutex ); + ldap_pvt_thread_mutex_lock( &conn_nextid_mutex ); id = conn_nextid; - ldap_pvt_thread_mutex_unlock( &connections_mutex ); + ldap_pvt_thread_mutex_unlock( &conn_nextid_mutex ); return id; } Connection* connection_first( ber_socket_t *index ) { +#ifdef SLAP_MULTI_CONN_ARRAY + int conn_array_id; +#endif + assert( connections != NULL ); assert( index != NULL ); +#ifdef SLAP_MULTI_CONN_ARRAY + for ( conn_array_id = 0; + conn_array_id < NUM_CONNECTION_ARRAY; + conn_array_id++ ) + { + ldap_pvt_thread_mutex_lock( &connections_mutex[ conn_array_id ] ); + } +#else ldap_pvt_thread_mutex_lock( &connections_mutex ); +#endif *index = 0; @@ -839,6 +1037,50 @@ Connection* connection_first( ber_socket_t *index ) } Connection* connection_next( Connection *c, ber_socket_t *index ) +#ifdef SLAP_MULTI_CONN_ARRAY +{ + Connection* conn; + + assert( connections != NULL ); + assert( index != NULL ); + assert( *index <= (dtblsize/NUM_CONNECTION_ARRAY) ); + + if( c != NULL ) { + ldap_pvt_thread_mutex_unlock( &c->c_mutex ); + } + + c = NULL; + + for(; *index < dtblsize; (*index)++) { + conn = MCA_GET_CONNECTION(*index); + if( conn->c_struct_state == SLAP_C_UNINITIALIZED ) { + assert( conn->c_conn_state == SLAP_C_INVALID ); +#ifndef HAVE_WINSOCK + continue; +#else + break; +#endif + } + + if( conn->c_struct_state == SLAP_C_USED ) { + assert( conn->c_conn_state != SLAP_C_INVALID ); + c = conn; + (*index)++; + break; + } + + assert( conn->c_struct_state == SLAP_C_UNUSED ); + assert( conn->c_conn_state == SLAP_C_INVALID ); + } + + if( c != NULL ) { + ldap_pvt_thread_mutex_lock( &c->c_mutex ); + } + + return c; + +} +#else { assert( connections != NULL ); assert( index != NULL ); @@ -876,16 +1118,30 @@ Connection* connection_next( Connection *c, ber_socket_t *index ) return c; } +#endif void connection_done( Connection *c ) { +#ifdef SLAP_MULTI_CONN_ARRAY + int conn_array_id; +#endif + assert( connections != NULL ); if( c != NULL ) { ldap_pvt_thread_mutex_unlock( &c->c_mutex ); } +#ifdef SLAP_MULTI_CONN_ARRAY + for ( conn_array_id = 0; + conn_array_id < NUM_CONNECTION_ARRAY; + conn_array_id++ ) + { + ldap_pvt_thread_mutex_unlock( &connections_mutex[ conn_array_id ] ); + } +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif } /* @@ -910,7 +1166,7 @@ void connection_done( Connection *c ) ldap_pvt_thread_mutex_unlock( &slap_counters.sc_ops_mutex ); \ } while (0) #else /* !SLAPD_MONITOR */ -#define INCR_OP_INITIATED(index) +#define INCR_OP_INITIATED(index) do { } while (0) #define INCR_OP_COMPLETED(index) \ do { \ ldap_pvt_thread_mutex_lock( &slap_counters.sc_ops_mutex ); \ @@ -1169,14 +1425,76 @@ void connection_client_stop( slapd_remove( s, 0, 1 ); } +#ifdef SLAP_LIGHTWEIGHT_LISTENER +void* connection_processing_thread( void* ctx, void* argv ) +{ + int rc ; + Operation* new_op = NULL; + ber_socket_t s = (ber_socket_t)argv; + + /* + * read incoming LDAP requests. If there is more than one, + * the first one is returned with new_op + */ + if( ( rc = connection_read( s, &new_op ) ) < 0 ) { + Debug( LDAP_DEBUG_TRACE, "connection_read(%d) error\n", s, 0, 0 ); + tcp_close( s ); + return (void*)rc; + } + + /* execute the queued request in the same thread */ + if( new_op ) { + rc = (int)connection_operation( + ldap_pvt_thread_pool_context(), new_op ); + } + + return (void*)rc; +} + +int connection_processing_activate( ber_socket_t s ) +{ + int status; + + /* + * suspend reading on this file descriptor until a connection processing + * thread reads data on it. Otherwise the listener thread will repeatedly + * submit the same event on it to the pool. + */ + if( !slapd_suspend( s ) ) return 0; + + status = ldap_pvt_thread_pool_submit( &connection_pool, + connection_processing_thread, (void *) s ); + + if( status != 0 ) { + Debug( LDAP_DEBUG_ANY, "connection_processing_activiate(%d): " + "ldap_pvt_thread_pool_submit failed\n", + s, 0, 0 ); + return -1; + } + + return 1; +} +#endif + +#ifdef SLAP_LIGHTWEIGHT_LISTENER +int connection_read( ber_socket_t s, Operation** op ) +#else int connection_read(ber_socket_t s) +#endif { int rc = 0; Connection *c; +#ifdef SLAP_LIGHTWEIGHT_LISTENER + int need_resume = 1; +#endif assert( connections != NULL ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_lock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_lock( &connections_mutex ); +#endif /* get (locked) connection */ c = connection_get( s ); @@ -1187,7 +1505,11 @@ int connection_read(ber_socket_t s) (long) s, 0, 0 ); slapd_remove(s, 1, 0); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return -1; } @@ -1198,16 +1520,32 @@ int connection_read(ber_socket_t s) "connection_read(%d): closing, ignoring input for id=%lu\n", s, c->c_connid, 0 ); connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif + +#ifdef SLAP_LIGHTWEIGHT_LISTENER + slapd_resume( s ); +#endif return 0; } if ( c->c_conn_state == SLAP_C_CLIENT ) { +#ifdef SLAP_LIGHTWEIGHT_LISTENER + slapd_resume( s ); +#endif slapd_clr_read( s, 0 ); ldap_pvt_thread_pool_submit( &connection_pool, c->c_clientfunc, c->c_clientarg ); + connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return 0; } @@ -1219,33 +1557,34 @@ int connection_read(ber_socket_t s) if ( c->c_is_tls && c->c_needs_tls_accept ) { rc = ldap_pvt_tls_accept( c->c_sb, slap_tls_ctx ); if ( rc < 0 ) { -#if 0 /* required by next #if 0 */ - struct timeval tv; - fd_set rfd; -#endif - Debug( LDAP_DEBUG_TRACE, "connection_read(%d): TLS accept error " "error=%d id=%lu, closing\n", s, rc, c->c_connid ); + c->c_needs_tls_accept = 0; /* connections_mutex and c_mutex are locked */ connection_closing( c, "TLS negotiation failure" ); #if 0 - /* Drain input before close, to allow SSL error codes - * to propagate to client. */ - FD_ZERO(&rfd); - FD_SET(s, &rfd); - for (rc=1; rc>0;) { - tv.tv_sec = 1; - tv.tv_usec = 0; - rc = select(s+1, &rfd, NULL, NULL, &tv); - if (rc == 1) { - ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_DRAIN, NULL); + { + struct timeval tv; + fd_set rfd; + /* Drain input before close, to allow SSL error codes + * to propagate to client. */ + FD_ZERO(&rfd); + FD_SET(s, &rfd); + for (rc=1; rc>0;) { + tv.tv_sec = 1; + tv.tv_usec = 0; + rc = select(s+1, &rfd, NULL, NULL, &tv); + if (rc == 1) { + ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_DRAIN, NULL); + } } } #endif + connection_close( c ); } else if ( rc == 0 ) { @@ -1280,7 +1619,15 @@ int connection_read(ber_socket_t s) !ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_DATA_READY, NULL ) ) { connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif + +#ifdef SLAP_LIGHTWEIGHT_LISTENER + slapd_resume( s ); +#endif return 0; } } @@ -1291,24 +1638,38 @@ int connection_read(ber_socket_t s) /* If previous layer is not removed yet, give up for now */ if ( !c->c_sasl_sockctx ) { connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif + +#ifdef SLAP_LIGHTWEIGHT_LISTENER + slapd_resume( s ); +#endif return 0; } c->c_sasl_layers = 0; rc = ldap_pvt_sasl_install( c->c_sb, c->c_sasl_sockctx ); - if( rc != LDAP_SUCCESS ) { Debug( LDAP_DEBUG_TRACE, "connection_read(%d): SASL install error " "error=%d id=%lu, closing\n", s, rc, c->c_connid ); /* connections_mutex and c_mutex are locked */ +#ifdef SLAP_LIGHTWEIGHT_LISTENER + slapd_resume( s ); +#endif connection_closing( c, "SASL layer install failure" ); connection_close( c ); connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return 0; } } @@ -1319,7 +1680,21 @@ int connection_read(ber_socket_t s) do { /* How do we do this without getting into a busy loop ? */ +#ifdef SLAP_LIGHTWEIGHT_LISTENER + rc = connection_input( c, op ); +#else rc = connection_input( c ); +#endif +#ifdef SLAP_LIGHTWEIGHT_LISTENER + if( *op && (*op)->o_tag == LDAP_REQ_UNBIND ) { + /* + * After the reception of an unbind request, + * no more incoming requests via the connection + * is expected. Therefore, don't resume connection reading. + */ + need_resume = 0; + } +#endif } #ifdef DATA_READY_LOOP while( !rc && ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_DATA_READY, NULL )); @@ -1333,14 +1708,23 @@ int connection_read(ber_socket_t s) Debug( LDAP_DEBUG_TRACE, "connection_read(%d): input error=%d id=%lu, closing.\n", s, rc, c->c_connid ); + /* connections_mutex and c_mutex are locked */ connection_closing( c, conn_lost_str ); connection_close( c ); connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return 0; } +#ifdef SLAP_LIGHTWEIGHT_LISTENER + if ( need_resume ) slapd_resume( s ); +#endif + if ( ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_NEEDS_READ, NULL ) ) { slapd_set_read( s, 1 ); } @@ -1350,13 +1734,22 @@ int connection_read(ber_socket_t s) } connection_return( c ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif + return 0; } +#ifdef SLAP_LIGHTWEIGHT_LISTENER static int -connection_input( - Connection *conn ) +connection_input( Connection *conn, Operation** c_op ) +#else +static int +connection_input( Connection *conn ) +#endif { Operation *op; ber_tag_t tag; @@ -1382,10 +1775,11 @@ connection_input( #ifdef LDAP_CONNECTIONLESS if ( conn->c_is_udp ) { char peername[sizeof("IP=255.255.255.255:65336")]; + len = ber_int_sb_read(conn->c_sb, &peeraddr, sizeof(struct sockaddr)); - if (len != sizeof(struct sockaddr)) - return 1; + if (len != sizeof(struct sockaddr)) return 1; + sprintf( peername, "IP=%s:%d", inet_ntoa( peeraddr.sa_in_addr.sin_addr ), (unsigned) ntohs( peeraddr.sa_in_addr.sin_port ) ); @@ -1394,6 +1788,7 @@ connection_input( conn->c_connid, peername, conn->c_sock_name.bv_val, 0, 0 ); } #endif + tag = ber_get_next( conn->c_sb, &len, conn->c_currentber ); if ( tag != LDAP_TAG_MESSAGE ) { int err = errno; @@ -1447,6 +1842,7 @@ connection_input( } } #endif + if(tag == LDAP_REQ_BIND) { /* immediately abandon all existing operations upon BIND */ connection_abandon( conn ); @@ -1543,7 +1939,21 @@ connection_input( } } else { conn->c_n_ops_executing++; - connection_op_activate( op ); +#ifdef SLAP_LIGHTWEIGHT_LISTENER + /* + * The first op will be processed in the same thread context, + * Subsequent ops will be submitted to the pool by + * calling connection_op_activate() + */ + if ( *c_op == NULL ) { + /* the first incoming request */ + connection_op_queue( op ); + *c_op = op; + } else +#endif + { + connection_op_activate( op ); + } } #ifdef NO_THREADS @@ -1568,7 +1978,11 @@ connection_resched( Connection *conn ) ber_sockbuf_ctrl( conn->c_sb, LBER_SB_OPT_GET_FD, &sd ); /* use trylock to avoid possible deadlock */ +#ifdef SLAP_MULTI_CONN_ARRAY + rc = ldap_pvt_thread_mutex_trylock( MCA_GET_CONN_MUTEX( sd ) ); +#else rc = ldap_pvt_thread_mutex_trylock( &connections_mutex ); +#endif if( rc ) { Debug( LDAP_DEBUG_TRACE, @@ -1580,7 +1994,11 @@ connection_resched( Connection *conn ) * so recheck state below. */ ldap_pvt_thread_mutex_unlock( &conn->c_mutex ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_lock( MCA_GET_CONN_MUTEX ( sd ) ); +#else ldap_pvt_thread_mutex_lock( &connections_mutex ); +#endif ldap_pvt_thread_mutex_lock( &conn->c_mutex ); } @@ -1595,7 +2013,11 @@ connection_resched( Connection *conn ) connection_close( conn ); } +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX( sd ) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return 0; } @@ -1630,15 +2052,15 @@ connection_init_log_prefix( Operation *op ) { if ( op->o_connid == (unsigned long)(-1) ) { snprintf( op->o_log_prefix, sizeof( op->o_log_prefix ), - "conn=-1 op=%lu", op->o_opid ); + "conn=-1 op=%lu", op->o_opid ); } else { snprintf( op->o_log_prefix, sizeof( op->o_log_prefix ), - "conn=%lu op=%lu", op->o_connid, op->o_opid ); + "conn=%lu op=%lu", op->o_connid, op->o_opid ); } } -static int connection_op_activate( Operation *op ) +static void connection_op_queue( Operation *op ) { int status; ber_tag_t tag = op->o_tag; @@ -1657,6 +2079,7 @@ static int connection_op_activate( Operation *op ) ber_dupbv( &op->o_ndn, &op->o_conn->c_sasl_authz_dn ); } } + op->o_authtype = op->o_conn->c_authtype; ber_dupbv( &op->o_authmech, &op->o_conn->c_authmech ); @@ -1664,6 +2087,7 @@ static int connection_op_activate( Operation *op ) op->o_protocol = op->o_conn->c_protocol ? op->o_conn->c_protocol : LDAP_VERSION3; } + if (op->o_conn->c_conn_state == SLAP_C_INACTIVE && op->o_protocol > LDAP_VERSION2) { @@ -1674,6 +2098,14 @@ static int connection_op_activate( Operation *op ) connection_init_log_prefix( op ); LDAP_STAILQ_INSERT_TAIL( &op->o_conn->c_ops, op, o_next ); +} + +static int connection_op_activate( Operation *op ) +{ + int status; + ber_tag_t tag = op->o_tag; + + connection_op_queue( op ); status = ldap_pvt_thread_pool_submit( &connection_pool, connection_operation, (void *) op ); @@ -1695,21 +2127,27 @@ int connection_write(ber_socket_t s) assert( connections != NULL ); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_lock( MCA_GET_CONN_MUTEX( s ) ); +#else ldap_pvt_thread_mutex_lock( &connections_mutex ); +#endif c = connection_get( s ); - if( c == NULL ) { Debug( LDAP_DEBUG_ANY, "connection_write(%ld): no connection!\n", (long)s, 0, 0 ); slapd_remove(s, 1, 0); +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX( s ) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif return -1; } slapd_clr_write( s, 0); - c->c_n_write++; Debug( LDAP_DEBUG_TRACE, @@ -1723,8 +2161,9 @@ int connection_write(ber_socket_t s) if ( ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_NEEDS_WRITE, NULL ) ) { slapd_set_write( s, 1 ); } - /* If there are ops pending because of a writewaiter, start - * one up. + + /* If there are ops pending because of a writewaiter, + * start one up. */ while ((op = LDAP_STAILQ_FIRST( &c->c_pending_ops )) != NULL) { if ( !c->c_writewaiter ) break; @@ -1744,7 +2183,13 @@ int connection_write(ber_socket_t s) break; } connection_return( c ); + +#ifdef SLAP_MULTI_CONN_ARRAY + ldap_pvt_thread_mutex_unlock( MCA_GET_CONN_MUTEX(s) ); +#else ldap_pvt_thread_mutex_unlock( &connections_mutex ); +#endif + return 0; } @@ -1784,4 +2229,3 @@ connection_assign_nextid( Connection *conn ) conn->c_connid = conn_nextid++; ldap_pvt_thread_mutex_unlock( &conn_nextid_mutex ); } - diff --git a/servers/slapd/daemon.c b/servers/slapd/daemon.c index c50db15639..8224859df3 100644 --- a/servers/slapd/daemon.c +++ b/servers/slapd/daemon.c @@ -46,10 +46,10 @@ #ifdef HAVE_TCPD #include -#define SLAP_STRING_UNKNOWN STRING_UNKNOWN - int allow_severity = LOG_INFO; int deny_severity = LOG_NOTICE; + +#define SLAP_STRING_UNKNOWN STRING_UNKNOWN #else /* ! TCP Wrappers */ #define SLAP_STRING_UNKNOWN "unknown" #endif /* ! TCP Wrappers */ @@ -82,9 +82,12 @@ static ber_socket_t wake_sds[2]; static int emfile; static int waking; -#define WAKE_LISTENER(w) do { \ - if ((w) && waking < 5) { waking++; tcp_write( wake_sds[1], "0", 1 ); } \ - } while(0) +#define WAKE_LISTENER(w) do { \ + if ((w) && waking < 5) { \ + waking++; \ + tcp_write( wake_sds[1], "0", 1 ); \ + } \ +} while(0) volatile sig_atomic_t slapd_shutdown = 0, slapd_gentle_shutdown = 0; volatile sig_atomic_t slapd_abrupt_shutdown = 0; @@ -95,6 +98,14 @@ static struct slap_daemon { ber_socket_t sd_nactives; int sd_nwriters; +#ifdef SLAP_LIGHTWEIGHT_LISTENER + /* + * 0: reading on a FD is suspended + * 1: reading on a FD is allowed + */ + int *sd_suspend; +#endif + #ifdef HAVE_EPOLL struct epoll_event *sd_epolls; int sd_nepolls; @@ -102,13 +113,11 @@ static struct slap_daemon { int sd_epfd; int sd_nfds; #else - #ifndef HAVE_WINSOCK /* In winsock, accept() returns values higher than dtblsize so don't bother with this optimization */ int sd_nfds; #endif - fd_set sd_actives; fd_set sd_readers; fd_set sd_writers; @@ -116,196 +125,257 @@ static struct slap_daemon { } slap_daemon; #ifdef HAVE_EPOLL -#define SLAP_EVENTS_ARE_INDEXED 0 -#define SLAP_SOCK_IX(s) (slap_daemon.sd_index[s]) -#define SLAP_SOCK_EP(s) (slap_daemon.sd_epolls[SLAP_SOCK_IX(s)]) -#define SLAP_SOCK_EV(s) (SLAP_SOCK_EP(s).events) -#define SLAP_SOCK_IS_ACTIVE(s) (SLAP_SOCK_IX(s) != -1) -#define SLAP_SOCK_NOT_ACTIVE(s) (SLAP_SOCK_IX(s) == -1) -#define SLAP_SOCK_IS_SET(s, mode) (SLAP_SOCK_EV(s) & mode) - -#define SLAP_SOCK_IS_READ(s) SLAP_SOCK_IS_SET(s, EPOLLIN) -#define SLAP_SOCK_IS_WRITE(s) SLAP_SOCK_IS_SET(s, EPOLLOUT) - -#define SLAP_SET_SOCK(s, mode) do { \ - if ((SLAP_SOCK_EV(s) & mode) != mode) { \ - SLAP_SOCK_EV(s) |= mode; \ - epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_MOD, s, \ - &SLAP_SOCK_EP(s)); \ - } \ +# define SLAP_EVENTS_ARE_INDEXED 0 +# define SLAP_SOCK_IX(s) (slap_daemon.sd_index[(s)]) +# define SLAP_SOCK_EP(s) (slap_daemon.sd_epolls[SLAP_SOCK_IX(s)]) +# define SLAP_SOCK_EV(s) (SLAP_SOCK_EP(s).events) +# define SLAP_SOCK_IS_ACTIVE(s) (SLAP_SOCK_IX(s) != -1) +# define SLAP_SOCK_NOT_ACTIVE(s) (SLAP_SOCK_IX(s) == -1) +# define SLAP_SOCK_IS_SET(s, mode) (SLAP_SOCK_EV(s) & (mode)) + +# define SLAP_SOCK_IS_READ(s) SLAP_SOCK_IS_SET((s), EPOLLIN) +# define SLAP_SOCK_IS_WRITE(s) SLAP_SOCK_IS_SET((s), EPOLLOUT) + +# define SLAP_SET_SOCK(s, mode) do { \ + if ((SLAP_SOCK_EV(s) & (mode)) != (mode)) { \ + SLAP_SOCK_EV(s) |= (mode); \ + epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_MOD, (s), \ + &SLAP_SOCK_EP(s)); \ + } \ } while(0) -#define SLAP_CLR_SOCK(s, mode) do { \ - if ((SLAP_SOCK_EV(s) & mode)) { \ - SLAP_SOCK_EV(s) &= ~mode; \ - epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_MOD, s, \ - &SLAP_SOCK_EP(s)); \ - } \ +# define SLAP_CLR_SOCK(s, mode) do { \ + if ((SLAP_SOCK_EV(s) & (mode))) { \ + SLAP_SOCK_EV(s) &= ~(mode); \ + epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_MOD, s, \ + &SLAP_SOCK_EP(s)); \ + } \ } while(0) -#define SLAP_SOCK_SET_READ(s) SLAP_SET_SOCK(s, EPOLLIN) -#define SLAP_SOCK_SET_WRITE(s) SLAP_SET_SOCK(s, EPOLLOUT) +# define SLAP_SOCK_SET_READ(s) SLAP_SET_SOCK(s, EPOLLIN) +# define SLAP_SOCK_SET_WRITE(s) SLAP_SET_SOCK(s, EPOLLOUT) -#define SLAP_SOCK_CLR_READ(s) SLAP_CLR_SOCK(s, EPOLLIN) -#define SLAP_SOCK_CLR_WRITE(s) SLAP_CLR_SOCK(s, EPOLLOUT) +# ifdef SLAP_LIGHTWEIGHT_LISTENER +# define SLAP_SOCK_SET_SUSPEND(s) \ + ( slap_daemon.sd_suspend[SLAP_SOCK_IX(s)] = 1 ) +# define SLAP_SOCK_CLR_SUSPEND(s) \ + ( slap_daemon.sd_suspend[SLAP_SOCK_IX(s)] = 0 ) +# define SLAP_SOCK_IS_SUSPEND(s) \ + ( slap_daemon.sd_suspend[SLAP_SOCK_IX(s)] == 1 ) +# endif -#define SLAP_CLR_EVENT(i, mode) (revents[i].events &= ~mode) +# define SLAP_SOCK_CLR_READ(s) SLAP_CLR_SOCK((s), EPOLLIN) +# define SLAP_SOCK_CLR_WRITE(s) SLAP_CLR_SOCK((s), EPOLLOUT) +# define SLAP_CLR_EVENT(i, mode) (revents[(i)].events &= ~(mode)) -#define SLAP_EVENT_MAX slap_daemon.sd_nfds +# define SLAP_EVENT_MAX slap_daemon.sd_nfds /* If a Listener address is provided, store that as the epoll data. * Otherwise, store the address of this socket's slot in the * index array. If we can't do this add, the system is out of * resources and we need to shutdown. */ -#define SLAP_ADD_SOCK(s, l) do { \ - int rc; \ - SLAP_SOCK_IX(s) = slap_daemon.sd_nfds; \ - SLAP_SOCK_EP(s).data.ptr = (l) ? (l) : (void *)(&SLAP_SOCK_IX(s)); \ - SLAP_SOCK_EV(s) = EPOLLIN; \ - rc = epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_ADD, s, \ - &SLAP_SOCK_EP(s)); \ - if ( rc == 0 ) slap_daemon.sd_nfds++; \ - else { \ - Debug( LDAP_DEBUG_ANY, "daemon: epoll_ctl ADD failed, errno %d, shutting down\n", \ - errno, 0, 0 ); \ - slapd_shutdown = 2; \ - } \ -} while(0) - -#define SLAP_EV_LISTENER(ptr) (((int *)(ptr) >= slap_daemon.sd_index && \ - (int *)(ptr) <= (slap_daemon.sd_index+dtblsize)) ? 0 : 1) - -#define SLAP_EV_PTRFD(ptr) (SLAP_EV_LISTENER(ptr) ? \ - ((Listener *)ptr)->sl_sd : (int *)(ptr) - slap_daemon.sd_index) - -#define SLAP_DEL_SOCK(s) do { \ - int fd, rc, index = SLAP_SOCK_IX(s); \ - rc = epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_DEL, s, \ - &SLAP_SOCK_EP(s)); \ - slap_daemon.sd_epolls[index] = slap_daemon.sd_epolls[slap_daemon.sd_nfds-1]; \ - fd = SLAP_EV_PTRFD(slap_daemon.sd_epolls[index].data.ptr); \ - slap_daemon.sd_index[fd] = index; \ - slap_daemon.sd_index[s] = -1; \ - slap_daemon.sd_nfds--; \ -} while(0) - -#define SLAP_EVENT_CLR_READ(i) SLAP_CLR_EVENT(i, EPOLLIN) -#define SLAP_EVENT_CLR_WRITE(i) SLAP_CLR_EVENT(i, EPOLLOUT) - -#define SLAP_CHK_EVENT(i, mode) (revents[i].events & mode) - -#define SLAP_EVENT_IS_READ(i) SLAP_CHK_EVENT(i, EPOLLIN) -#define SLAP_EVENT_IS_WRITE(i) SLAP_CHK_EVENT(i, EPOLLOUT) -#define SLAP_EVENT_IS_LISTENER(i) SLAP_EV_LISTENER(revents[i].data.ptr) -#define SLAP_EVENT_LISTENER(i) (revents[i].data.ptr) - -#define SLAP_EVENT_FD(i) SLAP_EV_PTRFD(revents[i].data.ptr) -#define SLAP_SOCK_SET_MUTE(s) SLAP_SOCK_CLR_READ(s) -#define SLAP_SOCK_CLR_MUTE(s) SLAP_SOCK_SET_READ(s) -#define SLAP_SOCK_IS_MUTE(s) (!SLAP_SOCK_IS_READ(s)) - -#define SLAP_SOCK_SET_INIT \ - slap_daemon.sd_epolls = ch_calloc(1, sizeof(struct epoll_event) * dtblsize * 2); \ - slap_daemon.sd_index = ch_malloc(sizeof(int) * dtblsize); \ - slap_daemon.sd_epfd = epoll_create( dtblsize ); \ - for (i=0; itv_sec * 1000 : -1 ) +# define SLAP_ADD_SOCK(s, l) do { \ + int rc; \ + SLAP_SOCK_IX((s)) = slap_daemon.sd_nfds; \ + SLAP_SOCK_EP((s)).data.ptr = (l) ? (l) : (void *)(&SLAP_SOCK_IX(s)); \ + SLAP_SOCK_EV((s)) = EPOLLIN; \ + rc = epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_ADD, \ + (s), &SLAP_SOCK_EP((s))); \ + if ( rc == 0 ) { \ + slap_daemon.sd_nfds++; \ + } else { \ + Debug( LDAP_DEBUG_ANY, \ + "daemon: epoll_ctl ADD failed, errno %d, shutting down\n", \ + errno, 0, 0 ); \ + slapd_shutdown = 2; \ + } \ +} while (0) + +# define SLAP_EV_LISTENER(ptr) (((int *)(ptr) >= slap_daemon.sd_index && \ + (int *)(ptr) <= (slap_daemon.sd_index+dtblsize)) ? 0 : 1 ) + +# define SLAP_EV_PTRFD(ptr) (SLAP_EV_LISTENER(ptr) ? \ + ((Listener *)ptr)->sl_sd : (int *)(ptr) - slap_daemon.sd_index) + +# ifdef SLAP_LIGHTWEIGHT_LISTENER +# define SLAP_DEL_SOCK(s) do { \ + int fd, rc, suspend, index = SLAP_SOCK_IX((s)); \ + rc = epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_DEL, \ + (s), &SLAP_SOCK_EP((s))); \ + slap_daemon.sd_epolls[index] = \ + slap_daemon.sd_epolls[slap_daemon.sd_nfds-1]; \ + fd = SLAP_EV_PTRFD(slap_daemon.sd_epolls[index].data.ptr); \ + slap_daemon.sd_suspend[index] = \ + slap_daemon.sd_suspend[slap_daemon.sd_nfds-1]; \ + slap_daemon.sd_suspend[slap_daemon.sd_nfds-1] = 0; \ + slap_daemon.sd_index[fd] = index; \ + slap_daemon.sd_index[(s)] = -1; \ + slap_daemon.sd_nfds--; \ +} while (0) +# else +# define SLAP_DEL_SOCK(s) do { \ + int fd, rc, index = SLAP_SOCK_IX((s)); \ + rc = epoll_ctl(slap_daemon.sd_epfd, EPOLL_CTL_DEL, \ + (s), &SLAP_SOCK_EP((s))); \ + slap_daemon.sd_epolls[index] = \ + slap_daemon.sd_epolls[slap_daemon.sd_nfds-1]; \ + fd = SLAP_EV_PTRFD(slap_daemon.sd_epolls[index].data.ptr); \ + slap_daemon.sd_index[fd] = index; \ + slap_daemon.sd_index[(s)] = -1; \ + slap_daemon.sd_nfds--; \ +} while (0) +# endif + +# define SLAP_EVENT_CLR_READ(i) SLAP_CLR_EVENT((i), EPOLLIN) +# define SLAP_EVENT_CLR_WRITE(i) SLAP_CLR_EVENT((i), EPOLLOUT) + +# define SLAP_CHK_EVENT(i, mode) (revents[(i)].events & mode) + +# define SLAP_EVENT_IS_READ(i) SLAP_CHK_EVENT((i), EPOLLIN) +# define SLAP_EVENT_IS_WRITE(i) SLAP_CHK_EVENT((i), EPOLLOUT) +# define SLAP_EVENT_IS_LISTENER(i) SLAP_EV_LISTENER(revents[(i)].data.ptr) +# define SLAP_EVENT_LISTENER(i) (revents[(i)].data.ptr) + +# define SLAP_EVENT_FD(i) SLAP_EV_PTRFD(revents[(i)].data.ptr) +# define SLAP_SOCK_SET_MUTE(s) SLAP_SOCK_CLR_READ((s)) +# define SLAP_SOCK_CLR_MUTE(s) SLAP_SOCK_SET_READ((s)) +# define SLAP_SOCK_IS_MUTE(s) (!SLAP_SOCK_IS_READ((s))) + +# ifdef SLAP_LIGHTWEIGHT_LISTENER +# define SLAP_SOCK_SET_INIT do { \ + slap_daemon.sd_epolls = \ + ch_malloc( sizeof(struct epoll_event) * dtblsize * 2 ); \ + slap_daemon.sd_index = ch_malloc(sizeof(int) * dtblsize); \ + slap_daemon.sd_epfd = epoll_create( dtblsize ); \ + for (i=0; itv_sec * 1000 : -1 ) #else - /* select */ -#define SLAP_EVENTS_ARE_INDEXED 1 -#define SLAP_EVENT_DECL \ + +# define SLAP_EVENTS_ARE_INDEXED 1 +# define SLAP_EVENT_DECL \ fd_set readfds, writefds -#define SLAP_EVENT_INIT \ +# define SLAP_EVENT_INIT do { \ AC_MEMCPY( &readfds, &slap_daemon.sd_readers, sizeof(fd_set) ); \ - if ( nwriters ) \ + if ( nwriters ) { \ AC_MEMCPY( &writefds, &slap_daemon.sd_writers, sizeof(fd_set) ); \ - else \ - FD_ZERO( &writefds ) - -#ifdef FD_SETSIZE -#define CHK_SETSIZE \ - if (dtblsize > FD_SETSIZE) dtblsize = FD_SETSIZE -#else -#define CHK_SETSIZE -#endif - -#define SLAP_SOCK_SET_INIT \ - CHK_SETSIZE; \ - FD_ZERO(&slap_daemon.sd_readers); \ - FD_ZERO(&slap_daemon.sd_writers) - -#define SLAP_SOCK_IS_ACTIVE(fd) FD_ISSET(fd, &slap_daemon.sd_actives) - -#define SLAP_SOCK_IS_READ(fd) FD_ISSET(fd, &slap_daemon.sd_readers) -#define SLAP_SOCK_IS_WRITE(fd) FD_ISSET(fd, &slap_daemon.sd_writers) - -#define SLAP_SOCK_NOT_ACTIVE(fd) (!SLAP_SOCK_IS_ACTIVE(fd) && \ + } else { \ + FD_ZERO( &writefds ); \ + } \ +} while (0) + +# ifdef FD_SETSIZE +# define CHK_SETSIZE do { \ + if (dtblsize > FD_SETSIZE) dtblsize = FD_SETSIZE; \ +} while (0) +# else +# define CHK_SETSIZE do { ; } while (0) +# endif + +# ifdef SLAP_LIGHTWEIGHT_LISTENER +# define SLAP_SOCK_SET_INIT do { \ + CHK_SETSIZE; \ + slap_daemon.sd_suspend = ch_malloc(sizeof(int) * dtblsize); \ + for (i=0; i= slap_daemon.sd_nfds) slap_daemon.sd_nfds = s+1 -#endif - -#define SLAP_SOCK_CLR_READ(fd) FD_CLR(fd, &slap_daemon.sd_readers) -#define SLAP_SOCK_CLR_WRITE(fd) FD_CLR(fd, &slap_daemon.sd_writers) - -#define SLAP_ADD_SOCK(s, l) do { \ - SLAP_ADDTEST(s); \ - FD_SET(s, &slap_daemon.sd_actives); \ - FD_SET(s, &slap_daemon.sd_readers); \ +# define SLAP_ADDTEST(s) do { } while 0 +# define SLAP_EVENT_MAX dtblsize +# else +# define SLAP_SOCK_SET_READ(fd) FD_SET((fd), &slap_daemon.sd_readers) +# define SLAP_SOCK_SET_WRITE(fd) FD_SET((fd), &slap_daemon.sd_writers) + +# define SLAP_EVENT_MAX slap_daemon.sd_nfds +# define SLAP_ADDTEST(s) do { \ + if ((s) >= slap_daemon.sd_nfds) slap_daemon.sd_nfds = (s)+1; \ +} while (0) +# endif + +# define SLAP_SOCK_CLR_READ(fd) FD_CLR((fd), &slap_daemon.sd_readers) +# define SLAP_SOCK_CLR_WRITE(fd) FD_CLR((fd), &slap_daemon.sd_writers) + +# define SLAP_ADD_SOCK(s, l) do { \ + SLAP_ADDTEST((s)); \ + FD_SET((s), &slap_daemon.sd_actives); \ + FD_SET((s), &slap_daemon.sd_readers); \ } while(0) -#define SLAP_DEL_SOCK(s) do { \ - FD_CLR(s, &slap_daemon.sd_actives); \ - FD_CLR(s, &slap_daemon.sd_readers); \ - FD_CLR(s, &slap_daemon.sd_writers); \ +# define SLAP_DEL_SOCK(s) do { \ + FD_CLR((s), &slap_daemon.sd_actives); \ + FD_CLR((s), &slap_daemon.sd_readers); \ + FD_CLR((s), &slap_daemon.sd_writers); \ } while(0) -#define SLAP_EVENT_IS_READ(fd) FD_ISSET(fd, &readfds) -#define SLAP_EVENT_IS_WRITE(fd) FD_ISSET(fd, &writefds) +# define SLAP_EVENT_IS_READ(fd) FD_ISSET((fd), &readfds) +# define SLAP_EVENT_IS_WRITE(fd) FD_ISSET((fd), &writefds) -#define SLAP_EVENT_CLR_READ(fd) FD_CLR(fd, &readfds) -#define SLAP_EVENT_CLR_WRITE(fd) FD_CLR(fd, &writefds) +# define SLAP_EVENT_CLR_READ(fd) FD_CLR((fd), &readfds) +# define SLAP_EVENT_CLR_WRITE(fd) FD_CLR((fd), &writefds) -#define SLAP_EVENT_WAIT(tvp) \ - select( SLAP_EVENT_MAX, &readfds, \ - nwriters > 0 ? &writefds : NULL, NULL, tvp ) - -#define SLAP_SOCK_SET_MUTE(s) FD_CLR(s, &readfds) -#define SLAP_SOCK_CLR_MUTE(s) FD_SET(s, &readfds) -#define SLAP_SOCK_IS_MUTE(s) (!FD_ISSET(s, &readfds)) +# define SLAP_EVENT_WAIT(tvp) \ + select( SLAP_EVENT_MAX, &readfds, \ + nwriters > 0 ? &writefds : NULL, NULL, (tvp) ) +# define SLAP_SOCK_SET_MUTE(s) FD_CLR((s), &readfds) +# define SLAP_SOCK_CLR_MUTE(s) FD_SET((s), &readfds) +# define SLAP_SOCK_IS_MUTE(s) (!FD_ISSET((s), &readfds)) #endif - #ifdef HAVE_SLP /* * SLP related functions @@ -426,27 +496,33 @@ static void slapd_add(ber_socket_t s, int isactive, Listener *sl) { assert( SLAP_SOCK_NOT_ACTIVE(s) ); - if ( isactive ) { - slap_daemon.sd_nactives++; - } + if ( isactive ) slap_daemon.sd_nactives++; SLAP_ADD_SOCK(s, sl); Debug( LDAP_DEBUG_CONNS, "daemon: added %ldr\n", (long) s, 0, 0 ); + ldap_pvt_thread_mutex_unlock( &slap_daemon.sd_mutex ); + +#ifdef SLAP_LIGHTWEIGHT_LISTENER + WAKE_LISTENER(1); +#endif } /* * Remove the descriptor from daemon control */ -void slapd_remove(ber_socket_t s, int wasactive, int wake) { +void slapd_remove( + ber_socket_t s, + int wasactive, + int wake ) +{ int waswriter; + ldap_pvt_thread_mutex_lock( &slap_daemon.sd_mutex ); - if ( wasactive ) { - slap_daemon.sd_nactives--; - } + if ( wasactive ) slap_daemon.sd_nactives--; waswriter = SLAP_SOCK_IS_WRITE(s); Debug( LDAP_DEBUG_CONNS, "daemon: removing %ld%s%s\n", @@ -454,8 +530,11 @@ void slapd_remove(ber_socket_t s, int wasactive, int wake) { waswriter ? "w" : "" ); if ( waswriter ) slap_daemon.sd_nwriters--; - SLAP_DEL_SOCK(s); +#ifdef SLAP_LIGHTWEIGHT_LISTENER + SLAP_SOCK_CLR_SUSPEND(s); +#endif + SLAP_DEL_SOCK(s); /* If we ran out of file descriptors, we dropped a listener from * the select() loop. Now that we're removing a session from our * control, we can try to resume a dropped listener to use. @@ -475,17 +554,56 @@ void slapd_remove(ber_socket_t s, int wasactive, int wake) { /* Walked the entire list without enabling anything; emfile * counter is stale. Reset it. */ - if ( slap_listeners[i] == NULL ) - emfile = 0; + if ( slap_listeners[i] == NULL ) emfile = 0; } ldap_pvt_thread_mutex_unlock( &slap_daemon.sd_mutex ); WAKE_LISTENER(wake || slapd_gentle_shutdown == 2); } +#ifdef SLAP_LIGHTWEIGHT_LISTENER +/* + * Temporarily suspend submitting events on the descriptor to the pool. + * Reading on the descriptor will be resumed by a connection procseeing thread + * when data (LDAP requests) on it are read. + * slapd_suspend() returns 1 when it is suspended otherwise returns 0 + */ +int slapd_suspend(ber_socket_t s) { + int rc = 0; + + ldap_pvt_thread_mutex_lock( &slap_daemon.sd_mutex ); + + if ( !SLAP_SOCK_IS_SUSPEND( s ) && SLAP_SOCK_IS_ACTIVE( s ) && + SLAP_SOCK_IS_READ( s ) ) + { + SLAP_SOCK_SET_SUSPEND( s ); + SLAP_SOCK_CLR_READ( s ); + rc = 1; + } + + ldap_pvt_thread_mutex_unlock( &slap_daemon.sd_mutex ); + return rc; +} + +void slapd_resume ( ber_socket_t s ) { + ldap_pvt_thread_mutex_lock( &slap_daemon.sd_mutex ); + + SLAP_SOCK_SET_READ( s ); + + assert( SLAP_SOCK_IS_SUSPEND( s ) ); + + SLAP_SOCK_CLR_SUSPEND ( s ); + + ldap_pvt_thread_mutex_unlock( &slap_daemon.sd_mutex ); + + WAKE_LISTENER(1); +} +#endif + void slapd_clr_write(ber_socket_t s, int wake) { ldap_pvt_thread_mutex_lock( &slap_daemon.sd_mutex ); assert( SLAP_SOCK_IS_ACTIVE( s )); + if ( SLAP_SOCK_IS_WRITE( s )) { SLAP_SOCK_CLR_WRITE( s ); slap_daemon.sd_nwriters--; @@ -499,6 +617,7 @@ void slapd_set_write(ber_socket_t s, int wake) { ldap_pvt_thread_mutex_lock( &slap_daemon.sd_mutex ); assert( SLAP_SOCK_IS_ACTIVE( s )); + if ( !SLAP_SOCK_IS_WRITE( s )) { SLAP_SOCK_SET_WRITE( s ); slap_daemon.sd_nwriters++; @@ -539,9 +658,7 @@ static void slap_free_listener_addresses(struct sockaddr **sal) { struct sockaddr **sap; - if (sal == NULL) { - return; - } + if (sal == NULL) return; for (sap = sal; *sap != NULL; sap++) { ch_free(*sap); @@ -572,18 +689,17 @@ static int get_url_perms( type++; } - if ( strncasecmp( type, LDAPI_MOD_URLEXT "=", sizeof(LDAPI_MOD_URLEXT "=") - 1 ) == 0 ) { - char *value = type - + ( sizeof(LDAPI_MOD_URLEXT "=") - 1 ); - mode_t p = 0; - int j; + if ( strncasecmp( type, LDAPI_MOD_URLEXT "=", + sizeof(LDAPI_MOD_URLEXT "=") - 1 ) == 0 ) + { + char *value = type + ( sizeof(LDAPI_MOD_URLEXT "=") - 1 ); + mode_t p = 0; + int j; switch (strlen(value)) { case 4: /* skip leading '0' */ - if ( value[ 0 ] != '0' ) { - return LDAP_OTHER; - } + if ( value[ 0 ] != '0' ) return LDAP_OTHER; value++; case 3: @@ -592,9 +708,7 @@ static int get_url_perms( v = value[ j ] - '0'; - if ( v < 0 || v > 7 ) { - return LDAP_OTHER; - } + if ( v < 0 || v > 7 ) return LDAP_OTHER; p |= v << 3*(2-j); } @@ -644,21 +758,19 @@ static int slap_get_listener_addresses( #ifdef LDAP_PF_LOCAL if ( port == 0 ) { *sal = ch_malloc(2 * sizeof(void *)); - if (*sal == NULL) { - return -1; - } + if (*sal == NULL) return -1; sap = *sal; *sap = ch_malloc(sizeof(struct sockaddr_un)); - if (*sap == NULL) - goto errexit; + if (*sap == NULL) goto errexit; sap[1] = NULL; if ( strlen(host) > - (sizeof(((struct sockaddr_un *)*sap)->sun_path) - 1) ) { + (sizeof(((struct sockaddr_un *)*sap)->sun_path) - 1) ) + { Debug( LDAP_DEBUG_ANY, - "daemon: domain socket path (%s) too long in URL", - host, 0, 0); + "daemon: domain socket path (%s) too long in URL", + host, 0, 0); goto errexit; } @@ -690,9 +802,7 @@ static int slap_get_listener_addresses( /* EMPTY */ ; } *sal = ch_calloc(n, sizeof(void *)); - if (*sal == NULL) { - return -1; - } + if (*sal == NULL) return -1; sap = *sal; *sap = NULL; @@ -751,7 +861,7 @@ static int slap_get_listener_addresses( he = gethostbyname( host ); if( he == NULL ) { Debug( LDAP_DEBUG_ANY, - "daemon: invalid host %s", host, 0, 0); + "daemon: invalid host %s", host, 0, 0); return -1; } for (n = 0; he->h_addr_list[n]; n++) ; @@ -765,16 +875,16 @@ static int slap_get_listener_addresses( sap = *sal; for ( i = 0; isa_family = AF_INET; ((struct sockaddr_in *)sap[i])->sin_port = htons(port); if (he) { - AC_MEMCPY( &((struct sockaddr_in *)sap[i])->sin_addr, he->h_addr_list[i], sizeof(struct in_addr) ); + AC_MEMCPY( &((struct sockaddr_in *)sap[i])->sin_addr, + he->h_addr_list[i], sizeof(struct in_addr) ); } else { - AC_MEMCPY( &((struct sockaddr_in *)sap[i])->sin_addr, &in, sizeof(struct in_addr) ); + AC_MEMCPY( &((struct sockaddr_in *)sap[i])->sin_addr, + &in, sizeof(struct in_addr) ); } } sap[i] = NULL; @@ -831,9 +941,7 @@ static int slap_open_listener( return -1; } - if(! lud->lud_port ) { - lud->lud_port = LDAP_PORT; - } + if(! lud->lud_port ) lud->lud_port = LDAP_PORT; #else l.sl_is_tls = ldap_pvt_url_scheme2tls( lud->lud_scheme ); @@ -869,6 +977,7 @@ static int slap_open_listener( err = slap_get_listener_addresses(lud->lud_host, port, &sal); } } + #ifdef LDAP_CONNECTIONLESS l.sl_is_udp = ( tmp == LDAP_PROTO_UDP ); #endif @@ -882,17 +991,16 @@ static int slap_open_listener( #endif /* LDAP_PF_LOCAL || SLAP_X_LISTENER_MOD */ ldap_free_urldesc( lud ); - if ( err ) { - return -1; - } + if ( err ) return -1; /* If we got more than one address returned, we need to make space * for it in the slap_listeners array. */ - for ( num=0; sal[num]; num++ ); + for ( num=0; sal[num]; num++ ) /* empty */; if ( num > 1 ) { *listeners += num-1; - slap_listeners = ch_realloc( slap_listeners, (*listeners + 1) * sizeof(Listener *) ); + slap_listeners = ch_realloc( slap_listeners, + (*listeners + 1) * sizeof(Listener *) ); } psal = sal; @@ -916,9 +1024,11 @@ static int slap_open_listener( sal++; continue; } + #ifdef LDAP_CONNECTIONLESS if( l.sl_is_udp ) socktype = SOCK_DGRAM; #endif + l.sl_sd = socket( (*sal)->sa_family, socktype, 0); if ( l.sl_sd == AC_SOCKET_INVALID ) { int err = sock_errno(); @@ -928,6 +1038,7 @@ static int slap_open_listener( sal++; continue; } + #ifndef HAVE_WINSOCK if ( l.sl_sd >= dtblsize ) { Debug( LDAP_DEBUG_ANY, @@ -938,6 +1049,7 @@ static int slap_open_listener( continue; } #endif + #ifdef LDAP_PF_LOCAL if ( (*sal)->sa_family == AF_LOCAL ) { unlink ( ((struct sockaddr_un *)*sal)->sun_path ); @@ -951,9 +1063,9 @@ static int slap_open_listener( (char *) &tmp, sizeof(tmp) ); if ( rc == AC_SOCKET_ERROR ) { int err = sock_errno(); - Debug( LDAP_DEBUG_ANY, - "slapd(%ld): setsockopt(SO_REUSEADDR) failed errno=%d (%s)\n", - (long) l.sl_sd, err, sock_errstr(err) ); + Debug( LDAP_DEBUG_ANY, "slapd(%ld): " + "setsockopt(SO_REUSEADDR) failed errno=%d (%s)\n", + (long) l.sl_sd, err, sock_errstr(err) ); } #endif } @@ -968,17 +1080,18 @@ static int slap_open_listener( /* Try to use IPv6 sockets for IPv6 only */ tmp = 1; rc = setsockopt( l.sl_sd, IPPROTO_IPV6, IPV6_V6ONLY, - (char *) &tmp, sizeof(tmp) ); + (char *) &tmp, sizeof(tmp) ); if ( rc == AC_SOCKET_ERROR ) { int err = sock_errno(); - Debug( LDAP_DEBUG_ANY, - "slapd(%ld): setsockopt(IPV6_V6ONLY) failed errno=%d (%s)\n", - (long) l.sl_sd, err, sock_errstr(err) ); + Debug( LDAP_DEBUG_ANY, "slapd(%ld): " + "setsockopt(IPV6_V6ONLY) failed errno=%d (%s)\n", + (long) l.sl_sd, err, sock_errstr(err) ); } #endif addrlen = sizeof(struct sockaddr_in6); break; #endif + #ifdef LDAP_PF_LOCAL case AF_LOCAL: #ifdef LOCAL_CREDS @@ -994,76 +1107,76 @@ static int slap_open_listener( if (bind(l.sl_sd, *sal, addrlen)) { err = sock_errno(); - Debug( LDAP_DEBUG_ANY, "daemon: bind(%ld) failed errno=%d (%s)\n", - (long) l.sl_sd, err, sock_errstr(err) ); + Debug( LDAP_DEBUG_ANY, + "daemon: bind(%ld) failed errno=%d (%s)\n", + (long) l.sl_sd, err, sock_errstr(err) ); tcp_close( l.sl_sd ); sal++; continue; } - switch ( (*sal)->sa_family ) { + switch ( (*sal)->sa_family ) { #ifdef LDAP_PF_LOCAL - case AF_LOCAL: { - char *addr = ((struct sockaddr_un *)*sal)->sun_path; - l.sl_name.bv_len = strlen(addr) + sizeof("PATH=") - 1; - l.sl_name.bv_val = ber_memalloc( l.sl_name.bv_len + 1 ); - snprintf( l.sl_name.bv_val, l.sl_name.bv_len + 1, + case AF_LOCAL: { + char *addr = ((struct sockaddr_un *)*sal)->sun_path; + l.sl_name.bv_len = strlen(addr) + sizeof("PATH=") - 1; + l.sl_name.bv_val = ber_memalloc( l.sl_name.bv_len + 1 ); + snprintf( l.sl_name.bv_val, l.sl_name.bv_len + 1, "PATH=%s", addr ); - } break; + } break; #endif /* LDAP_PF_LOCAL */ - case AF_INET: { - char *s; + case AF_INET: { + char *s; #if defined( HAVE_GETADDRINFO ) && defined( HAVE_INET_NTOP ) - char addr[INET_ADDRSTRLEN]; - inet_ntop( AF_INET, &((struct sockaddr_in *)*sal)->sin_addr, - addr, sizeof(addr) ); - s = addr; + char addr[INET_ADDRSTRLEN]; + inet_ntop( AF_INET, &((struct sockaddr_in *)*sal)->sin_addr, + addr, sizeof(addr) ); + s = addr; #else - s = inet_ntoa( ((struct sockaddr_in *) *sal)->sin_addr ); + s = inet_ntoa( ((struct sockaddr_in *) *sal)->sin_addr ); #endif - port = ntohs( ((struct sockaddr_in *)*sal) ->sin_port ); - l.sl_name.bv_val = ber_memalloc( sizeof("IP=255.255.255.255:65535") ); - snprintf( l.sl_name.bv_val, sizeof("IP=255.255.255.255:65535"), - "IP=%s:%d", - s != NULL ? s : SLAP_STRING_UNKNOWN, port ); - l.sl_name.bv_len = strlen( l.sl_name.bv_val ); - } break; + port = ntohs( ((struct sockaddr_in *)*sal) ->sin_port ); + l.sl_name.bv_val = + ber_memalloc( sizeof("IP=255.255.255.255:65535") ); + snprintf( l.sl_name.bv_val, sizeof("IP=255.255.255.255:65535"), + "IP=%s:%d", + s != NULL ? s : SLAP_STRING_UNKNOWN, port ); + l.sl_name.bv_len = strlen( l.sl_name.bv_val ); + } break; #ifdef LDAP_PF_INET6 - case AF_INET6: { - char addr[INET6_ADDRSTRLEN]; - inet_ntop( AF_INET6, &((struct sockaddr_in6 *)*sal)->sin6_addr, - addr, sizeof addr); - port = ntohs( ((struct sockaddr_in6 *)*sal)->sin6_port ); - l.sl_name.bv_len = strlen(addr) + sizeof("IP= 65535"); - l.sl_name.bv_val = ber_memalloc( l.sl_name.bv_len ); - snprintf( l.sl_name.bv_val, l.sl_name.bv_len, "IP=%s %d", + case AF_INET6: { + char addr[INET6_ADDRSTRLEN]; + inet_ntop( AF_INET6, &((struct sockaddr_in6 *)*sal)->sin6_addr, + addr, sizeof addr); + port = ntohs( ((struct sockaddr_in6 *)*sal)->sin6_port ); + l.sl_name.bv_len = strlen(addr) + sizeof("IP= 65535"); + l.sl_name.bv_val = ber_memalloc( l.sl_name.bv_len ); + snprintf( l.sl_name.bv_val, l.sl_name.bv_len, "IP=%s %d", addr, port ); - l.sl_name.bv_len = strlen( l.sl_name.bv_val ); - } break; + l.sl_name.bv_len = strlen( l.sl_name.bv_val ); + } break; #endif /* LDAP_PF_INET6 */ - default: - Debug( LDAP_DEBUG_ANY, "daemon: unsupported address family (%d)\n", - (int) (*sal)->sa_family, 0, 0 ); - break; - } - - AC_MEMCPY(&l.sl_sa, *sal, addrlen); - ber_str2bv( url, 0, 1, &l.sl_url); - li = ch_malloc( sizeof( Listener ) ); - *li = l; - slap_listeners[*cur] = li; - (*cur)++; - sal++; + default: + Debug( LDAP_DEBUG_ANY, "daemon: unsupported address family (%d)\n", + (int) (*sal)->sa_family, 0, 0 ); + break; + } - } /* while ( *sal != NULL ) */ + AC_MEMCPY(&l.sl_sa, *sal, addrlen); + ber_str2bv( url, 0, 1, &l.sl_url); + li = ch_malloc( sizeof( Listener ) ); + *li = l; + slap_listeners[*cur] = li; + (*cur)++; + sal++; + } slap_free_listener_addresses(psal); - if ( l.sl_url.bv_val == NULL ) - { + if ( l.sl_url.bv_val == NULL ) { Debug( LDAP_DEBUG_TRACE, "slap_open_listener: failed on %s\n", url, 0, 0 ); return -1; @@ -1157,6 +1270,7 @@ int slapd_daemon_init( const char *urls ) ldap_charray_free( u ); ldap_pvt_thread_mutex_init( &slap_daemon.sd_mutex ); + return !i; } @@ -1182,26 +1296,31 @@ slapd_daemon_destroy(void) static void close_listeners( - int remove -) + int remove ) { int l; for ( l = 0; slap_listeners[l] != NULL; l++ ) { if ( slap_listeners[l]->sl_sd != AC_SOCKET_INVALID ) { - if ( remove ) - slapd_remove( slap_listeners[l]->sl_sd, 0, 0 ); + if ( remove ) slapd_remove( slap_listeners[l]->sl_sd, 0, 0 ); + #ifdef LDAP_PF_LOCAL if ( slap_listeners[l]->sl_sa.sa_addr.sa_family == AF_LOCAL ) { unlink( slap_listeners[l]->sl_sa.sa_un_addr.sun_path ); } #endif /* LDAP_PF_LOCAL */ + slapd_close( slap_listeners[l]->sl_sd ); } - if ( slap_listeners[l]->sl_url.bv_val ) + + if ( slap_listeners[l]->sl_url.bv_val ) { ber_memfree( slap_listeners[l]->sl_url.bv_val ); - if ( slap_listeners[l]->sl_name.bv_val ) + } + + if ( slap_listeners[l]->sl_name.bv_val ) { ber_memfree( slap_listeners[l]->sl_name.bv_val ); + } + free ( slap_listeners[l] ); slap_listeners[l] = NULL; } @@ -1209,8 +1328,7 @@ close_listeners( static int slapd_handle_listener( - Listener *sl -) + Listener *sl ) { Sockaddr from; @@ -1257,6 +1375,13 @@ slapd_handle_listener( # endif /* LDAP_PF_LOCAL */ s = accept( sl->sl_sd, (struct sockaddr *) &from, &len ); +#ifdef SLAP_LIGHTWEIGHT_LISTENER + /* + * As soon as a TCP connection is accepted, the listener FD is resumed + * for concurrent-processing of incoming TCP connections. + */ + slapd_resume( sl->sl_sd ); +#endif if ( s == AC_SOCKET_INVALID ) { int err = sock_errno(); @@ -1414,7 +1539,9 @@ slapd_handle_listener( #ifdef LDAP_PF_INET6 || ( from.sa_addr.sa_family == AF_INET6 ) #endif - ) { + ) + { + dnsname = NULL; #ifdef SLAPD_RLOOKUPS if ( use_reverse_lookup ) { char *herr; @@ -1424,15 +1551,13 @@ slapd_handle_listener( dnsname = hbuf; } } -#else - dnsname = NULL; #endif /* SLAPD_RLOOKUPS */ #ifdef HAVE_TCPD if ( !hosts_ctl("slapd", - dnsname != NULL ? dnsname : SLAP_STRING_UNKNOWN, - peeraddr != NULL ? peeraddr : SLAP_STRING_UNKNOWN, - SLAP_STRING_UNKNOWN )) + dnsname != NULL ? dnsname : SLAP_STRING_UNKNOWN, + peeraddr != NULL ? peeraddr : SLAP_STRING_UNKNOWN, + SLAP_STRING_UNKNOWN )) { /* DENY ACCESS */ Statslog( LDAP_DEBUG_STATS, @@ -1477,13 +1602,52 @@ slapd_handle_listener( return 0; } +#ifdef SLAP_LIGHTWEIGHT_LISTENER +void* +slapd_handle_listener_thread( + void* ctx, + void* ptr ) +{ + int rc; + + rc = slapd_handle_listener( (Listener*)ptr ); + + if( rc != LDAP_SUCCESS ) { + Debug( LDAP_DEBUG_ANY, + "slapd_handle_listener_thread: failed", 0, 0, 0 ); + } + + return (void*)NULL; +} + +static int +new_connection_activate( + Listener* sl ) +{ + int status; + + if( !slapd_suspend( sl->sl_sd ) ) return (0); + + status = ldap_pvt_thread_pool_submit( &connection_pool, + slapd_handle_listener_thread, (void *) sl ); + + if( status != 0 ) { + Debug( LDAP_DEBUG_ANY, + "new_connection_activate: ldap_pvt_thread_pool_submit failed\n", + 0, 0, 0 ); + return (-1); + } + + return (0); +} +#endif + static void * slapd_daemon_task( - void *ptr -) + void *ptr ) { int l; - time_t last_idle_check = 0; + time_t last_idle_check = 0; struct timeval idle; int ebadf = 0; @@ -1495,8 +1659,9 @@ slapd_daemon_task( * Don't just truncate, preserve the fractions of * seconds to prevent sleeping for zero time. */ - idle.tv_sec = global_idletimeout/SLAPD_IDLE_CHECK_LIMIT; - idle.tv_usec = global_idletimeout - idle.tv_sec * SLAPD_IDLE_CHECK_LIMIT; + idle.tv_sec = global_idletimeout / SLAPD_IDLE_CHECK_LIMIT; + idle.tv_usec = global_idletimeout - \ + ( idle.tv_sec * SLAPD_IDLE_CHECK_LIMIT ); idle.tv_usec *= 1000000 / SLAPD_IDLE_CHECK_LIMIT; } else { idle.tv_sec = 0; @@ -1506,8 +1671,8 @@ slapd_daemon_task( slapd_add( wake_sds[0], 0, NULL ); for ( l = 0; slap_listeners[l] != NULL; l++ ) { - if ( slap_listeners[l]->sl_sd == AC_SOCKET_INVALID ) - continue; + if ( slap_listeners[l]->sl_sd == AC_SOCKET_INVALID ) continue; + #ifdef LDAP_CONNECTIONLESS /* Since this is connectionless, the data port is the * listening port. The listen() and accept() calls @@ -1537,15 +1702,19 @@ slapd_daemon_task( for ( i = 0 ; i < l; i++ ) { sa6 = slap_listeners[i]->sl_sa.sa_in6_addr; if ( sa6.sin6_family == AF_INET6 && - !memcmp( &sa6.sin6_addr, &in6addr_any, sizeof(struct in6_addr) ) ) + !memcmp( &sa6.sin6_addr, &in6addr_any, + sizeof(struct in6_addr) ) ) + { break; + } } if ( i < l ) { /* We are already listening to in6addr_any */ Debug( LDAP_DEBUG_CONNS, - "daemon: Attempt to listen to 0.0.0.0 failed, already listening on ::, assuming IPv4 included\n", - 0, 0, 0 ); + "daemon: Attempt to listen to 0.0.0.0 failed, " + "already listening on ::, assuming IPv4 included\n", + 0, 0, 0 ); slapd_close( slap_listeners[l]->sl_sd ); slap_listeners[l]->sl_sd = AC_SOCKET_INVALID; continue; @@ -1557,17 +1726,38 @@ slapd_daemon_task( "daemon: listen(%s, 5) failed errno=%d (%s)\n", slap_listeners[l]->sl_url.bv_val, err, sock_errstr(err) ); - return( (void*)-1 ); + return (void*)-1; } +#ifdef SLAP_LIGHTWEIGHT_LISTENER + /* make the listening socket non-blocking */ + if ( ber_pvt_socket_set_nonblock( slap_listeners[l]->sl_sd, 1 ) < 0 ) { + Debug( LDAP_DEBUG_ANY, "slapd_daemon_task: " + "set nonblocking on a listening socket failed\n", + 0, 0, 0 ); + slapd_shutdown = 2; + return (void*)-1; + } +#endif + slapd_add( slap_listeners[l]->sl_sd, 0, slap_listeners[l] ); } #ifdef HAVE_NT_SERVICE_MANAGER - if ( started_event != NULL ) { + if ( started_event != NULL ) } ldap_pvt_thread_cond_signal( &started_event ); } #endif + +#ifdef SLAP_SEM_LOAD_CONTROL + /* + * initialize count and lazyness of a semaphore + */ + (void) ldap_lazy_sem_init( + SLAP_MAX_WORKER_THREADS + 4 /* max workers + margin */, + 4 /* lazyness */ ); +#endif + /* initialization complete. Here comes the loop. */ while ( !slapd_shutdown ) { @@ -1640,8 +1830,7 @@ slapd_daemon_task( nfds = SLAP_EVENT_MAX; - if ( global_idletimeout && slap_daemon.sd_nactives ) - at = 1; + if ( global_idletimeout && slap_daemon.sd_nactives ) at = 1; ldap_pvt_thread_mutex_unlock( &slap_daemon.sd_mutex ); @@ -1650,9 +1839,11 @@ slapd_daemon_task( && ( tv.tv_sec || tv.tv_usec ) #endif ) + { tvp = &tv; - else + } else { tvp = NULL; + } ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex ); rtask = ldap_pvt_runqueue_next_sched( &slapd_rq, &cat ); @@ -1673,8 +1864,7 @@ slapd_daemon_task( if ( cat && cat->tv_sec ) { time_t diff = difftime( cat->tv_sec, now ); - if ( diff == 0 ) - diff = tdelta; + if ( diff == 0 ) diff = tdelta; if ( tvp == NULL || diff < tv.tv_sec ) { tv.tv_sec = diff; tv.tv_usec = 0; @@ -1685,16 +1875,17 @@ slapd_daemon_task( for ( l = 0; slap_listeners[l] != NULL; l++ ) { if ( slap_listeners[l]->sl_sd == AC_SOCKET_INVALID || slap_listeners[l]->sl_is_mute ) + { continue; + } Debug( LDAP_DEBUG_CONNS, "daemon: select: listen=%d active_threads=%d tvp=%s\n", - slap_listeners[l]->sl_sd, at, - tvp == NULL ? "NULL" : "zero" ); + slap_listeners[l]->sl_sd, at, + tvp == NULL ? "NULL" : "zero" ); } - switch(ns = SLAP_EVENT_WAIT(tvp)) - { + switch(ns = SLAP_EVENT_WAIT(tvp)) { case -1: { /* failure - try again */ int err = sock_errno(); @@ -1761,7 +1952,12 @@ slapd_daemon_task( if ( !SLAP_EVENT_IS_READ( slap_listeners[l]->sl_sd )) continue; + +#ifdef SLAP_LIGHTWEIGHT_LISTENER + rc = new_connection_activate(slap_listeners[l]); +#else rc = slapd_handle_listener(slap_listeners[l]); +#endif #ifdef LDAP_CONNECTIONLESS /* This is a UDP session, let the data loop process it */ @@ -1879,9 +2075,13 @@ slapd_daemon_task( * active. */ +#ifdef SLAP_LIGHTWEIGHT_LISTENER + connection_processing_activate( rd ); +#else if ( connection_read( rd ) < 0 ) { slapd_close( rd ); } +#endif } #else /* !SLAP_EVENTS_ARE_INDEXED */ /* FIXME */ @@ -1932,11 +2132,16 @@ slapd_daemon_task( } } #endif + for (i=0; iarg; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; - Operation *op = (Operation *)opbuf; + OperationBuffer opbuf; + Operation *op = (Operation *) &opbuf; SlapReply rs = {REP_RESULT}; slap_callback cb = { NULL, log_old_lookup, NULL, NULL }; Filter f; diff --git a/servers/slapd/overlays/pcache.c b/servers/slapd/overlays/pcache.c index 73ae35a272..3b594f96f4 100644 --- a/servers/slapd/overlays/pcache.c +++ b/servers/slapd/overlays/pcache.c @@ -1462,7 +1462,7 @@ consistency_check( cache_manager *cm = on->on_bi.bi_private; query_manager *qm = cm->qm; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; SlapReply rs = {REP_RESULT}; @@ -1470,7 +1470,7 @@ consistency_check( int i, return_val, pause = 1; QueryTemplate* templ; - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, ctx ); op->o_bd = &cm->db; diff --git a/servers/slapd/overlays/syncprov.c b/servers/slapd/overlays/syncprov.c index 5c0b7cf4ef..4070d5dd09 100644 --- a/servers/slapd/overlays/syncprov.c +++ b/servers/slapd/overlays/syncprov.c @@ -861,11 +861,11 @@ syncprov_qtask( void *ctx, void *arg ) struct re_s *rtask = arg; syncops *so = rtask->arg; slap_overinst *on = so->s_op->o_private; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; BackendDB be; - op = (Operation *)opbuf; + op = (Operation *) &opbuf; *op = *so->s_op; op->o_hdr = (Opheader *)(op+1); op->o_controls = (void **)(op->o_hdr+1); @@ -2217,9 +2217,9 @@ syncprov_db_open( syncprov_info_t *si = (syncprov_info_t *)on->on_bi.bi_private; Connection conn; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; char ctxcsnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE]; - Operation *op = (Operation *)opbuf; + Operation *op = (Operation *) &opbuf; Entry *e; Attribute *a; int rc; @@ -2307,8 +2307,8 @@ syncprov_db_close( } if ( si->si_numops ) { Connection conn; - char opbuf[OPERATION_BUFFER_SIZE]; - Operation *op = (Operation *)opbuf; + OperationBuffer opbuf; + Operation *op = (Operation *) &opbuf; SlapReply rs = {REP_RESULT}; void *thrctx; diff --git a/servers/slapd/proto-slap.h b/servers/slapd/proto-slap.h index e110c3865f..995bc05359 100644 --- a/servers/slapd/proto-slap.h +++ b/servers/slapd/proto-slap.h @@ -646,7 +646,11 @@ LDAP_SLAPD_F (const char *) connection_state2str LDAP_P(( int state )) LDAP_GCCATTR((const)); LDAP_SLAPD_F (int) connection_write LDAP_P((ber_socket_t s)); +#ifdef SLAP_LIGHTWEIGHT_LISTENER +LDAP_SLAPD_F (int) connection_read LDAP_P((ber_socket_t, Operation **)); +#else LDAP_SLAPD_F (int) connection_read LDAP_P((ber_socket_t s)); +#endif LDAP_SLAPD_F (unsigned long) connections_nextid(void); diff --git a/servers/slapd/slap.h b/servers/slapd/slap.h index 0a628f79be..a0c66a555c 100644 --- a/servers/slapd/slap.h +++ b/servers/slapd/slap.h @@ -60,6 +60,10 @@ LDAP_BEGIN_DECL #ifdef LDAP_DEVEL +#define SLAP_LIGHTWEIGHT_LISTENER /* experimental slapd architecture */ +#define SLAP_SEM_LOAD_CONTROL /* must also be defined in libldap_r/tpool.c */ +#define SLAP_MULTI_CONN_ARRAY + #define SLAP_ACL_HONOR_DISCLOSE /* partially implemented */ #define SLAP_ACL_HONOR_MANAGE /* not yet implemented */ #define SLAP_DYNACL @@ -2555,7 +2559,11 @@ typedef struct slap_op { LDAP_STAILQ_ENTRY(slap_op) o_next; /* next operation in list */ } Operation; -#define OPERATION_BUFFER_SIZE (sizeof(Operation)+sizeof(Opheader)+SLAP_MAX_CIDS*sizeof(void *)) +#define OPERATION_BUFFER_SIZE ( sizeof(Operation) + sizeof(Opheader) + \ + SLAP_MAX_CIDS*sizeof(void *) ) + +typedef LBER_ALIGNED_BUFFER(operation_buffer_u,OPERATION_BUFFER_SIZE) + OperationBuffer; #define send_ldap_error( op, rs, err, text ) do { \ (rs)->sr_err = err; (rs)->sr_text = text; \ diff --git a/servers/slapd/slapacl.c b/servers/slapd/slapacl.c index 13dfccc882..a30d033891 100644 --- a/servers/slapd/slapacl.c +++ b/servers/slapd/slapacl.c @@ -65,7 +65,7 @@ slapacl( int argc, char **argv ) const char *progname = "slapacl"; Connection conn = { 0 }; Listener listener; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op = NULL; Entry e = { 0 }, *ep = &e; char *attr = NULL; @@ -94,7 +94,7 @@ slapacl( int argc, char **argv ) argv = &argv[ optind ]; argc -= optind; - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, &conn ); conn.c_listener = &listener; diff --git a/servers/slapd/slapadd.c b/servers/slapd/slapadd.c index b36208fdbe..2b4cd4f934 100644 --- a/servers/slapd/slapadd.c +++ b/servers/slapd/slapadd.c @@ -56,8 +56,8 @@ slapadd( int argc, char **argv ) Attribute *attr; Entry *ctxcsn_e; ID ctxcsn_id, id; + OperationBuffer opbuf; Operation *op; - char opbuf[OPERATION_BUFFER_SIZE]; int match; int ret; @@ -69,8 +69,8 @@ slapadd( int argc, char **argv ) slap_tool_init( progname, SLAPADD, argc, argv ); - memset( opbuf, 0, sizeof(opbuf) ); - op = (Operation *)opbuf; + memset( &opbuf, 0, sizeof(opbuf) ); + op = (Operation *) &opbuf; if( !be->be_entry_open || !be->be_entry_close || diff --git a/servers/slapd/slapauth.c b/servers/slapd/slapauth.c index 6e61ff163e..d0c041873b 100644 --- a/servers/slapd/slapauth.c +++ b/servers/slapd/slapauth.c @@ -80,7 +80,7 @@ slapauth( int argc, char **argv ) int rc = EXIT_SUCCESS; const char *progname = "slapauth"; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; slap_tool_init( progname, SLAPAUTH, argc, argv ); @@ -88,7 +88,7 @@ slapauth( int argc, char **argv ) argv = &argv[ optind ]; argc -= optind; - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, &conn ); conn.c_sasl_bind_mech = mech; diff --git a/servers/slapd/syncrepl.c b/servers/slapd/syncrepl.c index 59ba0302aa..deeacc5f49 100644 --- a/servers/slapd/syncrepl.c +++ b/servers/slapd/syncrepl.c @@ -902,7 +902,7 @@ do_syncrepl( struct re_s* rtask = arg; syncinfo_t *si = ( syncinfo_t * ) rtask->arg; Connection conn = {0}; - char opbuf[OPERATION_BUFFER_SIZE]; + OperationBuffer opbuf; Operation *op; int rc = LDAP_SUCCESS; int first = 0; @@ -938,7 +938,7 @@ do_syncrepl( return NULL; } - op = (Operation *)opbuf; + op = (Operation *) &opbuf; connection_fake_init( &conn, op, ctx ); /* use global malloc for now */