Merge branch 'master' of 10.1.1.27:/shared/ctdb/ctdb-master

(This used to be ctdb commit 0452f212f2ddc1089dce83339bad0d73763f238f)

3 files changed, 70 insertions, 9 deletions

diff --git a/ctdb/include/ctdb.h b/ctdb/include/ctdb.h
index 4cdbc0f76a..35ccf83e32 100644
--- a/ctdb/include/ctdb.h
+++ b/ctdb/include/ctdb.h
@@ -315,6 +315,24 @@ bool ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
 			       ctdb_rrl_callback_t callback, void *cbdata);
 
 /**
+ * ctdb_readonlyrecordlock_async - read and lock a record for read-only access
+ * @ctdb_db: the database handle from ctdb_attachdb/ctdb_attachdb_recv.
+ * @key: the key of the record to lock.
+ * @callback: the callback once the record is locked (typesafe).
+ * @cbdata: the argument to callback()
+ *
+ * This returns true on success.  Commonly, we can obtain the record
+ * immediately and so the callback will be invoked.  Otherwise a request
+ * will be queued to ctdbd for the record.
+ *
+ * If failure is immediate, false is returned.  Otherwise, the callback
+ * may receive a NULL lock arg to indicate asynchronous failure.
+ */
+bool ctdb_readonlyrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
+			       ctdb_rrl_callback_t callback, void *cbdata);
+
+
+/**
  * ctdb_writerecord - write a locked record in a TDB
  * @ctdb_db: the database handle from ctdb_attachdb/ctdb_attachdb_recv.
  * @lock: the lock from ctdb_readrecordlock/ctdb_readrecordlock_recv
diff --git a/ctdb/libctdb/ctdb.c b/ctdb/libctdb/ctdb.c
index 72841dbd74..b555360e63 100644
--- a/ctdb/libctdb/ctdb.c
+++ b/ctdb/libctdb/ctdb.c
@@ -2,6 +2,7 @@
    core of libctdb
 
    Copyright (C) Rusty Russell 2010
+   Copyright (C) Ronnie Sahlberg 2011
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -35,6 +36,7 @@
 /* Remove type-safety macros. */
 #undef ctdb_attachdb_send
 #undef ctdb_readrecordlock_async
+#undef ctdb_readonlyrecordlock_async
 #undef ctdb_connect
 
 struct ctdb_lock {
@@ -43,6 +45,9 @@ struct ctdb_lock {
 	struct ctdb_db *ctdb_db;
 	TDB_DATA key;
 
+	/* Is this a request for read-only lock ? */
+	bool readonly;
+
 	/* This will always be set by the time user sees this. */
 	unsigned long held_magic;
 	struct ctdb_ltdb_header *hdr;
@@ -784,6 +789,14 @@ static bool try_readrecordlock(struct ctdb_lock *lock, TDB_DATA *data)
 	}
 
 	hdr = ctdb_local_fetch(lock->ctdb_db->tdb, lock->key, data);
+	if (hdr && lock->readonly && (hdr->flags & CTDB_REC_RO_HAVE_READONLY) ) {
+		DEBUG(lock->ctdb_db->ctdb, LOG_DEBUG,
+		      "ctdb_readrecordlock_async: got local lock for ro");
+		lock->held_magic = lock_magic(lock);
+		lock->hdr = hdr;
+		add_lock(lock->ctdb_db->ctdb, lock);
+		return true;
+	}
 	if (hdr && hdr->dmaster == lock->ctdb_db->ctdb->pnn) {
 		DEBUG(lock->ctdb_db->ctdb, LOG_DEBUG,
 		      "ctdb_readrecordlock_async: got local lock");
@@ -812,13 +825,13 @@ static void readrecordlock_retry(struct ctdb_connection *ctdb,
 	struct ctdb_reply_call *reply;
 	TDB_DATA data;
 
-	/* OK, we've received reply to noop migration */
-	reply = unpack_reply_call(req, CTDB_NULL_FUNC);
+	/* OK, we've received reply to fetch-with-header migration */
+	reply = unpack_reply_call(req, CTDB_FETCH_WITH_HEADER_FUNC);
 	if (!reply || reply->status != 0) {
 		if (reply) {
 			DEBUG(ctdb, LOG_ERR,
 			      "ctdb_readrecordlock_async(async):"
-			      " NULL_FUNC returned %i", reply->status);
+			      " FETCH_WITH_HEADER_FUNC returned %i", reply->status);
 		}
 		lock->callback(lock->ctdb_db, NULL, tdb_null, private);
 		ctdb_request_free(req); /* Also frees lock. */
@@ -839,9 +852,10 @@ static void readrecordlock_retry(struct ctdb_connection *ctdb,
 	DLIST_ADD(ctdb->outq, req);
 }
 
-bool
-ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
-			  ctdb_rrl_callback_t callback, void *cbdata)
+static bool
+ctdb_readrecordlock_internal(struct ctdb_db *ctdb_db, TDB_DATA key,
+			     bool readonly,
+			     ctdb_rrl_callback_t callback, void *cbdata)
 {
 	struct ctdb_request *req;
 	struct ctdb_lock *lock;
@@ -866,6 +880,7 @@ ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
 	lock->ctdb_db = ctdb_db;
 	lock->hdr = NULL;
 	lock->held_magic = 0;
+	lock->readonly = readonly;
 
 	/* Fast path. */
 	if (try_readrecordlock(lock, &data)) {
@@ -892,9 +907,13 @@ ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
 	io_elem_init_req_header(req->io, CTDB_REQ_CALL, CTDB_CURRENT_NODE,
 				new_reqid(ctdb_db->ctdb));
 
-	req->hdr.call->flags = CTDB_IMMEDIATE_MIGRATION;
+	if (lock->readonly) {
+		req->hdr.call->flags = CTDB_WANT_READONLY;
+	} else {
+		req->hdr.call->flags = CTDB_IMMEDIATE_MIGRATION;
+	}
 	req->hdr.call->db_id = ctdb_db->id;
-	req->hdr.call->callid = CTDB_NULL_FUNC;
+	req->hdr.call->callid = CTDB_FETCH_WITH_HEADER_FUNC;
 	req->hdr.call->hopcount = 0;
 	req->hdr.call->keylen = key.dsize;
 	req->hdr.call->calldatalen = 0;
@@ -903,6 +922,24 @@ ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
 	return true;
 }
 
+bool
+ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
+			  ctdb_rrl_callback_t callback, void *cbdata)
+{
+	return ctdb_readrecordlock_internal(ctdb_db, key,
+			false,
+			callback, cbdata);
+}
+
+bool
+ctdb_readonlyrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
+			  ctdb_rrl_callback_t callback, void *cbdata)
+{
+	return ctdb_readrecordlock_internal(ctdb_db, key,
+			true,
+			callback, cbdata);
+}
+
 bool ctdb_writerecord(struct ctdb_db *ctdb_db,
 		      struct ctdb_lock *lock, TDB_DATA data)
 {
diff --git a/ctdb/server/ctdb_call.c b/ctdb/server/ctdb_call.c
index 9fc8b33901..23d10b9210 100644
--- a/ctdb/server/ctdb_call.c
+++ b/ctdb/server/ctdb_call.c
@@ -741,8 +741,14 @@ void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 			goto finished_ro;
 		}
 
+		if (c->datalen < sizeof(struct ctdb_ltdb_header)) {
+			DEBUG(DEBUG_ERR,(__location__ " Got FETCH_WITH_HEADER reply with too little data: %d bytes\n", c->datalen));
+			ctdb_ltdb_unlock(ctdb_db, key);
+			goto finished_ro;
+		}
+
+		data.dsize = c->datalen - sizeof(struct ctdb_ltdb_header);
 		data.dptr  = &c->data[sizeof(struct ctdb_ltdb_header)];
-		data.dsize = sizeof(struct ctdb_ltdb_header);
 		ret = ctdb_ltdb_store(ctdb_db, key, header, data);
 		if (ret != 0) {
 			DEBUG(DEBUG_ERR, ("Failed to store new record in ctdb_reply_call\n"));