ReadOnly: Dont update the record header from the calling client. While it is convenie...

Remove the code in the example client code that writes the record to the local tdb.
Add code to the local ctdbd processing of replies to check if this reply contain a ro delegation and if so, spawn a child process to lock the tdb and then write the data.

diff --git a/client/ctdb_client.c b/client/ctdb_client.c
index f2359cafc7375dec18af673e7bd8100eef300825..f0172e423264cf744411ebbe96b6574d6d41605c 100644 (file)
--- a/client/ctdb_client.c
+++ b/client/ctdb_client.c
@@ -843,32 +843,6 @@ again:
                        goto again;
                }
 
-               if (h->header.rsn >= roheader->rsn) {
-                       DEBUG(DEBUG_ERR,("READONLY RECORD: Too small RSN, migrate and try again\n"));
-                       ctdb_ltdb_unlock(ctdb_db, key);
-
-                       ret = ctdb_client_force_migration(ctdb_db, key);
-                       if (ret != 0) {
-                               DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
-                               talloc_free(h);
-                               return NULL;
-                       }
-
-                       goto again;
-               }
-
-               if (ctdb_ltdb_store(ctdb_db, key, roheader, rodata) != 0) {
-                       ctdb_ltdb_unlock(ctdb_db, key);
-
-                       ret = ctdb_client_force_migration(ctdb_db, key);
-                       if (ret != 0) {
-                               DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
-                               talloc_free(h);
-                               return NULL;
-                       }
-
-                       goto again;
-               }
                return h;
        }
 

diff --git a/server/ctdb_call.c b/server/ctdb_call.c
index 9e8642221c462dc42531e3a08cc60f990acdcf62..9fc8b339012a06b9853b94ccd77cb000930cf938 100644 (file)
--- a/server/ctdb_call.c
+++ b/server/ctdb_call.c
@@ -700,6 +700,60 @@ void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
                return;
        }
 
+
+       /* read only delegation processing */
+       /* If we got a FETCH_WITH_HEADER we should check if this is a ro
+        * delegation since we may need to update the record header
+        */
+       if (state->c->callid == CTDB_FETCH_WITH_HEADER_FUNC) {
+               struct ctdb_db_context *ctdb_db = state->ctdb_db;
+               struct ctdb_ltdb_header *header = (struct ctdb_ltdb_header *)&c->data[0];
+               struct ctdb_ltdb_header oldheader;
+               TDB_DATA key, data, olddata;
+               int ret;
+
+               if (!(header->flags & CTDB_REC_RO_HAVE_READONLY)) {
+                       goto finished_ro;
+                       return;
+               }
+
+               key.dsize = state->c->keylen;
+               key.dptr  = state->c->data;
+               ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr,
+                                    ctdb_call_input_pkt, ctdb, False);
+               if (ret == -2) {
+                       return;
+               }
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR,(__location__ " Failed to get lock in ctdb_reply_call\n"));
+                       return;
+               }
+
+               ret = ctdb_ltdb_fetch(ctdb_db, key, &oldheader, state, &olddata);
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR, ("Failed to fetch old record in ctdb_reply_call\n"));
+                       ctdb_ltdb_unlock(ctdb_db, key);
+                       goto finished_ro;
+               }                       
+
+               if (header->rsn <= oldheader.rsn) {
+                       ctdb_ltdb_unlock(ctdb_db, key);
+                       goto finished_ro;
+               }
+
+               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"));
+                       ctdb_ltdb_unlock(ctdb_db, key);
+                       goto finished_ro;
+               }                       
+
+               ctdb_ltdb_unlock(ctdb_db, key);
+       }
+finished_ro:
+
        state->call->reply_data.dptr = c->data;
        state->call->reply_data.dsize = c->datalen;
        state->call->status = c->status;