Remove LACOUNT and LACCESSOR and migrate the records immediately.

[sahlberg/ctdb.git] / client / ctdb_client.c

diff --git a/client/ctdb_client.c b/client/ctdb_client.c
index 9a8917633b948089d60cde134b412a26f2d36315..1abea127a989cf9bc21b0afed89a340a33c365fb 100644 (file)
--- a/client/ctdb_client.c
+++ b/client/ctdb_client.c
@@ -22,7 +22,7 @@
 #include "db_wrap.h"
 #include "lib/tdb/include/tdb.h"
 #include "lib/util/dlinklist.h"
-#include "lib/events/events.h"
+#include "lib/tevent/tevent.h"
 #include "system/network.h"
 #include "system/filesys.h"
 #include "system/locale.h"
@@ -72,7 +72,7 @@ struct ctdb_req_header *_ctdbd_allocate_pkt(struct ctdb_context *ctdb,
 */
 int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call,
                    struct ctdb_ltdb_header *header, TALLOC_CTX *mem_ctx,
-                   TDB_DATA *data, uint32_t caller)
+                   TDB_DATA *data)
 {
        struct ctdb_call_info *c;
        struct ctdb_registered_call *fn;
@@ -105,15 +105,8 @@ int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call,
                return -1;
        }
 
-       if (header->laccessor != caller) {
-               header->lacount = 0;
-       }
-       header->laccessor = caller;
-       header->lacount++;
-
-       /* we need to force the record to be written out if this was a remote access,
-          so that the lacount is updated */
-       if (c->new_data == NULL && header->laccessor != ctdb->pnn) {
+       /* we need to force the record to be written out if this was a remote access */
+       if (c->new_data == NULL) {
                c->new_data = &c->record_data;
        }
 
@@ -281,7 +274,7 @@ int ctdb_socket_connect(struct ctdb_context *ctdb)
 
        ctdb->daemon.queue = ctdb_queue_setup(ctdb, ctdb, ctdb->daemon.sd, 
                                              CTDB_DS_ALIGNMENT, 
-                                             ctdb_client_read_cb, ctdb);
+                                             ctdb_client_read_cb, ctdb, "to-ctdbd");
        return 0;
 }
 
@@ -368,7 +361,7 @@ static struct ctdb_client_call_state *ctdb_client_call_local_send(struct ctdb_db
        *(state->call) = *call;
        state->ctdb_db = ctdb_db;
 
-       ret = ctdb_call_local(ctdb_db, state->call, header, state, data, ctdb->pnn);
+       ret = ctdb_call_local(ctdb_db, state->call, header, state, data);
 
        return state;
 }
@@ -474,8 +467,8 @@ int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
   tell the daemon what messaging srvid we will use, and register the message
   handler function in the client
 */
-int ctdb_set_message_handler(struct ctdb_context *ctdb, uint64_t srvid, 
-                            ctdb_message_fn_t handler,
+int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint64_t srvid, 
+                            ctdb_msg_fn_t handler,
                             void *private_data)
                                    
 {
@@ -496,7 +489,7 @@ int ctdb_set_message_handler(struct ctdb_context *ctdb, uint64_t srvid,
 /*
   tell the daemon we no longer want a srvid
 */
-int ctdb_remove_message_handler(struct ctdb_context *ctdb, uint64_t srvid, void *private_data)
+int ctdb_client_remove_message_handler(struct ctdb_context *ctdb, uint64_t srvid, void *private_data)
 {
        int res;
        int32_t status;
@@ -517,7 +510,7 @@ int ctdb_remove_message_handler(struct ctdb_context *ctdb, uint64_t srvid, void
 /*
   send a message - from client context
  */
-int ctdb_send_message(struct ctdb_context *ctdb, uint32_t pnn,
+int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t pnn,
                      uint64_t srvid, TDB_DATA data)
 {
        struct ctdb_req_message *r;
@@ -1849,7 +1842,7 @@ int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn, void *
        state.private_data = private_data;
        state.fn = fn;
 
-       ret = ctdb_set_message_handler(ctdb_db->ctdb, srvid, traverse_handler, &state);
+       ret = ctdb_client_set_message_handler(ctdb_db->ctdb, srvid, traverse_handler, &state);
        if (ret != 0) {
                DEBUG(DEBUG_ERR,("Failed to setup traverse handler\n"));
                return -1;
@@ -1866,7 +1859,7 @@ int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn, void *
                           data, NULL, NULL, &status, NULL, NULL);
        if (ret != 0 || status != 0) {
                DEBUG(DEBUG_ERR,("ctdb_traverse_all failed\n"));
-               ctdb_remove_message_handler(ctdb_db->ctdb, srvid, &state);
+               ctdb_client_remove_message_handler(ctdb_db->ctdb, srvid, &state);
                return -1;
        }
 
@@ -1874,7 +1867,7 @@ int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn, void *
                event_loop_once(ctdb_db->ctdb->ev);
        }
 
-       ret = ctdb_remove_message_handler(ctdb_db->ctdb, srvid, &state);
+       ret = ctdb_client_remove_message_handler(ctdb_db->ctdb, srvid, &state);
        if (ret != 0) {
                DEBUG(DEBUG_ERR,("Failed to remove ctdb_traverse handler\n"));
                return -1;
@@ -1906,7 +1899,7 @@ int ctdb_dumpdb_record(struct ctdb_context *ctdb, TDB_DATA key, TDB_DATA data, v
        fprintf(f, "dmaster: %u\n", h->dmaster);
        fprintf(f, "rsn: %llu\n", (unsigned long long)h->rsn);
 
-       fprintf(f, "data(%u) = \"", (unsigned)data.dsize);
+       fprintf(f, "data(%u) = \"", (unsigned)(data.dsize - sizeof(*h)));
        for (i=sizeof(*h);i<data.dsize;i++) {
                if (ISASCII(data.dptr[i])) {
                        fprintf(f, "%c", data.dptr[i]);
@@ -2916,6 +2909,8 @@ struct ctdb_context *ctdb_init(struct event_context *ev)
        }
        ctdb->ev  = ev;
        ctdb->idr = idr_init(ctdb);
+       /* Wrap early to exercise code. */
+       ctdb->lastid = INT_MAX-200;
        CTDB_NO_MEMORY_NULL(ctdb, ctdb->idr);
 
        ret = ctdb_set_socketname(ctdb, CTDB_PATH);
@@ -2925,6 +2920,8 @@ struct ctdb_context *ctdb_init(struct event_context *ev)
                return NULL;
        }
 
+       ctdb->statistics.statistics_start_time = timeval_current();
+
        return ctdb;
 }
 
@@ -2948,6 +2945,11 @@ int ctdb_set_socketname(struct ctdb_context *ctdb, const char *socketname)
        return 0;
 }
 
+const char *ctdb_get_socketname(struct ctdb_context *ctdb)
+{
+       return ctdb->daemon.name;
+}
+
 /*
   return the pnn of this node
 */
@@ -3818,9 +3820,15 @@ int ctdb_ctrl_recd_ping(struct ctdb_context *ctdb)
  * to the daemon as a client process, this function can be used to change
  * the ctdb context from daemon into client mode
  */
-int switch_from_server_to_client(struct ctdb_context *ctdb)
+int switch_from_server_to_client(struct ctdb_context *ctdb, const char *fmt, ...)
 {
        int ret;
+       va_list ap;
+
+       /* Add extra information so we can identify this in the logs */
+       va_start(ap, fmt);
+       debug_extra = talloc_strdup_append(talloc_vasprintf(NULL, fmt, ap), ":");
+       va_end(ap);
 
        /* shutdown the transport */
        if (ctdb->methods) {
@@ -3830,6 +3838,7 @@ int switch_from_server_to_client(struct ctdb_context *ctdb)
        /* get a new event context */
        talloc_free(ctdb->ev);
        ctdb->ev = event_context_init(ctdb);
+       tevent_loop_allow_nesting(ctdb->ev);
 
        close(ctdb->daemon.sd);
        ctdb->daemon.sd = -1;
@@ -4200,3 +4209,23 @@ int ctdb_ctrl_get_db_priority(struct ctdb_context *ctdb, struct timeval timeout,
 
        return 0;
 }
+
+int ctdb_ctrl_getstathistory(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, TALLOC_CTX *mem_ctx, struct ctdb_statistics_wire **stats)
+{
+       int ret;
+       TDB_DATA outdata;
+       int32_t res;
+
+       ret = ctdb_control(ctdb, destnode, 0, 
+                          CTDB_CONTROL_GET_STAT_HISTORY, 0, tdb_null, 
+                          mem_ctx, &outdata, &res, &timeout, NULL);
+       if (ret != 0 || res != 0 || outdata.dsize == 0) {
+               DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getstathistory failed ret:%d res:%d\n", ret, res));
+               return -1;
+       }
+
+       *stats = (struct ctdb_statistics_wire *)talloc_memdup(mem_ctx, outdata.dptr, outdata.dsize);
+       talloc_free(outdata.dptr);
+                   
+       return 0;
+}