/* Remove type-safety macros. */
#undef ctdb_attachdb_send
-#undef ctdb_readrecordlock_send
+#undef ctdb_readrecordlock_async
+#undef ctdb_connect
+
+struct ctdb_lock {
+ struct ctdb_lock *next, *prev;
+
+ struct ctdb_db *ctdb_db;
+ TDB_DATA key;
+
+ /* This will always be set by the time user sees this. */
+ unsigned long held_magic;
+ struct ctdb_ltdb_header *hdr;
+
+ /* For convenience, we stash original callback here. */
+ ctdb_rrl_callback_t callback;
+};
+
+static void remove_lock(struct ctdb_connection *ctdb, struct ctdb_lock *lock)
+{
+ DLIST_REMOVE(ctdb->locks, lock);
+}
+
+/* FIXME: for thread safety, need tid info too. */
+static bool holding_lock(struct ctdb_connection *ctdb)
+{
+ /* For the moment, you can't ever hold more than 1 lock. */
+ return (ctdb->locks != NULL);
+}
+
+static void add_lock(struct ctdb_connection *ctdb, struct ctdb_lock *lock)
+{
+ DLIST_ADD(ctdb->locks, lock);
+}
/* FIXME: Could be in shared util code with rest of ctdb */
static void close_noerr(int fd)
struct ctdb_request *req,
void *unused)
{
- if (ctdb_getpnn_recv(req, &ctdb->pnn) != 0) {
- /* FIXME: Report error. */
+ if (!ctdb_getpnn_recv(ctdb, req, &ctdb->pnn)) {
+ DEBUG(ctdb, LOG_CRIT,
+ "ctdb_connect(async): failed to get pnn");
ctdb->broken = true;
}
- ctdb_request_free(req);
+ ctdb_request_free(ctdb, req);
}
-struct ctdb_connection *ctdb_connect(const char *addr)
+struct ctdb_connection *ctdb_connect(const char *addr,
+ ctdb_log_fn_t log_fn, void *log_priv)
{
struct ctdb_connection *ctdb;
struct sockaddr_un sun;
ctdb = malloc(sizeof(*ctdb));
- if (!ctdb)
+ if (!ctdb) {
+ /* With no format string, we hope it doesn't use ap! */
+ va_list ap;
+ memset(&ap, 0, sizeof(ap));
+ errno = ENOMEM;
+ log_fn(log_priv, LOG_ERR, "ctdb_connect: no memory", ap);
goto fail;
+ }
ctdb->outq = NULL;
ctdb->doneq = NULL;
ctdb->in = NULL;
ctdb->message_handlers = NULL;
ctdb->next_id = 0;
ctdb->broken = false;
+ ctdb->log = log_fn;
+ ctdb->log_priv = log_priv;
+ ctdb->locks = NULL;
memset(&sun, 0, sizeof(sun));
sun.sun_family = AF_UNIX;
return req;
}
-void ctdb_request_free(struct ctdb_request *req)
+void ctdb_request_free(struct ctdb_connection *ctdb, struct ctdb_request *req)
{
if (req->extra_destructor) {
- req->extra_destructor(req);
+ req->extra_destructor(ctdb, req);
}
if (req->reply) {
free_io_elem(req->reply);
free(req);
}
-/* Sanity-checking wrapper for reply.
- * FIXME: logging! */
-static struct ctdb_reply_call *unpack_reply_call(struct ctdb_request *req,
+/* Sanity-checking wrapper for reply. */
+static struct ctdb_reply_call *unpack_reply_call(struct ctdb_connection *ctdb,
+ struct ctdb_request *req,
uint32_t callid)
{
size_t len;
struct ctdb_reply_call *inhdr = io_elem_data(req->reply, &len);
- /* ctdbd or our error if this isn't a reply call. */
- if (len < sizeof(*inhdr) || inhdr->hdr.operation != CTDB_REPLY_CALL) {
- errno = EIO;
+ /* Library user error if this isn't a reply to a call. */
+ if (req->hdr.hdr->operation != CTDB_REQ_CALL) {
+ errno = EINVAL;
+ DEBUG(ctdb, LOG_ERR,
+ "This was not a ctdbd call request: operation %u",
+ req->hdr.hdr->operation);
return NULL;
}
- /* Library user error if this isn't a reply to a call. */
- if (req->hdr.hdr->operation != CTDB_REQ_CALL
- || req->hdr.call->callid != callid) {
+ if (req->hdr.call->callid != callid) {
errno = EINVAL;
+ DEBUG(ctdb, LOG_ERR,
+ "This was not a ctdbd %u call request: %u",
+ callid, req->hdr.call->callid);
+ return NULL;
+ }
+
+ /* ctdbd or our error if this isn't a reply call. */
+ if (len < sizeof(*inhdr) || inhdr->hdr.operation != CTDB_REPLY_CALL) {
+ errno = EIO;
+ DEBUG(ctdb, LOG_CRIT,
+ "Invalid ctdbd call reply: len %zu, operation %u",
+ len, inhdr->hdr.operation);
return NULL;
}
return inhdr;
}
-/* Sanity-checking wrapper for reply.
- * FIXME: logging! */
-struct ctdb_reply_control *unpack_reply_control(struct ctdb_request *req,
+/* Sanity-checking wrapper for reply. */
+struct ctdb_reply_control *unpack_reply_control(struct ctdb_connection *ctdb,
+ struct ctdb_request *req,
enum ctdb_controls control)
{
size_t len;
struct ctdb_reply_control *inhdr = io_elem_data(req->reply, &len);
/* Library user error if this isn't a reply to a call. */
- if (len < sizeof(*inhdr)
- || req->hdr.hdr->operation != CTDB_REQ_CONTROL) {
+ if (len < sizeof(*inhdr)) {
+ errno = EINVAL;
+ DEBUG(ctdb, LOG_CRIT,
+ "Short ctdbd control reply: %zu bytes", len);
+ return NULL;
+ }
+ if (req->hdr.hdr->operation != CTDB_REQ_CONTROL) {
errno = EINVAL;
+ DEBUG(ctdb, LOG_ERR,
+ "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_ERR,
+ "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,
+ "Invalid ctdbd control reply: operation %u",
+ inhdr->hdr.operation);
return NULL;
}
return;
}
}
- /* FIXME: report this error. */
+ DEBUG(ctdb, LOG_WARNING,
+ "Unexpected ctdbd request reply: operation %u reqid %u",
+ hdr->operation, hdr->reqid);
free_io_elem(in);
}
return ret;
}
-int ctdb_service(struct ctdb_connection *ctdb, int revents)
+bool ctdb_service(struct ctdb_connection *ctdb, int revents)
{
if (ctdb->broken) {
- return -1;
+ return false;
+ }
+
+ if (holding_lock(ctdb)) {
+ DEBUG(ctdb, LOG_WARNING, "Do not block while holding lock!");
}
if (revents & POLLOUT) {
while (ctdb->outq) {
if (real_error(write_io_elem(ctdb->fd,
ctdb->outq->io)) < 0) {
+ DEBUG(ctdb, LOG_ERR,
+ "ctdb_service: error writing to ctdbd");
ctdb->broken = true;
- return -1;
+ return false;
}
if (io_elem_finished(ctdb->outq->io)) {
struct ctdb_request *done = ctdb->outq;
if (!ctdb->in) {
ctdb->in = new_io_elem(sizeof(struct ctdb_req_header));
if (!ctdb->in) {
+ DEBUG(ctdb, LOG_ERR,
+ "ctdb_service: allocating readbuf");
ctdb->broken = true;
- return -1;
+ return false;
}
}
/* They closed fd? */
if (ret == 0)
errno = EBADF;
+ DEBUG(ctdb, LOG_ERR,
+ "ctdb_service: error reading from ctdbd");
ctdb->broken = true;
- return -1;
+ return false;
} else if (ret < 0) {
/* No progress, stop loop. */
revents = 0;
}
}
- return 0;
+ return true;
}
/* This is inefficient. We could pull in idtree.c. */
CTDB_REQ_CONTROL, destnode, new_reqid(ctdb));
pkt = req->hdr.control;
+ pkt->pad = 0;
pkt->opcode = opcode;
pkt->srvid = 0;
pkt->client_id = 0;
struct ctdb_request *req,
void *unused)
{
- ctdb_request_free(req);
+ ctdb_request_free(ctdb, req);
}
-int ctdb_cancel(struct ctdb_request *req)
+void ctdb_cancel(struct ctdb_connection *ctdb, struct ctdb_request *req)
{
+ DEBUG(ctdb, LOG_DEBUG, "ctdb_cancel: %p (id %u)",
+ req, req->hdr.hdr ? req->hdr.hdr->reqid : 0);
+
/* FIXME: If it's not sent, we could just free it right now. */
req->callback = ctdb_cancel_callback;
- return 0;
}
struct ctdb_db {
uint32_t id;
struct tdb_context *tdb;
+ /* The lock we are holding, if any (we can only have one!) */
+ struct ctdb_lock *lock;
+
ctdb_callback_t callback;
void *private_data;
};
db->callback(ctdb, req->extra, db->private_data);
}
-struct ctdb_db *ctdb_attachdb_recv(struct ctdb_request *req)
+struct ctdb_db *ctdb_attachdb_recv(struct ctdb_connection *ctdb,
+ struct ctdb_request *req)
{
struct ctdb_request *dbpath_req = req->extra;
struct ctdb_reply_control *reply;
return NULL;
}
- reply = unpack_reply_control(dbpath_req, CTDB_CONTROL_GETDBPATH);
- if (!reply || reply->status != 0) {
+ reply = unpack_reply_control(ctdb, dbpath_req, CTDB_CONTROL_GETDBPATH);
+ if (!reply) {
+ return NULL;
+ }
+ if (reply->status != 0) {
+ DEBUG(db->ctdb, LOG_ERR,
+ "ctdb_attachdb_recv: reply status %i", reply->status);
return NULL;
}
tdb_flags = db->persistent ? TDB_DEFAULT : TDB_NOSYNC;
tdb_flags |= TDB_DISALLOW_NESTING;
+ /* FIXME: Setup logging to go through our logging. */
db->tdb = tdb_open((char *)reply->data, 0, tdb_flags, O_RDWR, 0);
if (db->tdb == NULL) {
+ DEBUG(db->ctdb, LOG_ERR,
+ "ctdb_attachdb_recv: failed to tdb_open %s",
+ (char *)reply->data);
return NULL;
}
/* Finally, separate the db from the request (see destroy_req_db). */
req->priv_data = NULL;
+ DEBUG(db->ctdb, LOG_DEBUG,
+ "ctdb_attachdb_recv: db %p, tdb %s", db, (char *)reply->data);
return db;
}
control = CTDB_CONTROL_DB_ATTACH_PERSISTENT;
}
- reply = unpack_reply_control(req, control);
+ reply = unpack_reply_control(ctdb, req, control);
if (!reply || reply->status != 0) {
+ if (reply) {
+ DEBUG(ctdb, LOG_ERR,
+ "ctdb_attachdb_send(async): DB_ATTACH status %i",
+ reply->status);
+ }
/* We failed. Hand request to user and have them discover it
* via ctdb_attachdb_recv. */
- db->callback(ctdb, req, db);
+ db->callback(ctdb, req, db->private_data);
return;
}
db->id = *(uint32_t *)reply->data;
&db->id, sizeof(db->id),
attachdb_getdbpath_done, db);
if (!req2) {
- db->callback(ctdb, req, db);
+ DEBUG(db->ctdb, LOG_ERR,
+ "ctdb_attachdb_send(async): failed to allocate");
+ db->callback(ctdb, req, db->private_data);
return;
}
req->extra = req2;
req2->extra = req;
+ DEBUG(db->ctdb, LOG_DEBUG,
+ "ctdb_attachdb_send(async): created getdbpath request");
}
-static void destroy_req_db(struct ctdb_request *req)
+static void destroy_req_db(struct ctdb_connection *ctdb,
+ struct ctdb_request *req)
{
/* Incomplete db is in priv_data. */
free(req->priv_data);
/* second request is chained off this one. */
if (req->extra) {
- ctdb_request_free(req->extra);
+ ctdb_request_free(ctdb, req->extra);
}
}
req = new_ctdb_control_request(ctdb, opcode, CTDB_CURRENT_NODE, name,
strlen(name) + 1, attachdb_done, db);
if (!req) {
+ DEBUG(db->ctdb, LOG_ERR,
+ "ctdb_attachdb_send: failed allocating DB_ATTACH");
free(db);
return NULL;
}
/* Flags get overloaded into srvid. */
req->hdr.control->srvid = tdb_flags;
+ DEBUG(db->ctdb, LOG_DEBUG,
+ "ctdb_attachdb_send: DB_ATTACH request %p", req);
return req;
}
-struct ctdb_lock {
- struct ctdb_db *ctdb_db;
- TDB_DATA key;
+static unsigned long lock_magic(struct ctdb_lock *lock)
+{
+ /* A non-zero magic specific to this structure. */
+ return ((unsigned long)lock->key.dptr
+ ^ (((unsigned long)lock->key.dptr) << 16)
+ ^ 0xBADC0FFEEBADC0DEULL)
+ | 1;
+}
- /* This will always be true by the time user sees this. */
- bool held;
- struct ctdb_ltdb_header *hdr;
- TDB_DATA data;
+/* This is only called on locks before they're held. */
+static void free_lock(struct ctdb_lock *lock)
+{
+ if (lock->held_magic) {
+ DEBUG(lock->ctdb_db->ctdb, LOG_CRIT,
+ "free_lock invalid lock %p", lock);
+ }
+ free(lock->hdr);
+ free(lock);
+}
- /* For convenience, we stash original callback here. */
- ctdb_callback_t callback;
-};
void ctdb_release_lock(struct ctdb_lock *lock)
{
- if (lock->held) {
+ if (lock->held_magic != lock_magic(lock)) {
+ DEBUG(lock->ctdb_db->ctdb, LOG_CRIT,
+ "ctdb_release_lock invalid lock %p", lock);
+ } else {
tdb_chainunlock(lock->ctdb_db->tdb, lock->key);
+ DEBUG(lock->ctdb_db->ctdb, LOG_DEBUG,
+ "ctdb_release_lock %p", lock);
+ remove_lock(lock->ctdb_db->ctdb, lock);
}
- free(lock->hdr); /* Also frees data */
- free(lock);
+ lock->held_magic = 0;
+ free_lock(lock);
}
+
/* We keep the lock if local node is the dmaster. */
-static bool try_readrecordlock(struct ctdb_lock *lock)
+static bool try_readrecordlock(struct ctdb_lock *lock, TDB_DATA *data)
{
struct ctdb_ltdb_header *hdr;
if (tdb_chainlock(lock->ctdb_db->tdb, lock->key) != 0) {
+ DEBUG(lock->ctdb_db->ctdb, LOG_WARNING,
+ "ctdb_readrecordlock_async: failed to chainlock");
return NULL;
}
- hdr = ctdb_local_fetch(lock->ctdb_db->tdb, lock->key, &lock->data);
+ hdr = ctdb_local_fetch(lock->ctdb_db->tdb, lock->key, data);
if (hdr && hdr->dmaster == lock->ctdb_db->ctdb->pnn) {
- lock->held = true;
+ DEBUG(lock->ctdb_db->ctdb, LOG_DEBUG,
+ "ctdb_readrecordlock_async: got local lock");
+ lock->held_magic = lock_magic(lock);
lock->hdr = hdr;
+ add_lock(lock->ctdb_db->ctdb, lock);
return true;
}
return NULL;
}
-/* If they cancel *before* we hand them the lock from
- * ctdb_readrecordlock_recv, we free it here. */
-static void destroy_lock(struct ctdb_request *req)
-{
- ctdb_release_lock(req->extra);
-}
-
-struct ctdb_lock *ctdb_readrecordlock_recv(struct ctdb_db *ctdb_db,
- struct ctdb_request *req,
- TDB_DATA *data)
+/* If they shutdown before we hand them the lock, we free it here. */
+static void destroy_lock(struct ctdb_connection *ctdb,
+ struct ctdb_request *req)
{
- struct ctdb_lock *lock = req->extra;
-
- if (!lock->held) {
- /* Something went wrong. */
- return NULL;
- }
-
- /* Now it's their responsibility to free! */
- req->extra_destructor = NULL;
- *data = lock->data;
- return lock;
+ free_lock(req->extra);
}
static void readrecordlock_retry(struct ctdb_connection *ctdb,
{
struct ctdb_lock *lock = req->extra;
struct ctdb_reply_call *reply;
+ TDB_DATA data;
/* OK, we've received reply to noop migration */
- reply = unpack_reply_call(req, CTDB_NULL_FUNC);
+ reply = unpack_reply_call(ctdb, req, CTDB_NULL_FUNC);
if (!reply || reply->status != 0) {
- lock->callback(ctdb, req, private);
+ if (reply) {
+ DEBUG(ctdb, LOG_ERR,
+ "ctdb_readrecordlock_async(async):"
+ " NULL_FUNC returned %i", reply->status);
+ }
+ lock->callback(lock->ctdb_db, NULL, tdb_null, private);
+ ctdb_request_free(ctdb, req); /* Also frees lock. */
return;
}
/* Can we get lock now? */
- if (try_readrecordlock(lock)) {
- lock->callback(ctdb, req, private);
+ if (try_readrecordlock(lock, &data)) {
+ /* Now it's their responsibility to free lock & request! */
+ req->extra_destructor = NULL;
+ lock->callback(lock->ctdb_db, lock, data, private);
return;
}
/* Retransmit the same request again (we lost race). */
io_elem_reset(req->io);
DLIST_ADD(ctdb->outq, req);
- return;
}
bool
-ctdb_readrecordlock_send(struct ctdb_db *ctdb_db, TDB_DATA key,
- struct ctdb_request **reqp,
- ctdb_callback_t callback, void *cbdata)
+ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
+ ctdb_rrl_callback_t callback, void *cbdata)
{
struct ctdb_request *req;
struct ctdb_lock *lock;
+ TDB_DATA data;
+
+ if (holding_lock(ctdb_db->ctdb)) {
+ DEBUG(ctdb_db->ctdb, LOG_ERR,
+ "ctdb_readrecordlock_async: already holding lock");
+ return false;
+ }
- /* Setup lock. */
+ /* Setup lock */
lock = malloc(sizeof(*lock) + key.dsize);
if (!lock) {
- return NULL;
+ DEBUG(ctdb_db->ctdb, LOG_ERR,
+ "ctdb_readrecordlock_async: lock allocation failed");
+ return false;
}
lock->key.dptr = (void *)(lock + 1);
memcpy(lock->key.dptr, key.dptr, key.dsize);
lock->key.dsize = key.dsize;
lock->ctdb_db = ctdb_db;
lock->hdr = NULL;
- lock->held = false;
+ lock->held_magic = 0;
+
+ /* Fast path. */
+ if (try_readrecordlock(lock, &data)) {
+ callback(ctdb_db, lock, data, cbdata);
+ return true;
+ }
- /* Get ready in case we need to send a migrate request. */
+ /* Slow path: create request. */
req = new_ctdb_request(offsetof(struct ctdb_req_call, data)
- + key.dsize, callback, cbdata);
+ + key.dsize, readrecordlock_retry, cbdata);
if (!req) {
- ctdb_release_lock(lock);
+ DEBUG(ctdb_db->ctdb, LOG_ERR,
+ "ctdb_readrecordlock_async: allocation failed");
+ free_lock(lock);
return NULL;
}
req->extra = lock;
req->extra_destructor = destroy_lock;
-
- if (try_readrecordlock(lock)) {
- *reqp = NULL;
- callback(ctdb_db->ctdb, req, cbdata);
- return true;
- }
-
/* We store the original callback in the lock, and use our own. */
lock->callback = callback;
- req->callback = readrecordlock_retry;
io_elem_init_req_header(req->io, CTDB_REQ_CALL, CTDB_CURRENT_NODE,
new_reqid(ctdb_db->ctdb));
req->hdr.call->calldatalen = 0;
memcpy(req->hdr.call->data, key.dptr, key.dsize);
DLIST_ADD(ctdb_db->ctdb->outq, req);
- *reqp = req;
return true;
}
int ctdb_writerecord(struct ctdb_lock *lock, TDB_DATA data)
{
+ if (lock->held_magic != lock_magic(lock)) {
+ errno = EBADF;
+ DEBUG(lock->ctdb_db->ctdb, LOG_ERR,
+ "ctdb_writerecord: Can not write. Lock has been released.");
+ return -1;
+ }
+
if (lock->ctdb_db->persistent) {
- /* FIXME: Report error. */
+ errno = EINVAL;
+ DEBUG(lock->ctdb_db->ctdb, LOG_ERR,
+ "ctdb_writerecord: cannot write to persistent db");
return -1;
}
git.samba.org / rusty / ctdb.git / blobdiff