Diff for /servers/slapd/overlays/syncprov.c between versions 1.147.2.77 and 1.147.2.89

version 1.147.2.77, 2010/12/12 03:54:10 version 1.147.2.89, 2011/01/26 18:10:41
Line 2 Line 2
 /* 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/>.
  *   *
  * Copyright 2004-2010 The OpenLDAP Foundation.   * Copyright 2004-2011 The OpenLDAP Foundation.
  * All rights reserved.   * All rights reserved.
  *   *
  * Redistribution and use in source and binary forms, with or without   * Redistribution and use in source and binary forms, with or without
Line 133  typedef struct syncprov_info_t { Line 133  typedef struct syncprov_info_t {
         int             si_numops;      /* number of ops since last checkpoint */          int             si_numops;      /* number of ops since last checkpoint */
         int             si_nopres;      /* Skip present phase */          int             si_nopres;      /* Skip present phase */
         int             si_usehint;     /* use reload hint */          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 */          time_t  si_chklast;     /* time of last checkpoint */
         Avlnode *si_mods;       /* entries being modified */          Avlnode *si_mods;       /* entries being modified */
         sessionlog      *si_logs;          sessionlog      *si_logs;
Line 783  syncprov_free_syncop( syncops *so ) Line 786  syncprov_free_syncop( syncops *so )
         GroupAssertion *ga, *gnext;          GroupAssertion *ga, *gnext;
   
         ldap_pvt_thread_mutex_lock( &so->s_mutex );          ldap_pvt_thread_mutex_lock( &so->s_mutex );
         if ( --so->s_inuse > 0 ) {          /* already being freed, or still in use */
           if ( !so->s_inuse || --so->s_inuse > 0 ) {
                 ldap_pvt_thread_mutex_unlock( &so->s_mutex );                  ldap_pvt_thread_mutex_unlock( &so->s_mutex );
                 return;                  return;
         }          }
Line 1376  syncprov_op_cleanup( Operation *op, Slap Line 1380  syncprov_op_cleanup( Operation *op, Slap
         syncmatches *sm, *snext;          syncmatches *sm, *snext;
         modtarget *mt, mtdummy;          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) {          for (sm = opc->smatches; sm; sm=snext) {
                 snext = sm->sm_next;                  snext = sm->sm_next;
                 syncprov_free_syncop( sm->sm_op );                  syncprov_free_syncop( sm->sm_op );
Line 1492  syncprov_add_slog( Operation *op ) Line 1501  syncprov_add_slog( Operation *op )
   
         sl = si->si_logs;          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. */                  /* 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 );                          op->o_csn.bv_len + 1 );
Line 1510  syncprov_add_slog( Operation *op ) Line 1536  syncprov_add_slog( Operation *op )
   
                 ldap_pvt_thread_mutex_lock( &sl->sl_mutex );                  ldap_pvt_thread_mutex_lock( &sl->sl_mutex );
                 if ( sl->sl_head ) {                  if ( sl->sl_head ) {
                         sl->sl_tail->se_next = se;                          /* 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;
                                           }
                                   }
                           }
                 } else {                  } else {
                         sl->sl_head = se;                          sl->sl_head = se;
                           sl->sl_tail = se;
                 }                  }
                 sl->sl_tail = se;  
                 sl->sl_num++;                  sl->sl_num++;
                 while ( sl->sl_num > sl->sl_size ) {                  while ( sl->sl_num > sl->sl_size ) {
                         se = sl->sl_head;                          se = sl->sl_head;
                         sl->sl_head = se->se_next;                          sl->sl_head = se->se_next;
                         strcpy( sl->sl_mincsn.bv_val, se->se_csn.bv_val );                          AC_MEMCPY( sl->sl_mincsn.bv_val, se->se_csn.bv_val, se->se_csn.bv_len );
                         sl->sl_mincsn.bv_len = se->se_csn.bv_len;                          sl->sl_mincsn.bv_len = se->se_csn.bv_len;
                         ch_free( se );                          ch_free( se );
                         sl->sl_num--;                          sl->sl_num--;
Line 1605  syncprov_playlog( Operation *op, SlapRep Line 1645  syncprov_playlog( Operation *op, SlapRep
                         delcsn[0].bv_len = se->se_csn.bv_len;                          delcsn[0].bv_len = se->se_csn.bv_len;
                         delcsn[0].bv_val[delcsn[0].bv_len] = '\0';                          delcsn[0].bv_val[delcsn[0].bv_len] = '\0';
                 } else {                  } else {
                           if ( se->se_tag == LDAP_REQ_ADD )
                                   continue;
                         nmods++;                          nmods++;
                         j = num - nmods;                          j = num - nmods;
                 }                  }
Line 1783  syncprov_op_response( Operation *op, Sla Line 1825  syncprov_op_response( Operation *op, Sla
                          * that changed, and only one can be passed in the csn queue.                           * that changed, and only one can be passed in the csn queue.
                          */                           */
                         Modifications *mod = op->orm_modlist;                          Modifications *mod = op->orm_modlist;
                         int i, j, sid;                          unsigned i;
                           int j, sid;
   
                         for ( i=0; i<mod->sml_numvals; i++ ) {                          for ( i=0; i<mod->sml_numvals; i++ ) {
                                 sid = slap_parse_csn_sid( &mod->sml_values[i] );                                  sid = slap_parse_csn_sid( &mod->sml_values[i] );
Line 1806  syncprov_op_response( Operation *op, Sla Line 1849  syncprov_op_response( Operation *op, Sla
                                         csn_changed = 1;                                          csn_changed = 1;
                                 }                                  }
                         }                          }
                           if ( csn_changed )
                                   si->si_dirty = 0;
                         ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );                          ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
   
                         if ( csn_changed ) {                          if ( csn_changed ) {
                                   syncops *ss;
                                 ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );                                  ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
                                 have_psearches = ( si->si_ops != NULL );                                  for ( ss = si->si_ops; ss; ss = ss->s_next ) {
                                 ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );                                          if ( ss->s_op->o_abandon )
                                                   continue;
                                 if ( have_psearches ) {                                          /* Send the updated csn to all syncrepl consumers,
                                         for ( sm = opc->smatches; sm; sm=sm->sm_next ) {                                           * including the server from which it originated.
                                                 if ( sm->sm_op->s_op->o_abandon )                                           * The syncrepl consumer and syncprov provider on
                                                         continue;                                           * the originating server may be configured to store
                                                 syncprov_qresp( opc, sm->sm_op, LDAP_SYNC_NEW_COOKIE );                                           * their csn values in different entries.
                                         }                                           */
                                           syncprov_qresp( opc, ss, LDAP_SYNC_NEW_COOKIE );
                                 }                                  }
                                   ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
                         }                          }
                         } else {                          } else {
                         ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );                          ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
Line 1849  syncprov_op_response( Operation *op, Sla Line 1897  syncprov_op_response( Operation *op, Sla
                                 }                                  }
                         }                          }
                 }                  }
                   si->si_dirty = !csn_changed;
                 ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );                  ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
   
                 if ( do_check ) {                  if ( do_check ) {
Line 1888  syncprov_op_response( Operation *op, Sla Line 1937  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 ) {
                         syncprov_add_slog( op );                          syncprov_add_slog( op );
                 }                  }
 leave:          ldap_pvt_thread_mutex_unlock( &si->si_resp_mutex );  leave:          ldap_pvt_thread_mutex_unlock( &si->si_resp_mutex );
Line 1976  syncprov_op_mod( Operation *op, SlapRepl Line 2025  syncprov_op_mod( Operation *op, SlapRepl
   
         ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );          ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
         have_psearches = ( si->si_ops != NULL );          have_psearches = ( si->si_ops != NULL );
           si->si_active++;
         ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );          ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
   
         cbsize = sizeof(slap_callback) + sizeof(opcookie) +          cbsize = sizeof(slap_callback) + sizeof(opcookie) +
Line 2281  syncprov_search_response( Operation *op, Line 2331  syncprov_search_response( Operation *op,
                 rs->sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2,                  rs->sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2,
                         op->o_tmpmemctx );                          op->o_tmpmemctx );
                 rs->sr_ctrls[1] = NULL;                  rs->sr_ctrls[1] = NULL;
                   rs->sr_flags |= REP_CTRLS_MUSTBEFREED;
                 /* If we're in delta-sync mode, always send a cookie */                  /* If we're in delta-sync mode, always send a cookie */
                 if ( si->si_nopres && si->si_usehint && a ) {                  if ( si->si_nopres && si->si_usehint && a ) {
                         struct berval cookie;                          struct berval cookie;
Line 2365  syncprov_op_search( Operation *op, SlapR Line 2416  syncprov_op_search( Operation *op, SlapR
         sync_control *srs;          sync_control *srs;
         BerVarray ctxcsn;          BerVarray ctxcsn;
         int i, *sids, numcsns;          int i, *sids, numcsns;
         struct berval mincsn;          struct berval mincsn, maxcsn;
           int dirty = 0;
   
         if ( !(op->o_sync_mode & SLAP_SYNC_REFRESH) ) return SLAP_CB_CONTINUE;          if ( !(op->o_sync_mode & SLAP_SYNC_REFRESH) ) return SLAP_CB_CONTINUE;
   
Line 2410  syncprov_op_search( Operation *op, SlapR Line 2462  syncprov_op_search( Operation *op, SlapR
                 sop->s_inuse = 1;                  sop->s_inuse = 1;
   
                 ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );                  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;                  sop->s_next = si->si_ops;
                 si->si_ops = sop;                  si->si_ops = sop;
                 ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );                  ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
Line 2427  syncprov_op_search( Operation *op, SlapR Line 2493  syncprov_op_search( Operation *op, SlapR
                 ctxcsn = NULL;                  ctxcsn = NULL;
                 sids = NULL;                  sids = NULL;
         }          }
           dirty = si->si_dirty;
         ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );          ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
                   
         /* If we have a cookie, handle the PRESENT lookups */          /* If we have a cookie, handle the PRESENT lookups */
Line 2464  syncprov_op_search( Operation *op, SlapR Line 2531  syncprov_op_search( Operation *op, SlapR
                         i++;                          i++;
                 }                  }
   
                 /* Find the smallest CSN */                  /* Find the smallest CSN which differs from contextCSN */
                 mincsn = srs->sr_state.ctxcsn[0];                  mincsn.bv_len = 0;
                 for ( i=1; i<srs->sr_state.numcsns; i++ ) {                  maxcsn.bv_len = 0;
                         if ( ber_bvcmp( &mincsn, &srs->sr_state.ctxcsn[i] ) > 0 )                  for ( i=0; i<srs->sr_state.numcsns; i++ ) {
                                 mincsn = srs->sr_state.ctxcsn[i];                          for ( j=0; j<numcsns; j++ ) {
                 }                                  if ( srs->sr_state.sids[i] != sids[j] )
                                           continue;
                                   if ( BER_BVISEMPTY( &maxcsn ) || ber_bvcmp( &maxcsn,
                                           &srs->sr_state.ctxcsn[i] ) < 0 ) {
                                           maxcsn = srs->sr_state.ctxcsn[i];
                                   }
                                   if ( ber_bvcmp( &srs->sr_state.ctxcsn[i], &ctxcsn[j] ) < 0) {
                                           if ( BER_BVISEMPTY( &mincsn ) || ber_bvcmp( &mincsn,
                                                   &srs->sr_state.ctxcsn[i] ) > 0 ) {
                                                   mincsn = srs->sr_state.ctxcsn[i];
                                           }
                                   }
                           }
                   }
                   if ( BER_BVISEMPTY( &mincsn ))
                           mincsn = maxcsn;
   
                 /* If nothing has changed, shortcut it */                  /* If nothing has changed, shortcut it */
                 if ( srs->sr_state.numcsns == numcsns ) {                  if ( srs->sr_state.numcsns == numcsns ) {
Line 2506  bailout: Line 2588  bailout:
                                 if ( changed )                                  if ( changed )
                                         break;                                          break;
                         }                          }
                         if ( !changed ) {                          if ( !changed && !dirty ) {
                                 do_present = 0;                                  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];                                          LDAPControl     *ctrls[2];
Line 2590  shortcut: Line 2672  shortcut:
         }          }
   
         /* If something changed, find the changes */          /* If something changed, find the changes */
         if ( gotstate && changed ) {          if ( gotstate && ( changed || dirty ) ) {
                 Filter *fand, *fava;                  Filter *fand, *fava;
   
                 fand = op->o_tmpalloc( sizeof(Filter), op->o_tmpmemctx );                  fand = op->o_tmpalloc( sizeof(Filter), op->o_tmpmemctx );
Line 2635  shortcut: Line 2717  shortcut:
          * the refresh phase, just invoke the response callback to transition           * the refresh phase, just invoke the response callback to transition
          * us into persist phase           * us into persist phase
          */           */
         if ( !changed ) {          if ( !changed && !dirty ) {
                 rs->sr_err = LDAP_SUCCESS;                  rs->sr_err = LDAP_SUCCESS;
                 rs->sr_nentries = 0;                  rs->sr_nentries = 0;
                 send_ldap_result( op, rs );                  send_ldap_result( op, rs );
Line 3004  syncprov_db_open( Line 3086  syncprov_db_open(
                 si->si_numops++;                  si->si_numops++;
         }          }
   
           /* Initialize the sessionlog mincsn */
           if ( si->si_logs && si->si_numcsns ) {
                   sessionlog *sl = si->si_logs;
                   int i;
                   /* If there are multiple, find the newest */
                   for ( i=0; i < si->si_numcsns; i++ ) {
                           if ( ber_bvcmp( &sl->sl_mincsn, &si->si_ctxcsn[i] ) < 0 ) {
                                   AC_MEMCPY( sl->sl_mincsn.bv_val, si->si_ctxcsn[i].bv_val,
                                           si->si_ctxcsn[i].bv_len );
                                   sl->sl_mincsn.bv_len = si->si_ctxcsn[i].bv_len;
                           }
                   }
           }
   
 out:  out:
         op->o_bd->bd_info = (BackendInfo *)on;          op->o_bd->bd_info = (BackendInfo *)on;
         return 0;          return 0;
Line 3095  syncprov_db_destroy( Line 3191  syncprov_db_destroy(
                                 se = se_next;                                  se = se_next;
                         }                          }
                                                                   
                           ldap_pvt_thread_mutex_destroy(&si->si_logs->sl_mutex);
                         ch_free( si->si_logs );                          ch_free( si->si_logs );
                 }                  }
                 if ( si->si_ctxcsn )                  if ( si->si_ctxcsn )

Removed from v.1.147.2.77  
changed lines
  Added in v.1.147.2.89


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