Commit afcb343c authored by Quanah Gibson-Mount's avatar Quanah Gibson-Mount
Browse files

Merge remote-tracking branch 'origin/master' into OPENLDAP_REL_ENG_2_5

parents e0ab0c38 092d0a89
Pipeline #721 passed with stage
in 28 minutes and 4 seconds
......@@ -282,6 +282,10 @@ LDAP_F( int )
ldap_pvt_thread_pool_pausecheck LDAP_P((
ldap_pvt_thread_pool_t *pool ));
LDAP_F( int )
ldap_pvt_thread_pool_pausecheck_native LDAP_P((
ldap_pvt_thread_pool_t *pool ));
LDAP_F( int )
ldap_pvt_thread_pool_pause LDAP_P((
ldap_pvt_thread_pool_t *pool ));
......
......@@ -1239,6 +1239,32 @@ ldap_pvt_thread_pool_pausecheck( ldap_pvt_thread_pool_t *tpool )
return handle_pause(tpool, PAUSE_ARG(CHECK_PAUSE));
}
/*
* Wait for a pause, from a non-pooled thread.
*/
int
ldap_pvt_thread_pool_pausecheck_native( ldap_pvt_thread_pool_t *tpool )
{
struct ldap_int_thread_pool_s *pool;
if (tpool == NULL)
return(-1);
pool = *tpool;
if (pool == NULL)
return(0);
if (!pool->ltp_pause)
return(0);
ldap_pvt_thread_mutex_lock(&pool->ltp_mutex);
while (pool->ltp_pause)
ldap_pvt_thread_cond_wait(&pool->ltp_cond, &pool->ltp_mutex);
ldap_pvt_thread_mutex_unlock(&pool->ltp_mutex);
return 1;
}
/*
* Pause the pool. The calling task must be active, not idle.
* Return when all other tasks are paused or idle.
......
......@@ -1958,7 +1958,7 @@ asyncmeta_back_cf_gen( ConfigArgs *c )
return 1;
}
c->ca_private = mt;
c->cleanup = asyncmeta_cf_cleanup;
config_push_cleanup( c, asyncmeta_cf_cleanup );
} break;
case LDAP_BACK_CFG_SUBTREE_EX:
case LDAP_BACK_CFG_SUBTREE_IN:
......
......@@ -445,7 +445,7 @@ mdb_cf_gen( ConfigArgs *c )
mdb->mi_flags |= MDB_RE_OPEN;
ch_free( mdb->mi_dbenv_home );
mdb->mi_dbenv_home = NULL;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
ldap_pvt_thread_pool_purgekey( mdb->mi_dbenv );
break;
case MDB_DBNOSYNC:
......@@ -462,7 +462,7 @@ mdb_cf_gen( ConfigArgs *c )
rc = mdb_env_set_flags( mdb->mi_dbenv, mdb_envflags[i].mask, 0 );
if ( rc ) {
mdb->mi_flags |= MDB_RE_OPEN;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
rc = 0;
}
mdb->mi_dbenv_flags ^= mdb_envflags[i].mask;
......@@ -474,7 +474,7 @@ mdb_cf_gen( ConfigArgs *c )
rc = mdb_env_set_flags( mdb->mi_dbenv, mdb_envflags[i].mask, 0 );
if ( rc ) {
mdb->mi_flags |= MDB_RE_OPEN;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
rc = 0;
}
mdb->mi_dbenv_flags ^= mdb_envflags[i].mask;
......@@ -498,7 +498,7 @@ mdb_cf_gen( ConfigArgs *c )
}
mdb->mi_defaultmask = 0;
mdb->mi_flags |= MDB_DEL_INDEX;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
} else {
struct berval bv, def = BER_BVC("default");
......@@ -535,7 +535,7 @@ mdb_cf_gen( ConfigArgs *c )
ai->ai_indexmask |= MDB_INDEX_DELETING;
mdb->mi_flags |= MDB_DEL_INDEX;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
}
bv.bv_val[ bv.bv_len ] = sep;
......@@ -735,7 +735,7 @@ mdb_cf_gen( ConfigArgs *c )
rc = 0;
if ( rc ) {
mdb->mi_flags |= MDB_RE_OPEN;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
rc = 0;
}
mdb->mi_dbenv_flags |= mdb_envflags[j].mask;
......@@ -757,7 +757,7 @@ mdb_cf_gen( ConfigArgs *c )
if( rc != LDAP_SUCCESS ) return 1;
if ( mdb->mi_flags & MDB_IS_OPEN ) {
mdb->mi_flags |= MDB_OPEN_INDEX;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
if ( !mdb->mi_index_task ) {
/* Start the task as soon as we finish here. Set a long
* interval (10 hours) so that it only gets scheduled once.
......@@ -791,7 +791,7 @@ mdb_cf_gen( ConfigArgs *c )
mdb->mi_readers = c->value_int;
if ( mdb->mi_flags & MDB_IS_OPEN ) {
mdb->mi_flags |= MDB_RE_OPEN;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
}
break;
......@@ -799,7 +799,7 @@ mdb_cf_gen( ConfigArgs *c )
mdb->mi_mapsize = c->value_ulong;
if ( mdb->mi_flags & MDB_IS_OPEN ) {
mdb->mi_flags |= MDB_RE_OPEN;
c->cleanup = mdb_cf_cleanup;
config_push_cleanup( c, mdb_cf_cleanup );
}
break;
......
......@@ -642,9 +642,30 @@ mdb_idl_delete_keys(
}
if ( lo2 >= hi2 ) {
/* The range has collapsed... */
rc = mdb_cursor_del( cursor, MDB_NODUPDATA );
/* delete the range marker */
rc = mdb_cursor_del( cursor, 0 );
if ( rc != 0 ) {
err = "c_del dup";
err = "c_del dup1";
goto fail;
}
/* skip past deleted marker */
rc = mdb_cursor_get( cursor, &key, &data, MDB_NEXT_DUP );
if ( rc != 0 ) {
err = "c_get dup1";
goto fail;
}
/* delete the requested id */
if ( id == hi ) {
/* skip lo */
rc = mdb_cursor_get( cursor, &key, &data, MDB_NEXT_DUP );
if ( rc != 0 ) {
err = "c_get dup2";
goto fail;
}
}
rc = mdb_cursor_del( cursor, 0 );
if ( rc != 0 ) {
err = "c_del dup2";
goto fail;
}
} else {
......
......@@ -2126,7 +2126,7 @@ meta_back_cf_gen( ConfigArgs *c )
return 1;
}
c->ca_private = mt;
c->cleanup = meta_cf_cleanup;
config_push_cleanup( c, meta_cf_cleanup );
} break;
case LDAP_BACK_CFG_SUBTREE_EX:
case LDAP_BACK_CFG_SUBTREE_IN:
......
......@@ -113,7 +113,7 @@ wt_cf_gen( ConfigArgs *c )
wi->wi_flags |= WT_OPEN_INDEX;
if ( wi->wi_flags & WT_IS_OPEN ) {
c->cleanup = wt_cf_cleanup;
config_push_cleanup( c, wt_cf_cleanup );
if ( !wi->wi_index_task ) {
/* Start the task as soon as we finish here. Set a long
......
......@@ -1043,6 +1043,14 @@ typedef struct ADlist {
static ADlist *sortVals;
static int new_daemon_threads;
static int
config_resize_lthreads(ConfigArgs *c)
{
return slapd_daemon_resize( new_daemon_threads );
}
static int
config_generic(ConfigArgs *c) {
int i;
......@@ -1806,7 +1814,7 @@ config_generic(ConfigArgs *c) {
case CFG_THREADQS:
if ( c->value_int < 1 ) {
snprintf( c->cr_msg, sizeof( c->cr_msg ),
"threadqueuess=%d smaller than minimum value 1",
"threadqueues=%d smaller than minimum value 1",
c->value_int );
Debug(LDAP_DEBUG_ANY, "%s: %s.\n",
c->log, c->cr_msg );
......@@ -1824,6 +1832,14 @@ config_generic(ConfigArgs *c) {
break;
case CFG_LTHREADS:
if ( c->value_uint < 1 ) {
snprintf( c->cr_msg, sizeof( c->cr_msg ),
"listenerthreads=%u smaller than minimum value 1",
c->value_uint );
Debug(LDAP_DEBUG_ANY, "%s: %s.\n",
c->log, c->cr_msg );
return 1;
}
{ int mask = 0;
/* use a power of two */
while (c->value_uint > 1) {
......@@ -1831,8 +1847,8 @@ config_generic(ConfigArgs *c) {
mask <<= 1;
mask |= 1;
}
slapd_daemon_mask = mask;
slapd_daemon_threads = mask+1;
new_daemon_threads = mask+1;
config_push_cleanup( c, config_resize_lthreads );
}
break;
......@@ -4195,11 +4211,11 @@ config_tls_option(ConfigArgs *c) {
if (c->op == SLAP_CONFIG_EMIT) {
return ldap_pvt_tls_get_option( ld, flag, berval ? (void *)&c->value_bv : (void *)&c->value_string );
} else if ( c->op == LDAP_MOD_DELETE ) {
c->cleanup = config_tls_cleanup;
config_push_cleanup( c, config_tls_cleanup );
return ldap_pvt_tls_set_option( ld, flag, NULL );
}
if ( !berval ) ch_free(c->value_string);
c->cleanup = config_tls_cleanup;
config_push_cleanup( c, config_tls_cleanup );
rc = ldap_pvt_tls_set_option(ld, flag, berval ? (void *)&c->value_bv : (void *)c->argv[1]);
if ( berval ) ch_free(c->value_bv.bv_val);
return rc;
......@@ -4223,11 +4239,11 @@ config_tls_config(ConfigArgs *c) {
return slap_tls_get_config( slap_tls_ld, flag, &c->value_string );
} else if ( c->op == LDAP_MOD_DELETE ) {
int i = 0;
c->cleanup = config_tls_cleanup;
config_push_cleanup( c, config_tls_cleanup );
return ldap_pvt_tls_set_option( slap_tls_ld, flag, &i );
}
ch_free( c->value_string );
c->cleanup = config_tls_cleanup;
config_push_cleanup( c, config_tls_cleanup );
if ( isdigit( (unsigned char)c->argv[1][0] ) && c->type != CFG_TLS_PROTOCOL_MIN ) {
if ( lutil_atoi( &i, c->argv[1] ) != 0 ) {
Debug(LDAP_DEBUG_ANY, "%s: "
......@@ -5613,8 +5629,8 @@ ok:
rc = ca->bi->bi_db_open( ca->be, &ca->reply );
ca->be->bd_info = bi_orig;
}
} else if ( ca->cleanup ) {
rc = ca->cleanup( ca );
} else if ( ca->num_cleanups ) {
rc = config_run_cleanup( ca );
}
if ( rc ) {
if (ca->cr_msg[0] == '\0')
......@@ -5684,8 +5700,8 @@ done:
overlay_destroy_one( ca->be, (slap_overinst *)ca->bi );
} else if ( coptr->co_type == Cft_Schema ) {
schema_destroy_one( ca, colst, nocs, last );
} else if ( ca->cleanup ) {
ca->cleanup( ca );
} else if ( ca->num_cleanups ) {
config_run_cleanup( ca );
}
}
done_noop:
......@@ -6224,8 +6240,8 @@ out:
ca->reply = msg;
}
if ( ca->cleanup ) {
i = ca->cleanup( ca );
if ( ca->num_cleanups ) {
i = config_run_cleanup( ca );
if (rc == LDAP_SUCCESS)
rc = i;
}
......
......@@ -563,6 +563,35 @@ config_get_vals(ConfigTable *cf, ConfigArgs *c)
return rc;
}
int
config_push_cleanup(ConfigArgs *ca, ConfigDriver *cleanup)
{
int i;
/* silently ignore redundant push */
for (i=0; i < ca->num_cleanups; i++) {
if ( ca->cleanups[i] == cleanup )
return 0;
}
if (ca->num_cleanups >= SLAP_CONFIG_CLEANUP_MAX)
return -1;
ca->cleanups[ca->num_cleanups++] = cleanup;
return 0;
}
int
config_run_cleanup(ConfigArgs *ca)
{
int i, rc = 0;
for (i=0; i < ca->num_cleanups; i++) {
rc = ca->cleanups[i](ca);
if (rc)
break;
}
return rc;
}
int
init_config_attrs(ConfigTable *ct) {
int i, code;
......
......@@ -168,8 +168,12 @@ typedef struct config_args_s {
BackendInfo *bi;
Entry *ca_entry; /* entry being modified */
void *ca_private; /* anything */
ConfigDriver *cleanup;
#ifndef SLAP_CONFIG_CLEANUP_MAX
#define SLAP_CONFIG_CLEANUP_MAX 16
#endif
ConfigDriver *cleanups[SLAP_CONFIG_CLEANUP_MAX];
ConfigType table; /* which config table did we come from */
int num_cleanups;
} ConfigArgs;
/* If lineno is zero, we have an actual LDAP Add request from a client.
......@@ -195,6 +199,9 @@ int config_del_vals(ConfigTable *cf, ConfigArgs *c);
int config_get_vals(ConfigTable *ct, ConfigArgs *c);
int config_add_vals(ConfigTable *ct, ConfigArgs *c);
int config_push_cleanup(ConfigArgs *c, ConfigDriver *cleanup);
int config_run_cleanup(ConfigArgs *c);
void init_config_argv( ConfigArgs *c );
int init_config_attrs(ConfigTable *ct);
int init_config_ocs( ConfigOCs *ocs );
......
......@@ -706,6 +706,17 @@ connection_destroy( Connection *c )
}
}
int connection_is_active( ber_socket_t s )
{
Connection *c;
assert( s < dtblsize );
c = &connections[s];
return c->c_conn_state == SLAP_C_CLOSING ||
c->c_conn_state == SLAP_C_BINDING ||
c->c_conn_state == SLAP_C_ACTIVE ;
}
int connection_valid( Connection *c )
{
/* c_mutex must be locked by caller */
......
......@@ -81,9 +81,6 @@ ber_socket_t dtblsize;
slap_ssf_t local_ssf = LDAP_PVT_SASL_LOCAL_SSF;
struct runqueue_s slapd_rq;
#ifndef SLAPD_MAX_DAEMON_THREADS
#define SLAPD_MAX_DAEMON_THREADS 16
#endif
int slapd_daemon_threads = 1;
int slapd_daemon_mask;
......@@ -94,7 +91,6 @@ int slapd_tcp_wmem;
Listener **slap_listeners = NULL;
static volatile sig_atomic_t listening = 1; /* 0 when slap_listeners closed */
static ldap_pvt_thread_t *listener_tid;
#ifndef SLAPD_LISTEN_BACKLOG
#define SLAPD_LISTEN_BACKLOG 2048
......@@ -102,7 +98,9 @@ static ldap_pvt_thread_t *listener_tid;
#define DAEMON_ID(fd) (fd & slapd_daemon_mask)
static ber_socket_t wake_sds[SLAPD_MAX_DAEMON_THREADS][2];
typedef ber_socket_t sdpair[2];
static sdpair *wake_sds;
static ldap_pvt_thread_mutex_t emfile_mutex;
static int emfile;
......@@ -136,6 +134,7 @@ typedef struct slap_daemon_st {
ber_socket_t sd_nactives;
int sd_nwriters;
int sd_nfds;
ldap_pvt_thread_t sd_tid;
#if defined(HAVE_KQUEUE)
uint8_t* sd_fdmodes; /* indexed by fd */
......@@ -173,7 +172,7 @@ typedef struct slap_daemon_st {
#endif /* ! kqueue && ! epoll && ! /dev/poll */
} slap_daemon_st;
static slap_daemon_st slap_daemon[SLAPD_MAX_DAEMON_THREADS];
static slap_daemon_st *slap_daemon;
/*
* NOTE: naming convention for macros:
......@@ -1881,11 +1880,13 @@ slapd_daemon_init( const char *urls )
Debug( LDAP_DEBUG_ARGS, "daemon_init: %s\n",
urls ? urls : "<null>" );
for ( i=0; i<SLAPD_MAX_DAEMON_THREADS; i++ ) {
wake_sds = ch_malloc( slapd_daemon_threads * sizeof( sdpair ));
for ( i=0; i<slapd_daemon_threads; i++ ) {
wake_sds[i][0] = AC_SOCKET_INVALID;
wake_sds[i][1] = AC_SOCKET_INVALID;
}
slap_daemon = ch_calloc( slapd_daemon_threads, sizeof( slap_daemon_st ));
ldap_pvt_thread_mutex_init( &slap_daemon[0].sd_mutex );
#ifdef HAVE_TCPD
ldap_pvt_thread_mutex_init( &sd_tcpd_mutex );
......@@ -1972,6 +1973,61 @@ slapd_daemon_init( const char *urls )
return !i;
}
/* transfer control of active sockets from old to new listener threads */
static void
slapd_socket_realloc( int newnum )
{
int i, j, oldid, newid;
int newmask = newnum - 1;
Listener *sl;
int num_listeners;
for ( i=0; slap_listeners[i] != NULL; i++ ) ;
num_listeners = i;
for ( i=0; i<dtblsize; i++ ) {
int skip = 0;
/* don't bother with wake_sds, they're assigned independent of mask */
for (j=0; j<slapd_daemon_threads; j++) {
if ( i == wake_sds[j][0] || i == wake_sds[j][1] ) {
skip = 1;
break;
}
}
if ( skip ) continue;
oldid = DAEMON_ID(i);
newid = i & newmask;
if ( oldid == newid ) continue;
if ( !SLAP_SOCK_IS_ACTIVE( oldid, i )) continue;
sl = NULL;
if ( num_listeners ) {
for ( j=0; slap_listeners[j] != NULL; j++ ) {
if ( slap_listeners[j]->sl_sd == i ) {
sl = slap_listeners[j];
num_listeners--;
break;
}
}
}
SLAP_SOCK_ADD( newid, i, sl );
if ( SLAP_SOCK_IS_READ( oldid, i )) {
SLAP_SOCK_SET_READ( newid, i );
}
if ( SLAP_SOCK_IS_WRITE( oldid, i )) {
SLAP_SOCK_SET_WRITE( newid, i );
slap_daemon[oldid].sd_nwriters--;
slap_daemon[newid].sd_nwriters++;
}
if ( connection_is_active( i )) {
slap_daemon[oldid].sd_nactives--;
slap_daemon[newid].sd_nactives++;
}
SLAP_SOCK_DEL( oldid, i );
}
}
int
slapd_daemon_destroy( void )
......@@ -2409,7 +2465,8 @@ slapd_daemon_task(
int l;
time_t last_idle_check = 0;
int ebadf = 0;
int tid = (ldap_pvt_thread_t *) ptr - listener_tid;
int tid = (slap_daemon_st *) ptr - slap_daemon;
int old_threads = slapd_daemon_threads;
#define SLAPD_IDLE_CHECK_LIMIT 4
......@@ -2783,6 +2840,8 @@ loop:
continue;
}
if ( DAEMON_ID( lr->sl_sd ) != tid ) continue;
if ( lr->sl_mute ) {
Debug( LDAP_DEBUG_CONNS,
"daemon: " SLAP_EVENT_FNAME ": "
......@@ -2870,6 +2929,7 @@ loop:
if ( ns <= 0 ) break;
if ( slap_listeners[l]->sl_sd == AC_SOCKET_INVALID ) continue;
if ( DAEMON_ID( slap_listeners[l]->sl_sd ) != tid ) continue;
#ifdef LDAP_CONNECTIONLESS
if ( slap_listeners[l]->sl_is_udp ) continue;
#endif /* LDAP_CONNECTIONLESS */
......@@ -3088,6 +3148,13 @@ loop:
}
#endif /* SLAP_EVENTS_ARE_INDEXED */
/* Was number of listener threads decreased? */
if ( ldap_pvt_thread_pool_pausecheck_native( &connection_pool )) {
/* decreased, let this thread finish */
if ( tid >= slapd_daemon_threads )
break;
}
#ifndef HAVE_YIELDING_SELECT
ldap_pvt_thread_yield();
#endif /* ! HAVE_YIELDING_SELECT */
......@@ -3136,6 +3203,107 @@ loop:
return NULL;
}
typedef struct slap_tid_waiter {
int num_tids;
ldap_pvt_thread_t tids[0];
} slap_tid_waiter;
static void *
slapd_daemon_tid_cleanup(
void *ctx,
void *ptr )
{
slap_tid_waiter *tids = ptr;
int i;
for ( i=0; i<tids->num_tids; i++ )
ldap_pvt_thread_join( tids->tids[i], (void *)NULL );
ch_free( ptr );
return NULL;
}
int
slapd_daemon_resize( int newnum )
{
int i, rc;
if ( newnum == slapd_daemon_threads )
return 0;
/* wake up all current listener threads */
for ( i=0; i<slapd_daemon_threads; i++ )
WAKE_LISTENER(i,1);
/* mutexes may not survive realloc, so destroy & recreate later */
for ( i=0; i<slapd_daemon_threads; i++ )
ldap_pvt_thread_mutex_destroy( &slap_daemon[i].sd_mutex );
if ( newnum > slapd_daemon_threads ) {
wake_sds = ch_realloc( wake_sds, newnum * sizeof( sdpair ));
slap_daemon = ch_realloc( slap_daemon, newnum * sizeof( slap_daemon_st ));
for ( i=slapd_daemon_threads; i<newnum; i++ )
{
memset( &slap_daemon[i], 0, sizeof( slap_daemon_st ));
if( (rc = lutil_pair( wake_sds[i] )) < 0 ) {
Debug( LDAP_DEBUG_ANY,
"daemon: lutil_pair() failed rc=%d\n", rc );
return rc;
}
ber_pvt_socket_set_nonblock( wake_sds[i][1], 1 );
SLAP_SOCK_INIT(i);
}
for ( i=0; i<newnum; i++ )
ldap_pvt_thread_mutex_init( &slap_daemon[i].sd_mutex );
slapd_socket_realloc( newnum );
for ( i=slapd_daemon_threads; i<newnum; i++ )
{
/* listener as a separate THREAD */
rc = ldap_pvt_thread_create( &slap_daemon[i].sd_tid,
0, slapd_daemon_task, &slap_daemon[i] );
if ( rc != 0 ) {
Debug( LDAP_DEBUG_ANY,
"listener ldap_pvt_thread_create failed (%d)\n", rc );
return rc;
}
}
} else {
int j;
slap_tid_waiter *tids = ch_malloc( sizeof(slap_tid_waiter) +
((slapd_daemon_threads - newnum) * sizeof(ldap_pvt_thread_t )));