diff options
author | Andrew Tridgell <tridge@samba.org> | 2007-04-04 04:57:30 +0000 |
---|---|---|
committer | Gerald (Jerry) Carter <jerry@samba.org> | 2007-10-10 14:49:47 -0500 |
commit | a78be2150ba06738f4c7e85ca2980200d4a3c533 (patch) | |
tree | bc30c417b22a8cd51692635134f4c865a6023235 /source4/cluster/ctdb/common/ctdb_call.c | |
parent | 690df3ccd98c586610cc5e528b05f7863b334a3c (diff) | |
download | samba-a78be2150ba06738f4c7e85ca2980200d4a3c533.tar.gz samba-a78be2150ba06738f4c7e85ca2980200d4a3c533.tar.bz2 samba-a78be2150ba06738f4c7e85ca2980200d4a3c533.zip |
r22070: merge in the changes from the bzr ctdb tree, and convert the brlock
ctdb backend to use the updated multi-database API
(This used to be commit 44dcac9e4d81bfc078512248967b6240db9d1bd8)
Diffstat (limited to 'source4/cluster/ctdb/common/ctdb_call.c')
-rw-r--r-- | source4/cluster/ctdb/common/ctdb_call.c | 80 |
1 files changed, 58 insertions, 22 deletions
diff --git a/source4/cluster/ctdb/common/ctdb_call.c b/source4/cluster/ctdb/common/ctdb_call.c index fb29aad9ac..75355f7ae7 100644 --- a/source4/cluster/ctdb/common/ctdb_call.c +++ b/source4/cluster/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); } |