X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=servers%2Fslapd%2Foverlays%2Fsyncprov.c;h=1d960673452c27eb9b12c4130bb76d56f4f433cf;hb=065b315f0da07e2d30308316ae38fe62dd488539;hp=d2da4c718133228053e2bf4bbcccd82d5812f1af;hpb=2000f771f6a7a8beb389c1a7b6f03afe8ff6ac98;p=openldap
diff --git a/servers/slapd/overlays/syncprov.c b/servers/slapd/overlays/syncprov.c
index d2da4c7181..1d96067345 100644
--- a/servers/slapd/overlays/syncprov.c
+++ b/servers/slapd/overlays/syncprov.c
@@ -2,7 +2,7 @@
/* syncprov.c - syncrepl provider */
/* This work is part of OpenLDAP Software .
*
- * Copyright 2004-2015 The OpenLDAP Foundation.
+ * Copyright 2004-2017 The OpenLDAP Foundation.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -28,7 +28,9 @@
#include "config.h"
#include "ldap_rq.h"
+#ifdef LDAP_DEVEL
#define CHECK_CSN 1
+#endif
/* A modify request on a particular entry */
typedef struct modinst {
@@ -87,6 +89,7 @@ typedef struct syncops {
int s_inuse; /* reference count */
struct syncres *s_res;
struct syncres *s_restail;
+ void *s_pool_cookie;
ldap_pvt_thread_mutex_t s_mutex;
} syncops;
@@ -792,20 +795,25 @@ static void free_resinfo( syncres *sr )
}
}
+#define FS_UNLINK 1
+#define FS_LOCK 2
+
static int
-syncprov_free_syncop( syncops *so, int unlink )
+syncprov_free_syncop( syncops *so, int flags )
{
syncres *sr, *srnext;
GroupAssertion *ga, *gnext;
- ldap_pvt_thread_mutex_lock( &so->s_mutex );
+ if ( flags & FS_LOCK )
+ ldap_pvt_thread_mutex_lock( &so->s_mutex );
/* already being freed, or still in use */
if ( !so->s_inuse || --so->s_inuse > 0 ) {
- ldap_pvt_thread_mutex_unlock( &so->s_mutex );
+ if ( flags & FS_LOCK )
+ ldap_pvt_thread_mutex_unlock( &so->s_mutex );
return 0;
}
ldap_pvt_thread_mutex_unlock( &so->s_mutex );
- if ( unlink ) {
+ if (( flags & FS_UNLINK ) && so->s_si ) {
syncops **sop;
ldap_pvt_thread_mutex_lock( &so->s_si->si_ops_mutex );
for ( sop = &so->s_si->si_ops; *sop; sop = &(*sop)->s_next ) {
@@ -915,7 +923,6 @@ syncprov_qstart( syncops *so );
static int
syncprov_qplay( Operation *op, syncops *so )
{
- slap_overinst *on = LDAP_SLIST_FIRST(&so->s_op->o_extra)->oe_key;
syncres *sr;
int rc = 0;
@@ -961,11 +968,8 @@ syncprov_qplay( Operation *op, syncops *so )
if ( rc == 0 && so->s_res ) {
syncprov_qstart( so );
- } else {
- so->s_flags ^= PS_TASK_QUEUED;
}
- ldap_pvt_thread_mutex_unlock( &so->s_mutex );
return rc;
}
@@ -1001,8 +1005,17 @@ syncprov_qtask( void *ctx, void *arg )
rc = syncprov_qplay( op, so );
+ /* if an error occurred, or no responses left, task is no longer queued */
+ if ( !rc && !so->s_res )
+ rc = 1;
+
/* decrement use count... */
- syncprov_free_syncop( so, 1 );
+ if ( !syncprov_free_syncop( so, FS_UNLINK )) {
+ if ( rc )
+ /* if we didn't unlink, and task is no longer queued, clear flag */
+ so->s_flags ^= PS_TASK_QUEUED;
+ ldap_pvt_thread_mutex_unlock( &so->s_mutex );
+ }
return NULL;
}
@@ -1013,8 +1026,8 @@ syncprov_qstart( syncops *so )
{
so->s_flags |= PS_TASK_QUEUED;
so->s_inuse++;
- ldap_pvt_thread_pool_submit( &connection_pool,
- syncprov_qtask, so );
+ ldap_pvt_thread_pool_submit2( &connection_pool,
+ syncprov_qtask, so, &so->s_pool_cookie );
}
/* Queue a persistent search response */
@@ -1129,7 +1142,7 @@ syncprov_drop_psearch( syncops *so, int lock )
if ( lock )
ldap_pvt_thread_mutex_unlock( &so->s_op->o_conn->c_mutex );
}
- return syncprov_free_syncop( so, 0 );
+ return syncprov_free_syncop( so, FS_LOCK );
}
static int
@@ -1362,7 +1375,7 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
* with saveit == TRUE
*/
snext = ss->s_next;
- if ( syncprov_free_syncop( ss, 0 ) ) {
+ if ( syncprov_free_syncop( ss, FS_LOCK ) ) {
*pss = snext;
gonext = 0;
}
@@ -1407,7 +1420,7 @@ syncprov_op_cleanup( Operation *op, SlapReply *rs )
for (sm = opc->smatches; sm; sm=snext) {
snext = sm->sm_next;
- syncprov_free_syncop( sm->sm_op, 1 );
+ syncprov_free_syncop( sm->sm_op, FS_LOCK|FS_UNLINK );
op->o_tmpfree( sm, op->o_tmpmemctx );
}
@@ -1492,6 +1505,7 @@ syncprov_checkpoint( Operation *op, slap_overinst *on )
opm.o_bd->bd_info = on->on_info->oi_orig;
opm.o_managedsait = SLAP_CONTROL_NONCRITICAL;
opm.o_no_schema_check = 1;
+ opm.o_opid = -1;
opm.o_bd->be_modify( &opm, &rsm );
if ( rsm.sr_err == LDAP_NO_SUCH_OBJECT &&
@@ -1548,7 +1562,7 @@ syncprov_add_slog( Operation *op )
}
/* Allocate a record. UUIDs are not NUL-terminated. */
- se = ch_malloc( sizeof( slog_entry ) + opc->suuid.bv_len +
+ se = ch_malloc( sizeof( slog_entry ) + opc->suuid.bv_len +
op->o_csn.bv_len + 1 );
se->se_next = NULL;
se->se_tag = op->o_tag;
@@ -1594,7 +1608,7 @@ syncprov_add_slog( Operation *op )
}
sl->sl_num++;
while ( sl->sl_num > sl->sl_size ) {
- int i, j;
+ int i;
se = sl->sl_head;
sl->sl_head = se->se_next;
for ( i=0; isl_numcsns; i++ )
@@ -1915,10 +1929,13 @@ syncprov_op_response( Operation *op, SlapReply *rs )
} else {
ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
}
+ if ( csn_changed )
+ si->si_numops++;
goto leave;
}
- si->si_numops++;
+ if ( csn_changed )
+ si->si_numops++;
if ( si->si_chkops || si->si_chktime ) {
/* Never checkpoint adding the context entry,
* it will deadlock
@@ -1975,6 +1992,8 @@ syncprov_op_response( Operation *op, SlapReply *rs )
continue;
syncprov_qresp( opc, sm->sm_op, LDAP_SYNC_DELETE );
}
+ if ( opc->ssres.s_info )
+ free_resinfo( &opc->ssres );
break;
}
}
@@ -2429,7 +2448,7 @@ syncprov_search_response( Operation *op, SlapReply *rs )
} else {
/* It's RefreshAndPersist, transition to Persist phase */
syncprov_sendinfo( op, rs, ( ss->ss_flags & SS_PRESENT ) ?
- LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE,
+ LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE,
( ss->ss_flags & SS_CHANGED ) ? &cookie : NULL,
1, NULL, 0 );
if ( !BER_BVISNULL( &cookie ))
@@ -2496,7 +2515,7 @@ syncprov_op_search( Operation *op, SlapReply *rs )
syncops so = {0};
fbase_cookie fc;
opcookie opc;
- slap_callback sc;
+ slap_callback sc = {0};
fc.fss = &so;
fc.fbase = 0;
@@ -2519,7 +2538,6 @@ syncprov_op_search( Operation *op, SlapReply *rs )
}
sop = ch_malloc( sizeof( syncops ));
*sop = so;
- ldap_pvt_thread_mutex_init( &sop->s_mutex );
sop->s_rid = srs->sr_state.rid;
sop->s_sid = srs->sr_state.sid;
/* set refcount=2 to prevent being freed out from under us
@@ -2544,13 +2562,21 @@ syncprov_op_search( Operation *op, SlapReply *rs )
ldap_pvt_thread_yield();
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
}
+ if ( op->o_abandon ) {
+ ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
+ ch_free( sop );
+ return SLAPD_ABANDON;
+ }
+ ldap_pvt_thread_mutex_init( &sop->s_mutex );
sop->s_next = si->si_ops;
sop->s_si = si;
si->si_ops = sop;
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
}
- /* snapshot the ctxcsn */
+ /* snapshot the ctxcsn
+ * Note: this must not be done before the psearch setup. (ITS#8365)
+ */
ldap_pvt_thread_rdwr_rlock( &si->si_csn_rwlock );
numcsns = si->si_numcsns;
if ( numcsns ) {
@@ -2564,17 +2590,19 @@ syncprov_op_search( Operation *op, SlapReply *rs )
}
dirty = si->si_dirty;
ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
-
+
/* If we have a cookie, handle the PRESENT lookups */
if ( srs->sr_state.ctxcsn ) {
sessionlog *sl;
int i, j;
- /* If we don't have any CSN of our own yet, pretend nothing
- * has changed.
+ /* If we don't have any CSN of our own yet, bail out.
*/
- if ( !numcsns )
- goto no_change;
+ if ( !numcsns ) {
+ rs->sr_err = LDAP_UNWILLING_TO_PERFORM;
+ rs->sr_text = "consumer has state info but provider doesn't!";
+ goto bailout;
+ }
if ( !si->si_nopres )
do_present = SS_PRESENT;
@@ -2640,6 +2668,10 @@ syncprov_op_search( Operation *op, SlapReply *rs )
/* our state is older, complain to consumer */
rs->sr_err = LDAP_UNWILLING_TO_PERFORM;
rs->sr_text = "consumer state is newer than provider!";
+ Log4( LDAP_DEBUG_SYNC, ldap_syslog_level,
+ "consumer %d state %s is newer than provider %d state %s\n",
+ sids[i], srs->sr_state.ctxcsn[i].bv_val, sids[j], /* == slap_serverID */
+ ctxcsn[j].bv_val);
bailout:
if ( sop ) {
syncops **sp = &si->si_ops;
@@ -2655,7 +2687,7 @@ bailout:
send_ldap_result( op, rs );
return rs->sr_err;
}
- }
+ }
if ( BER_BVISEMPTY( &mincsn )) {
mincsn = maxcsn;
minsid = maxsid;
@@ -2664,7 +2696,7 @@ bailout:
/* If nothing has changed, shortcut it */
if ( !changed && !dirty ) {
do_present = 0;
-no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
+no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
LDAPControl *ctrls[2];
ctrls[0] = NULL;
@@ -2751,6 +2783,9 @@ no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
}
}
} else {
+ /* The consumer knows nothing, we know nothing. OK. */
+ if (!numcsns)
+ goto no_change;
/* No consumer state, assume something has changed */
changed = SS_CHANGED;
}
@@ -3155,7 +3190,7 @@ syncprov_db_open(
char csnbuf[ LDAP_PVT_CSNSTR_BUFSIZE ];
struct berval csn;
- if ( SLAP_SYNC_SHADOW( op->o_bd )) {
+ if ( slap_serverID || SLAP_SYNC_SHADOW( op->o_bd )) {
/* If we're also a consumer, then don't generate anything.
* Wait for our provider to send it to us, or for a local
* modify if we have multimaster.
@@ -3198,8 +3233,8 @@ syncprov_db_close(
ConfigReply *cr
)
{
- slap_overinst *on = (slap_overinst *) be->bd_info;
- syncprov_info_t *si = (syncprov_info_t *)on->on_bi.bi_private;
+ slap_overinst *on = (slap_overinst *) be->bd_info;
+ syncprov_info_t *si = (syncprov_info_t *)on->on_bi.bi_private;
#ifdef SLAP_CONFIG_DELETE
syncops *so, *sonext;
#endif /* SLAP_CONFIG_DELETE */
@@ -3230,7 +3265,10 @@ syncprov_db_close(
rs.sr_err = LDAP_UNAVAILABLE;
send_ldap_result( so->s_op, &rs );
sonext=so->s_next;
- syncprov_drop_psearch( so, 0);
+ if ( so->s_flags & PS_TASK_QUEUED )
+ ldap_pvt_thread_pool_retract( so->s_pool_cookie );
+ if ( !syncprov_drop_psearch( so, 0 ))
+ so->s_si = NULL;
}
si->si_ops=NULL;
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
@@ -3238,7 +3276,7 @@ syncprov_db_close(
overlay_unregister_control( be, LDAP_CONTROL_SYNC );
#endif /* SLAP_CONFIG_DELETE */
- return 0;
+ return 0;
}
static int