r20945: fixed the bug causing locktest to fail with the ctdb backend after a

couple of hundred opertations. Also removed a lot of the debug code I
was using to track this down.
(This used to be commit 7622e7b4b2694ec14062d0f6de035b946106a6aa)

diff --git a/source4/cluster/ctdb/brlock_ctdb.c b/source4/cluster/ctdb/brlock_ctdb.c
index 0ae5c36094100cd0f70bf7784a65e8192af61f4b..a25d996131dfa5f0482c7041f99dd233f49eac66 100644 (file)
--- a/source4/cluster/ctdb/brlock_ctdb.c
+++ b/source4/cluster/ctdb/brlock_ctdb.c
@@ -101,8 +101,6 @@ static struct brl_context *brl_ctdb_init(TALLOC_CTX *mem_ctx, struct server_id s
        brl->server = server;
        brl->messaging_ctx = messaging_ctx;
 
-       DEBUG(0,("brl_ctdb_init: brl=%p\n", brl));
-
        return brl;
 }
 
@@ -116,9 +114,6 @@ static struct brl_handle *brl_ctdb_create_handle(TALLOC_CTX *mem_ctx, struct ntv
                return NULL;
        }
 
-       DEBUG(0,("file_key\n"));
-       dump_data(0,file_key->data, file_key->length);
-
        brlh->key = *file_key;
        brlh->ntvfs = ntvfs;
        ZERO_STRUCT(brlh->last_lock);
@@ -261,23 +256,6 @@ static NTSTATUS brl_ctdb_lock_failed(struct brl_handle *brlh, struct lock_struct
        return NT_STATUS_LOCK_NOT_GRANTED;
 }
 
-
-static void show_locks(const char *op, struct lock_struct *locks, int count)
-{
-       int i;
-       DEBUG(0,("OP: %s\n", op));
-       for (i=0;i<count;i++) {
-               DEBUG(0,("%2d: %4d %4d %d.%d.%d %p %p\n",
-                        i, (int)locks[i].start, (int)locks[i].size, 
-                        locks[i].context.server.node,
-                        locks[i].context.server.id,
-                        locks[i].context.smbpid,
-                        locks[i].context.ctx,
-                        locks[i].ntvfs));
-       }
-}
-
-
 struct ctdb_lock_req {
        uint16_t smbpid;
        uint64_t start;
@@ -333,23 +311,11 @@ static int brl_ctdb_lock_func(struct ctdb_call *call)
        lock.lock_type = req->lock_type;
        lock.notify_ptr = req->notify_ptr;
 
-       {
-               int xlen = sizeof(lock);
-               uint8_t *xx = &lock;
-               int ii, fd = open("/dev/null", O_WRONLY);
-               for (ii=0;ii<xlen;ii++) {
-                       write(fd, &xx[ii], 1);
-               }
-               close(fd);
-       }
-
        if (dbuf.dptr) {
                /* there are existing locks - make sure they don't conflict */
                locks = (struct lock_struct *)dbuf.dptr;
                count = dbuf.dsize / sizeof(*locks);
 
-               show_locks("lock", locks, count);
-
                for (i=0; i<count; i++) {
                        if (brl_ctdb_conflict(&locks[i], &lock)) {
                                status = NT_STATUS_LOCK_NOT_GRANTED;
@@ -375,8 +341,6 @@ static int brl_ctdb_lock_func(struct ctdb_call *call)
                status = NT_STATUS_LOCK_NOT_GRANTED;
        }
 
-       DEBUG(0,("lock: size now %d\n", call->new_data->dsize));
-
 reply:
        call->reply_data = talloc(call, TDB_DATA);
        if (call->reply_data == NULL) {
@@ -428,10 +392,9 @@ static NTSTATUS brl_ctdb_lock(struct brl_context *brl,
        req.server = brl->server;
        req.brl = brl;
        req.ntvfs = brlh->ntvfs;
-               
+
        ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_LOCK, &rbuf, &sbuf);
        if (ret == -1) {
-               DEBUG(0,("ctdb_call failed - %s\n", __location__));
                return NT_STATUS_INTERNAL_DB_CORRUPTION;
        }
 
@@ -523,8 +486,6 @@ static int brl_ctdb_unlock_func(struct ctdb_call *call)
        locks = (struct lock_struct *)dbuf.dptr;
        count = dbuf.dsize / sizeof(*locks);
 
-       show_locks("unlock", locks, count);
-
        for (i=0; i<count; i++) {
                struct lock_struct *lock = &locks[i];
                
@@ -561,10 +522,6 @@ static int brl_ctdb_unlock_func(struct ctdb_call *call)
                }
        }
 
-       if (call->new_data) {
-               DEBUG(0,("unlock: size now %d\n", call->new_data->dsize));
-       }
-       
        if (i == count) {
                /* we didn't find it */
                status = NT_STATUS_RANGE_NOT_LOCKED;
@@ -650,8 +607,6 @@ static int brl_ctdb_remove_pending_func(struct ctdb_call *call)
        locks = (struct lock_struct *)dbuf.dptr;
        count = dbuf.dsize / sizeof(*locks);
 
-       show_locks("remove_pending", locks, count);
-
        for (i=0; i<count; i++) {
                struct lock_struct *lock = &locks[i];
                
@@ -676,10 +631,6 @@ static int brl_ctdb_remove_pending_func(struct ctdb_call *call)
                }
        }
        
-       if (call->new_data) {
-               DEBUG(0,("remove_pending: size now %d\n", call->new_data->dsize));
-       }
-
        if (i == count) {
                /* we didn't find it */
                status = NT_STATUS_RANGE_NOT_LOCKED;
@@ -769,8 +720,6 @@ static int brl_ctdb_locktest_func(struct ctdb_call *call)
        locks = (struct lock_struct *)dbuf.dptr;
        count = dbuf.dsize / sizeof(*locks);
 
-       show_locks("locktest", locks, count);
-
        for (i=0; i<count; i++) {
                if (brl_ctdb_conflict_other(&locks[i], &lock)) {
                        status = NT_STATUS_FILE_LOCK_CONFLICT;
@@ -860,11 +809,6 @@ static int brl_ctdb_close_func(struct ctdb_call *call)
        locks = (struct lock_struct *)dbuf.dptr;
        count = dbuf.dsize / sizeof(*locks);
 
-       show_locks("close", locks, count);
-
-       DEBUG(0,("closing ctx=%p server=%d.%d ntvfs=%p\n",
-                req->brl, req->server.node, req->server.id, req->ntvfs));
-
        for (i=0; i<count; i++) {
                struct lock_struct *lock = &locks[i];
 
@@ -897,12 +841,6 @@ static int brl_ctdb_close_func(struct ctdb_call *call)
                memcpy(call->new_data->dptr, locks, count*sizeof(struct lock_struct));
        }
 
-       if (call->new_data) {
-               DEBUG(0,("close: size now %d\n", call->new_data->dsize));
-       }
-
-       DEBUG(0,("brl_ctdb_close_func dcount=%d count=%d\n", dcount, count));
-       
        call->reply_data = talloc(call, TDB_DATA);
        if (call->reply_data == NULL) {
                return CTDB_ERR_NOMEM;
@@ -940,8 +878,6 @@ static NTSTATUS brl_ctdb_close(struct brl_context *brl,
        req.server = brl->server;
        req.ntvfs = brlh->ntvfs;
 
-       DEBUG(0,("brl_ctdb_close %u.%u %p\n", req.server.node, req.server.id, brl));
-
        ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_CLOSE, &rbuf, &sbuf);
        if (ret == -1) {
                DEBUG(0,("ctdb_call failed - %s\n", __location__));

diff --git a/source4/cluster/ctdb/common/ctdb.c b/source4/cluster/ctdb/common/ctdb.c
index dd9a6db770c0fcbf88332335483c772c914f1ae2..ba00e7caa063f536bf513289e54561fbeb131729 100644 (file)
--- a/source4/cluster/ctdb/common/ctdb.c
+++ b/source4/cluster/ctdb/common/ctdb.c
@@ -186,8 +186,6 @@ static void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t len
                return;
        }
 
-       DEBUG(0,("got ctdb op %d reqid %d\n", hdr->operation, hdr->reqid));
-
        switch (hdr->operation) {
        case CTDB_REQ_CALL:
                ctdb_request_call(ctdb, hdr);

diff --git a/source4/cluster/ctdb/common/ctdb_call.c b/source4/cluster/ctdb/common/ctdb_call.c
index 648b2b0bc9a624768def68de53930c26221986b9..aace5971e27cd1258e000befe264f533089879d1 100644 (file)
--- a/source4/cluster/ctdb/common/ctdb_call.c
+++ b/source4/cluster/ctdb/common/ctdb_call.c
@@ -28,14 +28,12 @@
 #include "system/filesys.h"
 #include "cluster/ctdb/include/ctdb_private.h"
 
-
 /*
   queue a packet or die
 */
 static void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 {
        struct ctdb_node *node;
-       DEBUG(0,("queueing destnode=%u srcnode=%u\n", hdr->destnode, hdr->srcnode));
        node = ctdb->nodes[hdr->destnode];
        if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
                ctdb_fatal(ctdb, "Unable to queue packet\n");
@@ -210,7 +208,6 @@ static void ctdb_call_send_dmaster(struct ctdb_context *ctdb,
 
        if (r->hdr.destnode == ctdb->vnn) {
                /* we are the lmaster - don't send to ourselves */
-               DEBUG(0,("XXXX local ctdb_req_dmaster\n"));
                ctdb_request_dmaster(ctdb, &r->hdr);
        } else {
                ctdb_queue_packet(ctdb, &r->hdr);
@@ -234,7 +231,7 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 {
        struct ctdb_req_dmaster *c = (struct ctdb_req_dmaster *)hdr;
        struct ctdb_reply_dmaster *r;
-       TDB_DATA key, data;
+       TDB_DATA key, data, data2;
        struct ctdb_ltdb_header header;
        int ret, len;
 
@@ -243,13 +240,8 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
        data.dptr = c->data + c->keylen;
        data.dsize = c->datalen;
 
-       DEBUG(0,("request dmaster reqid=%d\n", hdr->reqid));
-
-       DEBUG(0,("change dmaster: stage 2 - new dmaster will be %d\n",
-                c->dmaster));
-
        /* fetch the current record */
-       ret = ctdb_ltdb_fetch(ctdb, key, &header, &data);
+       ret = ctdb_ltdb_fetch(ctdb, key, &header, &data2);
        if (ret != 0) {
                ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
                return;
@@ -261,8 +253,6 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
                return;
        }
 
-       DEBUG(0,("request dmaster reqid=%d %s\n", hdr->reqid, __location__));
-
        header.dmaster = c->dmaster;
        if (ctdb_ltdb_store(ctdb, key, &header, data) != 0) {
                ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster");
@@ -281,13 +271,10 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
        r->datalen       = data.dsize;
        memcpy(&r->data[0], data.dptr, data.dsize);
 
-       DEBUG(0,("request dmaster reqid=%d %s\n", hdr->reqid, __location__));
-
        if (r->hdr.destnode == r->hdr.srcnode) {
                ctdb_reply_dmaster(ctdb, &r->hdr);
        } else {
                ctdb_queue_packet(ctdb, &r->hdr);
-               DEBUG(0,("request dmaster reqid=%d %s\n", hdr->reqid, __location__));
        }
 
        talloc_free(r);
@@ -331,8 +318,6 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
           then give them the record */
        if (header.laccessor == c->hdr.srcnode &&
            header.lacount >= ctdb->max_lacount) {
-               DEBUG(0,("change dmaster: stage 1 - new dmaster will be %d\n",
-                        header.laccessor));
                ctdb_call_send_dmaster(ctdb, c, &header, &key, &data);
                talloc_free(data.dptr);
                return;
@@ -423,9 +408,6 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 
        talloc_steal(state, c);
 
-       DEBUG(0,("change dmaster: stage 3 - new dmaster is %d\n",
-                ctdb->vnn));
-
        /* we're now the dmaster - update our local ltdb with new header
           and data */
        state->header.dmaster = ctdb->vnn;
@@ -494,7 +476,7 @@ void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 */
 static int ctdb_call_destructor(struct ctdb_call_state *state)
 {
-//     idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
+       idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
        return 0;
 }
 
@@ -582,7 +564,6 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb,
        state->c->hdr.srcnode   = ctdb->vnn;
        /* this limits us to 16k outstanding messages - not unreasonable */
        state->c->hdr.reqid     = idr_get_new(ctdb->idr, state, 0xFFFF);
-       DEBUG(0,("Allocate reqid %u\n", state->c->hdr.reqid));
        state->c->callid        = call_id;
        state->c->keylen        = key.dsize;
        state->c->calldatalen   = call_data?call_data->dsize:0;

diff --git a/source4/cluster/ctdb/common/ctdb_ltdb.c b/source4/cluster/ctdb/common/ctdb_ltdb.c
index cc49aa501619952737fbe492395accda85b62b2b..fbf6a3e567a6549af5b9ec1b56cf83d6184e509e 100644 (file)
--- a/source4/cluster/ctdb/common/ctdb_ltdb.c
+++ b/source4/cluster/ctdb/common/ctdb_ltdb.c
@@ -24,6 +24,8 @@
 #include "system/network.h"
 #include "system/filesys.h"
 #include "cluster/ctdb/include/ctdb_private.h"
+#include "db_wrap.h"
+
 
 /*
   attach to a specific database
@@ -34,7 +36,7 @@ int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags,
        /* when we have a separate daemon this will need to be a real
           file, not a TDB_INTERNAL, so the parent can access it to
           for ltdb bypass */
-       ctdb->ltdb = tdb_open(name, 0, /* tdb_flags */ TDB_INTERNAL, open_flags, mode);
+       ctdb->ltdb = tdb_wrap_open(ctdb, name, 0, TDB_INTERNAL, open_flags, mode);
        if (ctdb->ltdb == NULL) {
                ctdb_set_error(ctdb, "Failed to open tdb %s\n", name);
                return -1;
@@ -76,7 +78,7 @@ int ctdb_ltdb_fetch(struct ctdb_context *ctdb,
 {
        TDB_DATA rec;
 
-       rec = tdb_fetch(ctdb->ltdb, key);
+       rec = tdb_fetch(ctdb->ltdb->tdb, key);
        if (rec.dsize < sizeof(*header)) {
                /* return an initial header */
                free(rec.dptr);
@@ -116,7 +118,7 @@ int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key,
        memcpy(rec.dptr, header, sizeof(*header));
        memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
 
-       ret = tdb_store(ctdb->ltdb, key, rec, TDB_REPLACE);
+       ret = tdb_store(ctdb->ltdb->tdb, key, rec, TDB_REPLACE);
        talloc_free(rec.dptr);
 
        return ret;

diff --git a/source4/cluster/ctdb/include/ctdb_private.h b/source4/cluster/ctdb/include/ctdb_private.h
index bace97afe7befe7145332aa5a4a67496e0f3dae7..db7da8fcaf66bef0a8905fd0997f0c397bc60901 100644 (file)
--- a/source4/cluster/ctdb/include/ctdb_private.h
+++ b/source4/cluster/ctdb/include/ctdb_private.h
@@ -89,7 +89,7 @@ struct ctdb_context {
        struct ctdb_node **nodes; /* array of nodes in the cluster - indexed by vnn */
        struct ctdb_registered_call *calls; /* list of registered calls */
        char *err_msg;
-       struct tdb_context *ltdb;
+       struct tdb_wrap *ltdb;
        const struct ctdb_methods *methods; /* transport methods */
        const struct ctdb_upcalls *upcalls; /* transport upcalls */
        void *private; /* private to transport */