Split CTDB into sub contexts to handle multiple concurrent databases within the same...

(This used to be ctdb commit d995103143f6f13f59118549d93ab4b29c27ec89)

diff --git a/ctdb/common/ctdb.c b/ctdb/common/ctdb.c
index 6dc78fa22a57d8b920dfb2b1707e80104427a556..e4f7289550472d9aaa35e14bfb6d7280d6bec58b 100644 (file)
--- a/ctdb/common/ctdb.c
+++ b/ctdb/common/ctdb.c
@@ -150,15 +150,15 @@ int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
 /*
   add a node to the list of active nodes
 */
-int ctdb_set_call(struct ctdb_context *ctdb, ctdb_fn_t fn, int id)
+int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, int id)
 {
        struct ctdb_registered_call *call;
 
-       call = talloc(ctdb, struct ctdb_registered_call);
+       call = talloc(ctdb_db, struct ctdb_registered_call);
        call->fn = fn;
        call->id = id;
 
-       DLIST_ADD(ctdb->calls, call);   
+       DLIST_ADD(ctdb_db->calls, call);        
        return 0;
 }
 

diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index fb29aad9ac7e77aaebba8b1360abb2d0223270ac..75355f7ae7edc9355d921f74f0e09730c22f74a5 100644 (file)
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -31,12 +31,13 @@
 /*
   local version of ctdb_call
 */
-static int ctdb_call_local(struct ctdb_context *ctdb, struct ctdb_call *call,
+static int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call,
                           struct ctdb_ltdb_header *header, TDB_DATA *data,
                           uint32_t caller)
 {
        struct ctdb_call_info *c;
        struct ctdb_registered_call *fn;
+       struct ctdb_context *ctdb = ctdb_db->ctdb;
 
        c = talloc(ctdb, struct ctdb_call_info);
        CTDB_NO_MEMORY(ctdb, c);
@@ -50,7 +51,7 @@ static int ctdb_call_local(struct ctdb_context *ctdb, struct ctdb_call *call,
        c->reply_data = NULL;
        c->status = 0;
 
-       for (fn=ctdb->calls;fn;fn=fn->next) {
+       for (fn=ctdb_db->calls;fn;fn=fn->next) {
                if (fn->id == call->call_id) break;
        }
        if (fn == NULL) {
@@ -76,7 +77,7 @@ static int ctdb_call_local(struct ctdb_context *ctdb, struct ctdb_call *call,
        }
 
        if (c->new_data) {
-               if (ctdb_ltdb_store(ctdb, call->key, header, *c->new_data) != 0) {
+               if (ctdb_ltdb_store(ctdb_db, call->key, header, *c->new_data) != 0) {
                        ctdb_set_error(ctdb, "ctdb_call tdb_store failed\n");
                        return -1;
                }
@@ -170,12 +171,13 @@ static void ctdb_call_send_redirect(struct ctdb_context *ctdb,
   always knows who the dmaster is. The lmaster will then send a
   CTDB_REPLY_DMASTER to the new dmaster
 */
-static void ctdb_call_send_dmaster(struct ctdb_context *ctdb, 
+static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db, 
                                   struct ctdb_req_call *c, 
                                   struct ctdb_ltdb_header *header,
                                   TDB_DATA *key, TDB_DATA *data)
 {
        struct ctdb_req_dmaster *r;
+       struct ctdb_context *ctdb = ctdb_db->ctdb;
        int len;
        
        len = offsetof(struct ctdb_req_dmaster, data) + key->dsize + data->dsize;
@@ -186,6 +188,7 @@ static void ctdb_call_send_dmaster(struct ctdb_context *ctdb,
        r->hdr.destnode  = ctdb_lmaster(ctdb, key);
        r->hdr.srcnode   = ctdb->vnn;
        r->hdr.reqid     = c->hdr.reqid;
+       r->db_id         = c->db_id;
        r->dmaster       = header->laccessor;
        r->keylen        = key->dsize;
        r->datalen       = data->dsize;
@@ -200,7 +203,7 @@ static void ctdb_call_send_dmaster(struct ctdb_context *ctdb,
 
                /* update the ltdb to record the new dmaster */
                header->dmaster = r->hdr.destnode;
-               ctdb_ltdb_store(ctdb, *key, header, *data);
+               ctdb_ltdb_store(ctdb_db, *key, header, *data);
        }
 
        talloc_free(r);
@@ -219,6 +222,7 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
        struct ctdb_reply_dmaster *r;
        TDB_DATA key, data, data2;
        struct ctdb_ltdb_header header;
+       struct ctdb_db_context *ctdb_db;
        int ret, len;
 
        key.dptr = c->data;
@@ -226,8 +230,18 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
        data.dptr = c->data + c->keylen;
        data.dsize = c->datalen;
 
+       for (ctdb_db=ctdb->db_list; ctdb_db; ctdb_db=ctdb_db->next) {
+               if (ctdb_db->db_id == c->db_id) {
+                       break;
+               }
+       }
+       if (!ctdb_db) {
+               ctdb_send_error(ctdb, hdr, ret, "Unknown database in request. db_id==0x%08x",c->db_id);
+               return;
+       }
+
        /* fetch the current record */
-       ret = ctdb_ltdb_fetch(ctdb, key, &header, &data2);
+       ret = ctdb_ltdb_fetch(ctdb_db, key, &header, &data2);
        if (ret != 0) {
                ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
                return;
@@ -240,7 +254,7 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
        }
 
        header.dmaster = c->dmaster;
-       if (ctdb_ltdb_store(ctdb, key, &header, data) != 0) {
+       if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
                ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster");
                return;
        }
@@ -278,6 +292,17 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
        int ret, len;
        struct ctdb_ltdb_header header;
        struct ctdb_call call;
+       struct ctdb_db_context *ctdb_db;
+
+       for (ctdb_db=ctdb->db_list; ctdb_db; ctdb_db=ctdb_db->next) {
+               if (ctdb_db->db_id == c->db_id) {
+                       break;
+               }
+       }
+       if (!ctdb_db) {
+               ctdb_send_error(ctdb, hdr, ret, "Unknown database in request. db_id==0x%08x",c->db_id);
+               return;
+       }
 
        call.call_id  = c->callid;
        call.key.dptr = c->data;
@@ -289,7 +314,7 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
           fetches the record data (if any), thus avoiding a 2nd fetch of the data 
           if the call will be answered locally */
 
-       ret = ctdb_ltdb_fetch(ctdb, call.key, &header, &data);
+       ret = ctdb_ltdb_fetch(ctdb_db, call.key, &header, &data);
        if (ret != 0) {
                ctdb_send_error(ctdb, hdr, ret, "ltdb fetch failed in ctdb_request_call");
                return;
@@ -307,12 +332,12 @@ 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) {
-               ctdb_call_send_dmaster(ctdb, c, &header, &call.key, &data);
+               ctdb_call_send_dmaster(ctdb_db, c, &header, &call.key, &data);
                talloc_free(data.dptr);
                return;
        }
 
-       ctdb_call_local(ctdb, &call, &header, &data, c->hdr.srcnode);
+       ctdb_call_local(ctdb_db, &call, &header, &data, c->hdr.srcnode);
 
        len = offsetof(struct ctdb_reply_call, data) + call.reply_data.dsize;
        r = ctdb->methods->allocate_pkt(ctdb, len);
@@ -342,6 +367,7 @@ enum call_state {CTDB_CALL_WAIT, CTDB_CALL_DONE, CTDB_CALL_ERROR};
 struct ctdb_call_state {
        enum call_state state;
        struct ctdb_req_call *c;
+       struct ctdb_db_context *ctdb_db;
        struct ctdb_node *node;
        const char *errmsg;
        struct ctdb_call call;
@@ -384,10 +410,15 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 {
        struct ctdb_reply_dmaster *c = (struct ctdb_reply_dmaster *)hdr;
        struct ctdb_call_state *state;
+       struct ctdb_db_context *ctdb_db;
        TDB_DATA data;
 
        state = idr_find(ctdb->idr, hdr->reqid);
-       if (state == NULL) return;
+       if (state == NULL) {
+               return;
+       }
+       ctdb_db = state->ctdb_db;
+
 
        data.dptr = c->data;
        data.dsize = c->datalen;
@@ -398,12 +429,12 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
           and data */
        state->header.dmaster = ctdb->vnn;
 
-       if (ctdb_ltdb_store(ctdb, state->call.key, &state->header, data) != 0) {
+       if (ctdb_ltdb_store(ctdb_db, state->call.key, &state->header, data) != 0) {
                ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
                return;
        }
 
-       ctdb_call_local(ctdb, &state->call, &state->header, &data, ctdb->vnn);
+       ctdb_call_local(ctdb_db, &state->call, &state->header, &data, ctdb->vnn);
 
        state->state = CTDB_CALL_DONE;
 }
@@ -483,22 +514,24 @@ void ctdb_call_timeout(struct event_context *ev, struct timed_event *te,
   this is used so that locally processed ctdb_call requests are processed
   in an event driven manner
 */
-struct ctdb_call_state *ctdb_call_local_send(struct ctdb_context *ctdb, 
+struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, 
                                             struct ctdb_call *call,
                                             struct ctdb_ltdb_header *header,
                                             TDB_DATA *data)
 {
        struct ctdb_call_state *state;
+       struct ctdb_context *ctdb = ctdb_db->ctdb;
        int ret;
 
-       state = talloc_zero(ctdb, struct ctdb_call_state);
+       state = talloc_zero(ctdb_db, struct ctdb_call_state);
        CTDB_NO_MEMORY_NULL(ctdb, state);
 
        state->state = CTDB_CALL_DONE;
        state->node = ctdb->nodes[ctdb->vnn];
        state->call = *call;
+       state->ctdb_db = ctdb_db;
 
-       ret = ctdb_call_local(ctdb, &state->call, header, data, ctdb->vnn);
+       ret = ctdb_call_local(ctdb_db, &state->call, header, data, ctdb->vnn);
 
        return state;
 }
@@ -510,13 +543,14 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_context *ctdb,
   This constructs a ctdb_call request and queues it for processing. 
   This call never blocks.
 */
-struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_call *call)
+struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
 {
        uint32_t len;
        struct ctdb_call_state *state;
        int ret;
        struct ctdb_ltdb_header header;
        TDB_DATA data;
+       struct ctdb_context *ctdb = ctdb_db->ctdb;
 
        /*
          if we are the dmaster for this key then we don't need to
@@ -524,14 +558,14 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_ca
          locally. To find out if we are the dmaster we need to look
          in our ltdb
        */
-       ret = ctdb_ltdb_fetch(ctdb, call->key, &header, &data);
+       ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, &data);
        if (ret != 0) return NULL;
 
        if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
-               return ctdb_call_local_send(ctdb, call, &header, &data);
+               return ctdb_call_local_send(ctdb_db, call, &header, &data);
        }
 
-       state = talloc_zero(ctdb, struct ctdb_call_state);
+       state = talloc_zero(ctdb_db, struct ctdb_call_state);
        CTDB_NO_MEMORY_NULL(ctdb, state);
 
        len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
@@ -544,6 +578,7 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_ca
        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);
+       state->c->db_id         = ctdb_db->db_id;
        state->c->callid        = call->call_id;
        state->c->keylen        = call->key.dsize;
        state->c->calldatalen   = call->call_data.dsize;
@@ -557,6 +592,7 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_ca
        state->node   = ctdb->nodes[header.dmaster];
        state->state  = CTDB_CALL_WAIT;
        state->header = header;
+       state->ctdb_db = ctdb_db;
 
        talloc_set_destructor(state, ctdb_call_destructor);
 
@@ -601,9 +637,9 @@ int ctdb_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 /*
   full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv()
 */
-int ctdb_call(struct ctdb_context *ctdb, struct ctdb_call *call)
+int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
 {
        struct ctdb_call_state *state;
-       state = ctdb_call_send(ctdb, call);
+       state = ctdb_call_send(ctdb_db, call);
        return ctdb_call_recv(state, call);
 }

diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index 10bcde43b53c4af4dd98ce1d0171fce5d0c3f00c..87550ff408d617cdf635e2e2a9f069fb714479bf 100644 (file)
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -30,18 +30,44 @@
 /*
   attach to a specific database
 */
-int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
-               int open_flags, mode_t mode)
+struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
+                                   int open_flags, mode_t mode)
 {
+       struct ctdb_db_context *ctdb_db, *tmp_db;
+       TDB_DATA data;
+
+       ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
+       CTDB_NO_MEMORY_NULL(ctdb, ctdb_db);
+
+       ctdb_db->ctdb = ctdb;
+       ctdb_db->db_name = talloc_strdup(ctdb_db, name);
+       CTDB_NO_MEMORY_NULL(ctdb, ctdb_db->db_name);
+
+       data.dptr = name;
+       data.dsize = strlen(name);
+       ctdb_db->db_id = ctdb_hash(&data);
+
+       for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
+               if (tmp_db->db_id == ctdb_db->db_id) {
+                       ctdb_set_error(ctdb, "CTDB database hash collission '%s' : '%s'",
+                                       name, tmp_db->db_name);
+                       talloc_free(ctdb_db);
+                       return NULL;
+               }
+       }
+
        /* 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_wrap_open(ctdb, name, 0, TDB_INTERNAL, open_flags, mode);
-       if (ctdb->ltdb == NULL) {
+       ctdb_db->ltdb = tdb_wrap_open(ctdb, name, 0, TDB_INTERNAL, open_flags, mode);
+       if (ctdb_db->ltdb == NULL) {
                ctdb_set_error(ctdb, "Failed to open tdb %s\n", name);
-               return -1;
+               talloc_free(ctdb_db);
+               return NULL;
        }
-       return 0;
+
+       DLIST_ADD(ctdb->db_list, ctdb_db);
+       return ctdb_db;
 }
 
 /*
@@ -56,13 +82,13 @@ uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key)
 /*
   construct an initial header for a record with no ltdb header yet
 */
-static void ltdb_initial_header(struct ctdb_context *ctdb, 
+static void ltdb_initial_header(struct ctdb_db_context *ctdb_db, 
                                TDB_DATA key,
                                struct ctdb_ltdb_header *header)
 {
        header->rsn = 0;
        /* initial dmaster is the lmaster */
-       header->dmaster = ctdb_lmaster(ctdb, &key);
+       header->dmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
        header->laccessor = header->dmaster;
        header->lacount = 0;
 }
@@ -73,16 +99,17 @@ static void ltdb_initial_header(struct ctdb_context *ctdb,
   and returning the body of the record. A valid (initial) header is
   returned if the record is not present
 */
-int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
+int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, 
                    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data)
 {
        TDB_DATA rec;
+       struct ctdb_context *ctdb = ctdb_db->ctdb;
 
-       rec = tdb_fetch(ctdb->ltdb->tdb, key);
+       rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
        if (rec.dsize < sizeof(*header)) {
                /* return an initial header */
                free(rec.dptr);
-               ltdb_initial_header(ctdb, key, header);
+               ltdb_initial_header(ctdb_db, key, header);
                data->dptr = NULL;
                data->dsize = 0;
                return 0;
@@ -91,7 +118,7 @@ int ctdb_ltdb_fetch(struct ctdb_context *ctdb,
        *header = *(struct ctdb_ltdb_header *)rec.dptr;
 
        data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
-       data->dptr = talloc_memdup(ctdb, sizeof(struct ctdb_ltdb_header)+rec.dptr,
+       data->dptr = talloc_memdup(ctdb_db, sizeof(struct ctdb_ltdb_header)+rec.dptr,
                                   data->dsize);
        free(rec.dptr);
        CTDB_NO_MEMORY(ctdb, data->dptr);
@@ -105,9 +132,10 @@ int ctdb_ltdb_fetch(struct ctdb_context *ctdb,
   and returning the body of the record. A valid (initial) header is
   returned if the record is not present
 */
-int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key, 
+int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
                    struct ctdb_ltdb_header *header, TDB_DATA data)
 {
+       struct ctdb_context *ctdb = ctdb_db->ctdb;
        TDB_DATA rec;
        int ret;
 
@@ -118,7 +146,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->tdb, key, rec, TDB_REPLACE);
+       ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
        talloc_free(rec.dptr);
 
        return ret;

diff --git a/ctdb/ctdb_bench.c b/ctdb/ctdb_bench.c
index 60d8b7990422bc33c6129fe31860b7ed97da3901..023c76e7f9091c718134de26f6f8724b2705b003 100644 (file)
--- a/ctdb/ctdb_bench.c
+++ b/ctdb/ctdb_bench.c
@@ -80,7 +80,7 @@ static int fetch_func(struct ctdb_call_info *call)
 /*
   benchmark incrementing an integer
 */
-static void bench_incr(struct ctdb_context *ctdb)
+static void bench_incr(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_db)
 {
        int loops=0;
        int ret, i;
@@ -98,7 +98,7 @@ static void bench_incr(struct ctdb_context *ctdb)
                call.key.dsize = 4;
 
                for (i=0;i<num_repeats;i++) {
-                       ret = ctdb_call(ctdb, &call);
+                       ret = ctdb_call(ctdb_db, &call);
                        if (ret != 0) {
                                printf("incr call failed - %s\n", ctdb_errstr(ctdb));
                                return;
@@ -113,7 +113,7 @@ static void bench_incr(struct ctdb_context *ctdb)
 
        call.call_id = FUNC_FETCH;
 
-       ret = ctdb_call(ctdb, &call);
+       ret = ctdb_call(ctdb_db, &call);
        if (ret == -1) {
                printf("ctdb_call FUNC_FETCH failed - %s\n", ctdb_errstr(ctdb));
                return;
@@ -195,6 +195,7 @@ static void bench_ring(struct ctdb_context *ctdb, struct event_context *ev)
 int main(int argc, const char *argv[])
 {
        struct ctdb_context *ctdb;
+       struct ctdb_db_context *ctdb_db;
        const char *nlist = NULL;
        const char *transport = "tcp";
        const char *myaddress = NULL;
@@ -274,17 +275,17 @@ int main(int argc, const char *argv[])
                exit(1);
        }
 
-       /* setup a ctdb call function */
-       ret = ctdb_set_call(ctdb, incr_func,  FUNC_INCR);
-       ret = ctdb_set_call(ctdb, fetch_func, FUNC_FETCH);
-
        /* attach to a specific database */
-       ret = ctdb_attach(ctdb, "test.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
-       if (ret == -1) {
+       ctdb_db = ctdb_attach(ctdb, "test.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
+       if (!ctdb_db) {
                printf("ctdb_attach failed - %s\n", ctdb_errstr(ctdb));
                exit(1);
        }
 
+       /* setup a ctdb call function */
+       ret = ctdb_set_call(ctdb_db, incr_func,  FUNC_INCR);
+       ret = ctdb_set_call(ctdb_db, fetch_func, FUNC_FETCH);
+
        ctdb_set_message_handler(ctdb, ring_message_handler, &msg_count);
 
        /* start the protocol running */

diff --git a/ctdb/ctdb_test.c b/ctdb/ctdb_test.c
index faf51f723bd96addc5577650321ffedc54f45aa1..752acd373def9b404e44b8565f3690b637b0c944 100644 (file)
--- a/ctdb/ctdb_test.c
+++ b/ctdb/ctdb_test.c
@@ -76,6 +76,7 @@ static int fetch_func(struct ctdb_call_info *call)
 int main(int argc, const char *argv[])
 {
        struct ctdb_context *ctdb;
+       struct ctdb_db_context *ctdb_db;
        const char *nlist = NULL;
        const char *transport = "tcp";
        const char *myaddress = NULL;
@@ -153,17 +154,17 @@ int main(int argc, const char *argv[])
                exit(1);
        }
 
-       /* setup a ctdb call function */
-       ret = ctdb_set_call(ctdb, sort_func,  FUNC_SORT);
-       ret = ctdb_set_call(ctdb, fetch_func, FUNC_FETCH);
-
        /* attach to a specific database */
-       ret = ctdb_attach(ctdb, "test.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
-       if (ret == -1) {
+       ctdb_db = ctdb_attach(ctdb, "test.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
+       if (!ctdb_db) {
                printf("ctdb_attach failed - %s\n", ctdb_errstr(ctdb));
                exit(1);
        }
 
+       /* setup a ctdb call function */
+       ret = ctdb_set_call(ctdb_db, sort_func,  FUNC_SORT);
+       ret = ctdb_set_call(ctdb_db, fetch_func, FUNC_FETCH);
+
        /* start the protocol running */
        ret = ctdb_start(ctdb);
 
@@ -183,7 +184,7 @@ int main(int argc, const char *argv[])
                call.call_data.dptr = (uint8_t *)&v;
                call.call_data.dsize = sizeof(v);
 
-               ret = ctdb_call(ctdb, &call);
+               ret = ctdb_call(ctdb_db, &call);
                if (ret == -1) {
                        printf("ctdb_call FUNC_SORT failed - %s\n", ctdb_errstr(ctdb));
                        exit(1);
@@ -195,7 +196,7 @@ int main(int argc, const char *argv[])
        call.call_data.dptr = NULL;
        call.call_data.dsize = 0;
 
-       ret = ctdb_call(ctdb, &call);
+       ret = ctdb_call(ctdb_db, &call);
        if (ret == -1) {
                printf("ctdb_call FUNC_FETCH failed - %s\n", ctdb_errstr(ctdb));
                exit(1);

diff --git a/ctdb/include/ctdb.h b/ctdb/include/ctdb.h
index efb12c5daac46d16e27b7af40af6d6defc763794..2859733044af9d80e69e5cc8453b8b77452134de 100644 (file)
--- a/ctdb/include/ctdb.h
+++ b/ctdb/include/ctdb.h
@@ -88,6 +88,12 @@ int ctdb_set_nlist(struct ctdb_context *ctdb, const char *nlist);
 */
 int ctdb_start(struct ctdb_context *ctdb);
 
+/*
+  attach to a ctdb database
+*/
+struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
+                                   int open_flags, mode_t mode);
+
 /*
   error string for last ctdb error
 */
@@ -99,20 +105,15 @@ typedef int (*ctdb_fn_t)(struct ctdb_call_info *);
 /*
   setup a ctdb call function
 */
-int ctdb_set_call(struct ctdb_context *ctdb, ctdb_fn_t fn, int id);
+int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, int id);
 
-/*
-  attach to a ctdb database
-*/
-int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
-               int open_flags, mode_t mode);
 
 
 /*
   make a ctdb call. The associated ctdb call function will be called on the DMASTER
   for the given record
 */
-int ctdb_call(struct ctdb_context *ctdb, struct ctdb_call *call);
+int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call);
 
 /*
   wait for all nodes to be connected - useful for test code

diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index c863781c50e75a81914a17af917d3e52c7c0503c..1e2244ff7de49d49d10126a01cace70cb2fdbb7a 100644 (file)
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -76,6 +76,7 @@ struct ctdb_upcalls {
        void (*node_connected)(struct ctdb_node *);
 };
 
+
 /* main state of the ctdb daemon */
 struct ctdb_context {
        struct event_context *ev;
@@ -87,15 +88,23 @@ struct ctdb_context {
        unsigned flags;
        struct idr_context *idr;
        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_wrap *ltdb;
        const struct ctdb_methods *methods; /* transport methods */
        const struct ctdb_upcalls *upcalls; /* transport upcalls */
        void *private; /* private to transport */
        unsigned max_lacount;
        ctdb_message_fn_t message_handler;
        void *message_private;
+       struct ctdb_db_context *db_list;
+};
+
+struct ctdb_db_context {
+       struct ctdb_db_context *next, *prev;
+       struct ctdb_context *ctdb;
+       uint32_t db_id;
+       const char *db_name;
+       struct tdb_wrap *ltdb;
+       struct ctdb_registered_call *calls; /* list of registered calls */
 };
 
 #define CTDB_NO_MEMORY(ctdb, p) do { if (!(p)) { \
@@ -157,6 +166,7 @@ struct ctdb_req_header {
 
 struct ctdb_req_call {
        struct ctdb_req_header hdr;
+       uint32_t db_id;
        uint32_t callid;
        uint32_t keylen;
        uint32_t calldatalen;
@@ -184,6 +194,7 @@ struct ctdb_reply_redirect {
 
 struct ctdb_req_dmaster {
        struct ctdb_req_header hdr;
+       uint32_t db_id;
        uint32_t dmaster;
        uint32_t keylen;
        uint32_t datalen;
@@ -220,9 +231,9 @@ void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 
 uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key);
-int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
+int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, 
                    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data);
-int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key, 
+int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
                    struct ctdb_ltdb_header *header, TDB_DATA data);
 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);