Diff for /servers/slapd/overlays/syncprov.c between versions 1.19 and 1.23

version 1.19, 2004/11/27 00:26:12 version 1.23, 2004/11/27 06:47:24
Line 25 Line 25
 #include "lutil.h"  #include "lutil.h"
 #include "slap.h"  #include "slap.h"
   
   /* A modify request on a particular entry */
   typedef struct modinst {
           struct modinst *mi_next;
           Operation *mi_op;
   } modinst;
   
   typedef struct modtarget {
           struct modinst *mt_mods;
           struct modinst *mt_tail;
           Operation *mt_op;
           ldap_pvt_thread_mutex_t mt_mutex;
   } modtarget;
   
 /* A queued result of a persistent search */  /* A queued result of a persistent search */
 typedef struct syncres {  typedef struct syncres {
         struct syncres *s_next;          struct syncres *s_next;
Line 46  typedef struct syncops { Line 59  typedef struct syncops {
         long    s_rid;          long    s_rid;
         struct berval s_filterstr;          struct berval s_filterstr;
         int             s_flags;        /* search status */          int             s_flags;        /* search status */
           int             s_inuse;        /* reference count */
         struct syncres *s_res;          struct syncres *s_res;
         struct syncres *s_restail;          struct syncres *s_restail;
         ldap_pvt_thread_mutex_t s_mutex;          ldap_pvt_thread_mutex_t s_mutex;
Line 76  typedef struct syncmatches { Line 90  typedef struct syncmatches {
         syncops *sm_op;          syncops *sm_op;
 } syncmatches;  } syncmatches;
   
   /* The main state for this overlay */
 typedef struct syncprov_info_t {  typedef struct syncprov_info_t {
         syncops         *si_ops;          syncops         *si_ops;
         struct berval   si_ctxcsn;      /* ldapsync context */          struct berval   si_ctxcsn;      /* ldapsync context */
         int             si_gotcsn;      /* is our ctxcsn up to date? */          int             si_gotcsn;      /* is our ctxcsn up to date? */
           Avlnode *si_mods;       /* entries being modified */
         ldap_pvt_thread_mutex_t si_csn_mutex;          ldap_pvt_thread_mutex_t si_csn_mutex;
         ldap_pvt_thread_mutex_t si_ops_mutex;          ldap_pvt_thread_mutex_t si_ops_mutex;
           ldap_pvt_thread_mutex_t si_mods_mutex;
         char            si_ctxcsnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];          char            si_ctxcsnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
 } syncprov_info_t;  } syncprov_info_t;
   
Line 105  typedef struct fbase_cookie { Line 122  typedef struct fbase_cookie {
 static AttributeName csn_anlist[2];  static AttributeName csn_anlist[2];
 static AttributeName uuid_anlist[2];  static AttributeName uuid_anlist[2];
   
   /* Build a LDAPsync intermediate state control */
 static int  static int
 syncprov_state_ctrl(  syncprov_state_ctrl(
         Operation       *op,          Operation       *op,
Line 164  syncprov_state_ctrl( Line 182  syncprov_state_ctrl(
         return LDAP_SUCCESS;          return LDAP_SUCCESS;
 }  }
   
   /* Build a LDAPsync final state control */
 static int  static int
 syncprov_done_ctrl(  syncprov_done_ctrl(
         Operation       *op,          Operation       *op,
Line 209  syncprov_done_ctrl( Line 228  syncprov_done_ctrl(
         return LDAP_SUCCESS;          return LDAP_SUCCESS;
 }  }
   
   #if 0
   /* Generate state based on session log - not implemented yet */
 static int  static int
 syncprov_state_ctrl_from_slog(  syncprov_state_ctrl_from_slog(
         Operation       *op,          Operation       *op,
Line 262  syncprov_state_ctrl_from_slog( Line 282  syncprov_state_ctrl_from_slog(
   
         return LDAP_SUCCESS;          return LDAP_SUCCESS;
 }  }
   #endif
   
 static int  static int
 syncprov_sendinfo(  syncprov_sendinfo(
Line 327  syncprov_sendinfo( Line 348  syncprov_sendinfo(
                 return ret;                  return ret;
         }          }
   
           rs->sr_rspoid = LDAP_SYNC_INFO;
         rs->sr_rspdata = &rspdata;          rs->sr_rspdata = &rspdata;
         send_ldap_intermediate( op, rs );          send_ldap_intermediate( op, rs );
         rs->sr_rspdata = NULL;          rs->sr_rspdata = NULL;
Line 334  syncprov_sendinfo( Line 356  syncprov_sendinfo(
   
         return LDAP_SUCCESS;          return LDAP_SUCCESS;
 }  }
   
   /* Find a modtarget in an AVL tree */
   static int
   sp_avl_cmp( const void *c1, const void *c2 )
   {
           const modtarget *m1, *m2;
           int rc;
   
           m1 = c1; m2 = c2;
           rc = m1->mt_op->o_req_ndn.bv_len - m2->mt_op->o_req_ndn.bv_len;
   
           if ( rc ) return rc;
           return ber_bvcmp( &m1->mt_op->o_req_ndn, &m2->mt_op->o_req_ndn );
   }
   
 /* syncprov_findbase:  /* syncprov_findbase:
  *   finds the true DN of the base of a search (with alias dereferencing) and   *   finds the true DN of the base of a search (with alias dereferencing) and
  * checks to make sure the base entry doesn't get replaced with a different   * checks to make sure the base entry doesn't get replaced with a different
Line 513  findpres_cb( Operation *op, SlapReply *r Line 550  findpres_cb( Operation *op, SlapReply *r
                         pc->num++;                          pc->num++;
                         ret = LDAP_SUCCESS;                          ret = LDAP_SUCCESS;
                         if ( pc->num == SLAP_SYNCUUID_SET_SIZE ) {                          if ( pc->num == SLAP_SYNCUUID_SET_SIZE ) {
                                 rs->sr_rspoid = LDAP_SYNC_INFO;  
                                 ret = syncprov_sendinfo( op, rs, LDAP_TAG_SYNC_ID_SET, NULL,                                  ret = syncprov_sendinfo( op, rs, LDAP_TAG_SYNC_ID_SET, NULL,
                                         0, pc->uuids, 0 );                                          0, pc->uuids, 0 );
                                 ber_bvarray_free_x( pc->uuids, op->o_tmpmemctx );                                  ber_bvarray_free_x( pc->uuids, op->o_tmpmemctx );
Line 526  findpres_cb( Operation *op, SlapReply *r Line 562  findpres_cb( Operation *op, SlapReply *r
         } else if ( rs->sr_type == REP_RESULT ) {          } else if ( rs->sr_type == REP_RESULT ) {
                 ret = rs->sr_err;                  ret = rs->sr_err;
                 if ( pc->num ) {                  if ( pc->num ) {
                         rs->sr_rspoid = LDAP_SYNC_INFO;  
                         ret = syncprov_sendinfo( op, rs, LDAP_TAG_SYNC_ID_SET, NULL,                          ret = syncprov_sendinfo( op, rs, LDAP_TAG_SYNC_ID_SET, NULL,
                                 0, pc->uuids, 0 );                                  0, pc->uuids, 0 );
                         ber_bvarray_free_x( pc->uuids, op->o_tmpmemctx );                          ber_bvarray_free_x( pc->uuids, op->o_tmpmemctx );
Line 746  syncprov_sendresp( Operation *op, opcook Line 781  syncprov_sendresp( Operation *op, opcook
 }  }
   
 static void  static void
   syncprov_free_syncop( syncops *so )
   {
           syncres *sr, *srnext;
   
           ldap_pvt_thread_mutex_lock( &so->s_mutex );
           so->s_inuse--;
           if ( so->s_inuse > 0 ) {
                   ldap_pvt_thread_mutex_unlock( &so->s_mutex );
                   return;
           }
           ldap_pvt_thread_mutex_unlock( &so->s_mutex );
           filter_free( so->s_op->ors_filter );
           ch_free( so->s_op );
           ch_free( so->s_base.bv_val );
           for ( sr=so->s_res; sr; sr=srnext ) {
                   srnext = sr->s_next;
                   ch_free( sr );
           }
           ldap_pvt_thread_mutex_destroy( &so->s_mutex );
           ch_free( so );
   }
   
   static int
   syncprov_drop_psearch( syncops *so )
   {
           ldap_pvt_thread_mutex_lock( &so->s_op->o_conn->c_mutex );
           so->s_op->o_conn->c_n_ops_executing--;
           so->s_op->o_conn->c_n_ops_completed++;
           ldap_pvt_thread_mutex_unlock( &so->s_op->o_conn->c_mutex );
           syncprov_free_syncop( so );
   }
   
   static int
   syncprov_op_abandon( Operation *op, SlapReply *rs )
   {
           slap_overinst           *on = (slap_overinst *)op->o_bd->bd_info;
           syncprov_info_t         *si = on->on_bi.bi_private;
           syncops *so, *soprev;
   
           ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
           for ( so=si->si_ops, soprev = (syncops *)&si->si_ops; so;
                   soprev=so, so=so->s_next ) {
                   if ( so->s_op->o_connid == op->o_connid &&
                           so->s_op->o_msgid == op->orn_msgid ) {
                                   so->s_op->o_abandon = 1;
                                   soprev->s_next = so->s_next;
                                   break;
                   }
           }
           ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
           if ( so ) {
                   /* Is this really a Cancel exop? */
                   if ( op->o_tag != LDAP_REQ_ABANDON ) {
                           rs->sr_err = LDAP_CANCELLED;
                           send_ldap_result( so->s_op, rs );
                   }
                   syncprov_drop_psearch( so );
           }
           return SLAP_CB_CONTINUE;
   }
   
   /* Find which persistent searches are affected by this operation */
   static void
 syncprov_matchops( Operation *op, opcookie *opc, int saveit )  syncprov_matchops( Operation *op, opcookie *opc, int saveit )
 {  {
         slap_overinst *on = opc->son;          slap_overinst *on = opc->son;
         syncprov_info_t         *si = on->on_bi.bi_private;          syncprov_info_t         *si = on->on_bi.bi_private;
   
         fbase_cookie fc;          fbase_cookie fc;
         syncops *ss;          syncops *ss, *sprev, *snext;
         Entry *e;          Entry *e;
         Attribute *a;          Attribute *a;
         int rc;          int rc;
Line 788  syncprov_matchops( Operation *op, opcook Line 886  syncprov_matchops( Operation *op, opcook
         }          }
   
         ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );          ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
         for (ss = si->si_ops; ss; ss=ss->s_next)          for (ss = si->si_ops, sprev = (syncops *)&si->si_ops; ss;
                   sprev = ss, ss=snext)
         {          {
                 syncmatches *sm;                  syncmatches *sm;
                 int found = 0;                  int found = 0;
   
                   snext = ss->s_next;
                 /* validate base */                  /* validate base */
                 fc.fss = ss;                  fc.fss = ss;
                 fc.fbase = 0;                  fc.fbase = 0;
                 fc.fscope = 0;                  fc.fscope = 0;
   
                   /* If the base of the search is missing, signal a refresh */
                 rc = syncprov_findbase( op, &fc );                  rc = syncprov_findbase( op, &fc );
                 if ( rc != LDAP_SUCCESS ) continue;                  if ( rc != LDAP_SUCCESS ) {
                           SlapReply rs = {REP_RESULT};
                           send_ldap_error( ss->s_op, &rs, LDAP_SYNC_REFRESH_REQUIRED,
                                   "search base has changed" );
                           sprev->s_next = snext;
                           syncprov_drop_psearch( ss );
                           continue;
                   }
   
                 /* If we're sending results now, look for this op in old matches */                  /* If we're sending results now, look for this op in old matches */
                 if ( !saveit ) {                  if ( !saveit ) {
Line 821  syncprov_matchops( Operation *op, opcook Line 930  syncprov_matchops( Operation *op, opcook
                                 sm = op->o_tmpalloc( sizeof(syncmatches), op->o_tmpmemctx );                                  sm = op->o_tmpalloc( sizeof(syncmatches), op->o_tmpmemctx );
                                 sm->sm_next = opc->smatches;                                  sm->sm_next = opc->smatches;
                                 sm->sm_op = ss;                                  sm->sm_op = ss;
                                   ss->s_inuse++;
                                 opc->smatches = sm;                                  opc->smatches = sm;
                         } else {                          } else {
                                 /* if found send UPDATE else send ADD */                                  /* if found send UPDATE else send ADD */
Line 845  syncprov_op_cleanup( Operation *op, Slap Line 955  syncprov_op_cleanup( Operation *op, Slap
 {  {
         slap_callback *cb = op->o_callback;          slap_callback *cb = op->o_callback;
         opcookie *opc = cb->sc_private;          opcookie *opc = cb->sc_private;
           slap_overinst *on = opc->son;
           syncprov_info_t         *si = on->on_bi.bi_private;
         syncmatches *sm, *snext;          syncmatches *sm, *snext;
           modtarget *mt, mtdummy;
   
         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 );
                 op->o_tmpfree( sm, op->o_tmpmemctx );                  op->o_tmpfree( sm, op->o_tmpmemctx );
         }          }
   
           /* Remove op from lock table */
           mtdummy.mt_op = op;
           ldap_pvt_thread_mutex_lock( &si->si_mods_mutex );
           mt = avl_find( si->si_mods, &mtdummy, sp_avl_cmp );
           if ( mt ) {
                   modinst *mi = mt->mt_mods;
                   
                   /* If there are more, promote the next one */
                   ldap_pvt_thread_mutex_lock( &mt->mt_mutex );
                   if ( mi->mi_next ) {
                           mt->mt_mods = mi->mi_next;
                           mt->mt_op = mt->mt_mods->mi_op;
                           ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
                   } else {
                           avl_delete( &si->si_mods, mt, sp_avl_cmp );
                           ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
                           ldap_pvt_thread_mutex_destroy( &mt->mt_mutex );
                           ch_free( mt );
                   }
           }
           ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
         op->o_callback = cb->sc_next;          op->o_callback = cb->sc_next;
         op->o_tmpfree(cb, op->o_tmpmemctx);          op->o_tmpfree(cb, op->o_tmpmemctx);
 }  }
Line 868  syncprov_op_response( Operation *op, Sla Line 1004  syncprov_op_response( Operation *op, Sla
                 struct berval maxcsn;                  struct berval maxcsn;
                 char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];                  char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
   
                   /* 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 );
Line 884  syncprov_op_response( Operation *op, Sla Line 1021  syncprov_op_response( Operation *op, Sla
                 opc->sctxcsn.bv_len = maxcsn.bv_len;                  opc->sctxcsn.bv_len = maxcsn.bv_len;
                 opc->sctxcsn.bv_val = cbuf;                  opc->sctxcsn.bv_val = cbuf;
   
                   /* Handle any persistent searches */
                 if ( si->si_ops ) {                  if ( si->si_ops ) {
                         switch(op->o_tag) {                          switch(op->o_tag) {
                         case LDAP_REQ_ADD:                          case LDAP_REQ_ADD:
Line 896  syncprov_op_response( Operation *op, Sla Line 1034  syncprov_op_response( Operation *op, Sla
                                 /* for each match in opc->smatches:                                  /* for each match in opc->smatches:
                                  *   send DELETE msg                                   *   send DELETE msg
                                  */                                   */
                                   ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
                                 for ( sm = opc->smatches; sm; sm=sm->sm_next ) {                                  for ( sm = opc->smatches; sm; sm=sm->sm_next ) {
                                           if ( sm->sm_op->s_op->o_abandon )
                                                   continue;
                                         syncprov_sendresp( op, opc, sm->sm_op, NULL,                                          syncprov_sendresp( op, opc, sm->sm_op, NULL,
                                                 LDAP_SYNC_DELETE, 1 );                                                  LDAP_SYNC_DELETE, 1 );
                                 }                                  }
                                   ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
                                 break;                                  break;
                         }                          }
                 }                  }
Line 908  syncprov_op_response( Operation *op, Sla Line 1050  syncprov_op_response( Operation *op, Sla
         return SLAP_CB_CONTINUE;          return SLAP_CB_CONTINUE;
 }  }
   
 static void  
 syncprov_free_syncop( syncops *so )  
 {  
         syncres *sr, *srnext;  
   
         filter_free( so->s_op->ors_filter );  
         ch_free( so->s_op );  
         ch_free( so->s_base.bv_val );  
         for ( sr=so->s_res; sr; sr=srnext ) {  
                 srnext = sr->s_next;  
                 ch_free( sr );  
         }  
         ldap_pvt_thread_mutex_destroy( &so->s_mutex );  
         ch_free( so );  
 }  
   
 static int  
 syncprov_op_abandon( Operation *op, SlapReply *rs )  
 {  
         slap_overinst           *on = (slap_overinst *)op->o_bd->bd_info;  
         syncprov_info_t         *si = on->on_bi.bi_private;  
         syncops *so, *soprev;  
   
         ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );  
         for ( so=si->si_ops, soprev = (syncops *)&si->si_ops; so;  
                 soprev=so, so=so->s_next ) {  
                 if ( so->s_op->o_connid == op->o_connid &&  
                         so->s_op->o_msgid == op->orn_msgid ) {  
                                 soprev->s_next = so->s_next;  
                                 break;  
                 }  
         }  
         ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );  
         if ( so ) {  
                 ldap_pvt_thread_mutex_lock( &op->o_conn->c_mutex );  
                 op->o_conn->c_n_ops_executing--;  
                 op->o_conn->c_n_ops_completed++;  
                 ldap_pvt_thread_mutex_unlock( &op->o_conn->c_mutex );  
                 /* Is this really a Cancel exop? */  
                 if ( op->o_tag != LDAP_REQ_ABANDON ) {  
                         rs->sr_err = LDAP_CANCELLED;  
                         send_ldap_result( so->s_op, rs );  
                 }  
                 syncprov_free_syncop( so );  
         }  
         return SLAP_CB_CONTINUE;  
 }  
   
 #if 0  #if 0
   /* We don't use a subentry to store the context CSN any more. But
    * we ought to expose the current context CSN as an operational attribute
    * somewhere, like on the suffix entry.
    */
 static int  static int
 syncprov_op_compare( Operation *op, SlapReply *rs )  syncprov_op_compare( Operation *op, SlapReply *rs )
 {  {
Line 1024  syncprov_op_mod( Operation *op, SlapRepl Line 1122  syncprov_op_mod( Operation *op, SlapRepl
         slap_overinst           *on = (slap_overinst *)op->o_bd->bd_info;          slap_overinst           *on = (slap_overinst *)op->o_bd->bd_info;
         syncprov_info_t         *si = on->on_bi.bi_private;          syncprov_info_t         *si = on->on_bi.bi_private;
   
         slap_callback *cb = op->o_tmpcalloc(1, sizeof(slap_callback)+sizeof(opcookie), op->o_tmpmemctx);          slap_callback *cb = op->o_tmpcalloc(1, sizeof(slap_callback)+
                   sizeof(opcookie) +
                   (si->si_ops ? sizeof(modinst) : 0 ),
                   op->o_tmpmemctx);
         opcookie *opc = (opcookie *)(cb+1);          opcookie *opc = (opcookie *)(cb+1);
         opc->son = on;          opc->son = on;
         cb->sc_response = syncprov_op_response;          cb->sc_response = syncprov_op_response;
Line 1033  syncprov_op_mod( Operation *op, SlapRepl Line 1134  syncprov_op_mod( Operation *op, SlapRepl
         cb->sc_next = op->o_callback;          cb->sc_next = op->o_callback;
         op->o_callback = cb;          op->o_callback = cb;
   
         if ( si->si_ops && op->o_tag != LDAP_REQ_ADD )          /* If there are active persistent searches, lock this operation.
                 syncprov_matchops( op, opc, 1 );           * See seqmod.c for the locking logic on its own.
            */
           if ( si->si_ops ) {
                   modtarget *mt, mtdummy;
                   modinst *mi;
   
                   mi = (modinst *)(opc+1);
                   mi->mi_op = op;
   
                   /* See if we're already modifying this entry... */
                   mtdummy.mt_op = op;
                   ldap_pvt_thread_mutex_lock( &si->si_mods_mutex );
                   mt = avl_find( si->si_mods, &mtdummy, sp_avl_cmp );
                   if ( mt ) {
                           ldap_pvt_thread_mutex_lock( &mt->mt_mutex );
                           ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
                           mt->mt_tail->mi_next = mi;
                           mt->mt_tail = mi;
                           /* wait for this op to get to head of list */
                           while ( mt->mt_mods != mi ) {
                                   ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
                                   ldap_pvt_thread_yield();
                                   ldap_pvt_thread_mutex_lock( &mt->mt_mutex );
                           }
                   } else {
                           /* Record that we're modifying this entry now */
                           mt = malloc( sizeof(modtarget) );
                           mt->mt_mods = mi;
                           mt->mt_tail = mi;
                           mt->mt_op = mi->mi_op;
                           ldap_pvt_thread_mutex_init( &mt->mt_mutex );
                           avl_insert( &si->si_mods, mt, sp_avl_cmp, avl_dup_error );
                           ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
                   }
   
                   if ( op->o_tag != LDAP_REQ_ADD )
                           syncprov_matchops( op, opc, 1 );
           }
   
         return SLAP_CB_CONTINUE;          return SLAP_CB_CONTINUE;
 }  }
Line 1161  syncprov_search_response( Operation *op, Line 1299  syncprov_search_response( Operation *op,
                 } else {                  } else {
                         int locked = 0;                          int locked = 0;
                 /* It's RefreshAndPersist, transition to Persist phase */                  /* It's RefreshAndPersist, transition to Persist phase */
                         rs->sr_rspoid = LDAP_SYNC_INFO;  
                         syncprov_sendinfo( op, rs, rs->sr_nentries ?                          syncprov_sendinfo( op, rs, 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 );
Line 1271  syncprov_op_search( Operation *op, SlapR Line 1408  syncprov_op_search( Operation *op, SlapR
                 sop->s_sid = srs->sr_state.sid;                  sop->s_sid = srs->sr_state.sid;
                 sop->s_rid = srs->sr_state.rid;                  sop->s_rid = srs->sr_state.rid;
                 sop->s_next = si->si_ops;                  sop->s_next = si->si_ops;
                   sop->s_inuse = 1;
                 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 );
         }          }

Removed from v.1.19  
changed lines
  Added in v.1.23


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