Diff for /servers/slapd/overlays/syncprov.c between versions 1.109 and 1.117

version 1.109, 2005/10/02 08:48:28 version 1.117, 2005/10/12 21:17:30
Line 1 Line 1
 /* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.108 2005/10/02 03:01:45 hyc Exp $ */  /* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.116 2005/10/06 22:31:38 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 69  typedef struct syncops { Line 69  typedef struct syncops {
         int             s_inuse;        /* reference count */          int             s_inuse;        /* reference count */
         struct syncres *s_res;          struct syncres *s_res;
         struct syncres *s_restail;          struct syncres *s_restail;
         void    *s_qtask;       /* task for playing psearch responses */          struct re_s     *s_qtask;       /* task for playing psearch responses */
   #define RUNQ_INTERVAL   36000   /* a long time */
         ldap_pvt_thread_mutex_t s_mutex;          ldap_pvt_thread_mutex_t s_mutex;
 } syncops;  } syncops;
   
Line 696  again: Line 697  again:
         return rc;          return rc;
 }  }
   
   static void
   syncprov_free_syncop( syncops *so )
   {
           syncres *sr, *srnext;
           GroupAssertion *ga, *gnext;
   
           ldap_pvt_thread_mutex_lock( &so->s_mutex );
           if ( --so->s_inuse > 0 ) {
                   ldap_pvt_thread_mutex_unlock( &so->s_mutex );
                   return;
           }
           ldap_pvt_thread_mutex_unlock( &so->s_mutex );
           if ( so->s_flags & PS_IS_DETACHED ) {
                   filter_free( so->s_op->ors_filter );
                   for ( ga = so->s_op->o_groups; ga; ga=gnext ) {
                           gnext = ga->ga_next;
                           ch_free( ga );
                   }
                   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 );
   }
   
 /* Send a persistent search response */  /* Send a persistent search response */
 static int  static int
 syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry **e, int mode)  syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry **e, int mode)
Line 719  syncprov_sendresp( Operation *op, opcook Line 749  syncprov_sendresp( Operation *op, opcook
         a_uuid.a_nvals = &opc->suuid;          a_uuid.a_nvals = &opc->suuid;
         rs.sr_err = syncprov_state_ctrl( op, &rs, &e_uuid,          rs.sr_err = syncprov_state_ctrl( op, &rs, &e_uuid,
                 mode, ctrls, 0, 1, &cookie );                  mode, ctrls, 0, 1, &cookie );
           op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
   
         rs.sr_ctrls = ctrls;          rs.sr_ctrls = ctrls;
         op->o_bd->bd_info = (BackendInfo *)on->on_info;          op->o_bd->bd_info = (BackendInfo *)on->on_info;
Line 761  syncprov_sendresp( Operation *op, opcook Line 792  syncprov_sendresp( Operation *op, opcook
         default:          default:
                 assert(0);                  assert(0);
         }          }
         op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx );          /* In case someone else freed it already? */
         rs.sr_ctrls = NULL;          if ( rs.sr_ctrls ) {
                   op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx );
                   rs.sr_ctrls = NULL;
           }
   
         return rs.sr_err;          return rs.sr_err;
 }  }
Line 788  syncprov_qplay( Operation *op, slap_over Line 822  syncprov_qplay( Operation *op, slap_over
                         so->s_restail = NULL;                          so->s_restail = NULL;
                 ldap_pvt_thread_mutex_unlock( &so->s_mutex );                  ldap_pvt_thread_mutex_unlock( &so->s_mutex );
   
                 if ( !sr )                  if ( !sr || so->s_op->o_abandon )
                         break;                          break;
   
                 if ( !so->s_op->o_abandon ) {                  opc.sdn = sr->s_dn;
                         opc.sdn = sr->s_dn;                  opc.sndn = sr->s_ndn;
                         opc.sndn = sr->s_ndn;                  opc.suuid = sr->s_uuid;
                         opc.suuid = sr->s_uuid;                  opc.sctxcsn = sr->s_csn;
                         opc.sctxcsn = sr->s_csn;                  opc.sreference = sr->s_isreference;
                         opc.sreference = sr->s_isreference;                  e = NULL;
                         e = NULL;  
                   if ( sr->s_mode != LDAP_SYNC_DELETE ) {
                         if ( sr->s_mode != LDAP_SYNC_DELETE ) {                          rc = be_entry_get_rw( op, &opc.sndn, NULL, NULL, 0, &e );
                                 rc = be_entry_get_rw( op, &opc.sndn, NULL, NULL, 0, &e );                          if ( rc ) {
                                 if ( rc ) {                                  ch_free( sr );
                                         ch_free( sr );                                  continue;
                                         continue;  
                                 }  
                         }                          }
                         rc = syncprov_sendresp( op, &opc, so, &e, sr->s_mode );                  }
                   rc = syncprov_sendresp( op, &opc, so, &e, sr->s_mode );
   
                         if ( e ) {                  if ( e ) {
                                 be_entry_release_rw( op, e, 0 );                          be_entry_release_rw( op, e, 0 );
                         }  
                         if ( rc )  
                                 break;  
                 }                  }
   
                 ch_free( sr );                  ch_free( sr );
   
                   if ( rc )
                           break;
         }          }
         op->o_bd->bd_info = (BackendInfo *)on;          op->o_bd->bd_info = (BackendInfo *)on;
         return rc;          return rc;
Line 828  syncprov_qtask( void *ctx, void *arg ) Line 861  syncprov_qtask( void *ctx, void *arg )
         struct re_s *rtask = arg;          struct re_s *rtask = arg;
         syncops *so = rtask->arg;          syncops *so = rtask->arg;
         slap_overinst *on = so->s_op->o_private;          slap_overinst *on = so->s_op->o_private;
         char opbuf[OPERATION_BUFFER_SIZE];          OperationBuffer opbuf;
         Operation *op;          Operation *op;
         BackendDB be;          BackendDB be;
   
         op = (Operation *)opbuf;          op = (Operation *) &opbuf;
         memset( op, 0, sizeof(opbuf));          *op = *so->s_op;
         op->o_hdr = (Opheader *)(op+1);          op->o_hdr = (Opheader *)(op+1);
         op->o_controls = (void **)(op->o_hdr+1);          op->o_controls = (void **)(op->o_hdr+1);
           memset( op->o_controls, 0, SLAP_MAX_CIDS * sizeof(void *));
   
         *op->o_hdr = *so->s_op->o_hdr;          *op->o_hdr = *so->s_op->o_hdr;
   
Line 852  syncprov_qtask( void *ctx, void *arg ) Line 886  syncprov_qtask( void *ctx, void *arg )
   
         syncprov_qplay( op, on, so );          syncprov_qplay( op, on, so );
   
           /* decrement use count... */
           syncprov_free_syncop( so );
   
         /* wait until we get explicitly scheduled again */          /* wait until we get explicitly scheduled again */
         ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );          ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
         ldap_pvt_runqueue_stoptask( &slapd_rq, so->s_qtask );          ldap_pvt_runqueue_stoptask( &slapd_rq, so->s_qtask );
Line 874  syncprov_qresp( opcookie *opc, syncops * Line 911  syncprov_qresp( opcookie *opc, syncops *
         sr->s_dn.bv_len = opc->sdn.bv_len;          sr->s_dn.bv_len = opc->sdn.bv_len;
         sr->s_mode = mode;          sr->s_mode = mode;
         sr->s_isreference = opc->sreference;          sr->s_isreference = opc->sreference;
         sr->s_ndn.bv_val = lutil_strcopy( sr->s_dn.bv_val, opc->sdn.bv_val );          sr->s_ndn.bv_val = lutil_strcopy( sr->s_dn.bv_val,
                    opc->sdn.bv_val ) + 1;
         sr->s_ndn.bv_len = opc->sndn.bv_len;          sr->s_ndn.bv_len = opc->sndn.bv_len;
         *(sr->s_ndn.bv_val++) = '\0';          sr->s_uuid.bv_val = lutil_strcopy( sr->s_ndn.bv_val,
         sr->s_uuid.bv_val = lutil_strcopy( sr->s_ndn.bv_val, opc->sndn.bv_val );                   opc->sndn.bv_val ) + 1;
         sr->s_uuid.bv_len = opc->suuid.bv_len;          sr->s_uuid.bv_len = opc->suuid.bv_len;
         *(sr->s_uuid.bv_val++) = '\0';          AC_MEMCPY( sr->s_uuid.bv_val, opc->suuid.bv_val, opc->suuid.bv_len );
         sr->s_csn.bv_val = lutil_strcopy( sr->s_uuid.bv_val, opc->suuid.bv_val );          sr->s_csn.bv_val = sr->s_uuid.bv_val + sr->s_uuid.bv_len + 1;
         sr->s_csn.bv_len = opc->sctxcsn.bv_len;          sr->s_csn.bv_len = opc->sctxcsn.bv_len;
         strcpy( sr->s_csn.bv_val, opc->sctxcsn.bv_val );          strcpy( sr->s_csn.bv_val, opc->sctxcsn.bv_val );
   
Line 900  syncprov_qresp( opcookie *opc, syncops * Line 938  syncprov_qresp( opcookie *opc, syncops *
         if ( so->s_flags & PS_IS_DETACHED ) {          if ( so->s_flags & PS_IS_DETACHED ) {
                 ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );                  ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
                 if ( !so->s_qtask ) {                  if ( !so->s_qtask ) {
                         so->s_qtask = ldap_pvt_runqueue_insert( &slapd_rq, 0,                          so->s_qtask = ldap_pvt_runqueue_insert( &slapd_rq, RUNQ_INTERVAL,
                                 syncprov_qtask, so, "syncprov_qtask",                                  syncprov_qtask, so, "syncprov_qtask",
                                 so->s_op->o_conn->c_peer_name.bv_val );                                  so->s_op->o_conn->c_peer_name.bv_val );
                           ++so->s_inuse;
                 } else {                  } else {
                         if (!ldap_pvt_runqueue_isrunning( &slapd_rq, so->s_qtask )) {                          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 );                                  ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 0 );
                                   so->s_qtask->interval.tv_sec = RUNQ_INTERVAL;
                                   ++so->s_inuse;
                         }                          }
                 }                  }
                 ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );                  ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
Line 914  syncprov_qresp( opcookie *opc, syncops * Line 957  syncprov_qresp( opcookie *opc, syncops *
         return LDAP_SUCCESS;          return LDAP_SUCCESS;
 }  }
   
 static void  
 syncprov_free_syncop( syncops *so )  
 {  
         syncres *sr, *srnext;  
         GroupAssertion *ga, *gnext;  
   
         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 );  
         if ( so->s_flags & PS_IS_DETACHED ) {  
                 filter_free( so->s_op->ors_filter );  
                 for ( ga = so->s_op->o_groups; ga; ga=gnext ) {  
                         gnext = ga->ga_next;  
                         ch_free( ga );  
                 }  
                 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  static int
 syncprov_drop_psearch( syncops *so, int lock )  syncprov_drop_psearch( syncops *so, int lock )
 {  {
Line 1119  syncprov_matchops( Operation *op, opcook Line 1132  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++;                                  ldap_pvt_thread_mutex_lock( &ss->s_mutex );
                                   ++ss->s_inuse;
                                   ldap_pvt_thread_mutex_unlock( &ss->s_mutex );
                                 opc->smatches = sm;                                  opc->smatches = sm;
                         } else {                          } else {
                                 /* if found send UPDATE else send ADD */                                  /* if found send UPDATE else send ADD */
                                 ss->s_inuse++;  
                                 syncprov_qresp( opc, ss,                                  syncprov_qresp( opc, ss,
                                         found ? LDAP_SYNC_MODIFY : LDAP_SYNC_ADD );                                          found ? LDAP_SYNC_MODIFY : LDAP_SYNC_ADD );
                                 ss->s_inuse--;  
                         }                          }
                 } else if ( !saveit && found ) {                  } else if ( !saveit && found ) {
                         /* send DELETE */                          /* send DELETE */
Line 1799  syncprov_search_response( Operation *op, Line 1812  syncprov_search_response( Operation *op,
                         rs->sr_err = syncprov_done_ctrl( op, rs, rs->sr_ctrls,                          rs->sr_err = syncprov_done_ctrl( op, rs, rs->sr_ctrls,
                                 0, 1, &cookie, ss->ss_present ?  LDAP_SYNC_REFRESH_PRESENTS :                                  0, 1, &cookie, ss->ss_present ?  LDAP_SYNC_REFRESH_PRESENTS :
                                         LDAP_SYNC_REFRESH_DELETES );                                          LDAP_SYNC_REFRESH_DELETES );
                           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 && 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 );
                         /* Flush any queued persist messages */                          op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
                         if ( ss->ss_so->s_res ) {  
                                 syncprov_qplay( op, on, ss->ss_so );  
                         }  
   
                         /* Detach this Op from frontend control */                          /* Detach this Op from frontend control */
                         ldap_pvt_thread_mutex_lock( &ss->ss_so->s_mutex );                          ldap_pvt_thread_mutex_lock( &ss->ss_so->s_mutex );
Line 2206  syncprov_db_open( Line 2217  syncprov_db_open(
         syncprov_info_t *si = (syncprov_info_t *)on->on_bi.bi_private;          syncprov_info_t *si = (syncprov_info_t *)on->on_bi.bi_private;
   
         Connection conn;          Connection conn;
         char opbuf[OPERATION_BUFFER_SIZE];          OperationBuffer opbuf;
         char ctxcsnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];          char ctxcsnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
         Operation *op = (Operation *)opbuf;          Operation *op = (Operation *) &opbuf;
         Entry *e;          Entry *e;
         Attribute *a;          Attribute *a;
         int rc;          int rc;
Line 2296  syncprov_db_close( Line 2307  syncprov_db_close(
         }          }
         if ( si->si_numops ) {          if ( si->si_numops ) {
                 Connection conn;                  Connection conn;
                 char opbuf[OPERATION_BUFFER_SIZE];                  OperationBuffer opbuf;
                 Operation *op = (Operation *)opbuf;                  Operation *op = (Operation *) &opbuf;
                 SlapReply rs = {REP_RESULT};                  SlapReply rs = {REP_RESULT};
                 void *thrctx;                  void *thrctx;
   

Removed from v.1.109  
changed lines
  Added in v.1.117


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