Diff for /servers/slapd/overlays/syncprov.c between versions 1.56.2.39 and 1.153

version 1.56.2.39, 2006/12/16 00:26:41 version 1.153, 2006/07/27 08:44:23
Line 1 Line 1
 /* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.56.2.38 2006/10/31 18:32:36 hyc Exp $ */  /* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.152 2006/07/23 22:32:27 hyc 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 912  syncprov_qtask( void *ctx, void *arg ) Line 911  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 977  syncprov_qresp( opcookie *opc, syncops * Line 949  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 ) {
                 syncprov_qstart( 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();
         }          }
         ldap_pvt_thread_mutex_unlock( &so->s_mutex );          ldap_pvt_thread_mutex_unlock( &so->s_mutex );
         return LDAP_SUCCESS;          return LDAP_SUCCESS;
Line 1243  syncprov_op_cleanup( Operation *op, Slap Line 1235  syncprov_op_cleanup( Operation *op, Slap
 }  }
   
 static void  static void
 syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on,  syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on )
         struct berval *csn )  
 {  {
           syncprov_info_t         *si = on->on_bi.bi_private;
         Modifications mod;          Modifications mod;
         Operation opm;          Operation opm;
         SlapReply rsm = { 0 };          SlapReply rsm = { 0 };
Line 1253  syncprov_checkpoint( Operation *op, Slap Line 1245  syncprov_checkpoint( Operation *op, Slap
         slap_callback cb = {0};          slap_callback cb = {0};
   
         /* If ctxcsn is empty, delete it */          /* If ctxcsn is empty, delete it */
         if ( BER_BVISEMPTY( csn )) {          if ( BER_BVISEMPTY( &si->si_ctxcsn )) {
                 mod.sml_values = NULL;                  mod.sml_values = NULL;
         } else {          } else {
                 mod.sml_values = bv;                  mod.sml_values = bv;
                 bv[1].bv_val = NULL;                  bv[1].bv_val = NULL;
                 bv[0] = *csn;                  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;
Line 1284  syncprov_checkpoint( Operation *op, Slap Line 1276  syncprov_checkpoint( Operation *op, Slap
 }  }
   
 static void  static void
 syncprov_add_slog( Operation *op )  syncprov_add_slog( Operation *op, struct berval *csn )
 {  {
         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 1296  syncprov_add_slog( Operation *op ) Line 1288  syncprov_add_slog( Operation *op )
         {          {
                 /* 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 + 
                         op->o_csn.bv_len + 1 );                          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 1305  syncprov_add_slog( Operation *op ) Line 1297  syncprov_add_slog( Operation *op )
                 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, op->o_csn.bv_val, op->o_csn.bv_len );                  AC_MEMCPY( se->se_csn.bv_val, csn->bv_val, csn->bv_len );
                 se->se_csn.bv_val[op->o_csn.bv_len] = '\0';                  se->se_csn.bv_val[csn->bv_len] = '\0';
                 se->se_csn.bv_len = op->o_csn.bv_len;                  se->se_csn.bv_len = 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 1491  syncprov_op_response( Operation *op, Sla Line 1483  syncprov_op_response( Operation *op, Sla
   
         if ( rs->sr_err == LDAP_SUCCESS )          if ( rs->sr_err == LDAP_SUCCESS )
         {          {
                 struct berval maxcsn = BER_BVNULL;                  struct berval maxcsn = BER_BVNULL, curcsn = BER_BVNULL;
                 char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];                  char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
                 int do_check = 0;  
   
                 /* 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 );                  slap_get_commit_csn( op, &maxcsn, &curcsn );
                 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 1516  syncprov_op_response( Operation *op, Sla Line 1507  syncprov_op_response( Operation *op, Sla
   
                 si->si_numops++;                  si->si_numops++;
                 if ( si->si_chkops || si->si_chktime ) {                  if ( si->si_chkops || si->si_chktime ) {
                           int do_check=0;
                         if ( si->si_chkops && si->si_numops >= si->si_chkops ) {                          if ( si->si_chkops && si->si_numops >= si->si_chkops ) {
                                 do_check = 1;                                  do_check = 1;
                                 si->si_numops = 0;                                  si->si_numops = 0;
Line 1525  syncprov_op_response( Operation *op, Sla Line 1517  syncprov_op_response( Operation *op, Sla
                                 do_check = 1;                                  do_check = 1;
                                 si->si_chklast = op->o_time;                                  si->si_chklast = op->o_time;
                         }                          }
                           if ( do_check ) {
                                   syncprov_checkpoint( op, rs, on );
                           }
                 }                  }
                 ldap_pvt_thread_mutex_unlock( &si->si_csn_mutex );                  ldap_pvt_thread_mutex_unlock( &si->si_csn_mutex );
   
                 opc->sctxcsn.bv_len = maxcsn.bv_len;                  opc->sctxcsn.bv_len = maxcsn.bv_len;
                 opc->sctxcsn.bv_val = cbuf;                  opc->sctxcsn.bv_val = cbuf;
   
                 if ( do_check ) {  
                         syncprov_checkpoint( op, rs, on, &opc->sctxcsn );  
                 }  
   
                 /* Handle any persistent searches */                  /* Handle any persistent searches */
                 if ( si->si_ops ) {                  if ( si->si_ops ) {
                         switch(op->o_tag) {                          switch(op->o_tag) {
Line 1559  syncprov_op_response( Operation *op, Sla Line 1550  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 );                          syncprov_add_slog( op, &curcsn );
                 }                  }
   
         }          }
Line 1794  syncprov_detach_op( Operation *op, synco Line 1785  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 1894  syncprov_search_response( Operation *op, Line 1877  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 ?                          syncprov_sendinfo( op, rs, ( ss->ss_present && rs->sr_nentries ) ?
                                 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 1906  syncprov_search_response( Operation *op, Line 1889  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 2017  syncprov_op_search( Operation *op, SlapR Line 1996  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 );
                         /* Are there any log entries, and is the consumer state                          if ( ber_bvcmp( &srs->sr_state.ctxcsn, &sl->sl_mincsn ) >= 0 ) {
                          * 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 2074  shortcut: Line 2050  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 2459  syncprov_db_close( Line 2433  syncprov_db_close(
                 op->o_bd = be;                  op->o_bd = be;
                 op->o_dn = be->be_rootdn;                  op->o_dn = be->be_rootdn;
                 op->o_ndn = be->be_rootndn;                  op->o_ndn = be->be_rootndn;
                 syncprov_checkpoint( op, &rs, on, &si->si_ctxcsn );                  syncprov_checkpoint( op, &rs, on );
                 ldap_pvt_thread_pool_context_reset( thrctx );                  ldap_pvt_thread_pool_context_reset( thrctx );
         }          }
   
Line 2474  syncprov_db_init( Line 2448  syncprov_db_init(
         slap_overinst   *on = (slap_overinst *)be->bd_info;          slap_overinst   *on = (slap_overinst *)be->bd_info;
         syncprov_info_t *si;          syncprov_info_t *si;
   
         if ( SLAP_ISGLOBALOVERLAY( be ) ) {  
                 Debug( LDAP_DEBUG_ANY,  
                         "syncprov must be instantiated within a database.\n",  
                         0, 0, 0 );  
                 return 1;  
         }  
   
         si = ch_calloc(1, sizeof(syncprov_info_t));          si = ch_calloc(1, sizeof(syncprov_info_t));
         on->on_bi.bi_private = si;          on->on_bi.bi_private = si;
         ldap_pvt_thread_mutex_init( &si->si_csn_mutex );          ldap_pvt_thread_mutex_init( &si->si_csn_mutex );

Removed from v.1.56.2.39  
changed lines
  Added in v.1.153


______________
© Copyright 1998-2020, OpenLDAP Foundation, info@OpenLDAP.org