Bug: 147157

Fix Description: There was already a flag to disable the tombstone reap
thread, but there was no way to set it from outside the replica code.
I added some functions to set the flag, and to query it to show it
in the status entry.  When a total update request comes in, the flag
is set to disable, and when the total update ends for whatever reason
(success, error, connection terminated), the flag is set back to
enable.
In addition, the tombstone reap thread was being started after 1 hour -
this is too long and needed to be shorter.  We already have the
tombstone reap interval, so I changed it so that the reap thread
starts after this interval, then runs after every interval elapses.

ldap/servers/plugins/replication/repl5.h

@@ -429,6 +429,7 @@ void replica_write_ruv (Replica *r);
 #define REPLICA_AGREEMENTS_DISABLED 8 /* Replica is offline */
 PRBool replica_is_state_flag_set(Replica *r, PRInt32 flag);
 void replica_set_state_flag (Replica *r, PRUint32 flag, PRBool clear); 
+void replica_set_tombstone_reap_stop(Replica *r, PRBool val);
 void replica_enable_replication (Replica *r);
 void replica_disable_replication (Replica *r, Object *r_obj);
 int replica_start_agreement(Replica *r, Repl_Agmt *ra);

ldap/servers/plugins/replication/repl5_replica.c

@@ -17,7 +17,6 @@ int g_get_shutdown();
 
 #define RUV_SAVE_INTERVAL (30 * 1000) /* 30 seconds */
 #define START_UPDATE_DELAY 2 /* 2 second */
-#define START_REAP_DELAY 3600 /* 1 hour */
 
 #define REPLICA_RDN				 "cn=replica"
 #define CHANGELOG_RDN            "cn=legacy changelog"
@@ -214,7 +213,9 @@ replica_new_from_entry (Slapi_Entry *e, char *errortext, PRBool is_add_operation
 		 * This will allow the server to fully start before consuming resources.
 		 */
 		repl_name = slapi_ch_strdup (r->repl_name);
-		r->repl_eqcxt_tr = slapi_eq_repeat(eq_cb_reap_tombstones, repl_name, current_time() + START_REAP_DELAY, 1000 * r->tombstone_reap_interval);
+		r->repl_eqcxt_tr = slapi_eq_repeat(eq_cb_reap_tombstones, repl_name,
+										   current_time() + r->tombstone_reap_interval,
+										   1000 * r->tombstone_reap_interval);
 	}
 
     if (r->legacy_consumer)
@@ -2335,7 +2336,18 @@ int process_reap_entry (Slapi_Entry *entry, void *cb_data)
 	unsigned long *num_entriesp = &((reap_callback_data *)cb_data)->num_entries;
 	unsigned long *num_purged_entriesp = &((reap_callback_data *)cb_data)->num_purged_entries;
 	CSN *purge_csn = ((reap_callback_data *)cb_data)->purge_csn;
+	/* this is a pointer into the actual value in the Replica object - so that
+	   if the value is set in the replica, we will know about it immediately */
 	PRBool *tombstone_reap_stop = ((reap_callback_data *)cb_data)->tombstone_reap_stop;
+
+	/* abort reaping if we've been told to stop or we're shutting down */
+	if (*tombstone_reap_stop || g_get_shutdown()) {
+		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
+						"_replica_reap_tombstones: the tombstone reap process "
+						" has been stopped\n");
+		return -1;
+	}
+
 	/* we only ask for the objectclass in the search - the deletion csn is in the 
 	   objectclass attribute values - if we need more attributes returned by the
 	   search in the future, see _replica_reap_tombstones below and add more to the
@@ -2361,9 +2373,6 @@ int process_reap_entry (Slapi_Entry *entry, void *cb_data)
 		"%s\n", escape_string(slapi_entry_get_dn(entry),ebuf));
 	}
 	(*num_entriesp)++;
-	if (*tombstone_reap_stop || g_get_shutdown()) { 
-		return -1;
-	}
 
 	return 0;
 }
@@ -2453,6 +2462,9 @@ _replica_reap_tombstones(void *arg)
 		cb_data.num_entries = 0UL;
 		cb_data.num_purged_entries = 0UL;
 		cb_data.purge_csn = purge_csn;
+		/* set the cb data pointer to point to the actual memory address in
+		   the actual Replica object - so that when the value in the Replica
+		   is set, the reap process will know about it immediately */
 		cb_data.tombstone_reap_stop = &(replica->tombstone_reap_stop);
 
 		slapi_search_internal_callback_pb (pb, &cb_data /* callback data */,
@@ -2489,11 +2501,11 @@ _replica_reap_tombstones(void *arg)
 						replica_name ? replica_name : "(null)");
 	}
 
+done:
 	PR_Lock(replica->repl_lock);
 	replica->tombstone_reap_active = PR_FALSE;
 	PR_Unlock(replica->repl_lock);
 
-done:
 	if (NULL != purge_csn)
 	{
 		csn_free(&purge_csn);
@@ -2978,7 +2990,9 @@ replica_set_tombstone_reap_interval (Replica *r, long interval)
 	if ( interval > 0 && r->repl_eqcxt_tr == NULL )
 	{
 		repl_name = slapi_ch_strdup (r->repl_name);
-		r->repl_eqcxt_tr = slapi_eq_repeat (eq_cb_reap_tombstones, repl_name, current_time() + START_REAP_DELAY, 1000 * r->tombstone_reap_interval);
+		r->repl_eqcxt_tr = slapi_eq_repeat (eq_cb_reap_tombstones, repl_name,
+											current_time() + r->tombstone_reap_interval,
+											1000 * r->tombstone_reap_interval);
 		slapi_log_error (SLAPI_LOG_REPL, NULL,
 			"tombstone_reap event (interval=%d) was %s\n",
 			r->tombstone_reap_interval, (r->repl_eqcxt_tr ? "scheduled" : "not scheduled successfully"));
@@ -3123,6 +3137,22 @@ replica_set_state_flag (Replica *r, PRUint32 flag, PRBool clear)
 	PR_Unlock(r->repl_lock);
 }
 
+/**
+ * Use this to tell the tombstone reap process to stop.  This will
+ * typically be used when we (consumer) get a request to do a
+ * total update.
+ */
+void
+replica_set_tombstone_reap_stop(Replica *r, PRBool val)
+{
+    if (r == NULL)
+        return;
+
+	PR_Lock(r->repl_lock);
+	r->tombstone_reap_stop = val;
+	PR_Unlock(r->repl_lock);
+}
+
 /* replica just came back online, probably after data was reloaded */
 void 
 replica_enable_replication (Replica *r)

ldap/servers/plugins/replication/repl5_replica_config.c

@@ -442,28 +442,32 @@ replica_config_search (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter
 {
     multimaster_mtnode_extension *mtnode_ext;   
     int changeCount = 0;
+	PRBool reapActive = PR_FALSE;
     char val [64];
 
 	/* add attribute that contains number of entries in the changelog for this replica */
 	
     PR_Lock (s_configLock);
     
-	/* if we have no changelog - we have no changes */
-	if (cl5GetState () == CL5_STATE_OPEN)
-	{    
-		mtnode_ext = _replica_config_get_mtnode_ext (e);
-		PR_ASSERT (mtnode_ext);
+	mtnode_ext = _replica_config_get_mtnode_ext (e);
+	PR_ASSERT (mtnode_ext);
     
-		if (mtnode_ext->replica)
-		{
-            object_acquire (mtnode_ext->replica);
-			changeCount = cl5GetOperationCount (mtnode_ext->replica);        
-			object_release (mtnode_ext->replica);
+	if (mtnode_ext->replica) {
+		Replica *replica;
+		object_acquire (mtnode_ext->replica);
+		if (cl5GetState () == CL5_STATE_OPEN) {    
+			changeCount = cl5GetOperationCount (mtnode_ext->replica);
 		}
-    }
+		replica = (Replica*)object_get_data (mtnode_ext->replica);
+		if (replica) {
+			reapActive = replica_get_tombstone_reap_active(replica);
+		}
+		object_release (mtnode_ext->replica);
+	}
 
     sprintf (val, "%d", changeCount);
     slapi_entry_add_string (e, type_replicaChangeCount, val);
+	slapi_entry_attr_set_int(e, "nsds5replicaReapActive", (int)reapActive);
 
     PR_Unlock (s_configLock);
 

ldap/servers/plugins/replication/repl_connext.c

@@ -75,6 +75,9 @@ void consumer_connection_extension_destructor (void *ext, void *object, void *pa
 						"of replicated area.\n");
 				}
 				slapi_pblock_destroy(pb);
+
+				/* allow reaping again */
+				replica_set_tombstone_reap_stop(r, PR_FALSE);
 			}
 			replica_relinquish_exclusive_access(r, connid, -1);
             object_release ((Object*)connext->replica_acquired);

ldap/servers/plugins/replication/repl_extop.c

@@ -776,6 +776,10 @@ multimaster_extop_StartNSDS50ReplicationRequest(Slapi_PBlock *pb)
 		char *mtnstate = slapi_mtn_get_state(repl_root_sdn);
 		char **mtnreferral = slapi_mtn_get_referral(repl_root_sdn);
 
+		/* richm 20041118 - we do not want to reap tombstones while there is
+		   a total update in progress, so shut it down */
+		replica_set_tombstone_reap_stop(replica, PR_TRUE);
+
 		/* richm 20010831 - set the mapping tree to the referral state *before*
 		   we invoke slapi_start_bulk_import - see bug 556992 -
 		   slapi_start_bulk_import sets the database offline, if an operation comes
@@ -839,6 +843,9 @@ send_response:
 			connid, opid,
 			(replica ? slapi_sdn_get_dn(replica_get_root(replica)) : "unknown"),
 			protocol_response2string (response), purlstr);
+
+		/* enable tombstone reap again since the total update failed */
+		replica_set_tombstone_reap_stop(replica, PR_FALSE);
     }
 	/* Send the response */
 	if ((resp_bere = der_alloc()) == NULL)
@@ -1043,6 +1050,10 @@ multimaster_extop_EndNSDS50ReplicationRequest(Slapi_PBlock *pb)
 			    /* ONREPL code that dealt with new RUV, etc was moved into the code
                    that enables replication when a backend comes back online. This
                    code is called once the bulk import is finished */
+
+				/* allow reaping again */
+				replica_set_tombstone_reap_stop(r, PR_FALSE);
+
 			}
 			else if (connext->repl_protocol_version == REPL_PROTOCOL_50_INCREMENTAL)
 			{