Revert "server: when we migrate off a record with data, set the MIGRATED_WITH_DATA...

[sahlberg/ctdb.git] / server / ctdb_call.c

diff --git a/server/ctdb_call.c b/server/ctdb_call.c
index 180d9905ea5db68d995ca99eace0127c8667f190..d6c08666a6c28fc22403f3745dc48853daf1eeaa 100644 (file)
--- a/server/ctdb_call.c
+++ b/server/ctdb_call.c
@@ -21,7 +21,7 @@
   protocol design and packet details
 */
 #include "includes.h"
-#include "lib/events/events.h"
+#include "lib/tevent/tevent.h"
 #include "lib/tdb/include/tdb.h"
 #include "lib/util/dlinklist.h"
 #include "system/network.h"
@@ -70,7 +70,7 @@ static void ctdb_send_error(struct ctdb_context *ctdb,
        int msglen, len;
 
        if (ctdb->methods == NULL) {
-               DEBUG(DEBUG_ERR,(__location__ " Failed to send error. Transport is DOWN\n"));
+               DEBUG(DEBUG_INFO,(__location__ " Failed to send error. Transport is DOWN\n"));
                return;
        }
 
@@ -276,6 +276,16 @@ static void ctdb_become_dmaster(struct ctdb_db_context *ctdb_db,
                return;
        }
 
+       if (key.dsize != state->call->key.dsize || memcmp(key.dptr, state->call->key.dptr, key.dsize)) {
+               DEBUG(DEBUG_ERR, ("Got bogus DMASTER packet reqid:%u from node %u. Key does not match key held in matching idr.\n", hdr->reqid, hdr->srcnode));
+
+               ret = ctdb_ltdb_unlock(ctdb_db, key);
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+               }
+               return;
+       }
+
        if (hdr->reqid != state->reqid) {
                /* we found a record  but it was the wrong one */
                DEBUG(DEBUG_ERR, ("Dropped orphan in ctdb_become_dmaster with reqid:%u\n from node %u", hdr->reqid, hdr->srcnode));
@@ -287,9 +297,9 @@ static void ctdb_become_dmaster(struct ctdb_db_context *ctdb_db,
                return;
        }
 
-       ctdb_call_local(ctdb_db, state->call, &header, state, &data, ctdb->pnn);
+       ctdb_call_local(ctdb_db, state->call, &header, state, &data);
 
-       ret = ctdb_ltdb_unlock(ctdb_db, key);
+       ret = ctdb_ltdb_unlock(ctdb_db, state->call->key);
        if (ret != 0) {
                DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
        }
@@ -380,7 +390,11 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
                ctdb_become_dmaster(ctdb_db, hdr, key, data, c->rsn);
        } else {
                ctdb_send_dmaster_reply(ctdb_db, &header, key, data, c->dmaster, hdr->reqid);
-               ctdb_ltdb_unlock(ctdb_db, key);
+
+               ret = ctdb_ltdb_unlock(ctdb_db, key);
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+               }
        }
 }
 
@@ -399,7 +413,7 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
        struct ctdb_db_context *ctdb_db;
 
        if (ctdb->methods == NULL) {
-               DEBUG(DEBUG_ERR,(__location__ " Failed ctdb_request_call. Transport is DOWN\n"));
+               DEBUG(DEBUG_INFO,(__location__ " Failed ctdb_request_call. Transport is DOWN\n"));
                return;
        }
 
@@ -441,22 +455,21 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
        if (header.dmaster != ctdb->pnn) {
                talloc_free(data.dptr);
                ctdb_call_send_redirect(ctdb, call->key, c, &header);
-               ctdb_ltdb_unlock(ctdb_db, call->key);
+
+               ret = ctdb_ltdb_unlock(ctdb_db, call->key);
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+               }
                return;
        }
 
-       if (c->hopcount > ctdb->statistics.max_hop_count) {
-               ctdb->statistics.max_hop_count = c->hopcount;
-       }
+       CTDB_UPDATE_STAT(ctdb, max_hop_count, c->hopcount);
 
-       /* if this nodes has done enough consecutive calls on the same record
-          then give them the record
-          or if the node requested an immediate migration
-       */
-       if ( c->hdr.srcnode != ctdb->pnn &&
-            ((header.laccessor == c->hdr.srcnode
-              && header.lacount >= ctdb->tunable.max_lacount)
-             || (c->flags & CTDB_IMMEDIATE_MIGRATION)) ) {
+       /* Try if possible to migrate the record off to the caller node.
+        * From the clients perspective a fetch of the data is just as 
+        * expensive as a migration.
+        */
+       if (c->hdr.srcnode != ctdb->pnn) {
                if (ctdb_db->transaction_active) {
                        DEBUG(DEBUG_INFO, (__location__ " refusing migration"
                              " of key %s while transaction is active\n",
@@ -466,14 +479,21 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
                                 ctdb->pnn, ctdb_hash(&(call->key)), c->hdr.srcnode));
                        ctdb_call_send_dmaster(ctdb_db, c, &header, &(call->key), &data);
                        talloc_free(data.dptr);
-                       ctdb_ltdb_unlock(ctdb_db, call->key);
+
+                       ret = ctdb_ltdb_unlock(ctdb_db, call->key);
+                       if (ret != 0) {
+                               DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+                       }
                        return;
                }
        }
 
-       ctdb_call_local(ctdb_db, call, &header, hdr, &data, c->hdr.srcnode);
+       ctdb_call_local(ctdb_db, call, &header, hdr, &data);
 
-       ctdb_ltdb_unlock(ctdb_db, call->key);
+       ret = ctdb_ltdb_unlock(ctdb_db, call->key);
+       if (ret != 0) {
+               DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+       }
 
        len = offsetof(struct ctdb_reply_call, data) + call->reply_data.dsize;
        r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REPLY_CALL, len, 
@@ -684,7 +704,7 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *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);
 
        event_add_timed(ctdb->ev, state, timeval_zero(), call_local_trigger, state);
 
@@ -707,7 +727,7 @@ struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctd
        struct ctdb_context *ctdb = ctdb_db->ctdb;
 
        if (ctdb->methods == NULL) {
-               DEBUG(DEBUG_ERR,(__location__ " Failed send packet. Transport is down\n"));
+               DEBUG(DEBUG_INFO,(__location__ " Failed send packet. Transport is down\n"));
                return NULL;
        }
 
@@ -791,7 +811,7 @@ void ctdb_send_keepalive(struct ctdb_context *ctdb, uint32_t destnode)
        struct ctdb_req_keepalive *r;
        
        if (ctdb->methods == NULL) {
-               DEBUG(DEBUG_ERR,(__location__ " Failed to send keepalive. Transport is DOWN\n"));
+               DEBUG(DEBUG_INFO,(__location__ " Failed to send keepalive. Transport is DOWN\n"));
                return;
        }
 
@@ -802,7 +822,7 @@ void ctdb_send_keepalive(struct ctdb_context *ctdb, uint32_t destnode)
        r->hdr.destnode  = destnode;
        r->hdr.reqid     = 0;
        
-       ctdb->statistics.keepalive_packets_sent++;
+       CTDB_INCREMENT_STAT(ctdb, keepalive_packets_sent);
 
        ctdb_queue_packet(ctdb, &r->hdr);