From 59f03e332061b2c68bb53eed5949ddcfdc563300 Mon Sep 17 00:00:00 2001

From: Mark Reynolds <mreynolds@redhat.com>

Date: Wed, 7 Aug 2019 20:36:53 -0400

Subject: [PATCH] Issue 50538 - cleanAllRUV task limit is not enforced for

 replicated tasks

Bug Description:

There is a hard limit of 64 concurrent cleanAllRUV tasks, but this limit is

only enforced when creating "new" tasks. It was not enforced when a task was

received via an extended operation. There were also race conditions in the

existing logic that allowed the array of cleaned rids to get corrupted . This

allowed for a very large number of task threads to be created.

Fix Description:

Maintain a new counter to keep track of the number of clean and abort threads

to make sure it never over runs the rid array buffers.

relates: https://pagure.io/389-ds-base/issue/50538

Reviewed by: lkrispenz(Thanks!)

---

 .../suites/replication/cleanallruv_test.py    |  47 +++-

 ldap/servers/plugins/replication/repl5.h      |   7 +-

 .../replication/repl5_replica_config.c        | 247 ++++++++++--------

 ldap/servers/plugins/replication/repl_extop.c |  19 +-

 4 files changed, 202 insertions(+), 118 deletions(-)

diff --git a/dirsrvtests/tests/suites/replication/cleanallruv_test.py b/dirsrvtests/tests/suites/replication/cleanallruv_test.py

index 620a53e1a..43801dd52 100644

--- a/dirsrvtests/tests/suites/replication/cleanallruv_test.py

+++ b/dirsrvtests/tests/suites/replication/cleanallruv_test.py

@@ -1,5 +1,5 @@

 # --- BEGIN COPYRIGHT BLOCK ---

-# Copyright (C) 2016 Red Hat, Inc.

+# Copyright (C) 2019 Red Hat, Inc.

 # All rights reserved.

 #

 # License: GPL (version 3 or any later version).

@@ -7,7 +7,6 @@

 # --- END COPYRIGHT BLOCK ---

 #

 import threading

-

 import pytest

 from lib389.tasks import *

 from lib389.utils import *

@@ -859,6 +858,50 @@ def test_multiple_tasks_with_force(topology_m4):

     restore_master4(topology_m4)

+def test_max_tasks(topology_m4):

+    """Test we can not create more than 64 cleaning tasks

+

+    :id: c34d0b40-3c3e-4f53-8656-5e4c2a310a1f

+    :setup: Replication setup with four masters

+    :steps:

+        1. Stop masters 3 & 4

+        2. Create over 64 tasks between m1 and m2

+        3. Check logs to see if (>65) tasks were rejected

+

+    :expectedresults:

+        1. Success

+        2. Success

+        3. Success

+    """

+

+    # Stop masters 3 & 4

+    m1 = topology_m4.ms["master1"]

+    m2 = topology_m4.ms["master2"]

+    m3 = topology_m4.ms["master3"]

+    m4 = topology_m4.ms["master4"]

+    m3.stop()

+    m4.stop()

+

+    # Add over 64 tasks between master1 & 2 to try to exceed the 64 task limit

+    for i in range(1, 64):

+        cruv_task = CleanAllRUVTask(m1)

+        cruv_task.create(properties={

+            'replica-id': str(i),

+            'replica-base-dn': DEFAULT_SUFFIX,

+            'replica-force-cleaning': 'no',  # This forces these tasks to stick around

+        })

+        cruv_task = CleanAllRUVTask(m2)

+        cruv_task.create(properties={

+            'replica-id': "10" + str(i),

+            'replica-base-dn': DEFAULT_SUFFIX,

+            'replica-force-cleaning': 'yes',  # This allows the tasks to propagate

+        })

+

+    # Check the errors log for our error message in master 1

+    assert m1.searchErrorsLog('Exceeded maximum number of active CLEANALLRUV tasks')

+>>>>>>> ab24aa4cb... Issue 50538 - cleanAllRUV task limit is not enforced for replicated tasks

+

+

 if __name__ == '__main__':

     # Run isolated

     # -s for DEBUG mode

diff --git a/ldap/servers/plugins/replication/repl5.h b/ldap/servers/plugins/replication/repl5.h

index 3c7f06f36..b06c6fbf4 100644

--- a/ldap/servers/plugins/replication/repl5.h

+++ b/ldap/servers/plugins/replication/repl5.h

@@ -80,6 +80,8 @@

 #define CLEANRUV_FINISHED  "finished"

 #define CLEANRUV_CLEANING  "cleaning"

 #define CLEANRUV_NO_MAXCSN "no maxcsn"

+#define CLEANALLRUV_ID "CleanAllRUV Task"

+#define ABORT_CLEANALLRUV_ID "Abort CleanAllRUV Task"

 /* DS 5.0 replication protocol error codes */

 #define NSDS50_REPL_REPLICA_READY             0x00  /* Replica ready, go ahead */

@@ -784,6 +786,7 @@ void multimaster_mtnode_construct_replicas(void);

 void multimaster_be_state_change(void *handle, char *be_name, int old_be_state, int new_be_state);

 #define CLEANRIDSIZ 64 /* maximum number for concurrent CLEANALLRUV tasks */

+#define CLEANRID_BUFSIZ 128

 typedef struct _cleanruv_data

 {

@@ -815,6 +818,8 @@ int get_replica_type(Replica *r);

 int replica_execute_cleanruv_task_ext(Object *r, ReplicaId rid);

 void add_cleaned_rid(cleanruv_data *data, char *maxcsn);

 int is_cleaned_rid(ReplicaId rid);

+int32_t check_and_set_cleanruv_task_count(ReplicaId rid);

+int32_t check_and_set_abort_cleanruv_task_count(void);

 int replica_cleanall_ruv_abort(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *eAfter, int *returncode, char *returntext, void *arg);

 void replica_cleanallruv_thread_ext(void *arg);

 void stop_ruv_cleaning(void);

@@ -833,8 +838,6 @@ void set_cleaned_rid(ReplicaId rid);

 void cleanruv_log(Slapi_Task *task, int rid, char *task_type, int sev_level, char *fmt, ...);

 char *replica_cleanallruv_get_local_maxcsn(ReplicaId rid, char *base_dn);

-

-

 /* replutil.c */

 LDAPControl *create_managedsait_control(void);

 LDAPControl *create_backend_control(Slapi_DN *sdn);

diff --git a/ldap/servers/plugins/replication/repl5_replica_config.c b/ldap/servers/plugins/replication/repl5_replica_config.c

index 62bfcf6ce..80a079784 100644

--- a/ldap/servers/plugins/replication/repl5_replica_config.c

+++ b/ldap/servers/plugins/replication/repl5_replica_config.c

@@ -30,17 +30,18 @@

 #define CLEANALLRUV "CLEANALLRUV"

 #define CLEANALLRUVLEN 11

 #define REPLICA_RDN "cn=replica"

-#define CLEANALLRUV_ID "CleanAllRUV Task"

-#define ABORT_CLEANALLRUV_ID "Abort CleanAllRUV Task"

 int slapi_log_urp = SLAPI_LOG_REPL;

-static ReplicaId cleaned_rids[CLEANRIDSIZ + 1] = {0};

-static ReplicaId pre_cleaned_rids[CLEANRIDSIZ + 1] = {0};

-static ReplicaId aborted_rids[CLEANRIDSIZ + 1] = {0};

-static Slapi_RWLock *rid_lock = NULL;

-static Slapi_RWLock *abort_rid_lock = NULL;

+static ReplicaId cleaned_rids[CLEANRID_BUFSIZ] = {0};

+static ReplicaId pre_cleaned_rids[CLEANRID_BUFSIZ] = {0};

+static ReplicaId aborted_rids[CLEANRID_BUFSIZ] = {0};

+static PRLock *rid_lock = NULL;

+static PRLock *abort_rid_lock = NULL;

 static PRLock *notify_lock = NULL;

 static PRCondVar *notify_cvar = NULL;

+static PRLock *task_count_lock = NULL;

+static int32_t clean_task_count = 0;

+static int32_t abort_task_count = 0;

 /* Forward Declartions */

 static int replica_config_add(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *entryAfter, int *returncode, char *returntext, void *arg);

@@ -67,8 +68,6 @@ static int replica_cleanallruv_send_abort_extop(Repl_Agmt *ra, Slapi_Task *task,

 static int replica_cleanallruv_check_maxcsn(Repl_Agmt *agmt, char *basedn, char *rid_text, char *maxcsn, Slapi_Task *task);

 static int replica_cleanallruv_replica_alive(Repl_Agmt *agmt);

 static int replica_cleanallruv_check_ruv(char *repl_root, Repl_Agmt *ra, char *rid_text, Slapi_Task *task, char *force);

-static int get_cleanruv_task_count(void);

-static int get_abort_cleanruv_task_count(void);

 static int replica_cleanup_task(Object *r, const char *task_name, char *returntext, int apply_mods);

 static int replica_task_done(Replica *replica);

 static void delete_cleaned_rid_config(cleanruv_data *data);

@@ -114,20 +113,27 @@ replica_config_init()

                       PR_GetError());

         return -1;

     }

-    rid_lock = slapi_new_rwlock();

+    rid_lock = PR_NewLock();

     if (rid_lock == NULL) {

         slapi_log_err(SLAPI_LOG_ERR, repl_plugin_name, "replica_config_init - "

                                                        "Failed to create rid_lock; NSPR error - %d\n",

                       PR_GetError());

         return -1;

     }

-    abort_rid_lock = slapi_new_rwlock();

+    abort_rid_lock = PR_NewLock();

     if (abort_rid_lock == NULL) {

         slapi_log_err(SLAPI_LOG_ERR, repl_plugin_name, "replica_config_init - "

                                                        "Failed to create abort_rid_lock; NSPR error - %d\n",

                       PR_GetError());

         return -1;

     }

+    task_count_lock = PR_NewLock();

+    if (task_count_lock == NULL) {

+        slapi_log_err(SLAPI_LOG_ERR, repl_plugin_name, "replica_config_init - "

+                                                       "Failed to create task_count_lock; NSPR error - %d\n",

+                      PR_GetError());

+        return -1;

+    }

     if ((notify_lock = PR_NewLock()) == NULL) {

         slapi_log_err(SLAPI_LOG_ERR, repl_plugin_name, "replica_config_init - "

                                                        "Failed to create notify lock; NSPR error - %d\n",

@@ -1483,12 +1489,6 @@ replica_execute_cleanall_ruv_task(Object *r, ReplicaId rid, Slapi_Task *task, co

     cleanruv_log(pre_task, rid, CLEANALLRUV_ID, SLAPI_LOG_INFO, "Initiating CleanAllRUV Task...");

-    if (get_cleanruv_task_count() >= CLEANRIDSIZ) {

-        /* we are already running the maximum number of tasks */

-        cleanruv_log(pre_task, rid, CLEANALLRUV_ID, SLAPI_LOG_ERR,

-                     "Exceeded maximum number of active CLEANALLRUV tasks(%d)", CLEANRIDSIZ);

-        return LDAP_UNWILLING_TO_PERFORM;

-    }

     /*

      *  Grab the replica

      */

@@ -1540,6 +1540,13 @@ replica_execute_cleanall_ruv_task(Object *r, ReplicaId rid, Slapi_Task *task, co

         goto fail;

     }

+    if (check_and_set_cleanruv_task_count(rid) != LDAP_SUCCESS) {

+        cleanruv_log(NULL, rid, CLEANALLRUV_ID, SLAPI_LOG_ERR,

+                     "Exceeded maximum number of active CLEANALLRUV tasks(%d)", CLEANRIDSIZ);

+        rc = LDAP_UNWILLING_TO_PERFORM;

+        goto fail;

+    }

+

     /*

      *  Launch the cleanallruv thread.  Once all the replicas are cleaned it will release the rid

      */

@@ -1547,6 +1554,9 @@ replica_execute_cleanall_ruv_task(Object *r, ReplicaId rid, Slapi_Task *task, co

     if (data == NULL) {

         cleanruv_log(pre_task, rid, CLEANALLRUV_ID, SLAPI_LOG_ERR, "Failed to allocate cleanruv_data.  Aborting task.");

         rc = -1;

+        PR_Lock(task_count_lock);

+        clean_task_count--;

+        PR_Unlock(task_count_lock);

         goto fail;

     }

     data->repl_obj = r;

@@ -1629,13 +1639,13 @@ replica_cleanallruv_thread(void *arg)

     int aborted = 0;

     int rc = 0;

-    if (!data || slapi_is_shutting_down()) {

-        return; /* no data */

-    }

-

     /* Increase active thread count to prevent a race condition at server shutdown */

     g_incr_active_threadcnt();

+    if (!data || slapi_is_shutting_down()) {

+        goto done;

+    }

+

     if (data->task) {

         slapi_task_inc_refcount(data->task);

         slapi_log_err(SLAPI_LOG_PLUGIN, repl_plugin_name,

@@ -1682,16 +1692,13 @@ replica_cleanallruv_thread(void *arg)

         slapi_task_begin(data->task, 1);

     }

     /*

-     *  Presetting the rid prevents duplicate thread creation, but allows the db and changelog to still

-     *  process updates from the rid.

-     *  set_cleaned_rid() blocks updates, so we don't want to do that... yet unless we are in force mode.

-     *  If we are forcing a clean independent of state of other servers for this RID we can set_cleaned_rid()

+     *  We have already preset this rid, but if we are forcing a clean independent of state

+     *  of other servers for this RID we can set_cleaned_rid()

      */

     if (data->force) {

         set_cleaned_rid(data->rid);

-    } else {

-        preset_cleaned_rid(data->rid);

     }

+

     rid_text = slapi_ch_smprintf("%d", data->rid);

     csn_as_string(data->maxcsn, PR_FALSE, csnstr);

     /*

@@ -1861,6 +1868,9 @@ done:

     /*

      *  If the replicas are cleaned, release the rid

      */

+    if (slapi_is_shutting_down()) {

+        stop_ruv_cleaning();

+    }

     if (!aborted && !slapi_is_shutting_down()) {

         /*

          * Success - the rid has been cleaned!

@@ -1879,10 +1889,9 @@ done:

         } else {

             cleanruv_log(data->task, data->rid, CLEANALLRUV_ID, SLAPI_LOG_INFO, "Propagated task does not delete Keep alive entry (%d).", data->rid);

         }

-

         clean_agmts(data);

         remove_cleaned_rid(data->rid);

-        cleanruv_log(data->task, data->rid, CLEANALLRUV_ID, SLAPI_LOG_INFO, "Successfully cleaned rid(%d).", data->rid);

+        cleanruv_log(data->task, data->rid, CLEANALLRUV_ID, SLAPI_LOG_INFO, "Successfully cleaned rid(%d)", data->rid);

     } else {

         /*

          *  Shutdown or abort

@@ -1915,6 +1924,10 @@ done:

     slapi_ch_free_string(&data->force);

     slapi_ch_free_string(&rid_text);

     slapi_ch_free((void **)&data);

+    /* decrement task count */

+    PR_Lock(task_count_lock);

+    clean_task_count--;

+    PR_Unlock(task_count_lock);

     g_decr_active_threadcnt();

 }

@@ -2414,16 +2427,14 @@ replica_send_cleanruv_task(Repl_Agmt *agmt, cleanruv_data *clean_data)

 int

 is_cleaned_rid(ReplicaId rid)

 {

-    int i;

-

-    slapi_rwlock_rdlock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ && cleaned_rids[i] != 0; i++) {

+    PR_Lock(rid_lock);

+    for (size_t i = 0; i < CLEANRID_BUFSIZ; i++) {

         if (rid == cleaned_rids[i]) {

-            slapi_rwlock_unlock(rid_lock);

+            PR_Unlock(rid_lock);

             return 1;

         }

     }

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(rid_lock);

     return 0;

 }

@@ -2431,16 +2442,14 @@ is_cleaned_rid(ReplicaId rid)

 int

 is_pre_cleaned_rid(ReplicaId rid)

 {

-    int i;

-

-    slapi_rwlock_rdlock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ && pre_cleaned_rids[i] != 0; i++) {

+    PR_Lock(rid_lock);

+    for (size_t i = 0; i < CLEANRID_BUFSIZ; i++) {

         if (rid == pre_cleaned_rids[i]) {

-            slapi_rwlock_unlock(rid_lock);

+            PR_Unlock(rid_lock);

             return 1;

         }

     }

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(rid_lock);

     return 0;

 }

@@ -2453,14 +2462,14 @@ is_task_aborted(ReplicaId rid)

     if (rid == 0) {

         return 0;

     }

-    slapi_rwlock_rdlock(abort_rid_lock);

-    for (i = 0; i < CLEANRIDSIZ && aborted_rids[i] != 0; i++) {

+    PR_Lock(abort_rid_lock);

+    for (i = 0; i < CLEANRID_BUFSIZ && aborted_rids[i] != 0; i++) {

         if (rid == aborted_rids[i]) {

-            slapi_rwlock_unlock(abort_rid_lock);

+            PR_Unlock(abort_rid_lock);

             return 1;

         }

     }

-    slapi_rwlock_unlock(abort_rid_lock);

+    PR_Unlock(abort_rid_lock);

     return 0;

 }

@@ -2469,15 +2478,14 @@ preset_cleaned_rid(ReplicaId rid)

 {

     int i;

-    slapi_rwlock_wrlock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ; i++) {

+    PR_Lock(rid_lock);

+    for (i = 0; i < CLEANRID_BUFSIZ && pre_cleaned_rids[i] != rid; i++) {

         if (pre_cleaned_rids[i] == 0) {

             pre_cleaned_rids[i] = rid;

-            pre_cleaned_rids[i + 1] = 0;

             break;

         }

     }

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(rid_lock);

 }

 /*

@@ -2490,14 +2498,13 @@ set_cleaned_rid(ReplicaId rid)

 {

     int i;

-    slapi_rwlock_wrlock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ; i++) {

+    PR_Lock(rid_lock);

+    for (i = 0; i < CLEANRID_BUFSIZ && cleaned_rids[i] != rid; i++) {

         if (cleaned_rids[i] == 0) {

             cleaned_rids[i] = rid;

-            cleaned_rids[i + 1] = 0;

         }

     }

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(rid_lock);

 }

 /*

@@ -2569,15 +2576,14 @@ add_aborted_rid(ReplicaId rid, Replica *r, char *repl_root)

     int rc;

     int i;

-    slapi_rwlock_wrlock(abort_rid_lock);

-    for (i = 0; i < CLEANRIDSIZ; i++) {

+    PR_Lock(abort_rid_lock);

+    for (i = 0; i < CLEANRID_BUFSIZ; i++) {

         if (aborted_rids[i] == 0) {

             aborted_rids[i] = rid;

-            aborted_rids[i + 1] = 0;

             break;

         }

     }

-    slapi_rwlock_unlock(abort_rid_lock);

+    PR_Unlock(abort_rid_lock);

     /*

      *  Write the rid to the config entry

      */

@@ -2620,21 +2626,24 @@ delete_aborted_rid(Replica *r, ReplicaId rid, char *repl_root, int skip)

     char *data;

     char *dn;

     int rc;

-    int i;

     if (r == NULL)

         return;

     if (skip) {

         /* skip the deleting of the config, and just remove the in memory rid */

-        slapi_rwlock_wrlock(abort_rid_lock);

-        for (i = 0; i < CLEANRIDSIZ && aborted_rids[i] != rid; i++)

-            ; /* found rid, stop */

-        for (; i < CLEANRIDSIZ; i++) {

-            /* rewrite entire array */

-            aborted_rids[i] = aborted_rids[i + 1];

-        }

-        slapi_rwlock_unlock(abort_rid_lock);

+        ReplicaId new_abort_rids[CLEANRID_BUFSIZ] = {0};

+        int32_t idx = 0;

+

+        PR_Lock(abort_rid_lock);

+        for (size_t i = 0; i < CLEANRID_BUFSIZ; i++) {

+            if (aborted_rids[i] != rid) {

+                new_abort_rids[idx] = aborted_rids[i];

+                idx++;

+            }

+        }

+        memcpy(aborted_rids, new_abort_rids, sizeof(new_abort_rids));

+        PR_Unlock(abort_rid_lock);

     } else {

         /* only remove the config, leave the in-memory rid */

         dn = replica_get_dn(r);

@@ -2792,27 +2801,31 @@ bail:

 void

 remove_cleaned_rid(ReplicaId rid)

 {

-    int i;

-    /*

-     *  Remove this rid, and optimize the array

-     */

-    slapi_rwlock_wrlock(rid_lock);

+    ReplicaId new_cleaned_rids[CLEANRID_BUFSIZ] = {0};

+    ReplicaId new_pre_cleaned_rids[CLEANRID_BUFSIZ] = {0};

+    size_t idx = 0;

+

+    PR_Lock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ && cleaned_rids[i] != rid; i++)

-        ; /* found rid, stop */

-    for (; i < CLEANRIDSIZ; i++) {

-        /* rewrite entire array */

-        cleaned_rids[i] = cleaned_rids[i + 1];

+    for (size_t i = 0; i < CLEANRID_BUFSIZ; i++) {

+        if (cleaned_rids[i] != rid) {

+            new_cleaned_rids[idx] = cleaned_rids[i];

+            idx++;

+        }

     }

+    memcpy(cleaned_rids, new_cleaned_rids, sizeof(new_cleaned_rids));

+

     /* now do the preset cleaned rids */

-    for (i = 0; i < CLEANRIDSIZ && pre_cleaned_rids[i] != rid; i++)

-        ; /* found rid, stop */

-    for (; i < CLEANRIDSIZ; i++) {

-        /* rewrite entire array */

-        pre_cleaned_rids[i] = pre_cleaned_rids[i + 1];

+    idx = 0;

+    for (size_t i = 0; i < CLEANRID_BUFSIZ; i++) {

+        if (pre_cleaned_rids[i] != rid) {

+            new_pre_cleaned_rids[idx] = pre_cleaned_rids[i];

+            idx++;

+        }

     }

+    memcpy(pre_cleaned_rids, new_pre_cleaned_rids, sizeof(new_pre_cleaned_rids));

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(rid_lock);

 }

 /*

@@ -2840,16 +2853,6 @@ replica_cleanall_ruv_abort(Slapi_PBlock *pb __attribute__((unused)),

     char *ridstr = NULL;

     int rc = SLAPI_DSE_CALLBACK_OK;

-    if (get_abort_cleanruv_task_count() >= CLEANRIDSIZ) {

-        /* we are already running the maximum number of tasks */

-        PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE,

-                    "Exceeded maximum number of active ABORT CLEANALLRUV tasks(%d)",

-                    CLEANRIDSIZ);

-        cleanruv_log(task, -1, ABORT_CLEANALLRUV_ID, SLAPI_LOG_ERR, "%s", returntext);

-        *returncode = LDAP_OPERATIONS_ERROR;

-        return SLAPI_DSE_CALLBACK_ERROR;

-    }

-

     /* allocate new task now */

     task = slapi_new_task(slapi_entry_get_ndn(e));

@@ -2934,6 +2937,16 @@ replica_cleanall_ruv_abort(Slapi_PBlock *pb __attribute__((unused)),

          */

         certify_all = "no";

     }

+

+    if (check_and_set_abort_cleanruv_task_count() != LDAP_SUCCESS) {

+        /* we are already running the maximum number of tasks */

+        PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE,

+                    "Exceeded maximum number of active ABORT CLEANALLRUV tasks(%d)",

+                    CLEANRIDSIZ);

+        cleanruv_log(task, -1, ABORT_CLEANALLRUV_ID, SLAPI_LOG_ERR, "%s", returntext);

+        *returncode = LDAP_UNWILLING_TO_PERFORM;

+        goto out;

+    }

     /*

      *  Create payload

      */

@@ -3142,6 +3155,9 @@ done:

     slapi_ch_free_string(&data->certify);

     slapi_sdn_free(&data->sdn);

     slapi_ch_free((void **)&data);

+    PR_Lock(task_count_lock);

+    abort_task_count--;

+    PR_Unlock(task_count_lock);

     g_decr_active_threadcnt();

 }

@@ -3493,36 +3509,43 @@ replica_cleanallruv_check_ruv(char *repl_root, Repl_Agmt *agmt, char *rid_text,

     return rc;

 }

-static int

-get_cleanruv_task_count(void)

+/*

+ * Before starting a cleanAllRUV task make sure there are not

+ * too many task threads already running.  If everything is okay

+ * also pre-set the RID now so rebounding extended ops do not

+ * try to clean it over and over.

+ */

+int32_t

+check_and_set_cleanruv_task_count(ReplicaId rid)

 {

-    int i, count = 0;

+    int32_t rc = 0;

-    slapi_rwlock_wrlock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ; i++) {

-        if (pre_cleaned_rids[i] != 0) {

-            count++;

-        }

+    PR_Lock(task_count_lock);

+    if (clean_task_count >= CLEANRIDSIZ) {

+        rc = -1;

+    } else {

+        clean_task_count++;

+        preset_cleaned_rid(rid);

     }

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(task_count_lock);

-    return count;

+    return rc;

 }

-static int

-get_abort_cleanruv_task_count(void)

+int32_t

+check_and_set_abort_cleanruv_task_count(void)

 {

-    int i, count = 0;

+    int32_t rc = 0;

-    slapi_rwlock_wrlock(rid_lock);

-    for (i = 0; i < CLEANRIDSIZ; i++) {

-        if (aborted_rids[i] != 0) {

-            count++;

+    PR_Lock(task_count_lock);

+    if (abort_task_count > CLEANRIDSIZ) {

+            rc = -1;

+        } else {

+            abort_task_count++;

         }

-    }

-    slapi_rwlock_unlock(rid_lock);

+    PR_Unlock(task_count_lock);

-    return count;

+    return rc;

 }

 /*

diff --git a/ldap/servers/plugins/replication/repl_extop.c b/ldap/servers/plugins/replication/repl_extop.c

index 68e2544b4..0c2abb6d5 100644

--- a/ldap/servers/plugins/replication/repl_extop.c

+++ b/ldap/servers/plugins/replication/repl_extop.c

@@ -1393,6 +1393,12 @@ multimaster_extop_abort_cleanruv(Slapi_PBlock *pb)

         rc = LDAP_OPERATIONS_ERROR;

         goto out;

     }

+    if (check_and_set_abort_cleanruv_task_count() != LDAP_SUCCESS) {

+        cleanruv_log(NULL, rid, CLEANALLRUV_ID, SLAPI_LOG_ERR,

+                     "Exceeded maximum number of active abort CLEANALLRUV tasks(%d)", CLEANRIDSIZ);

+        rc = LDAP_UNWILLING_TO_PERFORM;

+        goto out;

+    }

     /*

      *  Prepare the abort data

      */

@@ -1499,6 +1505,7 @@ multimaster_extop_cleanruv(Slapi_PBlock *pb)

     if (force == NULL) {

         force = "no";

     }

+

     maxcsn = csn_new();

     csn_init_by_string(maxcsn, csnstr);

     /*

@@ -1535,13 +1542,21 @@ multimaster_extop_cleanruv(Slapi_PBlock *pb)

         goto free_and_return;

     }

+    if (check_and_set_cleanruv_task_count((ReplicaId)rid) != LDAP_SUCCESS) {

+        cleanruv_log(NULL, rid, CLEANALLRUV_ID, SLAPI_LOG_ERR,

+                     "Exceeded maximum number of active CLEANALLRUV tasks(%d)", CLEANRIDSIZ);

+        rc = LDAP_UNWILLING_TO_PERFORM;

+        goto free_and_return;

+    }

+

     if (replica_get_type(r) != REPLICA_TYPE_READONLY) {

         /*

          *  Launch the cleanruv monitoring thread.  Once all the replicas are cleaned it will release the rid

          *

          *  This will also release mtnode_ext->replica

          */

-        slapi_log_err(SLAPI_LOG_INFO, repl_plugin_name, "multimaster_extop_cleanruv - CleanAllRUV Task - Launching cleanAllRUV thread...\n");

+

+        cleanruv_log(NULL, rid, CLEANALLRUV_ID, SLAPI_LOG_ERR, "Launching cleanAllRUV thread...\n");

         data = (cleanruv_data *)slapi_ch_calloc(1, sizeof(cleanruv_data));

         if (data == NULL) {

             slapi_log_err(SLAPI_LOG_ERR, repl_plugin_name, "multimaster_extop_cleanruv - CleanAllRUV Task - Failed to allocate "

@@ -1635,7 +1650,7 @@ free_and_return:

         ber_printf(resp_bere, "{s}", CLEANRUV_ACCEPTED);

         ber_flatten(resp_bere, &resp_bval);

         slapi_pblock_set(pb, SLAPI_EXT_OP_RET_VALUE, resp_bval);

-        slapi_send_ldap_result(pb, LDAP_SUCCESS, NULL, NULL, 0, NULL);

+        slapi_send_ldap_result(pb, rc, NULL, NULL, 0, NULL);

         /* resp_bere */

         if (NULL != resp_bere) {

             ber_free(resp_bere, 1);

-- 

2.21.0