From 83948ee92f791c274bec195707a54ea784927338 Mon Sep 17 00:00:00 2001
From: Ronnie Sahlberg <ronniesahlberg@gmail.com>
Date: Mon, 24 Jan 2011 11:42:50 +1100
Subject: [PATCH 3/3] LockWait congestion.

Add a dlist to track all active lockwait child processes.
Everytime creating a new lockwait handle, check if there is already an
active lockwait process for this database/key and if so,
send the new request straight to the overflow queue.

This means we will only have one active lockwaic child process for a certain key,
even if there were thousands of fetch-lock requests for this key.

When the lockwait processing finishes for the original request, the processing in d_overflow() will automagically process all remaining keys as well.

Add back a --nosetsched argument to make it easier to run under gdb
(cherry picked from commit 8a143a97a313a2e50fb409f6382c759dbd14cdcd)

[ddiss@samba.org: rebased w/o dcb7779fa52dc8f6f689c925f6f62014d6466c3b]
[ddiss@samba.org: removed --nosetsched]
---
 include/ctdb_private.h |  1 +
 server/ctdb_lockwait.c | 38 +++++++++++++++++++++++++-------------
 2 files changed, 26 insertions(+), 13 deletions(-)

Index: include/ctdb_private.h
===================================================================
--- include/ctdb_private.h.orig
+++ include/ctdb_private.h
@@ -497,6 +497,7 @@ struct ctdb_db_context {
 				  struct ctdb_ltdb_header *header,
 				  TDB_DATA data);
 	int pending_requests;
+	struct lockwait_handle *lockwait_active;
 	struct lockwait_handle *lockwait_overflow;
 };
 
Index: server/ctdb_lockwait.c
===================================================================
--- server/ctdb_lockwait.c.orig
+++ server/ctdb_lockwait.c
@@ -70,6 +70,15 @@ static void do_overflow(struct ctdb_db_c
 	}
 }
 
+static int lockwait_destructor(struct lockwait_handle *h)
+{
+	h->ctdb->statistics.pending_lockwait_calls--;
+	kill(h->child, SIGKILL);
+	h->ctdb_db->pending_requests--;
+	DLIST_REMOVE(h->ctdb_db->lockwait_active, h);
+	return 0;
+}
+
 static void lockwait_handler(struct event_context *ev, struct fd_event *fde, 
 			     uint16_t flags, void *private_data)
 {
@@ -77,7 +86,6 @@ static void lockwait_handler(struct even
 						     struct lockwait_handle);
 	void (*callback)(void *) = h->callback;
 	void *p = h->private_data;
-	pid_t child = h->child;
 	TDB_DATA key = h->key;
 	struct tdb_context *tdb = h->ctdb_db->ltdb->tdb;
 	TALLOC_CTX *tmp_ctx = talloc_new(ev);
@@ -85,9 +93,7 @@ static void lockwait_handler(struct even
 	key.dptr = talloc_memdup(tmp_ctx, key.dptr, key.dsize);
 	h->ctdb_db->pending_requests--;
 
-	talloc_set_destructor(h, NULL);
 	ctdb_latency(h->ctdb_db, "lockwait", &h->ctdb->statistics.max_lockwait_latency, h->start_time);
-	h->ctdb->statistics.pending_lockwait_calls--;
 
 	/* the handle needs to go away when the context is gone - when
 	   the handle goes away this implicitly closes the pipe, which
@@ -107,18 +113,9 @@ static void lockwait_handler(struct even
 	}
 	tdb_chainlock_unmark(tdb, key);
 
-	kill(child, SIGKILL);
 	talloc_free(tmp_ctx);
 }
 
-static int lockwait_destructor(struct lockwait_handle *h)
-{
-	h->ctdb->statistics.pending_lockwait_calls--;
-	kill(h->child, SIGKILL);
-	h->ctdb_db->pending_requests--;
-	return 0;
-}
-
 static int overflow_lockwait_destructor(struct lockwait_handle *h)
 {
 	h->ctdb->statistics.pending_lockwait_calls--;
@@ -141,7 +138,7 @@ struct lockwait_handle *ctdb_lockwait(st
 				      void (*callback)(void *private_data),
 				      void *private_data)
 {
-	struct lockwait_handle *result;
+	struct lockwait_handle *result, *i;
 	int ret;
 	pid_t parent = getpid();
 
@@ -159,6 +156,18 @@ struct lockwait_handle *ctdb_lockwait(st
 	result->ctdb_db = ctdb_db;
 	result->key = key;
 
+	/* If we already have a lockwait child for this request, then put this
+	   request on the overflow queue straight away
+	 */
+	for (i = ctdb_db->lockwait_active; i; i = i->next) {
+		if (key.dsize == i->key.dsize
+		    && memcmp(key.dptr, i->key.dptr, key.dsize) == 0) {
+			DLIST_ADD_END(ctdb_db->lockwait_overflow, result, NULL);
+			talloc_set_destructor(result, overflow_lockwait_destructor);
+			return result;
+		}
+	}
+
 	/* Don't fire off too many children at once! */
 	if (ctdb_db->pending_requests > 200) {
 		DLIST_ADD_END(ctdb_db->lockwait_overflow, result, NULL);
@@ -201,6 +210,9 @@ struct lockwait_handle *ctdb_lockwait(st
 	close(result->fd[1]);
 	set_close_on_exec(result->fd[0]);
 
+	/* This is an active lockwait child process */
+	DLIST_ADD_END(ctdb_db->lockwait_active, result, NULL);
+
 	DEBUG(DEBUG_DEBUG, (__location__ " Created PIPE FD:%d to child lockwait process\n", result->fd[0]));
 
 	ctdb_db->pending_requests++;
