core of libctdb
Copyright (C) Rusty Russell 2010
+ Copyright (C) Ronnie Sahlberg 2011
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
/* Remove type-safety macros. */
#undef ctdb_attachdb_send
#undef ctdb_readrecordlock_async
+#undef ctdb_readonlyrecordlock_async
#undef ctdb_connect
struct ctdb_lock {
struct ctdb_db *ctdb_db;
TDB_DATA key;
+ /* Is this a request for read-only lock ? */
+ bool readonly;
+
/* This will always be set by the time user sees this. */
unsigned long held_magic;
struct ctdb_ltdb_header *hdr;
log_fn(log_priv, LOG_ERR, "ctdb_connect: no memory", ap);
goto fail;
}
+ ctdb->pnn = -1;
ctdb->outq = NULL;
ctdb->doneq = NULL;
ctdb->in = NULL;
}
/* Sanity-checking wrapper for reply. */
-struct ctdb_reply_control *unpack_reply_control(struct ctdb_connection *ctdb,
- struct ctdb_request *req,
+struct ctdb_reply_control *unpack_reply_control(struct ctdb_request *req,
enum ctdb_controls control)
{
size_t len;
/* Library user error if this isn't a reply to a call. */
if (len < sizeof(*inhdr)) {
errno = EINVAL;
- DEBUG(ctdb, LOG_ALERT,
+ DEBUG(req->ctdb, LOG_ALERT,
"Short ctdbd control reply: %zu bytes", len);
return NULL;
}
if (req->hdr.hdr->operation != CTDB_REQ_CONTROL) {
errno = EINVAL;
- DEBUG(ctdb, LOG_ALERT,
+ DEBUG(req->ctdb, LOG_ALERT,
"This was not a ctdbd control request: operation %u",
req->hdr.hdr->operation);
return NULL;
/* ... or if it was a different control from what we expected. */
if (req->hdr.control->opcode != control) {
errno = EINVAL;
- DEBUG(ctdb, LOG_ALERT,
+ DEBUG(req->ctdb, LOG_ALERT,
"This was not an opcode %u ctdbd control request: %u",
control, req->hdr.control->opcode);
return NULL;
/* ctdbd or our error if this isn't a reply call. */
if (inhdr->hdr.operation != CTDB_REPLY_CONTROL) {
errno = EIO;
- DEBUG(ctdb, LOG_CRIT,
+ DEBUG(req->ctdb, LOG_CRIT,
"Invalid ctdbd control reply: operation %u",
inhdr->hdr.operation);
return NULL;
control = CTDB_CONTROL_DB_ATTACH_PERSISTENT;
}
- reply = unpack_reply_control(ctdb, req, control);
+ reply = unpack_reply_control(req, control);
if (!reply || reply->status != 0) {
if (reply) {
DEBUG(ctdb, LOG_ERR,
return NULL;
}
- reply = unpack_reply_control(ctdb, dbpath_req, CTDB_CONTROL_GETDBPATH);
+ reply = unpack_reply_control(dbpath_req, CTDB_CONTROL_GETDBPATH);
if (!reply) {
return NULL;
}
}
hdr = ctdb_local_fetch(lock->ctdb_db->tdb, lock->key, data);
+ if (hdr && lock->readonly && (hdr->flags & CTDB_REC_RO_HAVE_READONLY) ) {
+ DEBUG(lock->ctdb_db->ctdb, LOG_DEBUG,
+ "ctdb_readrecordlock_async: got local lock for ro");
+ lock->held_magic = lock_magic(lock);
+ lock->hdr = hdr;
+ add_lock(lock->ctdb_db->ctdb, lock);
+ return true;
+ }
if (hdr && hdr->dmaster == lock->ctdb_db->ctdb->pnn) {
DEBUG(lock->ctdb_db->ctdb, LOG_DEBUG,
"ctdb_readrecordlock_async: got local lock");
return true;
}
+ /* we dont have the record locally,
+ * drop to writelock to force a migration
+ */
+ if (!hdr && lock->readonly) {
+ lock->readonly = false;
+ }
+
tdb_chainunlock(lock->ctdb_db->tdb, lock->key);
free(hdr);
return NULL;
struct ctdb_reply_call *reply;
TDB_DATA data;
- /* OK, we've received reply to noop migration */
- reply = unpack_reply_call(req, CTDB_NULL_FUNC);
+ /* OK, we've received reply to fetch-with-header migration */
+ reply = unpack_reply_call(req, CTDB_FETCH_WITH_HEADER_FUNC);
if (!reply || reply->status != 0) {
if (reply) {
DEBUG(ctdb, LOG_ERR,
"ctdb_readrecordlock_async(async):"
- " NULL_FUNC returned %i", reply->status);
+ " FETCH_WITH_HEADER_FUNC returned %i", reply->status);
}
lock->callback(lock->ctdb_db, NULL, tdb_null, private);
ctdb_request_free(req); /* Also frees lock. */
DLIST_ADD(ctdb->outq, req);
}
-bool
-ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
- ctdb_rrl_callback_t callback, void *cbdata)
+static bool
+ctdb_readrecordlock_internal(struct ctdb_db *ctdb_db, TDB_DATA key,
+ bool readonly,
+ ctdb_rrl_callback_t callback, void *cbdata)
{
struct ctdb_request *req;
struct ctdb_lock *lock;
lock->ctdb_db = ctdb_db;
lock->hdr = NULL;
lock->held_magic = 0;
+ lock->readonly = readonly;
/* Fast path. */
if (try_readrecordlock(lock, &data)) {
io_elem_init_req_header(req->io, CTDB_REQ_CALL, CTDB_CURRENT_NODE,
new_reqid(ctdb_db->ctdb));
- req->hdr.call->flags = CTDB_IMMEDIATE_MIGRATION;
+ if (lock->readonly) {
+ req->hdr.call->flags = CTDB_WANT_READONLY;
+ } else {
+ req->hdr.call->flags = CTDB_IMMEDIATE_MIGRATION;
+ }
req->hdr.call->db_id = ctdb_db->id;
- req->hdr.call->callid = CTDB_NULL_FUNC;
+ req->hdr.call->callid = CTDB_FETCH_WITH_HEADER_FUNC;
req->hdr.call->hopcount = 0;
req->hdr.call->keylen = key.dsize;
req->hdr.call->calldatalen = 0;
return true;
}
+bool
+ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
+ ctdb_rrl_callback_t callback, void *cbdata)
+{
+ return ctdb_readrecordlock_internal(ctdb_db, key,
+ false,
+ callback, cbdata);
+}
+
+bool
+ctdb_readonlyrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
+ ctdb_rrl_callback_t callback, void *cbdata)
+{
+ return ctdb_readrecordlock_internal(ctdb_db, key,
+ true,
+ callback, cbdata);
+}
+
bool ctdb_writerecord(struct ctdb_db *ctdb_db,
struct ctdb_lock *lock, TDB_DATA data)
{
return true;
}
+
+int ctdb_num_out_queue(struct ctdb_connection *ctdb)
+{
+ struct ctdb_request *req;
+ int i;
+
+ for (i = 0, req = ctdb->outq; req; req = req->next, i++)
+ ;
+
+ return i;
+}
+
+int ctdb_num_in_flight(struct ctdb_connection *ctdb)
+{
+ struct ctdb_request *req;
+ int i;
+
+ for (i = 0, req = ctdb->doneq; req; req = req->next, i++)
+ ;
+
+ return i;
+}
+
+int ctdb_num_active(struct ctdb_connection *ctdb)
+{
+ return ctdb_num_out_queue(ctdb)
+ + ctdb_num_in_flight(ctdb);
+}
+
git.samba.org / sahlberg / ctdb.git / blobdiff