--- servers/slapd/overlays/syncprov.c 2011/01/13 19:03:47 1.147.2.86 +++ servers/slapd/overlays/syncprov.c 2009/03/13 21:59:20 1.276 @@ -1,8 +1,8 @@ -/* $OpenLDAP$ */ +/* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.275 2009/03/13 07:33:34 hyc Exp $ */ /* syncprov.c - syncrepl provider */ /* This work is part of OpenLDAP Software . * - * Copyright 2004-2011 The OpenLDAP Foundation. + * Copyright 2004-2009 The OpenLDAP Foundation. * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -48,7 +48,6 @@ typedef struct modtarget { /* A queued result of a persistent search */ typedef struct syncres { struct syncres *s_next; - Entry *s_e; struct berval s_dn; struct berval s_ndn; struct berval s_uuid; @@ -113,6 +112,7 @@ typedef struct slog_entry { } slog_entry; typedef struct sessionlog { + struct berval sl_mincsn; int sl_num; int sl_size; slog_entry *sl_head; @@ -124,7 +124,6 @@ typedef struct sessionlog { typedef struct syncprov_info_t { syncops *si_ops; BerVarray si_ctxcsn; /* ldapsync context */ - struct berval si_contextdn; int *si_sids; int si_numcsns; int si_chkops; /* checkpointing info */ @@ -132,23 +131,18 @@ typedef struct syncprov_info_t { int si_numops; /* number of ops since last checkpoint */ int si_nopres; /* Skip present phase */ int si_usehint; /* use reload hint */ - int si_active; /* True if there are active mods */ - int si_dirty; /* True if the context is dirty, i.e changes - * have been made without updating the csn. */ time_t si_chklast; /* time of last checkpoint */ Avlnode *si_mods; /* entries being modified */ sessionlog *si_logs; ldap_pvt_thread_rdwr_t si_csn_rwlock; ldap_pvt_thread_mutex_t si_ops_mutex; ldap_pvt_thread_mutex_t si_mods_mutex; - ldap_pvt_thread_mutex_t si_resp_mutex; } syncprov_info_t; typedef struct opcookie { slap_overinst *son; syncmatches *smatches; modtarget *smt; - Entry *se; struct berval sdn; /* DN of entry, for deletes */ struct berval sndn; struct berval suuid; /* UUID of entry */ @@ -158,11 +152,6 @@ typedef struct opcookie { short sreference; /* Is the entry a reference? */ } opcookie; -typedef struct mutexint { - ldap_pvt_thread_mutex_t mi_mutex; - int mi_int; -} mutexint; - typedef struct fbase_cookie { struct berval *fdn; /* DN of a modified entry, for scope testing */ syncops *fss; /* persistent search we're testing against */ @@ -593,8 +582,8 @@ syncprov_findcsn( Operation *op, find_cs slap_callback cb = {0}; Operation fop; SlapReply frs = { REP_RESULT }; - char buf[LDAP_PVT_CSNSTR_BUFSIZE + STRLENOF("(entryCSN<=)")]; - char cbuf[LDAP_PVT_CSNSTR_BUFSIZE]; + char buf[LDAP_LUTIL_CSNSTR_BUFSIZE + STRLENOF("(entryCSN<=)")]; + char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE]; struct berval maxcsn; Filter cf; AttributeAssertion eq = ATTRIBUTEASSERTION_INIT; @@ -748,36 +737,6 @@ again: return rc; } -/* Should find a place to cache these */ -static mutexint *get_mutexint() -{ - mutexint *mi = ch_malloc( sizeof( mutexint )); - ldap_pvt_thread_mutex_init( &mi->mi_mutex ); - mi->mi_int = 1; - return mi; -} - -static void inc_mutexint( mutexint *mi ) -{ - ldap_pvt_thread_mutex_lock( &mi->mi_mutex ); - mi->mi_int++; - ldap_pvt_thread_mutex_unlock( &mi->mi_mutex ); -} - -/* return resulting counter */ -static int dec_mutexint( mutexint *mi ) -{ - int i; - ldap_pvt_thread_mutex_lock( &mi->mi_mutex ); - i = --mi->mi_int; - ldap_pvt_thread_mutex_unlock( &mi->mi_mutex ); - if ( !i ) { - ldap_pvt_thread_mutex_destroy( &mi->mi_mutex ); - ch_free( mi ); - } - return i; -} - static void syncprov_free_syncop( syncops *so ) { @@ -785,8 +744,7 @@ syncprov_free_syncop( syncops *so ) GroupAssertion *ga, *gnext; ldap_pvt_thread_mutex_lock( &so->s_mutex ); - /* already being freed, or still in use */ - if ( !so->s_inuse || --so->s_inuse > 0 ) { + if ( --so->s_inuse > 0 ) { ldap_pvt_thread_mutex_unlock( &so->s_mutex ); return; } @@ -802,12 +760,6 @@ syncprov_free_syncop( syncops *so ) ch_free( so->s_base.bv_val ); for ( sr=so->s_res; sr; sr=srnext ) { srnext = sr->s_next; - if ( sr->s_e ) { - if ( !dec_mutexint( sr->s_e->e_private )) { - sr->s_e->e_private = NULL; - entry_free( sr->s_e ); - } - } ch_free( sr ); } ldap_pvt_thread_mutex_destroy( &so->s_mutex ); @@ -816,13 +768,14 @@ syncprov_free_syncop( syncops *so ) /* Send a persistent search response */ static int -syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, int mode ) +syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, + Entry **e, int mode ) { slap_overinst *on = opc->son; SlapReply rs = { REP_SEARCH }; LDAPControl *ctrls[2]; - struct berval cookie = BER_BVNULL, csns[2]; + struct berval cookie, csns[2]; Entry e_uuid = {0}; Attribute a_uuid = {0}; @@ -830,19 +783,17 @@ syncprov_sendresp( Operation *op, opcook return SLAPD_ABANDON; ctrls[1] = NULL; - if ( !BER_BVISNULL( &opc->sctxcsn )) { - csns[0] = opc->sctxcsn; - BER_BVZERO( &csns[1] ); - slap_compose_sync_cookie( op, &cookie, csns, so->s_rid, slap_serverID ? slap_serverID : -1 ); - } + csns[0] = opc->sctxcsn; + BER_BVZERO( &csns[1] ); + slap_compose_sync_cookie( op, &cookie, csns, so->s_rid, slap_serverID ? slap_serverID : -1 ); #ifdef LDAP_DEBUG if ( so->s_sid > 0 ) { Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: to=%03x, cookie=%s\n", - so->s_sid, cookie.bv_val ? cookie.bv_val : "", 0 ); + so->s_sid, cookie.bv_val, 0 ); } else { Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: cookie=%s\n", - cookie.bv_val ? cookie.bv_val : "", 0, 0 ); + cookie.bv_val, 0, 0 ); } #endif @@ -851,34 +802,38 @@ syncprov_sendresp( Operation *op, opcook a_uuid.a_nvals = &opc->suuid; rs.sr_err = syncprov_state_ctrl( op, &rs, &e_uuid, mode, ctrls, 0, 1, &cookie ); - if ( !BER_BVISNULL( &cookie )) { - op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); - } + op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); rs.sr_ctrls = ctrls; - rs.sr_entry = &e_uuid; - if ( mode == LDAP_SYNC_ADD || mode == LDAP_SYNC_MODIFY ) { - e_uuid = *opc->se; - e_uuid.e_private = NULL; - } - + op->o_bd->bd_info = (BackendInfo *)on->on_info; switch( mode ) { case LDAP_SYNC_ADD: + rs.sr_entry = *e; + if ( rs.sr_entry->e_private ) + rs.sr_flags = REP_ENTRY_MUSTRELEASE; if ( opc->sreference && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) { rs.sr_ref = get_entry_referrals( op, rs.sr_entry ); rs.sr_err = send_search_reference( op, &rs ); ber_bvarray_free( rs.sr_ref ); + if ( !rs.sr_entry ) + *e = NULL; break; } /* fallthru */ case LDAP_SYNC_MODIFY: + rs.sr_entry = *e; + if ( rs.sr_entry->e_private ) + rs.sr_flags = REP_ENTRY_MUSTRELEASE; rs.sr_attrs = op->ors_attrs; rs.sr_err = send_search_entry( op, &rs ); + if ( !rs.sr_entry ) + *e = NULL; break; case LDAP_SYNC_DELETE: e_uuid.e_attrs = NULL; e_uuid.e_name = opc->sdn; e_uuid.e_nname = opc->sndn; + rs.sr_entry = &e_uuid; if ( opc->sreference && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) { struct berval bv = BER_BVNULL; rs.sr_ref = &bv; @@ -892,14 +847,7 @@ syncprov_sendresp( Operation *op, opcook } /* In case someone else freed it already? */ if ( rs.sr_ctrls ) { - int i; - for ( i=0; rs.sr_ctrls[i]; i++ ) { - if ( rs.sr_ctrls[i] == ctrls[0] ) { - op->o_tmpfree( ctrls[0]->ldctl_value.bv_val, op->o_tmpmemctx ); - ctrls[0]->ldctl_value.bv_val = NULL; - break; - } - } + op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx ); rs.sr_ctrls = NULL; } @@ -934,9 +882,9 @@ syncprov_qplay( Operation *op, syncops * ldap_pvt_thread_mutex_unlock( &so->s_mutex ); if ( sr->s_mode == LDAP_SYNC_NEW_COOKIE ) { - SlapReply rs = { REP_INTERMEDIATE }; + SlapReply rs = { REP_INTERMEDIATE }; - rc = syncprov_sendinfo( op, &rs, LDAP_TAG_SYNC_NEW_COOKIE, + rc = syncprov_sendinfo( op, &rs, LDAP_TAG_SYNC_NEW_COOKIE, &sr->s_csn, 0, NULL, 0 ); } else { opc.sdn = sr->s_dn; @@ -944,15 +892,22 @@ syncprov_qplay( Operation *op, syncops * opc.suuid = sr->s_uuid; opc.sctxcsn = sr->s_csn; opc.sreference = sr->s_isreference; - opc.se = sr->s_e; + e = NULL; - rc = syncprov_sendresp( op, &opc, so, sr->s_mode ); + if ( sr->s_mode != LDAP_SYNC_DELETE ) { + rc = overlay_entry_get_ov( op, &opc.sndn, NULL, NULL, 0, &e, on ); + if ( rc ) { + Debug( LDAP_DEBUG_SYNC, "syncprov_qplay: failed to get %s, " + "error (%d), ignoring...\n", opc.sndn.bv_val, rc, 0 ); + ch_free( sr ); + rc = 0; + continue; + } + } + rc = syncprov_sendresp( op, &opc, so, &e, sr->s_mode ); - } - if ( sr->s_e ) { - if ( !dec_mutexint( sr->s_e->e_private )) { - sr->s_e->e_private = NULL; - entry_free ( sr->s_e ); + if ( e ) { + overlay_entry_release_ov( op, e, 0, on ); } } @@ -1046,10 +1001,6 @@ syncprov_qresp( opcookie *opc, syncops * srsize += cookie.bv_len + 1; sr = ch_malloc( srsize ); sr->s_next = NULL; - sr->s_e = opc->se; - /* bump refcount on this entry */ - if ( opc->se ) - inc_mutexint( opc->se->e_private ); sr->s_dn.bv_val = (char *)(sr + 1); sr->s_dn.bv_len = opc->sdn.bv_len; sr->s_mode = mode; @@ -1193,12 +1144,9 @@ syncprov_matchops( Operation *op, opcook rc = overlay_entry_get_ov( op, fc.fdn, NULL, NULL, 0, &e, on ); /* If we're sending responses now, make a copy and unlock the DB */ if ( e && !saveit ) { - if ( !opc->se ) { - opc->se = entry_dup( e ); - opc->se->e_private = get_mutexint(); - } + Entry *e2 = entry_dup( e ); overlay_entry_release_ov( op, e, 0, on ); - e = opc->se; + e = e2; } if ( rc ) { op->o_bd = b0; @@ -1206,13 +1154,6 @@ syncprov_matchops( Operation *op, opcook } } else { e = op->ora_e; - if ( !saveit ) { - if ( !opc->se ) { - opc->se = entry_dup( e ); - opc->se->e_private = get_mutexint(); - } - e = opc->se; - } } if ( saveit || op->o_tag == LDAP_REQ_ADD ) { @@ -1242,18 +1183,22 @@ syncprov_matchops( Operation *op, opcook if ( ss->s_op->o_abandon ) continue; - /* Don't send ops back to the originator */ - if ( opc->osid > 0 && opc->osid == ss->s_sid ) { - Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping original sid %03x\n", - opc->osid, 0, 0 ); - continue; - } + /* First time thru, check for possible skips */ + if ( saveit || op->o_tag == LDAP_REQ_ADD ) { - /* Don't send ops back to the messenger */ - if ( opc->rsid > 0 && opc->rsid == ss->s_sid ) { - Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping relayed sid %03x\n", - opc->rsid, 0, 0 ); - continue; + /* Don't send ops back to the originator */ + if ( opc->osid > 0 && opc->osid == ss->s_sid ) { + Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping original sid %03x\n", + opc->osid, 0, 0 ); + continue; + } + + /* Don't send ops back to the messenger */ + if ( opc->rsid > 0 && opc->rsid == ss->s_sid ) { + Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping relayed sid %03x\n", + opc->rsid, 0, 0 ); + continue; + } } /* validate base */ @@ -1273,6 +1218,7 @@ syncprov_matchops( Operation *op, opcook continue; } + /* If we're sending results now, look for this op in old matches */ if ( !saveit ) { syncmatches *old; @@ -1296,23 +1242,13 @@ syncprov_matchops( Operation *op, opcook } if ( fc.fscope ) { - ldap_pvt_thread_mutex_lock( &ss->s_mutex ); op2 = *ss->s_op; oh = *op->o_hdr; oh.oh_conn = ss->s_op->o_conn; oh.oh_connid = ss->s_op->o_connid; - op2.o_bd = op->o_bd->bd_self; op2.o_hdr = &oh; op2.o_extra = op->o_extra; - op2.o_callback = NULL; - if (ss->s_flags & PS_FIX_FILTER) { - /* Skip the AND/GE clause that we stuck on in front. We - would lose deletes/mods that happen during the refresh - phase otherwise (ITS#6555) */ - op2.ors_filter = ss->s_op->ors_filter->f_and->f_next; - } - ldap_pvt_thread_mutex_unlock( &ss->s_mutex ); - rc = test_filter( &op2, e, op2.ors_filter ); + rc = test_filter( &op2, e, ss->s_op->ors_filter ); } Debug( LDAP_DEBUG_TRACE, "syncprov_matchops: sid %03x fscope %d rc %d\n", @@ -1352,17 +1288,9 @@ syncprov_matchops( Operation *op, opcook if ( !SLAP_ISOVERLAY( op->o_bd )) { op->o_bd = &db; } - if ( saveit ) - overlay_entry_release_ov( op, e, 0, on ); + overlay_entry_release_ov( op, e, 0, on ); op->o_bd = b0; } - if ( opc->se && !saveit ) { - if ( !dec_mutexint( opc->se->e_private )) { - opc->se->e_private = NULL; - entry_free( opc->se ); - opc->se = NULL; - } - } if ( freefdn ) { op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx ); } @@ -1379,11 +1307,6 @@ syncprov_op_cleanup( Operation *op, Slap syncmatches *sm, *snext; modtarget *mt, mtdummy; - ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); - if ( si->si_active ) - si->si_active--; - ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); - for (sm = opc->smatches; sm; sm=snext) { snext = sm->sm_next; syncprov_free_syncop( sm->sm_op ); @@ -1393,14 +1316,15 @@ syncprov_op_cleanup( Operation *op, Slap /* Remove op from lock table */ mt = opc->smt; if ( mt ) { - ldap_pvt_thread_mutex_lock( &mt->mt_mutex ); - mt->mt_mods = mt->mt_mods->mi_next; + modinst *mi = mt->mt_mods; + /* If there are more, promote the next one */ - if ( mt->mt_mods ) { + if ( mi->mi_next ) { + ldap_pvt_thread_mutex_lock( &mt->mt_mutex ); + mt->mt_mods = mi->mi_next; mt->mt_op = mt->mt_mods->mi_op; ldap_pvt_thread_mutex_unlock( &mt->mt_mutex ); } else { - ldap_pvt_thread_mutex_unlock( &mt->mt_mutex ); ldap_pvt_thread_mutex_lock( &si->si_mods_mutex ); avl_delete( &si->si_mods, mt, sp_avl_cmp ); ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex ); @@ -1429,8 +1353,6 @@ syncprov_checkpoint( Operation *op, Slap SlapReply rsm = { 0 }; slap_callback cb = {0}; BackendDB be; - BackendInfo *bi; - #ifdef CHECK_CSN Syntax *syn = slap_schema.si_ad_contextCSN->ad_type->sat_syntax; @@ -1457,28 +1379,12 @@ syncprov_checkpoint( Operation *op, Slap be = *on->on_info->oi_origdb; opm.o_bd = &be; } - opm.o_req_dn = si->si_contextdn; - opm.o_req_ndn = si->si_contextdn; - bi = opm.o_bd->bd_info; + opm.o_req_dn = opm.o_bd->be_suffix[0]; + opm.o_req_ndn = opm.o_bd->be_nsuffix[0]; opm.o_bd->bd_info = on->on_info->oi_orig; opm.o_managedsait = SLAP_CONTROL_NONCRITICAL; opm.o_no_schema_check = 1; opm.o_bd->be_modify( &opm, &rsm ); - - if ( rsm.sr_err == LDAP_NO_SUCH_OBJECT && - SLAP_SYNC_SUBENTRY( opm.o_bd )) { - const char *text; - char txtbuf[SLAP_TEXT_BUFLEN]; - size_t textlen = sizeof txtbuf; - Entry *e = slap_create_context_csn_entry( opm.o_bd, NULL ); - slap_mods2entry( &mod, &e, 0, 1, &text, txtbuf, textlen); - opm.ora_e = e; - opm.o_bd->be_add( &opm, &rsm ); - if ( e == opm.ora_e ) - be_entry_release_w( &opm, opm.ora_e ); - } - opm.o_bd->bd_info = bi; - if ( mod.sml_next != NULL ) { slap_mods_free( mod.sml_next, 1 ); } @@ -1500,23 +1406,6 @@ syncprov_add_slog( Operation *op ) sl = si->si_logs; { - if ( BER_BVISEMPTY( &op->o_csn ) ) { - /* During the syncrepl refresh phase we can receive operations - * without a csn. We cannot reliably determine the consumers - * state with respect to such operations, so we ignore them and - * wipe out anything in the log if we see them. - */ - ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); - while ( se = sl->sl_head ) { - sl->sl_head = se->se_next; - ch_free( se ); - } - sl->sl_tail = NULL; - sl->sl_num = 0; - ldap_pvt_thread_mutex_unlock( &sl->sl_mutex ); - return; - } - /* Allocate a record. UUIDs are not NUL-terminated. */ se = ch_malloc( sizeof( slog_entry ) + opc->suuid.bv_len + op->o_csn.bv_len + 1 ); @@ -1535,29 +1424,17 @@ syncprov_add_slog( Operation *op ) ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); if ( sl->sl_head ) { - /* Keep the list in csn order. */ - if ( ber_bvcmp( &sl->sl_tail->se_csn, &se->se_csn ) <= 0 ) { - sl->sl_tail->se_next = se; - sl->sl_tail = se; - } else { - slog_entry **sep; - - for ( sep = &sl->sl_head; *sep; sep = &(*sep)->se_next ) { - if ( ber_bvcmp( &se->se_csn, &(*sep)->se_csn ) < 0 ) { - se->se_next = *sep; - *sep = se; - break; - } - } - } + sl->sl_tail->se_next = se; } else { sl->sl_head = se; - sl->sl_tail = se; } + sl->sl_tail = se; sl->sl_num++; while ( sl->sl_num > sl->sl_size ) { se = sl->sl_head; sl->sl_head = se->se_next; + strcpy( sl->sl_mincsn.bv_val, se->se_csn.bv_val ); + sl->sl_mincsn.bv_len = se->se_csn.bv_len; ch_free( se ); sl->sl_num--; } @@ -1583,7 +1460,7 @@ syncprov_playlog( Operation *op, SlapRep slap_overinst *on = (slap_overinst *)op->o_bd->bd_info; slog_entry *se; int i, j, ndel, num, nmods, mmods; - char cbuf[LDAP_PVT_CSNSTR_BUFSIZE]; + char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE]; BerVarray uuids; struct berval delcsn[2]; @@ -1642,8 +1519,6 @@ syncprov_playlog( Operation *op, SlapRep delcsn[0].bv_len = se->se_csn.bv_len; delcsn[0].bv_val[delcsn[0].bv_len] = '\0'; } else { - if ( se->se_tag == LDAP_REQ_ADD ) - continue; nmods++; j = num - nmods; } @@ -1760,17 +1635,67 @@ syncprov_op_response( Operation *op, Sla if ( rs->sr_err == LDAP_SUCCESS ) { struct berval maxcsn; - char cbuf[LDAP_PVT_CSNSTR_BUFSIZE]; + char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE]; int do_check = 0, have_psearches, foundit, csn_changed = 0; - ldap_pvt_thread_mutex_lock( &si->si_resp_mutex ); - /* Update our context CSN */ cbuf[0] = '\0'; maxcsn.bv_val = cbuf; maxcsn.bv_len = sizeof(cbuf); ldap_pvt_thread_rdwr_wlock( &si->si_csn_rwlock ); +#if 0 + if ( op->o_dont_replicate && + op->orm_modlist->sml_op == LDAP_MOD_REPLACE && + op->orm_modlist->sml_desc == slap_schema.si_ad_contextCSN ) { + /* Catch contextCSN updates from syncrepl. We have to look at + * all the attribute values, as there may be more than one csn + * that changed, and only one can be passed in the csn queue. + */ + Modifications *mod = op->orm_modlist; + int i, j, sid; + + for ( i=0; isml_numvals; i++ ) { + sid = slap_parse_csn_sid( &mod->sml_values[i] ); + + for ( j=0; jsi_numcsns; j++ ) { + if ( sid == si->si_sids[j] ) { + if ( ber_bvcmp( &mod->sml_values[i], &si->si_ctxcsn[j] ) > 0 ) { + ber_bvreplace( &si->si_ctxcsn[j], &mod->sml_values[i] ); + csn_changed = 1; + } + break; + } + } + + if ( j == si->si_numcsns ) { + value_add_one( &si->si_ctxcsn, &mod->sml_values[i] ); + si->si_numcsns++; + si->si_sids = ch_realloc( si->si_sids, si->si_numcsns * + sizeof(int)); + si->si_sids[j] = sid; + csn_changed = 1; + } + } + ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock ); + + if ( csn_changed ) { + ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); + have_psearches = ( si->si_ops != NULL ); + ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); + + if ( have_psearches ) { + for ( sm = opc->smatches; sm; sm=sm->sm_next ) { + if ( sm->sm_op->s_op->o_abandon ) + continue; + syncprov_qresp( opc, sm->sm_op, LDAP_SYNC_NEW_COOKIE ); + } + } + } + return SLAP_CB_CONTINUE; + } +#endif + slap_get_commit_csn( op, &maxcsn, &foundit ); if ( BER_BVISEMPTY( &maxcsn ) && SLAP_GLUE_SUBORDINATE( op->o_bd )) { /* syncrepl queues the CSN values in the db where @@ -1810,91 +1735,34 @@ syncprov_op_response( Operation *op, Sla sizeof(int)); si->si_sids[i] = sid; } + } else if ( !foundit ) { + /* internal ops that aren't meant to be replicated */ + ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock ); + return SLAP_CB_CONTINUE; } /* Don't do any processing for consumer contextCSN updates */ if ( op->o_dont_replicate ) { - if ( op->o_tag == LDAP_REQ_MODIFY && - op->orm_modlist->sml_op == LDAP_MOD_REPLACE && - op->orm_modlist->sml_desc == slap_schema.si_ad_contextCSN ) { - /* Catch contextCSN updates from syncrepl. We have to look at - * all the attribute values, as there may be more than one csn - * that changed, and only one can be passed in the csn queue. - */ - Modifications *mod = op->orm_modlist; - unsigned i; - int j, sid; - - for ( i=0; isml_numvals; i++ ) { - sid = slap_parse_csn_sid( &mod->sml_values[i] ); - for ( j=0; jsi_numcsns; j++ ) { - if ( sid == si->si_sids[j] ) { - if ( ber_bvcmp( &mod->sml_values[i], &si->si_ctxcsn[j] ) > 0 ) { - ber_bvreplace( &si->si_ctxcsn[j], &mod->sml_values[i] ); - csn_changed = 1; - } - break; - } - } - - if ( j == si->si_numcsns ) { - value_add_one( &si->si_ctxcsn, &mod->sml_values[i] ); - si->si_numcsns++; - si->si_sids = ch_realloc( si->si_sids, si->si_numcsns * - sizeof(int)); - si->si_sids[j] = sid; - csn_changed = 1; - } - } - if ( csn_changed ) - si->si_dirty = 0; - ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock ); - - if ( csn_changed ) { - syncops *ss; - ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); - for ( ss = si->si_ops; ss; ss = ss->s_next ) { - if ( ss->s_op->o_abandon ) - continue; - /* Send the updated csn to all syncrepl consumers, - * including the server from which it originated. - * The syncrepl consumer and syncprov provider on - * the originating server may be configured to store - * their csn values in different entries. - */ - syncprov_qresp( opc, ss, LDAP_SYNC_NEW_COOKIE ); - } - ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); - } - } else { ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock ); - } - goto leave; + return SLAP_CB_CONTINUE; } si->si_numops++; if ( si->si_chkops || si->si_chktime ) { - /* Never checkpoint adding the context entry, - * it will deadlock - */ - if ( op->o_tag != LDAP_REQ_ADD || - !dn_match( &op->o_req_ndn, &si->si_contextdn )) { - if ( si->si_chkops && si->si_numops >= si->si_chkops ) { + if ( si->si_chkops && si->si_numops >= si->si_chkops ) { + do_check = 1; + si->si_numops = 0; + } + if ( si->si_chktime && + (op->o_time - si->si_chklast >= si->si_chktime )) { + if ( si->si_chklast ) { do_check = 1; - si->si_numops = 0; - } - if ( si->si_chktime && - (op->o_time - si->si_chklast >= si->si_chktime )) { - if ( si->si_chklast ) { - do_check = 1; - si->si_chklast = op->o_time; - } else { - si->si_chklast = 1; - } + si->si_chklast = op->o_time; + } else { + si->si_chklast = 1; } } } - si->si_dirty = !csn_changed; ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock ); if ( do_check ) { @@ -1934,10 +1802,10 @@ syncprov_op_response( Operation *op, Sla } /* Add any log records */ - if ( si->si_logs ) { + if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) { syncprov_add_slog( op ); } -leave: ldap_pvt_thread_mutex_unlock( &si->si_resp_mutex ); + } return SLAP_CB_CONTINUE; } @@ -1953,14 +1821,14 @@ syncprov_op_compare( Operation *op, Slap syncprov_info_t *si = on->on_bi.bi_private; int rc = SLAP_CB_CONTINUE; - if ( dn_match( &op->o_req_ndn, &si->si_contextdn ) && + if ( dn_match( &op->o_req_ndn, op->o_bd->be_nsuffix ) && op->oq_compare.rs_ava->aa_desc == slap_schema.si_ad_contextCSN ) { Entry e = {0}; Attribute a = {0}; - e.e_name = si->si_contextdn; - e.e_nname = si->si_contextdn; + e.e_name = op->o_bd->be_suffix[0]; + e.e_nname = op->o_bd->be_nsuffix[0]; e.e_attrs = &a; a.a_desc = slap_schema.si_ad_contextCSN; @@ -2022,7 +1890,6 @@ syncprov_op_mod( Operation *op, SlapRepl ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); have_psearches = ( si->si_ops != NULL ); - si->si_active++; ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); cbsize = sizeof(slap_callback) + sizeof(opcookie) + @@ -2065,15 +1932,6 @@ syncprov_op_mod( Operation *op, SlapRepl mt = avl_find( si->si_mods, &mtdummy, sp_avl_cmp ); if ( mt ) { ldap_pvt_thread_mutex_lock( &mt->mt_mutex ); - if ( mt->mt_mods == NULL ) { - /* Cannot reuse this mt, as another thread is about - * to release it in syncprov_op_cleanup. - */ - ldap_pvt_thread_mutex_unlock( &mt->mt_mutex ); - mt = NULL; - } - } - if ( mt ) { ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex ); mt->mt_tail->mi_next = mi; mt->mt_tail = mi; @@ -2339,10 +2197,9 @@ syncprov_search_response( Operation *op, LDAP_SYNC_ADD, rs->sr_ctrls, 0, 0, NULL ); } } else if ( rs->sr_type == REP_RESULT && rs->sr_err == LDAP_SUCCESS ) { - struct berval cookie = BER_BVNULL; + struct berval cookie; - if ( ( ss->ss_flags & SS_CHANGED ) && - ss->ss_ctxcsn && !BER_BVISNULL( &ss->ss_ctxcsn[0] )) { + if ( ss->ss_flags & SS_CHANGED ) { slap_compose_sync_cookie( op, &cookie, ss->ss_ctxcsn, srs->sr_state.rid, slap_serverID ? slap_serverID : -1 ); @@ -2366,7 +2223,7 @@ syncprov_search_response( Operation *op, LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE, ( ss->ss_flags & SS_CHANGED ) ? &cookie : NULL, 1, NULL, 0 ); - if ( !BER_BVISNULL( &cookie )) + if ( ss->ss_flags & SS_CHANGED ) op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); /* Detach this Op from frontend control */ @@ -2413,7 +2270,6 @@ syncprov_op_search( Operation *op, SlapR BerVarray ctxcsn; int i, *sids, numcsns; struct berval mincsn; - int dirty = 0; if ( !(op->o_sync_mode & SLAP_SYNC_REFRESH) ) return SLAP_CB_CONTINUE; @@ -2458,20 +2314,6 @@ syncprov_op_search( Operation *op, SlapR sop->s_inuse = 1; ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); - while ( si->si_active ) { - /* Wait for active mods to finish before proceeding, as they - * may already have inspected the si_ops list looking for - * consumers to replicate the change to. Using the log - * doesn't help, as we may finish playing it before the - * active mods gets added to it. - */ - ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); - if ( slapd_shutdown ) - return SLAPD_ABANDON; - if ( !ldap_pvt_thread_pool_pausecheck( &connection_pool )) - ldap_pvt_thread_yield(); - ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); - } sop->s_next = si->si_ops; si->si_ops = sop; ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); @@ -2489,7 +2331,6 @@ syncprov_op_search( Operation *op, SlapR ctxcsn = NULL; sids = NULL; } - dirty = si->si_dirty; ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock ); /* If we have a cookie, handle the PRESENT lookups */ @@ -2536,40 +2377,19 @@ syncprov_op_search( Operation *op, SlapR /* If nothing has changed, shortcut it */ if ( srs->sr_state.numcsns == numcsns ) { - int i, j, newer; + int i, j; for ( i=0; isr_state.numcsns; i++ ) { for ( j=0; jsr_state.sids[i] != sids[j] ) continue; - newer = ber_bvcmp( &srs->sr_state.ctxcsn[i], &ctxcsn[j] ); - /* If our state is newer, tell consumer about changes */ - if ( newer < 0 ) + if ( !bvmatch( &srs->sr_state.ctxcsn[i], &ctxcsn[j] )) changed = SS_CHANGED; - else if ( newer > 0 ) { - /* our state is older, complain to consumer */ - rs->sr_err = LDAP_UNWILLING_TO_PERFORM; - rs->sr_text = "consumer state is newer than provider!"; -bailout: - if ( sop ) { - syncops **sp = &si->si_ops; - - ldap_pvt_thread_mutex_lock( &si->si_ops_mutex ); - while ( *sp != sop ) - sp = &(*sp)->s_next; - *sp = sop->s_next; - ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex ); - ch_free( sop ); - } - rs->sr_ctrls = NULL; - send_ldap_result( op, rs ); - return rs->sr_err; - } break; } if ( changed ) break; } - if ( !changed && !dirty ) { + if ( !changed ) { do_present = 0; no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) { LDAPControl *ctrls[2]; @@ -2597,7 +2417,7 @@ no_change: if ( !(op->o_sync_mode & SLA /* Are there any log entries, and is the consumer state * present in the session log? */ - if ( sl->sl_num > 0 && ber_bvcmp( &mincsn, &sl->sl_head->se_csn ) >= 0 ) { + if ( sl->sl_num > 0 && ber_bvcmp( &mincsn, &sl->sl_mincsn ) >= 0 ) { do_present = 0; /* mutex is unlocked in playlog */ syncprov_playlog( op, rs, sl, srs, ctxcsn, numcsns, sids ); @@ -2614,9 +2434,8 @@ no_change: if ( !(op->o_sync_mode & SLA ber_bvarray_free_x( ctxcsn, op->o_tmpmemctx ); if ( sids ) op->o_tmpfree( sids, op->o_tmpmemctx ); - rs->sr_err = LDAP_SYNC_REFRESH_REQUIRED; - rs->sr_text = "sync cookie is stale"; - goto bailout; + send_ldap_error( op, rs, LDAP_SYNC_REFRESH_REQUIRED, "sync cookie is stale" ); + return rs->sr_err; } if ( srs->sr_state.ctxcsn ) { ber_bvarray_free_x( srs->sr_state.ctxcsn, op->o_tmpmemctx ); @@ -2636,7 +2455,8 @@ no_change: if ( !(op->o_sync_mode & SLA ber_bvarray_free_x( ctxcsn, op->o_tmpmemctx ); if ( sids ) op->o_tmpfree( sids, op->o_tmpmemctx ); - goto bailout; + send_ldap_result( op, rs ); + return rs->sr_err; } } } else { @@ -2653,7 +2473,7 @@ shortcut: } /* If something changed, find the changes */ - if ( gotstate && ( changed || dirty ) ) { + if ( gotstate && changed ) { Filter *fand, *fava; fand = op->o_tmpalloc( sizeof(Filter), op->o_tmpmemctx ); @@ -2669,14 +2489,10 @@ shortcut: #endif ber_dupbv_x( &fava->f_ava->aa_value, &mincsn, op->o_tmpmemctx ); fava->f_next = op->ors_filter; - if ( sop ) - ldap_pvt_thread_mutex_lock( &sop->s_mutex ); op->ors_filter = fand; filter2bv_x( op, op->ors_filter, &op->ors_filterstr ); - if ( sop ) { + if ( sop ) sop->s_flags |= PS_FIX_FILTER; - ldap_pvt_thread_mutex_unlock( &sop->s_mutex ); - } } /* Let our callback add needed info to returned entries */ @@ -2698,7 +2514,7 @@ shortcut: * the refresh phase, just invoke the response callback to transition * us into persist phase */ - if ( !changed && !dirty ) { + if ( !changed ) { rs->sr_err = LDAP_SUCCESS; rs->sr_nentries = 0; send_ldap_result( op, rs ); @@ -2722,7 +2538,7 @@ syncprov_operational( return SLAP_CB_CONTINUE; if ( rs->sr_entry && - dn_match( &rs->sr_entry->e_nname, &si->si_contextdn )) { + dn_match( &rs->sr_entry->e_nname, op->o_bd->be_nsuffix )) { if ( SLAP_OPATTRS( rs->sr_attr_flags ) || ad_inlist( slap_schema.si_ad_contextCSN, rs->sr_attrs )) { @@ -2829,11 +2645,8 @@ sp_cf_gen(ConfigArgs *c) case SP_CHKPT: if ( si->si_chkops || si->si_chktime ) { struct berval bv; - /* we assume si_chktime is a multiple of 60 - * because the parsed value was originally - * multiplied by 60 */ bv.bv_len = snprintf( c->cr_msg, sizeof( c->cr_msg ), - "%d %d", si->si_chkops, si->si_chktime/60 ); + "%d %d", si->si_chkops, si->si_chktime ); if ( bv.bv_len >= sizeof( c->cr_msg ) ) { rc = 1; } else { @@ -2939,7 +2752,9 @@ sp_cf_gen(ConfigArgs *c) } sl = si->si_logs; if ( !sl ) { - sl = ch_malloc( sizeof( sessionlog ) ); + sl = ch_malloc( sizeof( sessionlog ) + LDAP_LUTIL_CSNSTR_BUFSIZE ); + sl->sl_mincsn.bv_val = (char *)(sl+1); + sl->sl_mincsn.bv_len = 0; sl->sl_num = 0; sl->sl_head = sl->sl_tail = NULL; ldap_pvt_thread_mutex_init( &sl->sl_mutex ); @@ -3007,19 +2822,13 @@ syncprov_db_open( } thrctx = ldap_pvt_thread_pool_context(); - connection_fake_init2( &conn, &opbuf, thrctx, 0 ); + connection_fake_init( &conn, &opbuf, thrctx ); op = &opbuf.ob_op; op->o_bd = be; op->o_dn = be->be_rootdn; op->o_ndn = be->be_rootndn; - if ( SLAP_SYNC_SUBENTRY( be )) { - build_new_dn( &si->si_contextdn, be->be_nsuffix, - (struct berval *)&slap_ldapsync_cn_bv, NULL ); - } else { - si->si_contextdn = be->be_nsuffix[0]; - } - rc = overlay_entry_get_ov( op, &si->si_contextdn, NULL, + rc = overlay_entry_get_ov( op, be->be_nsuffix, NULL, slap_schema.si_ad_contextCSN, 0, &e, on ); if ( e ) { @@ -3043,7 +2852,7 @@ syncprov_db_open( /* Didn't find a contextCSN, should we generate one? */ if ( !si->si_ctxcsn ) { - char csnbuf[ LDAP_PVT_CSNSTR_BUFSIZE ]; + char csnbuf[ LDAP_LUTIL_CSNSTR_BUFSIZE ]; struct berval csn; if ( SLAP_SYNC_SHADOW( op->o_bd )) { @@ -3092,7 +2901,7 @@ syncprov_db_close( void *thrctx; thrctx = ldap_pvt_thread_pool_context(); - connection_fake_init2( &conn, &opbuf, thrctx, 0 ); + connection_fake_init( &conn, &opbuf, thrctx ); op = &opbuf.ob_op; op->o_bd = be; op->o_dn = be->be_rootdn; @@ -3124,7 +2933,6 @@ syncprov_db_init( ldap_pvt_thread_rdwr_init( &si->si_csn_rwlock ); ldap_pvt_thread_mutex_init( &si->si_ops_mutex ); ldap_pvt_thread_mutex_init( &si->si_mods_mutex ); - ldap_pvt_thread_mutex_init( &si->si_resp_mutex ); csn_anlist[0].an_desc = slap_schema.si_ad_entryCSN; csn_anlist[0].an_name = slap_schema.si_ad_entryCSN->ad_cname; @@ -3156,14 +2964,12 @@ syncprov_db_destroy( se = se_next; } - ldap_pvt_thread_mutex_destroy(&si->si_logs->sl_mutex); ch_free( si->si_logs ); } if ( si->si_ctxcsn ) ber_bvarray_free( si->si_ctxcsn ); if ( si->si_sids ) ch_free( si->si_sids ); - ldap_pvt_thread_mutex_destroy( &si->si_resp_mutex ); ldap_pvt_thread_mutex_destroy( &si->si_mods_mutex ); ldap_pvt_thread_mutex_destroy( &si->si_ops_mutex ); ldap_pvt_thread_rdwr_destroy( &si->si_csn_rwlock );