X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=servers%2Fslapd%2Foverlays%2Fsyncprov.c;h=7114eb90f895527c08fd6af0d03bc4a416b43e8e;hb=ee3f05000055619ec6d4eee48b57fa5cb6085127;hp=6f8fb7f3380cce0d2f370380a9f13a4de0822f16;hpb=13f52646b26e3afbb16e78c2487f19f7854a9180;p=openldap diff --git a/servers/slapd/overlays/syncprov.c b/servers/slapd/overlays/syncprov.c index 6f8fb7f338..7114eb90f8 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-2010 The OpenLDAP Foundation. + * Copyright 2004-2011 The OpenLDAP Foundation. * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -113,6 +113,9 @@ typedef struct slog_entry { } slog_entry; typedef struct sessionlog { + BerVarray sl_mincsn; + int *sl_sids; + int sl_numcsns; int sl_num; int sl_size; slog_entry *sl_head; @@ -123,8 +126,8 @@ typedef struct sessionlog { /* The main state for this overlay */ typedef struct syncprov_info_t { syncops *si_ops; - BerVarray si_ctxcsn; /* ldapsync context */ struct berval si_contextdn; + BerVarray si_ctxcsn; /* ldapsync context */ int *si_sids; int si_numcsns; int si_chkops; /* checkpointing info */ @@ -224,10 +227,10 @@ syncprov_state_ctrl( if ( ret < 0 ) { Debug( LDAP_DEBUG_TRACE, - "slap_build_sync_ctrl: ber_flatten2 failed\n", - 0, 0, 0 ); + "slap_build_sync_ctrl: ber_flatten2 failed (%d)\n", + ret, 0, 0 ); send_ldap_error( op, rs, LDAP_OTHER, "internal error" ); - return ret; + return LDAP_OTHER; } return LDAP_SUCCESS; @@ -270,10 +273,10 @@ syncprov_done_ctrl( if ( ret < 0 ) { Debug( LDAP_DEBUG_TRACE, - "syncprov_done_ctrl: ber_flatten2 failed\n", - 0, 0, 0 ); + "syncprov_done_ctrl: ber_flatten2 failed (%d)\n", + ret, 0, 0 ); send_ldap_error( op, rs, LDAP_OTHER, "internal error" ); - return ret; + return LDAP_OTHER; } return LDAP_SUCCESS; @@ -337,10 +340,10 @@ syncprov_sendinfo( if ( ret < 0 ) { Debug( LDAP_DEBUG_TRACE, - "syncprov_sendinfo: ber_flatten2 failed\n", - 0, 0, 0 ); + "syncprov_sendinfo: ber_flatten2 failed (%d)\n", + ret, 0, 0 ); send_ldap_error( op, rs, LDAP_OTHER, "internal error" ); - return ret; + return LDAP_OTHER; } rs->sr_rspoid = LDAP_SYNC_INFO; @@ -585,7 +588,7 @@ findpres_cb( Operation *op, SlapReply *rs ) } static int -syncprov_findcsn( Operation *op, find_csn_t mode ) +syncprov_findcsn( Operation *op, find_csn_t mode, struct berval *csn ) { slap_overinst *on = (slap_overinst *)op->o_bd->bd_info; syncprov_info_t *si = on->on_bi.bi_private; @@ -658,15 +661,12 @@ again: break; case FIND_CSN: if ( BER_BVISEMPTY( &cf.f_av_value )) { - cf.f_av_value = srs->sr_state.ctxcsn[0]; - /* If there are multiple CSNs, use the smallest */ - for ( i=1; isr_state.numcsns; i++ ) { - if ( ber_bvcmp( &cf.f_av_value, &srs->sr_state.ctxcsn[i] ) - > 0 ) { - cf.f_av_value = srs->sr_state.ctxcsn[i]; - } - } + cf.f_av_value = *csn; } + fop.o_dn = op->o_bd->be_rootdn; + fop.o_ndn = op->o_bd->be_rootndn; + fop.o_req_dn = op->o_bd->be_suffix[0]; + fop.o_req_ndn = op->o_bd->be_nsuffix[0]; /* Look for exact match the first time */ if ( findcsn_retry ) { cf.f_choice = LDAP_FILTER_EQUALITY; @@ -1552,11 +1552,29 @@ syncprov_add_slog( Operation *op ) } else { sl->sl_head = se; sl->sl_tail = se; + if ( !sl->sl_mincsn ) { + sl->sl_numcsns = 1; + sl->sl_mincsn = ch_malloc( 2*sizeof( struct berval )); + sl->sl_sids = ch_malloc( sizeof( int )); + sl->sl_sids[0] = se->se_sid; + ber_dupbv( sl->sl_mincsn, &se->se_csn ); + BER_BVZERO( &sl->sl_mincsn[1] ); + } } sl->sl_num++; while ( sl->sl_num > sl->sl_size ) { + int i, j; se = sl->sl_head; sl->sl_head = se->se_next; + for ( i=0; isl_numcsns; i++ ) + if ( sl->sl_sids[i] >= se->se_sid ) + break; + if ( i == sl->sl_numcsns || sl->sl_sids[i] != se->se_sid ) { + slap_insert_csn_sids( (struct sync_cookie *)sl, + i, se->se_sid, &se->se_csn ); + } else { + ber_bvreplace( &sl->sl_mincsn[i], &se->se_csn ); + } ch_free( se ); sl->sl_num--; } @@ -1641,6 +1659,8 @@ syncprov_playlog( Operation *op, SlapReply *rs, sessionlog *sl, 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; } @@ -1790,6 +1810,8 @@ syncprov_op_response( Operation *op, SlapReply *rs ) #endif sid = slap_parse_csn_sid( &maxcsn ); for ( i=0; isi_numcsns; i++ ) { + if ( sid < si->si_sids[i] ) + break; if ( sid == si->si_sids[i] ) { if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn[i] ) > 0 ) { ber_bvreplace( &si->si_ctxcsn[i], &maxcsn ); @@ -1799,13 +1821,10 @@ syncprov_op_response( Operation *op, SlapReply *rs ) } } /* It's a new SID for us */ - if ( i == si->si_numcsns ) { - value_add_one( &si->si_ctxcsn, &maxcsn ); + if ( i == si->si_numcsns || sid != si->si_sids[i] ) { + slap_insert_csn_sids((struct sync_cookie *)&(si->si_ctxcsn), + i, sid, &maxcsn ); csn_changed = 1; - si->si_numcsns++; - si->si_sids = ch_realloc( si->si_sids, si->si_numcsns * - sizeof(int)); - si->si_sids[i] = sid; } } @@ -1819,11 +1838,14 @@ syncprov_op_response( Operation *op, SlapReply *rs ) * that changed, and only one can be passed in the csn queue. */ Modifications *mod = op->orm_modlist; - int i, j, sid; + 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] ) + break; 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] ); @@ -1833,12 +1855,9 @@ syncprov_op_response( Operation *op, SlapReply *rs ) } } - 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; + if ( j == si->si_numcsns || sid != si->si_sids[j] ) { + slap_insert_csn_sids( (struct sync_cookie *)&si->si_ctxcsn, + j, sid, &mod->sml_values[i] ); csn_changed = 1; } } @@ -1930,7 +1949,7 @@ syncprov_op_response( Operation *op, SlapReply *rs ) } /* Add any log records */ - if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) { + if ( si->si_logs ) { syncprov_add_slog( op ); } leave: ldap_pvt_thread_mutex_unlock( &si->si_resp_mutex ); @@ -2045,6 +2064,9 @@ syncprov_op_mod( Operation *op, SlapReply *rs ) opc->rsid = scook->sid; } + if ( op->o_dont_replicate ) + return SLAP_CB_CONTINUE; + /* If there are active persistent searches, lock this operation. * See seqmod.c for the locking logic on its own. */ @@ -2324,6 +2346,7 @@ syncprov_search_response( Operation *op, SlapReply *rs ) rs->sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2, op->o_tmpmemctx ); rs->sr_ctrls[1] = NULL; + rs->sr_flags |= REP_CTRLS_MUSTBEFREED; /* If we're in delta-sync mode, always send a cookie */ if ( si->si_nopres && si->si_usehint && a ) { struct berval cookie; @@ -2408,7 +2431,8 @@ syncprov_op_search( Operation *op, SlapReply *rs ) sync_control *srs; BerVarray ctxcsn; int i, *sids, numcsns; - struct berval mincsn; + struct berval mincsn, maxcsn; + int minsid, maxsid; int dirty = 0; if ( !(op->o_sync_mode & SLAP_SYNC_REFRESH) ) return SLAP_CB_CONTINUE; @@ -2504,96 +2528,133 @@ syncprov_op_search( Operation *op, SlapReply *rs ) /* If there are SIDs we don't recognize in the cookie, drop them */ for (i=0; isr_state.numcsns; ) { - for (j=0; jsr_state.sids[i] == sids[j] ) { + for (j=i; jsr_state.sids[i] <= sids[j] ) { break; } } /* not found */ - if ( j == numcsns ) { - struct berval tmp = srs->sr_state.ctxcsn[i]; - j = srs->sr_state.numcsns - 1; - srs->sr_state.ctxcsn[i] = srs->sr_state.ctxcsn[j]; - tmp.bv_len = 0; - srs->sr_state.ctxcsn[j] = tmp; - srs->sr_state.numcsns = j; - srs->sr_state.sids[i] = srs->sr_state.sids[j]; + if ( j == numcsns || srs->sr_state.sids[i] != sids[j] ) { + char *tmp = srs->sr_state.ctxcsn[i].bv_val; + srs->sr_state.numcsns--; + for ( j=i; jsr_state.numcsns; j++ ) { + srs->sr_state.ctxcsn[j] = srs->sr_state.ctxcsn[j+1]; + srs->sr_state.sids[j] = srs->sr_state.sids[j+1]; + } + srs->sr_state.ctxcsn[j].bv_val = tmp; + srs->sr_state.ctxcsn[j].bv_len = 0; continue; } i++; } - /* Find the smallest CSN */ - mincsn = srs->sr_state.ctxcsn[0]; - for ( i=1; isr_state.numcsns; i++ ) { - if ( ber_bvcmp( &mincsn, &srs->sr_state.ctxcsn[i] ) > 0 ) - mincsn = srs->sr_state.ctxcsn[i]; - } - - /* If nothing has changed, shortcut it */ - if ( srs->sr_state.numcsns == numcsns ) { - int i, j, newer; - 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 ) - 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!"; + if (srs->sr_state.numcsns != numcsns) { + /* consumer doesn't have the right number of CSNs */ + changed = SS_CHANGED; + if ( srs->sr_state.ctxcsn ) { + ber_bvarray_free_x( srs->sr_state.ctxcsn, op->o_tmpmemctx ); + srs->sr_state.ctxcsn = NULL; + } + if ( srs->sr_state.sids ) { + slap_sl_free( srs->sr_state.sids, op->o_tmpmemctx ); + srs->sr_state.sids = NULL; + } + srs->sr_state.numcsns = 0; + goto shortcut; + } + + /* Find the smallest CSN which differs from contextCSN */ + mincsn.bv_len = 0; + maxcsn.bv_len = 0; + for ( i=0,j=0; isr_state.numcsns; i++ ) { + int newer; + while ( srs->sr_state.sids[i] != sids[j] ) j++; + if ( BER_BVISEMPTY( &maxcsn ) || ber_bvcmp( &maxcsn, + &srs->sr_state.ctxcsn[i] ) < 0 ) { + maxcsn = srs->sr_state.ctxcsn[i]; + maxsid = sids[j]; + } + newer = ber_bvcmp( &srs->sr_state.ctxcsn[i], &ctxcsn[j] ); + /* If our state is newer, tell consumer about changes */ + if ( newer < 0) { + changed = SS_CHANGED; + if ( BER_BVISEMPTY( &mincsn ) || ber_bvcmp( &mincsn, + &srs->sr_state.ctxcsn[i] ) > 0 ) { + mincsn = srs->sr_state.ctxcsn[i]; + minsid = sids[j]; + } + } 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 ( 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 ); } - if ( changed ) - break; + rs->sr_ctrls = NULL; + send_ldap_result( op, rs ); + return rs->sr_err; } - if ( !changed && !dirty ) { - do_present = 0; + } + if ( BER_BVISEMPTY( &mincsn )) { + mincsn = maxcsn; + minsid = maxsid; + } + + /* If nothing has changed, shortcut it */ + if ( !changed && !dirty ) { + do_present = 0; no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) { - LDAPControl *ctrls[2]; - - ctrls[0] = NULL; - ctrls[1] = NULL; - syncprov_done_ctrl( op, rs, ctrls, 0, 0, - NULL, LDAP_SYNC_REFRESH_DELETES ); - rs->sr_ctrls = ctrls; - rs->sr_err = LDAP_SUCCESS; - send_ldap_result( op, rs ); - rs->sr_ctrls = NULL; - return rs->sr_err; - } - goto shortcut; + LDAPControl *ctrls[2]; + + ctrls[0] = NULL; + ctrls[1] = NULL; + syncprov_done_ctrl( op, rs, ctrls, 0, 0, + NULL, LDAP_SYNC_REFRESH_DELETES ); + rs->sr_ctrls = ctrls; + rs->sr_err = LDAP_SUCCESS; + send_ldap_result( op, rs ); + rs->sr_ctrls = NULL; + return rs->sr_err; } - } else { - /* consumer doesn't have the right number of CSNs */ - changed = SS_CHANGED; + goto shortcut; } + /* Do we have a sessionlog for this search? */ sl=si->si_logs; if ( sl ) { + int do_play = 0; ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); /* 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 ) { + int i; + for ( i=0; isl_numcsns; i++ ) { + /* SID not present == new enough */ + if ( minsid < sl->sl_sids[i] ) { + do_play = 1; + break; + } + /* SID present and new enough */ + if ( minsid == sl->sl_sids[i] + && ber_bvcmp( &mincsn, &sl->sl_mincsn[i] ) >= 0 ) { + do_play = 1; + break; + } + } + /* SID not present == new enough */ + if ( i == sl->sl_numcsns ) + do_play = 1; + } + if ( do_play ) { do_present = 0; /* mutex is unlocked in playlog */ syncprov_playlog( op, rs, sl, srs, ctxcsn, numcsns, sids ); @@ -2602,7 +2663,7 @@ no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) { } } /* Is the CSN still present in the database? */ - if ( syncprov_findcsn( op, FIND_CSN ) != LDAP_SUCCESS ) { + if ( syncprov_findcsn( op, FIND_CSN, &mincsn ) != LDAP_SUCCESS ) { /* No, so a reload is required */ /* the 2.2 consumer doesn't send this hint */ if ( si->si_usehint && srs->sr_rhint == 0 ) { @@ -2626,7 +2687,7 @@ no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) { } else { gotstate = 1; /* If changed and doing Present lookup, send Present UUIDs */ - if ( do_present && syncprov_findcsn( op, FIND_PRESENT ) != + if ( do_present && syncprov_findcsn( op, FIND_PRESENT, 0 ) != LDAP_SUCCESS ) { if ( ctxcsn ) ber_bvarray_free_x( ctxcsn, op->o_tmpmemctx ); @@ -2740,17 +2801,7 @@ syncprov_operational( } if ( !ap ) { - if ( !(rs->sr_flags & REP_ENTRY_MODIFIABLE) ) { - Entry *e = entry_dup( rs->sr_entry ); - if ( rs->sr_flags & REP_ENTRY_MUSTRELEASE ) { - overlay_entry_release_ov( op, rs->sr_entry, 0, on ); - rs->sr_flags ^= REP_ENTRY_MUSTRELEASE; - } else if ( rs->sr_flags & REP_ENTRY_MUSTBEFREED ) { - entry_free( rs->sr_entry ); - } - rs->sr_entry = e; - rs->sr_flags |= - REP_ENTRY_MODIFIABLE|REP_ENTRY_MUSTBEFREED; + if ( rs_entry2modifiable( op, rs, on )) { a = attr_find( rs->sr_entry->e_attrs, slap_schema.si_ad_contextCSN ); } @@ -2935,8 +2986,11 @@ sp_cf_gen(ConfigArgs *c) } sl = si->si_logs; if ( !sl ) { - sl = ch_malloc( sizeof( sessionlog ) ); + sl = ch_malloc( sizeof( sessionlog )); + sl->sl_mincsn = NULL; + sl->sl_sids = NULL; sl->sl_num = 0; + sl->sl_numcsns = 0; sl->sl_head = sl->sl_tail = NULL; ldap_pvt_thread_mutex_init( &sl->sl_mutex ); si->si_logs = sl; @@ -2962,10 +3016,11 @@ syncprov_db_otask( void *ptr ) { - syncprov_findcsn( ptr, FIND_MAXCSN ); + syncprov_findcsn( ptr, FIND_MAXCSN, 0 ); return NULL; } + /* Read any existing contextCSN from the underlying db. * Then search for any entries newer than that. If no value exists, * just generate it. Cache whatever result. @@ -3026,6 +3081,7 @@ syncprov_db_open( ber_bvarray_dup_x( &si->si_ctxcsn, a->a_vals, NULL ); si->si_numcsns = a->a_numvals; si->si_sids = slap_parse_csn_sids( si->si_ctxcsn, a->a_numvals, NULL ); + slap_sort_csn_sids( si->si_ctxcsn, si->si_sids, si->si_numcsns, NULL ); } overlay_entry_release_ov( op, e, 0, on ); if ( si->si_ctxcsn && !SLAP_DBCLEAN( be )) { @@ -3061,6 +3117,17 @@ syncprov_db_open( si->si_numops++; } + /* Initialize the sessionlog mincsn */ + if ( si->si_logs && si->si_numcsns ) { + sessionlog *sl = si->si_logs; + int i; + ber_bvarray_dup_x( &sl->sl_mincsn, si->si_ctxcsn, NULL ); + sl->sl_numcsns = si->si_numcsns; + sl->sl_sids = ch_malloc( si->si_numcsns * sizeof(int) ); + for ( i=0; i < si->si_numcsns; i++ ) + sl->sl_sids[i] = si->si_sids[i]; + } + out: op->o_bd->bd_info = (BackendInfo *)on; return 0; @@ -3160,14 +3227,19 @@ syncprov_db_destroy( if ( si ) { if ( si->si_logs ) { - slog_entry *se = si->si_logs->sl_head; + sessionlog *sl = si->si_logs; + slog_entry *se = sl->sl_head; while ( se ) { slog_entry *se_next = se->se_next; ch_free( se ); se = se_next; } - + if ( sl->sl_mincsn ) + ber_bvarray_free( sl->sl_mincsn ); + if ( sl->sl_sids ) + ch_free( sl->sl_sids ); + ldap_pvt_thread_mutex_destroy(&si->si_logs->sl_mutex); ch_free( si->si_logs ); }