ITS#4720 release csn mutex before checkpointing.

[openldap] / servers / slapd / overlays / syncprov.c

diff --git a/servers/slapd/overlays/syncprov.c b/servers/slapd/overlays/syncprov.c
index 02ea4851cf0b8c3f1b2bbd5b6f6af37764d032b0..85aa485c3d5d61f663cee5d2b89d24b97be7c2dd 100644 (file)
--- a/servers/slapd/overlays/syncprov.c
+++ b/servers/slapd/overlays/syncprov.c
@@ -65,6 +65,7 @@ typedef struct syncops {
 #define        PS_IS_DETACHED          0x02
 #define        PS_WROTE_BASE           0x04
 #define        PS_FIND_BASE            0x08
+#define        PS_FIX_FILTER           0x10
 
        int             s_inuse;        /* reference count */
        struct syncres *s_res;
@@ -773,7 +774,7 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
                        rs.sr_flags = REP_ENTRY_MUSTRELEASE;
                if ( opc->sreference ) {
                        rs.sr_ref = get_entry_referrals( op, rs.sr_entry );
-                       send_search_reference( op, &rs );
+                       rs.sr_err = send_search_reference( op, &rs );
                        ber_bvarray_free( rs.sr_ref );
                        if ( !rs.sr_entry )
                                *e = NULL;
@@ -785,7 +786,7 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
                if ( rs.sr_entry->e_private )
                        rs.sr_flags = REP_ENTRY_MUSTRELEASE;
                rs.sr_attrs = op->ors_attrs;
-               send_search_entry( op, &rs );
+               rs.sr_err = send_search_entry( op, &rs );
                if ( !rs.sr_entry )
                        *e = NULL;
                break;
@@ -797,9 +798,9 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
                if ( opc->sreference ) {
                        struct berval bv = BER_BVNULL;
                        rs.sr_ref = &bv;
-                       send_search_reference( op, &rs );
+                       rs.sr_err = send_search_reference( op, &rs );
                } else {
-                       send_search_entry( op, &rs );
+                       rs.sr_err = send_search_entry( op, &rs );
                }
                break;
        default:
@@ -877,6 +878,7 @@ syncprov_qtask( void *ctx, void *arg )
        OperationBuffer opbuf;
        Operation *op;
        BackendDB be;
+       int rc;
 
        op = (Operation *) &opbuf;
        *op = *so->s_op;
@@ -897,17 +899,33 @@ syncprov_qtask( void *ctx, void *arg )
        op->o_private = NULL;
        op->o_callback = NULL;
 
-       (void)syncprov_qplay( op, on, so );
+       rc = syncprov_qplay( op, on, so );
 
        /* decrement use count... */
        syncprov_free_syncop( so );
 
        /* wait until we get explicitly scheduled again */
        ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
-       ldap_pvt_runqueue_stoptask( &slapd_rq, so->s_qtask );
-       ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 1 );
+       ldap_pvt_runqueue_stoptask( &slapd_rq, rtask );
+       if ( rc == 0 ) {
+               ldap_pvt_runqueue_resched( &slapd_rq, rtask, 1 );
+       } else {
+               /* bail out on any error */
+               ldap_pvt_runqueue_remove( &slapd_rq, rtask );
+       }
        ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
 
+#if 0  /* FIXME: connection_close isn't exported from slapd.
+                * should it be?
+                */
+       if ( rc ) {
+               ldap_pvt_thread_mutex_lock( &op->o_conn->c_mutex );
+               if ( connection_state_closing( op->o_conn )) {
+                       connection_close( op->o_conn );
+               }
+               ldap_pvt_thread_mutex_unlock( &op->o_conn->c_mutex );
+       }
+#endif
        return NULL;
 }
 
@@ -1242,9 +1260,9 @@ syncprov_op_cleanup( Operation *op, SlapReply *rs )
 }
 
 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;
        Operation opm;
        SlapReply rsm = { 0 };
@@ -1252,12 +1270,12 @@ syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on )
        slap_callback cb = {0};
 
        /* If ctxcsn is empty, delete it */
-       if ( BER_BVISEMPTY( &si->si_ctxcsn )) {
+       if ( BER_BVISEMPTY( csn )) {
                mod.sml_values = NULL;
        } else {
                mod.sml_values = bv;
                bv[1].bv_val = NULL;
-               bv[0] = si->si_ctxcsn;
+               bv[0] = *csn;
        }
        mod.sml_nvalues = NULL;
        mod.sml_desc = slap_schema.si_ad_contextCSN;
@@ -1492,6 +1510,7 @@ syncprov_op_response( Operation *op, SlapReply *rs )
        {
                struct berval maxcsn = BER_BVNULL;
                char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
+               int do_check = 0;
 
                /* Update our context CSN */
                cbuf[0] = '\0';
@@ -1514,7 +1533,6 @@ syncprov_op_response( Operation *op, SlapReply *rs )
 
                si->si_numops++;
                if ( si->si_chkops || si->si_chktime ) {
-                       int do_check=0;
                        if ( si->si_chkops && si->si_numops >= si->si_chkops ) {
                                do_check = 1;
                                si->si_numops = 0;
@@ -1524,15 +1542,16 @@ syncprov_op_response( Operation *op, SlapReply *rs )
                                do_check = 1;
                                si->si_chklast = op->o_time;
                        }
-                       if ( do_check ) {
-                               syncprov_checkpoint( op, rs, on );
-                       }
                }
                ldap_pvt_thread_mutex_unlock( &si->si_csn_mutex );
 
                opc->sctxcsn.bv_len = maxcsn.bv_len;
                opc->sctxcsn.bv_val = cbuf;
 
+               if ( do_check ) {
+                       syncprov_checkpoint( op, rs, on, &opc->sctxcsn );
+               }
+
                /* Handle any persistent searches */
                if ( si->si_ops ) {
                        switch(op->o_tag) {
@@ -1792,7 +1811,15 @@ syncprov_detach_op( Operation *op, syncops *so, slap_overinst *on )
        op2->ors_filterstr.bv_val = ptr;
        strcpy( ptr, so->s_filterstr.bv_val );
        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;
 
        /* Copy any cached group ACLs individually */
@@ -2064,6 +2091,8 @@ shortcut:
                fava->f_next = op->ors_filter;
                op->ors_filter = fand;
                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 */
@@ -2121,13 +2150,10 @@ syncprov_operational(
                        if ( !a ) {
                                for ( ap = &rs->sr_operational_attrs; *ap; ap=&(*ap)->a_next );
 
-                               a = ch_malloc( sizeof(Attribute));
-                               a->a_desc = slap_schema.si_ad_contextCSN;
+                               a = attr_alloc( slap_schema.si_ad_contextCSN );
                                a->a_vals = ch_malloc( 2 * sizeof(struct berval));
                                a->a_vals[1].bv_val = NULL;
                                a->a_nvals = a->a_vals;
-                               a->a_next = NULL;
-                               a->a_flags = 0;
                                *ap = a;
                        }
 
@@ -2447,7 +2473,7 @@ syncprov_db_close(
                op->o_bd = be;
                op->o_dn = be->be_rootdn;
                op->o_ndn = be->be_rootndn;
-               syncprov_checkpoint( op, &rs, on );
+               syncprov_checkpoint( op, &rs, on, &si->si_ctxcsn );
                ldap_pvt_thread_pool_context_reset( thrctx );
        }