From c5e89132f2ee017aa026f7b8009a2976b627fa3a Mon Sep 17 00:00:00 2001

From: Thierry Bordaz <tbordaz@redhat.com>

Date: Tue, 26 Apr 2016 13:00:44 +0300

Subject: [PATCH 2/4] backend: support backend shutdown for priming thread

 cancellation

Launching a separate thread to populate map cache has a side effect that

the thread could be scheduled to execute over a shutdown time. If LDAP server

received the request to shutdown, we need to stop processing the

original source and shut the priming thread.

Resolves: rhbz#1327197

---

 src/back-shr.c | 104 +++++++++++++++++++++++++++++++++++++++++++++++----------

 src/back-shr.h |   1 +

 src/backend.h  |   1 +

 src/plugin.h   |   3 ++

 4 files changed, 92 insertions(+), 17 deletions(-)

diff --git a/src/back-shr.c b/src/back-shr.c

index 0157582..7842e05 100644

--- a/src/back-shr.c

+++ b/src/back-shr.c

@@ -665,25 +665,34 @@ backend_shr_get_vattr_sdnlist(struct plugin_state *state,

 }

 struct backend_shr_data_init_cbdata {

-	Slapi_PBlock *parent_pb;

 	struct plugin_state *state;

 	const char *filter;

 };

 #define PLUGIN_SCAN_DELAY 5

-static void

-backend_shr_data_initialize_thread_cb(void *arg)

+static void *

+backend_shr_data_initialize_thread_cb(struct wrapped_thread *t)

 {

-	struct backend_shr_data_init_cbdata *cbdata = (struct backend_shr_data_init_cbdata *)arg;

+	struct backend_shr_data_init_cbdata *cbdata = wrap_thread_arg(t);

 	Slapi_PBlock *pb = NULL;

 	struct backend_set_config_entry_add_cbdata set_cbdata;

 	int result = 0, i = 0;

 	Slapi_Entry **entries = NULL;

 	struct plugin_state *state = NULL;

+	/* We have to be cautious here as the thread can be executed after

+	 * the plugin received a shutdown request, thus a number of checks here. */

+	if (slapi_is_shutting_down()) {

+		return NULL;

+	}

+

 	if (cbdata == NULL) {

-		return;

+		return NULL;

+	}

+

+	if ((cbdata->state == NULL) || (cbdata->state->plugin_base == NULL)) {

+		return NULL;

 	}

 	state = cbdata->state;

@@ -694,13 +703,23 @@ backend_shr_data_initialize_thread_cb(void *arg)

 	 * so we just wait some time. */

 	DS_Sleep(PR_SecondsToInterval(PLUGIN_SCAN_DELAY));

-	backend_update_params(cbdata->parent_pb, state);

+	if (slapi_is_shutting_down()) {

+		return NULL;

+	}

+

+	if (state->plugin_base == NULL) {

+		return NULL;

+	}

+

+	pb = wrap_pblock_new(NULL);

+	backend_update_params(pb, state);

+	slapi_pblock_destroy(pb);

 	slapi_log_error(SLAPI_LOG_PLUGIN,

 			state->plugin_desc->spd_id,

 			"searching under \"%s\" for configuration\n",

 			state->plugin_base);

-	pb = wrap_pblock_new(cbdata->parent_pb);

+	pb = wrap_pblock_new(NULL);

 	slapi_search_internal_set_pb(pb,

 				     state->plugin_base,

 				     LDAP_SCOPE_ONELEVEL,

@@ -717,10 +736,10 @@ backend_shr_data_initialize_thread_cb(void *arg)

 	/* Do a search and collect found entries to avoid locking the backends */

 	if (slapi_search_internal_pb(pb) == 0) {

 		if (map_wrlock() != 0) {

-			slapi_log_error(SLAPI_LOG_PLUGIN,

+			slapi_log_error(SLAPI_LOG_FATAL,

 				state->plugin_desc->spd_id,

 				"failed to search under \"%s\" for "

-				"configuration: failed to acquire a lock\n",

+				"configuration: failed to acquire a write lock to a map\n",

 				state->plugin_base);

 			goto done_with_lock;

 		}

@@ -728,6 +747,11 @@ backend_shr_data_initialize_thread_cb(void *arg)

 		if (result == 0) {

 			slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &entries);

 			for (i = 0; entries[i] != NULL; i++) {

+				/* We may be scheduled to run over shutdown time, exit early */

+				if (slapi_is_shutting_down()) {

+					map_unlock();

+					goto done_with_lock;

+				}

 				backend_set_config_entry_add_cb(entries[i], &set_cbdata);

 			}

 		}

@@ -746,6 +770,7 @@ done_with_lock:

         }

 	PR_AtomicSet(&state->ready_to_serve, 1);

+        return NULL;

 }

 static void

@@ -754,11 +779,28 @@ backend_shr_data_initialize_thread(time_t when, void *arg)

 	struct backend_shr_data_init_cbdata *cbdata = (struct backend_shr_data_init_cbdata *)arg;

 	PRThread *thread = NULL;

-	/* start data import as a separate thread */

-	thread = PR_CreateThread(PR_USER_THREAD, backend_shr_data_initialize_thread_cb,

-			(void *)arg, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,

-			PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);

-	if (thread == NULL) {

+	if (slapi_is_shutting_down()) {

+		return;

+	}

+

+        if (cbdata->state->priming_mutex == NULL) {

+            /* This mutex is allocated at plugin startup

+             * Without this mutex we can not enforce that shutdown wait for priming completion

+             * This is better to skip the priming

+             */

+            slapi_log_error(SLAPI_LOG_FATAL, cbdata->state->plugin_desc->spd_id, "priming_mutex not initialized. Priming fails\n");

+            return;

+        }

+        wrap_mutex_lock(cbdata->state->priming_mutex);

+

+        if (!cbdata->state->start_priming_thread) {

+            slapi_log_error(SLAPI_LOG_PLUGIN, cbdata->state->plugin_desc->spd_id,

+                "Likely a shutdown occurred before we started \n");

+            goto done;

+        }

+

+        cbdata->state->priming_tid = wrap_start_thread(&backend_shr_data_initialize_thread_cb, arg);

+        if (cbdata->state->priming_tid == NULL) {

 		slapi_log_error(SLAPI_LOG_FATAL,

 				cbdata->state->plugin_desc->spd_id,

 				"unable to create compatibility tree scan thread!\n");

@@ -769,6 +811,8 @@ backend_shr_data_initialize_thread(time_t when, void *arg)

 				cbdata->state->plugin_desc->spd_id, PLUGIN_SCAN_DELAY);

 	}

+done:

+        wrap_mutex_unlock(cbdata->state->priming_mutex);

 }

 /* Scan for the list of configured groups and sets. */

@@ -779,6 +823,14 @@ backend_shr_startup(struct plugin_state *state,

 {

 	struct backend_shr_data_init_cbdata *cbdata = NULL;

+	if (slapi_is_shutting_down()) {

+		slapi_log_error(SLAPI_LOG_FATAL,

+				state->plugin_desc->spd_id,

+				"task for populating compatibility tree will "

+				"not be created due to upcoming server shutdown\n");

+		return;

+	}

+

 	cbdata = (struct backend_shr_data_init_cbdata *) 

 		 slapi_ch_malloc(sizeof(struct backend_shr_data_init_cbdata));

@@ -792,10 +844,10 @@ backend_shr_startup(struct plugin_state *state,

 	PR_AtomicSet(&state->ready_to_serve, 0);

 	cbdata->state = state;

-	cbdata->parent_pb = parent_pb;

 	cbdata->filter = filter;

-	/* Schedule running a callback that will create a thread */

+	/* Schedule running a callback that will create a thread

+	 * but make sure it is called a first thing when event loop is created */

 	slapi_eq_once(backend_shr_data_initialize_thread,

 		      cbdata, PR_SecondsToInterval(PLUGIN_SCAN_DELAY));

@@ -808,6 +860,24 @@ backend_shr_startup(struct plugin_state *state,

 }

+void

+backend_shr_shutdown(struct plugin_state *state)

+{

+    /* Make sure the priming thread is stopped or will not start

+     * Note: priming_mutex will not be freed because the priming thread

+     * may access it independently of the server/plugin shutdown

+     */

+    wrap_mutex_lock(state->priming_mutex);

+    state->start_priming_thread = 0; /* prevent spawing of priming thread */

+    if (state->priming_tid == NULL) {

+        /* priming thread has not yet started or failed to start */

+        slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,

+                "At shutdown, priming thread not yet started or failed to start\n");

+    } else {

+        wrap_stop_thread(state->priming_tid);

+    }

+    wrap_mutex_unlock(state->priming_mutex);

+}

 /* Process a set configuration directory entry.  Pull out the group and set

  * names which are specified in the entry and delete each in turn. */

 int

@@ -2336,7 +2406,7 @@ backend_shr_modify_cb(Slapi_PBlock *pb)

 		backend_set_config_entry_add_cb(cbdata.e_post, &set_cbdata);

 	}

 	/* Lastly, if the entry is our own entry, re-read parameters. */

-	sdn = slapi_sdn_new_dn_byref(cbdata.state->plugin_base);

+	sdn = slapi_sdn_new_dn_byval(cbdata.state->plugin_base);

 	if (sdn != NULL) {

 		if ((strcmp(slapi_entry_get_ndn(cbdata.e_pre),

 			    slapi_sdn_get_ndn(sdn)) == 0) ||

diff --git a/src/back-shr.h b/src/back-shr.h

index 44c25fe..2caea5d 100644

--- a/src/back-shr.h

+++ b/src/back-shr.h

@@ -39,6 +39,7 @@ void backend_shr_free_sdnlist(const Slapi_DN **sdnlist);

 void backend_shr_startup(struct plugin_state *state,

 			 Slapi_PBlock *pb, const char *set_filter);

+void backend_shr_shutdown(struct plugin_state *state);

 int backend_shr_betxn_postop_init(Slapi_PBlock *pb,

 				  struct plugin_state *state);

 int backend_shr_postop_init(Slapi_PBlock *pb, struct plugin_state *state);

diff --git a/src/backend.h b/src/backend.h

index 7974aae..4608d2d 100644

--- a/src/backend.h

+++ b/src/backend.h

@@ -58,6 +58,7 @@ struct backend_shr_set_data {

 /* Startup/initialization functions called through the map. */

 void backend_startup(struct slapi_pblock *pb, struct plugin_state *state);

+void backend_shutdown(struct plugin_state *state);

 int backend_init_preop(struct slapi_pblock *pb, struct plugin_state *state);

 int backend_init_betxn_preop(struct slapi_pblock *pb,

 			     struct plugin_state *state);

diff --git a/src/plugin.h b/src/plugin.h

index 7a89ac7..56d672f 100644

--- a/src/plugin.h

+++ b/src/plugin.h

@@ -34,6 +34,9 @@ struct plugin_state {

 	Slapi_PluginDesc *plugin_desc;

 	unsigned int use_be_txns: 1;

 	PRInt32 ready_to_serve;

+        struct wrapped_mutex *priming_mutex;

+        unsigned int start_priming_thread: 1;   /* flag to allow spawning of the priming thread */

+        struct wrapped_thread *priming_tid; /* priming thread pid. use to join */

 	/* NIS-specific data. */

 	struct wrapped_thread *tid;

-- 

2.7.4