ReadOnly: If record does not exist, upgrade to write-lock

[sahlberg/ctdb.git] / libctdb / ctdb.c

diff --git a/libctdb/ctdb.c b/libctdb/ctdb.c
index d51f4a11d3e65ee40695fe2e6d780fc62cb69d9e..099ceac0877523720885625940e8c05c9c10d4db 100644 (file)
--- a/libctdb/ctdb.c
+++ b/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;
@@ -151,6 +156,7 @@ struct ctdb_connection *ctdb_connect(const char *addr,
                log_fn(log_priv, LOG_ERR, "ctdb_connect: no memory", ap);
                goto fail;
        }
+       ctdb->pnn = -1;
        ctdb->outq = NULL;
        ctdb->doneq = NULL;
        ctdb->in = NULL;
@@ -783,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");
@@ -792,6 +806,13 @@ static bool try_readrecordlock(struct ctdb_lock *lock, TDB_DATA *data)
                return true;
        }
 
+       /* we dont have the record locally,
+        * drop to writelock to force a migration
+        */
+       if (!hdr && lock->readonly) {
+               lock->readonly = false;
+       }
+
        tdb_chainunlock(lock->ctdb_db->tdb, lock->key);
        free(hdr);
        return NULL;
@@ -811,13 +832,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. */
@@ -838,9 +859,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;
@@ -865,6 +887,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)) {
@@ -891,9 +914,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;
@@ -902,6 +929,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)
 {