version 1.147.2.1, 2006/05/15 17:04:44
|
version 1.147.2.2, 2006/08/17 23:50:32
|
Line 1
|
Line 1
|
/* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.150 2006/05/11 08:46:44 hyc Exp $ */ |
/* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.147.2.1 2006/05/15 17:04:44 kurt Exp $ */ |
/* syncprov.c - syncrepl provider */ |
/* syncprov.c - syncrepl provider */ |
/* This work is part of OpenLDAP Software <http://www.openldap.org/>. |
/* This work is part of OpenLDAP Software <http://www.openldap.org/>. |
* |
* |
Line 65 typedef struct syncops {
|
Line 65 typedef struct syncops {
|
#define PS_IS_DETACHED 0x02 |
#define PS_IS_DETACHED 0x02 |
#define PS_WROTE_BASE 0x04 |
#define PS_WROTE_BASE 0x04 |
#define PS_FIND_BASE 0x08 |
#define PS_FIND_BASE 0x08 |
|
#define PS_FIX_FILTER 0x10 |
|
|
int s_inuse; /* reference count */ |
int s_inuse; /* reference count */ |
struct syncres *s_res; |
struct syncres *s_res; |
Line 911 syncprov_qtask( void *ctx, void *arg )
|
Line 912 syncprov_qtask( void *ctx, void *arg )
|
return NULL; |
return NULL; |
} |
} |
|
|
|
/* Start the task to play back queued psearch responses */ |
|
static void |
|
syncprov_qstart( syncops *so ) |
|
{ |
|
int wake=0; |
|
ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex ); |
|
if ( !so->s_qtask ) { |
|
so->s_qtask = ldap_pvt_runqueue_insert( &slapd_rq, RUNQ_INTERVAL, |
|
syncprov_qtask, so, "syncprov_qtask", |
|
so->s_op->o_conn->c_peer_name.bv_val ); |
|
++so->s_inuse; |
|
wake = 1; |
|
} else { |
|
if (!ldap_pvt_runqueue_isrunning( &slapd_rq, so->s_qtask ) && |
|
!so->s_qtask->next_sched.tv_sec ) { |
|
so->s_qtask->interval.tv_sec = 0; |
|
ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 0 ); |
|
so->s_qtask->interval.tv_sec = RUNQ_INTERVAL; |
|
++so->s_inuse; |
|
wake = 1; |
|
} |
|
} |
|
ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex ); |
|
if ( wake ) |
|
slap_wake_listener(); |
|
} |
|
|
/* Queue a persistent search response */ |
/* Queue a persistent search response */ |
static int |
static int |
syncprov_qresp( opcookie *opc, syncops *so, int mode ) |
syncprov_qresp( opcookie *opc, syncops *so, int mode ) |
Line 949 syncprov_qresp( opcookie *opc, syncops *
|
Line 977 syncprov_qresp( opcookie *opc, syncops *
|
so->s_flags |= PS_FIND_BASE; |
so->s_flags |= PS_FIND_BASE; |
} |
} |
if ( so->s_flags & PS_IS_DETACHED ) { |
if ( so->s_flags & PS_IS_DETACHED ) { |
int wake=0; |
syncprov_qstart( so ); |
ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex ); |
|
if ( !so->s_qtask ) { |
|
so->s_qtask = ldap_pvt_runqueue_insert( &slapd_rq, RUNQ_INTERVAL, |
|
syncprov_qtask, so, "syncprov_qtask", |
|
so->s_op->o_conn->c_peer_name.bv_val ); |
|
++so->s_inuse; |
|
wake = 1; |
|
} else { |
|
if (!ldap_pvt_runqueue_isrunning( &slapd_rq, so->s_qtask ) && |
|
!so->s_qtask->next_sched.tv_sec ) { |
|
so->s_qtask->interval.tv_sec = 0; |
|
ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 0 ); |
|
so->s_qtask->interval.tv_sec = RUNQ_INTERVAL; |
|
++so->s_inuse; |
|
wake = 1; |
|
} |
|
} |
|
ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex ); |
|
if ( wake ) |
|
slap_wake_listener(); |
|
} |
} |
ldap_pvt_thread_mutex_unlock( &so->s_mutex ); |
ldap_pvt_thread_mutex_unlock( &so->s_mutex ); |
return LDAP_SUCCESS; |
return LDAP_SUCCESS; |
Line 1055 syncprov_matchops( Operation *op, opcook
|
Line 1063 syncprov_matchops( Operation *op, opcook
|
int rc; |
int rc; |
struct berval newdn; |
struct berval newdn; |
int freefdn = 0; |
int freefdn = 0; |
|
BackendDB *b0 = op->o_bd, db; |
|
|
fc.fdn = &op->o_req_ndn; |
fc.fdn = &op->o_req_ndn; |
/* compute new DN */ |
/* compute new DN */ |
Line 1067 syncprov_matchops( Operation *op, opcook
|
Line 1076 syncprov_matchops( Operation *op, opcook
|
freefdn = 1; |
freefdn = 1; |
} |
} |
if ( op->o_tag != LDAP_REQ_ADD ) { |
if ( op->o_tag != LDAP_REQ_ADD ) { |
|
if ( !SLAP_ISOVERLAY( op->o_bd )) { |
|
db = *op->o_bd; |
|
op->o_bd = &db; |
|
} |
op->o_bd->bd_info = (BackendInfo *)on->on_info; |
op->o_bd->bd_info = (BackendInfo *)on->on_info; |
rc = be_entry_get_rw( op, fc.fdn, NULL, NULL, 0, &e ); |
rc = be_entry_get_rw( op, fc.fdn, NULL, NULL, 0, &e ); |
/* If we're sending responses now, make a copy and unlock the DB */ |
/* If we're sending responses now, make a copy and unlock the DB */ |
Line 1076 syncprov_matchops( Operation *op, opcook
|
Line 1089 syncprov_matchops( Operation *op, opcook
|
e = e2; |
e = e2; |
} |
} |
op->o_bd->bd_info = (BackendInfo *)on; |
op->o_bd->bd_info = (BackendInfo *)on; |
if ( rc ) return; |
if ( rc ) { |
|
op->o_bd = b0; |
|
return; |
|
} |
} else { |
} else { |
e = op->ora_e; |
e = op->ora_e; |
} |
} |
Line 1174 syncprov_matchops( Operation *op, opcook
|
Line 1190 syncprov_matchops( Operation *op, opcook
|
if ( freefdn ) { |
if ( freefdn ) { |
op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx ); |
op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx ); |
} |
} |
|
op->o_bd = b0; |
} |
} |
|
|
static int |
static int |
Line 1235 syncprov_checkpoint( Operation *op, Slap
|
Line 1252 syncprov_checkpoint( Operation *op, Slap
|
struct berval bv[2]; |
struct berval bv[2]; |
slap_callback cb = {0}; |
slap_callback cb = {0}; |
|
|
mod.sml_values = bv; |
/* If ctxcsn is empty, delete it */ |
bv[1].bv_val = NULL; |
if ( BER_BVISEMPTY( &si->si_ctxcsn )) { |
bv[0] = si->si_ctxcsn; |
mod.sml_values = NULL; |
|
} else { |
|
mod.sml_values = bv; |
|
bv[1].bv_val = NULL; |
|
bv[0] = si->si_ctxcsn; |
|
} |
mod.sml_nvalues = NULL; |
mod.sml_nvalues = NULL; |
mod.sml_desc = slap_schema.si_ad_contextCSN; |
mod.sml_desc = slap_schema.si_ad_contextCSN; |
mod.sml_op = LDAP_MOD_REPLACE; |
mod.sml_op = LDAP_MOD_REPLACE; |
Line 1262 syncprov_checkpoint( Operation *op, Slap
|
Line 1284 syncprov_checkpoint( Operation *op, Slap
|
} |
} |
|
|
static void |
static void |
syncprov_add_slog( Operation *op, struct berval *csn ) |
syncprov_add_slog( Operation *op ) |
{ |
{ |
opcookie *opc = op->o_callback->sc_private; |
opcookie *opc = op->o_callback->sc_private; |
slap_overinst *on = opc->son; |
slap_overinst *on = opc->son; |
Line 1274 syncprov_add_slog( Operation *op, struct
|
Line 1296 syncprov_add_slog( Operation *op, struct
|
{ |
{ |
/* Allocate a record. UUIDs are not NUL-terminated. */ |
/* 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 + |
csn->bv_len + 1 ); |
op->o_csn.bv_len + 1 ); |
se->se_next = NULL; |
se->se_next = NULL; |
se->se_tag = op->o_tag; |
se->se_tag = op->o_tag; |
|
|
Line 1283 syncprov_add_slog( Operation *op, struct
|
Line 1305 syncprov_add_slog( Operation *op, struct
|
se->se_uuid.bv_len = opc->suuid.bv_len; |
se->se_uuid.bv_len = opc->suuid.bv_len; |
|
|
se->se_csn.bv_val = se->se_uuid.bv_val + opc->suuid.bv_len; |
se->se_csn.bv_val = se->se_uuid.bv_val + opc->suuid.bv_len; |
AC_MEMCPY( se->se_csn.bv_val, csn->bv_val, csn->bv_len ); |
AC_MEMCPY( se->se_csn.bv_val, op->o_csn.bv_val, op->o_csn.bv_len ); |
se->se_csn.bv_val[csn->bv_len] = '\0'; |
se->se_csn.bv_val[op->o_csn.bv_len] = '\0'; |
se->se_csn.bv_len = csn->bv_len; |
se->se_csn.bv_len = op->o_csn.bv_len; |
|
|
ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); |
ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); |
if ( sl->sl_head ) { |
if ( sl->sl_head ) { |
Line 1357 syncprov_playlog( Operation *op, SlapRep
|
Line 1379 syncprov_playlog( Operation *op, SlapRep
|
i++; |
i++; |
AC_MEMCPY( cbuf, se->se_csn.bv_val, se->se_csn.bv_len ); |
AC_MEMCPY( cbuf, se->se_csn.bv_val, se->se_csn.bv_len ); |
delcsn.bv_len = se->se_csn.bv_len; |
delcsn.bv_len = se->se_csn.bv_len; |
|
delcsn.bv_val[delcsn.bv_len] = '\0'; |
} else { |
} else { |
nmods++; |
nmods++; |
j = num - nmods; |
j = num - nmods; |
Line 1468 syncprov_op_response( Operation *op, Sla
|
Line 1491 syncprov_op_response( Operation *op, Sla
|
|
|
if ( rs->sr_err == LDAP_SUCCESS ) |
if ( rs->sr_err == LDAP_SUCCESS ) |
{ |
{ |
struct berval maxcsn = BER_BVNULL, curcsn = BER_BVNULL; |
struct berval maxcsn = BER_BVNULL; |
char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE]; |
char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE]; |
|
|
/* Update our context CSN */ |
/* Update our context CSN */ |
cbuf[0] = '\0'; |
cbuf[0] = '\0'; |
ldap_pvt_thread_mutex_lock( &si->si_csn_mutex ); |
ldap_pvt_thread_mutex_lock( &si->si_csn_mutex ); |
slap_get_commit_csn( op, &maxcsn, &curcsn ); |
slap_get_commit_csn( op, &maxcsn ); |
if ( !BER_BVISNULL( &maxcsn ) ) { |
if ( !BER_BVISNULL( &maxcsn ) ) { |
strcpy( cbuf, maxcsn.bv_val ); |
strcpy( cbuf, maxcsn.bv_val ); |
if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn ) > 0 ) { |
if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn ) > 0 ) { |
Line 1535 syncprov_op_response( Operation *op, Sla
|
Line 1558 syncprov_op_response( Operation *op, Sla
|
|
|
/* Add any log records */ |
/* Add any log records */ |
if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) { |
if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) { |
syncprov_add_slog( op, &curcsn ); |
syncprov_add_slog( op ); |
} |
} |
|
|
} |
} |
Line 1770 syncprov_detach_op( Operation *op, synco
|
Line 1793 syncprov_detach_op( Operation *op, synco
|
op2->ors_filterstr.bv_val = ptr; |
op2->ors_filterstr.bv_val = ptr; |
strcpy( ptr, so->s_filterstr.bv_val ); |
strcpy( ptr, so->s_filterstr.bv_val ); |
op2->ors_filterstr.bv_len = so->s_filterstr.bv_len; |
op2->ors_filterstr.bv_len = so->s_filterstr.bv_len; |
op2->ors_filter = filter_dup( op->ors_filter, NULL ); |
|
|
/* Skip the AND/GE clause that we stuck on in front */ |
|
if ( so->s_flags & PS_FIX_FILTER ) { |
|
op2->ors_filter = op->ors_filter->f_and->f_next; |
|
so->s_flags ^= PS_FIX_FILTER; |
|
} else { |
|
op2->ors_filter = op->ors_filter; |
|
} |
|
op2->ors_filter = filter_dup( op2->ors_filter, NULL ); |
so->s_op = op2; |
so->s_op = op2; |
|
|
/* Copy any cached group ACLs individually */ |
/* Copy any cached group ACLs individually */ |
Line 1862 syncprov_search_response( Operation *op,
|
Line 1893 syncprov_search_response( Operation *op,
|
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); |
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); |
} else { |
} else { |
/* It's RefreshAndPersist, transition to Persist phase */ |
/* It's RefreshAndPersist, transition to Persist phase */ |
syncprov_sendinfo( op, rs, ( ss->ss_present && rs->sr_nentries ) ? |
syncprov_sendinfo( op, rs, ss->ss_present ? |
LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE, |
LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE, |
&cookie, 1, NULL, 0 ); |
&cookie, 1, NULL, 0 ); |
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); |
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx ); |
Line 1874 syncprov_search_response( Operation *op,
|
Line 1905 syncprov_search_response( Operation *op,
|
ss->ss_so->s_flags ^= PS_IS_REFRESHING; |
ss->ss_so->s_flags ^= PS_IS_REFRESHING; |
|
|
syncprov_detach_op( op, ss->ss_so, on ); |
syncprov_detach_op( op, ss->ss_so, on ); |
|
|
|
/* If there are queued responses, fire them off */ |
|
if ( ss->ss_so->s_res ) |
|
syncprov_qstart( ss->ss_so ); |
ldap_pvt_thread_mutex_unlock( &ss->ss_so->s_mutex ); |
ldap_pvt_thread_mutex_unlock( &ss->ss_so->s_mutex ); |
|
|
return LDAP_SUCCESS; |
return LDAP_SUCCESS; |
Line 1981 syncprov_op_search( Operation *op, SlapR
|
Line 2016 syncprov_op_search( Operation *op, SlapR
|
sl=si->si_logs; |
sl=si->si_logs; |
if ( sl ) { |
if ( sl ) { |
ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); |
ldap_pvt_thread_mutex_lock( &sl->sl_mutex ); |
if ( ber_bvcmp( &srs->sr_state.ctxcsn, &sl->sl_mincsn ) >= 0 ) { |
/* Are there any log entries, and is the consumer state |
|
* present in the session log? |
|
*/ |
|
if ( sl->sl_num > 0 && ber_bvcmp( &srs->sr_state.ctxcsn, &sl->sl_mincsn ) >= 0 ) { |
do_present = 0; |
do_present = 0; |
/* mutex is unlocked in playlog */ |
/* mutex is unlocked in playlog */ |
syncprov_playlog( op, rs, sl, srs, &ctxcsn ); |
syncprov_playlog( op, rs, sl, srs, &ctxcsn ); |
Line 2035 shortcut:
|
Line 2073 shortcut:
|
fava->f_next = op->ors_filter; |
fava->f_next = op->ors_filter; |
op->ors_filter = fand; |
op->ors_filter = fand; |
filter2bv_x( op, op->ors_filter, &op->ors_filterstr ); |
filter2bv_x( op, op->ors_filter, &op->ors_filterstr ); |
|
if ( sop ) |
|
sop->s_flags |= PS_FIX_FILTER; |
} |
} |
|
|
/* Let our callback add needed info to returned entries */ |
/* Let our callback add needed info to returned entries */ |
Line 2317 syncprov_db_open(
|
Line 2357 syncprov_db_open(
|
int rc; |
int rc; |
void *thrctx = NULL; |
void *thrctx = NULL; |
|
|
|
if ( !SLAP_LASTMOD( be )) { |
|
Debug( LDAP_DEBUG_ANY, |
|
"syncprov_db_open: invalid config, lastmod must be enabled\n", 0, 0, 0 ); |
|
return -1; |
|
} |
|
|
if ( slapMode & SLAP_TOOL_MODE ) { |
if ( slapMode & SLAP_TOOL_MODE ) { |
return 0; |
return 0; |
} |
} |
Line 2360 syncprov_db_open(
|
Line 2406 syncprov_db_open(
|
ldap_pvt_thread_create( &tid, 0, syncprov_db_otask, op ); |
ldap_pvt_thread_create( &tid, 0, syncprov_db_otask, op ); |
ldap_pvt_thread_join( tid, NULL ); |
ldap_pvt_thread_join( tid, NULL ); |
} |
} |
} else if ( SLAP_SYNC_SHADOW( op->o_bd )) { |
|
/* If we're also a consumer, and we didn't find the context entry, |
|
* then don't generate anything, wait for our provider to send it |
|
* to us. |
|
*/ |
|
goto out; |
|
} |
} |
|
|
if ( BER_BVISEMPTY( &si->si_ctxcsn ) ) { |
if ( BER_BVISEMPTY( &si->si_ctxcsn ) ) { |
|
if ( SLAP_SYNC_SHADOW( op->o_bd )) { |
|
/* If we're also a consumer, and we didn't get a contextCSN, |
|
* then don't generate anything, wait for our provider to send it |
|
* to us. |
|
*/ |
|
goto out; |
|
} |
si->si_ctxcsn.bv_len = sizeof( si->si_ctxcsnbuf ); |
si->si_ctxcsn.bv_len = sizeof( si->si_ctxcsnbuf ); |
slap_get_csn( op, &si->si_ctxcsn, 0 ); |
slap_get_csn( op, &si->si_ctxcsn, 0 ); |
} |
} |