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

(This used to be ctdb commit c7196d16e8e03bb2a64be164d15a7502300eae0e)

diff --git a/ctdb/client/ctdb_client.c b/ctdb/client/ctdb_client.c
index 55c32735d91c0282b1fec0f9df5b6088f44c273e..d5c2f55997e0b4ecd3feef45e8a39c526a5ed933 100644 (file)
--- a/ctdb/client/ctdb_client.c
+++ b/ctdb/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/ctdb/include/ctdb_protocol.h b/ctdb/include/ctdb_protocol.h
index 0422afeb7b73306ef93bf61d145bd86ee30ff822..b805182b6fee9e18dca204f77ca0de986510303b 100644 (file)
--- a/ctdb/include/ctdb_protocol.h
+++ b/ctdb/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/ctdb/server/ctdb_ltdb_server.c b/ctdb/server/ctdb_ltdb_server.c
index a93e2fa0c950a7cc3c50f2a838c3ff8a724a3ddf..f04b7de4a3aa352f838e198b430f38f032716ccd 100644 (file)
--- a/ctdb/server/ctdb_ltdb_server.c
+++ b/ctdb/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 "