ReadOnly records: Add a new RPC function FETCH_WITH_HEADER.

This function differs from the old FETCH in that this function will also fetch the record header and not just the record data

diff --git a/client/ctdb_client.c b/client/ctdb_client.c
index 55c32735d91c0282b1fec0f9df5b6088f44c273e..d5c2f55997e0b4ecd3feef45e8a39c526a5ed933 100644 (file)
--- a/client/ctdb_client.c
+++ b/client/ctdb_client.c
@@ -89,6 +89,7 @@ int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call,
        c->new_data = NULL;
        c->reply_data = NULL;
        c->status = 0;
+       c->header = header;
 
        for (fn=ctdb_db->calls;fn;fn=fn->next) {
                if (fn->id == call->call_id) break;
@@ -1693,6 +1694,27 @@ static int ctdb_fetch_func(struct ctdb_call_info *call)
        return 0;
 }
 
+/*
+  this is a plain fetch procedure that all databases support
+  this returns the full record including the ltdb header
+*/
+static int ctdb_fetch_with_header_func(struct ctdb_call_info *call)
+{
+       call->reply_data = talloc(call, TDB_DATA);
+       if (call->reply_data == NULL) {
+               return -1;
+       }
+       call->reply_data->dsize = sizeof(struct ctdb_ltdb_header) + call->record_data.dsize;
+       call->reply_data->dptr  = talloc_size(call->reply_data, call->reply_data->dsize);
+       if (call->reply_data->dptr == NULL) {
+               return -1;
+       }
+       memcpy(call->reply_data->dptr, call->header, sizeof(struct ctdb_ltdb_header));
+       memcpy(&call->reply_data->dptr[sizeof(struct ctdb_ltdb_header)], call->record_data.dptr, call->record_data.dsize);
+
+       return 0;
+}
+
 /*
   attach to a specific database - client call
 */
@@ -1758,6 +1780,7 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name,
        /* add well known functions */
        ctdb_set_call(ctdb_db, ctdb_null_func, CTDB_NULL_FUNC);
        ctdb_set_call(ctdb_db, ctdb_fetch_func, CTDB_FETCH_FUNC);
+       ctdb_set_call(ctdb_db, ctdb_fetch_with_header_func, CTDB_FETCH_WITH_HEADER_FUNC);
 
        return ctdb_db;
 }

diff --git a/include/ctdb_protocol.h b/include/ctdb_protocol.h
index 0422afeb7b73306ef93bf61d145bd86ee30ff822..b805182b6fee9e18dca204f77ca0de986510303b 100644 (file)
--- a/include/ctdb_protocol.h
+++ b/include/ctdb_protocol.h
@@ -30,8 +30,9 @@
 #define CTDB_DS_ALIGNMENT 8
 
 
-#define CTDB_NULL_FUNC      0xFF000001
-#define CTDB_FETCH_FUNC     0xFF000002
+#define CTDB_NULL_FUNC                  0xFF000001
+#define CTDB_FETCH_FUNC                 0xFF000002
+#define CTDB_FETCH_WITH_HEADER_FUNC     0xFF000003
 
 
 struct ctdb_call {
@@ -50,6 +51,7 @@ struct ctdb_call {
 */
 struct ctdb_call_info {
        TDB_DATA key;          /* record key */
+       struct ctdb_ltdb_header *header;
        TDB_DATA record_data;  /* current data in the record */
        TDB_DATA *new_data;    /* optionally updated record data */
        TDB_DATA *call_data;   /* optionally passed from caller */

diff --git a/server/ctdb_ltdb_server.c b/server/ctdb_ltdb_server.c
index a93e2fa0c950a7cc3c50f2a838c3ff8a724a3ddf..f04b7de4a3aa352f838e198b430f38f032716ccd 100644 (file)
--- a/server/ctdb_ltdb_server.c
+++ b/server/ctdb_ltdb_server.c
@@ -49,6 +49,27 @@ static int ctdb_fetch_func(struct ctdb_call_info *call)
        return 0;
 }
 
+/*
+  this is a plain fetch procedure that all databases support
+  this returns the full record including the ltdb header
+*/
+static int ctdb_fetch_with_header_func(struct ctdb_call_info *call)
+{
+       call->reply_data = talloc(call, TDB_DATA);
+       if (call->reply_data == NULL) {
+               return -1;
+       }
+       call->reply_data->dsize = sizeof(struct ctdb_ltdb_header) + call->record_data.dsize;
+       call->reply_data->dptr  = talloc_size(call->reply_data, call->reply_data->dsize);
+       if (call->reply_data->dptr == NULL) {
+               return -1;
+       }
+       memcpy(call->reply_data->dptr, call->header, sizeof(struct ctdb_ltdb_header));
+       memcpy(&call->reply_data->dptr[sizeof(struct ctdb_ltdb_header)], call->record_data.dptr, call->record_data.dsize);
+
+       return 0;
+}
+
 
 /**
  * write a record to a normal database
@@ -932,6 +953,17 @@ again:
                return -1;
        }
 
+       /* 
+          all databases support the "fetch_with_header" function. we need this
+          for efficient readonly record fetches
+       */
+       ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_fetch_with_header_func, CTDB_FETCH_WITH_HEADER_FUNC);
+       if (ret != 0) {
+               DEBUG(DEBUG_CRIT,("Failed to setup fetch function for '%s'\n", ctdb_db->db_name));
+               talloc_free(ctdb_db);
+               return -1;
+       }
+
        ret = ctdb_vacuum_init(ctdb_db);
        if (ret != 0) {
                DEBUG(DEBUG_CRIT,("Failed to setup vacuuming for "