src/delta_lease.c | 196 ++++++++++++++++++++++++++++---------------------
src/direct_lib.c | 2
src/diskio.c | 75 +++++++++---------
src/diskio.h | 17 +++-
src/host_id.c | 77 ++++---------------
src/host_id.h | 1
src/leader.h | 2
src/log.c | 18 ++--
src/log.h | 23 +++--
src/main.c | 95 ++++++++++++++++++++---
src/paxos_lease.c | 2
src/sanlock_internal.h | 3
src/task.c | 72 ++++++++++++++++--
src/token_manager.c | 1
14 files changed, 358 insertions(+), 226 deletions(-)
New commits:
commit e58da9c2426a712575de29d77453ef2fb6e48c79
Author: David Teigland <teigland(a)redhat.com>
Date: Fri Aug 12 11:18:17 2011 -0500
sanlock: read align_size in renewal
Read all delta leases when renewing our own, rather
than reading only our delta lease sector. This is
is preparation for using the info in the other delta
leases.
Also includes
- only permit -a 0|1 (not 2)
- remove bogus error message about corrupt renewal result
- record every new lockspace and resource in /var/log/sanlock.log
by default so the space_id/token_id values can later be
translated to actual lockspace/resource
- improvements to aio related error messages
- initializing task structs to zero
- don't reread sectors due to verify error to avoid
interfering with renewal aio
- free iobuf in find_callback_slot
- initialize our_host_name_global early in main so it
won't be blank if no cmd line option is given
- wait for all aio to complete in close_task_aio before
destroying aio context
diff --git a/src/delta_lease.c b/src/delta_lease.c
index bf5ad82..b210d7f 100644
--- a/src/delta_lease.c
+++ b/src/delta_lease.c
@@ -35,6 +35,12 @@
/* delta_leases are a series max_hosts leader_records, one leader per sector,
host N's delta_lease is the leader_record in sectors N-1 */
+/*
+ * variable names:
+ * rv: success is 0, failure is < 0
+ * error: success is 1 (SANLK_OK), failure is < 0
+ */
+
static void log_leader_error(int result,
char *space_name,
uint64_t host_id,
@@ -75,9 +81,8 @@ static int verify_leader(struct sync_disk *disk,
struct leader_record *lr,
const char *caller)
{
- struct leader_record leader_rr;
uint32_t sum;
- int result, rv;
+ int result;
if (lr->magic != DELTA_DISK_MAGIC) {
log_error("verify_leader %llu wrong magic %x %s",
@@ -126,6 +131,10 @@ static int verify_leader(struct sync_disk *disk,
fail:
log_leader_error(result, space_name, host_id, disk, lr, caller);
+ /*
+ struct leader_record leader_rr;
+ int rv;
+
memset(&leader_rr, 0, sizeof(leader_rr));
rv = read_sectors(disk, host_id - 1, 1, (char *)&leader_rr,
@@ -133,6 +142,7 @@ static int verify_leader(struct sync_disk *disk,
NULL, "delta_verify");
log_leader_error(rv, space_name, host_id, disk, &leader_rr, "delta_verify");
+ */
return result;
}
@@ -163,32 +173,6 @@ int delta_lease_leader_read(struct task *task,
return error;
}
-static int delta_lease_leader_reap(struct task *task,
- struct sync_disk *disk,
- char *space_name,
- uint64_t host_id,
- struct leader_record *leader_ret,
- const char *caller)
-{
- struct leader_record leader;
- int rv, error;
-
- /* host_id N is block offset N-1 */
-
- memset(&leader, 0, sizeof(struct leader_record));
- memset(leader_ret, 0, sizeof(struct leader_record));
-
- rv = read_sectors_reap(disk, host_id - 1, 1, (char *)&leader, sizeof(struct leader_record),
- task, "delta_leader");
- if (rv < 0)
- return rv;
-
- error = verify_leader(disk, space_name, host_id, &leader, caller);
-
- memcpy(leader_ret, &leader, sizeof(struct leader_record));
- return error;
-}
-
/*
* delta_lease_acquire:
* set the owner of host_id to our_host_name.
@@ -215,7 +199,7 @@ int delta_lease_acquire(struct task *task,
struct leader_record leader;
struct leader_record leader1;
uint64_t new_ts;
- int i, error, delay, delta_large_delay;
+ int i, error, rv, delay, delta_large_delay;
log_space(sp, "delta_acquire %llu begin", (unsigned long long)host_id);
@@ -300,10 +284,10 @@ int delta_lease_acquire(struct task *task,
(unsigned long long)leader.timestamp,
leader.resource_name);
- error = write_sector(disk, host_id - 1, (char *)&leader, sizeof(struct leader_record),
- task, "delta_leader");
- if (error < 0)
- return error;
+ rv = write_sector(disk, host_id - 1, (char *)&leader, sizeof(struct leader_record),
+ task, "delta_leader");
+ if (rv < 0)
+ return rv;
memcpy(&leader1, &leader, sizeof(struct leader_record));
@@ -345,41 +329,111 @@ int delta_lease_renew(struct task *task,
struct leader_record *leader_ret)
{
struct leader_record leader;
- uint64_t host_id;
+ uint64_t host_id, offset;
uint64_t new_ts;
- int io_timeout_save;
- int error;
+ char **p_iobuf;
+ int iobuf_len, io_timeout_save;
+ int rv;
if (!leader_last)
return -EINVAL;
host_id = leader_last->owner_id;
+ /* read all delta leases */
+ iobuf_len = direct_align(disk);
+ if (iobuf_len <= 0)
+ return -EINVAL;
+
+ /* offset of our leader_record */
+ offset = (host_id - 1) * disk->sector_size;
+ if (offset > iobuf_len)
+ return -EINVAL;
+
+
/* if the previous renew timed out in this initial read, and that read
is now complete, we can use that result here instead of discarding
it and doing another. */
- if (prev_result == SANLK_AIO_TIMEOUT && task->read_timeout) {
- error = delta_lease_leader_reap(task, disk, space_name, host_id,
- &leader, "delta_renew_reap");
+ if (prev_result == SANLK_AIO_TIMEOUT) {
+ if (!task->read_iobuf_timeout_aicb) {
+ /* shouldn't happen, when do_linux_aio returned AIO_TIMEOUT
+ it should have set read_iobuf_timeout_aicb */
+ log_erros(sp, "delta_renew reap no aicb");
+ goto skip_reap;
+ }
+
+ if (!task->iobuf) {
+ /* shouldn't happen */
+ log_erros(sp, "delta_renew reap no iobuf");
+ goto skip_reap;
+ }
- log_space(sp, "delta_renew %llu reap %d",
- (unsigned long long)host_id, error);
+ rv = read_iobuf_reap(disk->fd, disk->offset,
+ task->iobuf, iobuf_len, task);
- if (error == SANLK_OK) {
- task->read_timeout = NULL;
+ log_space(sp, "delta_renew reap %d", rv);
+
+ if (!rv) {
+ task->read_iobuf_timeout_aicb = NULL;
goto read_done;
}
+ skip_reap:
+ /* abandon the previous timed out read and try a new
+ one from scratch. the current task->iobuf mem will
+ freed when timeout_aicb completes sometime */
+
+ task->read_iobuf_timeout_aicb = NULL;
+ task->iobuf = NULL;
}
- task->read_timeout = NULL;
+ if (task->read_iobuf_timeout_aicb) {
+ /* this could happen get here if there was another read between
+ renewal reads, which timed out and caused
+ read_iobuf_timeout_aicb to be set; I don't think there are
+ any cases where that would happen, though. we could avoid
+ this confusion by passing back the timed out aicb along with
+ SANLK_AIO_TIMEOUT, and only save the timed out aicb when we
+ want to try to reap it later. */
+
+ log_space(sp, "delta_renew timeout_aicb is unexpectedly %p iobuf %p",
+ task->read_iobuf_timeout_aicb, task->iobuf);
+ task->read_iobuf_timeout_aicb = NULL;
+ task->iobuf = NULL;
+ }
- error = delta_lease_leader_read(task, disk, space_name, host_id,
- &leader, "delta_renew_read");
- if (error < 0)
- return error;
+ if (!task->iobuf) {
+ /* this will happen the first time renew is called, and after
+ a timed out renewal read fails to be reaped (see
+ task->iobuf = NULL above) */
+
+ p_iobuf = &task->iobuf;
+
+ rv = posix_memalign((void *)p_iobuf, getpagesize(), iobuf_len);
+ if (rv) {
+ log_erros(sp, "dela_renew memalign rv %d", rv);
+ rv = -ENOMEM;
+ }
+ }
+
+ rv = read_iobuf(disk->fd, disk->offset, task->iobuf, iobuf_len, task);
+ if (rv) {
+ /* the next time delta_lease_renew() is called, prev_result
+ will be this rv. If this rv is SANLK_AIO_TIMEOUT, we'll
+ try to reap the event */
+
+ log_erros(sp, "delta_renew read rv %d offset %llu %s",
+ rv, (unsigned long long)disk->offset, disk->path);
+ return rv;
+ }
read_done:
+ memcpy(&leader, task->iobuf+offset, sizeof(struct leader_record));
+
+ rv = verify_leader(disk, space_name, host_id, &leader, "delta_renew");
+ if (rv < 0)
+ return rv;
+
/* We can't always memcmp(&leader, leader_last) because previous writes
may have timed out and we don't know if they were actually written
or not. We can definately verify that we're still the owner,
@@ -388,7 +442,7 @@ int delta_lease_renew(struct task *task,
if (leader.owner_id != leader_last->owner_id ||
leader.owner_generation != leader_last->owner_generation ||
memcmp(leader.resource_name, leader_last->resource_name, NAME_ID_SIZE)) {
- log_erros(sp, "delta_renew %llu not owner", (unsigned long long)host_id);
+ log_erros(sp, "delta_renew not owner");
log_leader_error(0, space_name, host_id, disk, leader_last, "delta_renew_last");
log_leader_error(0, space_name, host_id, disk, &leader, "delta_renew_read");
return SANLK_RENEW_OWNER;
@@ -396,7 +450,7 @@ int delta_lease_renew(struct task *task,
if (prev_result == SANLK_OK &&
memcmp(&leader, leader_last, sizeof(struct leader_record))) {
- log_erros(sp, "delta_renew %llu reread mismatch", (unsigned long long)host_id);
+ log_erros(sp, "delta_renew reread mismatch");
log_leader_error(0, space_name, host_id, disk, leader_last, "delta_renew_last");
log_leader_error(0, space_name, host_id, disk, &leader, "delta_renew_read");
return SANLK_RENEW_DIFF;
@@ -419,41 +473,17 @@ int delta_lease_renew(struct task *task,
io_timeout_save = task->io_timeout_seconds;
task->io_timeout_seconds = task->host_dead_seconds;
- error = write_sector(disk, host_id - 1, (char *)&leader, sizeof(struct leader_record),
- task, "delta_leader");
+ rv = write_sector(disk, host_id - 1, (char *)&leader, sizeof(struct leader_record),
+ task, "delta_leader");
task->io_timeout_seconds = io_timeout_save;
- if (error < 0)
- return error;
+ if (rv < 0)
+ return rv;
-#if 0
/* the paper shows doing a delay and another read here, but it seems
unnecessary since we do the same at the beginning of the next renewal */
- delay = 2 * task->io_timeout_seconds;
- /* log_space(sp, "delta_renew sleep 2d %d", delay); */
- sleep(delay);
-
- error = delta_lease_leader_read(task, disk, space_name, host_id, &leader_read,
- "delta_renew_check");
- if (error < 0)
- return error;
-
- /*
- if ((leader.timestamp != new_ts) || (leader.owner_id != our_host_id))
- return SANLK_BAD_LEADER;
- */
-
- if (memcmp(&leader, &leader_read, sizeof(struct leader_record))) {
- log_erros(sp, "delta_renew %llu reread mismatch",
- (unsigned long long)host_id);
- log_leader_error(0, space_name, host_id, disk, &leader, "delta_renew_write");
- log_leader_error(0, space_name, host_id, disk, &leader_read, "delta_renew_reread");
- return SANLK_RENEW_DIFF;
- }
-#endif
-
memcpy(leader_ret, &leader, sizeof(struct leader_record));
return SANLK_OK;
}
@@ -467,7 +497,7 @@ int delta_lease_release(struct task *task,
{
struct leader_record leader;
uint64_t host_id;
- int error;
+ int rv;
if (!leader_last)
return -EINVAL;
@@ -480,10 +510,10 @@ int delta_lease_release(struct task *task,
leader.timestamp = LEASE_FREE;
leader.checksum = leader_checksum(&leader);
- error = write_sector(disk, host_id - 1, (char *)&leader, sizeof(struct leader_record),
- task, "delta_leader");
- if (error < 0)
- return error;
+ rv = write_sector(disk, host_id - 1, (char *)&leader, sizeof(struct leader_record),
+ task, "delta_leader");
+ if (rv < 0)
+ return rv;
memcpy(leader_ret, &leader, sizeof(struct leader_record));
return SANLK_OK;
diff --git a/src/direct_lib.c b/src/direct_lib.c
index 71b19e9..2cb79c3 100644
--- a/src/direct_lib.c
+++ b/src/direct_lib.c
@@ -22,9 +22,11 @@
#include "task.h"
void log_level(int space_id GNUC_UNUSED, int token_id GNUC_UNUSED,
+ char *name GNUC_UNUSED,
int level GNUC_UNUSED, const char *fmt GNUC_UNUSED, ...);
void log_level(int space_id GNUC_UNUSED, int token_id GNUC_UNUSED,
+ char *name GNUC_UNUSED,
int level GNUC_UNUSED, const char *fmt GNUC_UNUSED, ...)
{
}
diff --git a/src/diskio.c b/src/diskio.c
index bda0e47..c336b3b 100644
--- a/src/diskio.c
+++ b/src/diskio.c
@@ -306,10 +306,11 @@ static struct aicb *find_callback_slot(struct task *task)
struct iocb *ev_iocb = event.obj;
struct aicb *ev_aicb = container_of(ev_iocb, struct aicb, iocb);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld old free",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
ev_aicb->used = 0;
-
- log_error("aio %s clear iocb %p event result %ld %ld",
- task->name, ev_iocb, event.res, event.res2);
+ free(ev_aicb->buf);
+ ev_aicb->buf = NULL;
goto find;
}
return NULL;
@@ -349,7 +350,8 @@ static int do_linux_aio(int fd, uint64_t offset, char *buf, int len,
rv = io_submit(task->aio_ctx, 1, &iocb);
if (rv < 0) {
- log_error("aio %s io_submit error %d", task->name, rv);
+ log_taske(task, "aio submit %p:%p:%p rv %d fd %d cmd %d",
+ aicb, iocb, buf, rv, fd, cmd);
goto out;
}
@@ -368,7 +370,8 @@ static int do_linux_aio(int fd, uint64_t offset, char *buf, int len,
if (rv == -EINTR)
goto retry;
if (rv < 0) {
- log_error("aio %s io_getevents error %d", task->name, rv);
+ log_taske(task, "aio getevent %p:%p:%p rv %d",
+ aicb, iocb, buf, rv);
goto out;
}
if (rv == 1) {
@@ -378,21 +381,21 @@ static int do_linux_aio(int fd, uint64_t offset, char *buf, int len,
ev_aicb->used = 0;
if (ev_iocb != iocb) {
- log_error("aio %s other iocb %p event result %ld %ld",
- task->name, ev_iocb, event.res, event.res2);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld other free",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
free(ev_aicb->buf);
ev_aicb->buf = NULL;
goto retry;
}
if ((int)event.res < 0) {
- log_error("aio %s event result %ld %ld",
- task->name, event.res, event.res2);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld match res",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
rv = event.res;
goto out;
}
if (event.res != len) {
- log_error("aio %s event len %d result %lu %lu",
- task->name, len, event.res, event.res2);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld match len %d",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2, len);
rv = -EMSGSIZE;
goto out;
}
@@ -415,9 +418,8 @@ static int do_linux_aio(int fd, uint64_t offset, char *buf, int len,
task->to_count++;
- log_error("aio %s iocb %p timeout sec %d count %u ios %u",
- task->name, iocb, task->io_timeout_seconds,
- task->to_count, task->io_count);
+ log_taske(task, "aio timeout %p:%p:%p sec %d to_count %d",
+ aicb, iocb, buf, task->io_timeout_seconds, task->to_count);
rv = io_cancel(task->aio_ctx, iocb, &event);
if (!rv) {
@@ -428,7 +430,7 @@ static int do_linux_aio(int fd, uint64_t offset, char *buf, int len,
rv = SANLK_AIO_TIMEOUT;
if (cmd == IO_CMD_PREAD)
- task->read_timeout = aicb;
+ task->read_iobuf_timeout_aicb = aicb;
}
out:
return rv;
@@ -438,6 +440,7 @@ static int do_write_aio_linux(int fd, uint64_t offset, char *buf, int len, struc
{
return do_linux_aio(fd, offset, buf, len, task, IO_CMD_PWRITE);
}
+
static int do_read_aio_linux(int fd, uint64_t offset, char *buf, int len, struct task *task)
{
return do_linux_aio(fd, offset, buf, len, task, IO_CMD_PREAD);
@@ -689,32 +692,26 @@ int read_sectors(const struct sync_disk *disk, uint64_t sector_nr,
return rv;
}
-/* Try to reap the event of a previously timed out read_sectors.
- A task's last timed out read is saved in task->read_timeout. */
+/* Try to reap the event of a previously timed out read_iobuf.
+ The aicb used in a task's last timed out read_iobuf is
+ task->read_iobuf_timeout_aicb . */
-int read_sectors_reap(const struct sync_disk *disk, uint64_t sector_nr,
- uint32_t sector_count, char *data, int data_len,
- struct task *task, const char *blktype GNUC_UNUSED)
+int read_iobuf_reap(int fd, uint64_t offset, char *iobuf, int iobuf_len, struct task *task)
{
struct timespec ts;
struct aicb *aicb;
struct iocb *iocb;
struct io_event event;
- char *iobuf;
- uint64_t offset;
- int iobuf_len;
int rv;
- iobuf_len = sector_count * disk->sector_size;
- offset = disk->offset + (sector_nr * disk->sector_size);
-
- aicb = task->read_timeout;
+ aicb = task->read_iobuf_timeout_aicb;
iocb = &aicb->iocb;
- iobuf = iocb->u.c.buf;
if (!aicb->used)
return -EINVAL;
- if (iocb->aio_fildes != disk->fd)
+ if (iocb->aio_fildes != fd)
+ return -EINVAL;
+ if (iocb->u.c.buf != iobuf)
return -EINVAL;
if (iocb->u.c.nbytes != iobuf_len)
return -EINVAL;
@@ -732,7 +729,8 @@ int read_sectors_reap(const struct sync_disk *disk, uint64_t sector_nr,
if (rv == -EINTR)
goto retry;
if (rv < 0) {
- log_error("reap aio %s io_getevents error %d", task->name, rv);
+ log_taske(task, "aio getevent %p:%p:%p rv %d r",
+ aicb, iocb, iobuf, rv);
goto out;
}
if (rv == 1) {
@@ -742,28 +740,29 @@ int read_sectors_reap(const struct sync_disk *disk, uint64_t sector_nr,
ev_aicb->used = 0;
if (ev_iocb != iocb) {
- log_error("reap aio %s other iocb %p event result %ld %ld",
- task->name, ev_iocb, event.res, event.res2);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld other free r",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
free(ev_aicb->buf);
ev_aicb->buf = NULL;
goto retry;
}
if ((int)event.res < 0) {
- log_error("reap aio %s event result %ld %ld",
- task->name, event.res, event.res2);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld match res r",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
rv = event.res;
goto out;
}
if (event.res != iobuf_len) {
- log_error("reap aio %s event len %d result %lu %lu",
- task->name, iobuf_len, event.res, event.res2);
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld match len %d r",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2, iobuf_len);
rv = -EMSGSIZE;
goto out;
}
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld match reap",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
+
rv = 0;
- memcpy(data, iobuf, data_len);
- free(iobuf);
goto out;
}
diff --git a/src/diskio.h b/src/diskio.h
index a54ffaf..4ce0632 100644
--- a/src/diskio.h
+++ b/src/diskio.h
@@ -14,12 +14,25 @@ int open_disk(struct sync_disk *disks);
int open_disks(struct sync_disk *disks, int num_disks);
int open_disks_fd(struct sync_disk *disks, int num_disks);
+/*
+ * iobuf functions require the caller to allocate iobuf using posix_memalign
+ * and pass it into the function
+ */
+
int write_iobuf(int fd, uint64_t offset, char *iobuf, int iobuf_len,
struct task *task);
int read_iobuf(int fd, uint64_t offset, char *iobuf, int iobuf_len,
struct task *task);
+int read_iobuf_reap(int fd, uint64_t offset, char *iobuf, int iobuf_len,
+ struct task *task);
+
+/*
+ * sector functions allocate an iobuf themselves, copy into it for read, use it
+ * for io, copy out of it for write, and free it
+ */
+
int write_sector(const struct sync_disk *disk, uint64_t sector_nr,
const char *data, int data_len,
struct task *task, const char *blktype);
@@ -31,8 +44,4 @@ int write_sectors(const struct sync_disk *disk, uint64_t sector_nr,
int read_sectors(const struct sync_disk *disk, uint64_t sector_nr,
uint32_t sector_count, char *data, int data_len,
struct task *task, const char *blktype);
-
-int read_sectors_reap(const struct sync_disk *disk, uint64_t sector_nr,
- uint32_t sector_count, char *data, int data_len,
- struct task *task, const char *blktype);
#endif
diff --git a/src/host_id.c b/src/host_id.c
index 7341d65..67e532a 100644
--- a/src/host_id.c
+++ b/src/host_id.c
@@ -21,9 +21,7 @@
#include <syslog.h>
#include <sys/types.h>
#include <sys/time.h>
-#include <sys/utsname.h>
#include <sys/un.h>
-#include <uuid/uuid.h>
#include "sanlock_internal.h"
#include "sanlock_sock.h"
@@ -36,10 +34,6 @@
static unsigned int space_id_counter = 1;
-static struct random_data rand_data;
-static char rand_state[32];
-static pthread_mutex_t rand_mutex = PTHREAD_MUTEX_INITIALIZER;
-
struct list_head spaces;
struct list_head spaces_add;
struct list_head spaces_rem;
@@ -241,6 +235,7 @@ static void *lockspace_thread(void *arg_in)
sp = (struct space *)arg_in;
+ memset(&task, 0, sizeof(struct task));
setup_task_timeouts(&task, main_task.io_timeout_seconds);
setup_task_aio(&task, main_task.use_aio, HOSTID_AIO_CB_SIZE);
memcpy(task.name, sp->space_name, NAME_ID_SIZE);
@@ -293,9 +288,6 @@ static void *lockspace_thread(void *arg_in)
sp->host_generation = leader.owner_generation;
while (1) {
- if (stop)
- break;
-
pthread_mutex_lock(&sp->mutex);
stop = sp->thread_stop;
pthread_mutex_unlock(&sp->mutex);
@@ -348,14 +340,18 @@ static void *lockspace_thread(void *arg_in)
result, delta_length,
(unsigned long long)sp->lease_status.renewal_last_success);
- if (!sp->lease_status.corrupt_result) {
+ if (!sp->lease_status.corrupt_result)
sp->lease_status.corrupt_result = corrupt_result(result);
- log_erros(sp, "renewal error %d is corruption",
- sp->lease_status.corrupt_result);
- }
}
- stop = sp->thread_stop;
pthread_mutex_unlock(&sp->mutex);
+
+ /* TODO: pass off all the delta leases we read (in task->iobuf)
+ for analysis by another thread */
+
+ /*
+ if (result == SANLK_OK)
+ queue_delta_lease_analysis(sp, task->iobuf);
+ */
}
/* unlink called below to get it done ASAP */
@@ -454,6 +450,14 @@ int add_lockspace(struct sanlk_lockspace *ls)
list_add(&sp->list, &spaces_add);
pthread_mutex_unlock(&spaces_mutex);
+ /* save a record of what this space_id is for later debugging */
+ log_level(sp->space_id, 0, NULL, LOG_WARNING,
+ "lockspace %.48s:%llu:%.256s:%llu",
+ sp->space_name,
+ (unsigned long long)sp->host_id,
+ sp->host_id_disk.path,
+ (unsigned long long)sp->host_id_disk.offset);
+
rv = pthread_create(&sp->thread, NULL, lockspace_thread, sp);
if (rv < 0) {
log_erros(sp, "add_lockspace create thread failed");
@@ -603,55 +607,10 @@ void free_lockspaces(int wait)
pthread_mutex_unlock(&spaces_mutex);
}
-/* return a random int between a and b inclusive */
-
-int get_rand(int a, int b)
-{
- int32_t val;
- int rv;
-
- pthread_mutex_lock(&rand_mutex);
- rv = random_r(&rand_data, &val);
- pthread_mutex_unlock(&rand_mutex);
- if (rv < 0)
- return rv;
-
- return a + (int) (((float)(b - a + 1)) * val / (RAND_MAX+1.0));
-}
-
void setup_spaces(void)
{
- struct utsname name;
- char uuid[37];
- uuid_t uu;
-
INIT_LIST_HEAD(&spaces);
INIT_LIST_HEAD(&spaces_add);
INIT_LIST_HEAD(&spaces_rem);
-
- memset(rand_state, 0, sizeof(rand_state));
- memset(&rand_data, 0, sizeof(rand_data));
-
- initstate_r(time(NULL), rand_state, sizeof(rand_state), &rand_data);
-
- /* use host name from command line */
-
- if (com.our_host_name[0]) {
- memcpy(our_host_name_global, com.our_host_name, SANLK_NAME_LEN);
- return;
- }
-
- /* make up something that's likely to be different among hosts */
-
- memset(&our_host_name_global, 0, sizeof(our_host_name_global));
- memset(&name, 0, sizeof(name));
- memset(&uuid, 0, sizeof(uuid));
-
- uname(&name);
- uuid_generate(uu);
- uuid_unparse_lower(uu, uuid);
-
- snprintf(our_host_name_global, NAME_ID_SIZE, "%s.%s",
- uuid, name.nodename);
}
diff --git a/src/host_id.h b/src/host_id.h
index 703cb3f..5cc8b50 100644
--- a/src/host_id.h
+++ b/src/host_id.h
@@ -18,6 +18,5 @@ int add_lockspace(struct sanlk_lockspace *ls);
int rem_lockspace(struct sanlk_lockspace *ls);
void free_lockspaces(int wait);
void setup_spaces(void);
-int get_rand(int a, int b);
#endif
diff --git a/src/leader.h b/src/leader.h
index e100bb6..dca9b62 100644
--- a/src/leader.h
+++ b/src/leader.h
@@ -21,7 +21,7 @@
#define DELTA_DISK_MAGIC 0x12212010
#define DELTA_DISK_VERSION_MAJOR 0x00030000
-#define DELTA_DISK_VERSION_MINOR 0x00000001
+#define DELTA_DISK_VERSION_MINOR 0x00000002
/* for all disk structures:
uint64 aligned on 8 byte boundaries,
diff --git a/src/log.c b/src/log.c
index e753ee6..3cb4064 100644
--- a/src/log.c
+++ b/src/log.c
@@ -100,7 +100,7 @@ static void _log_save_ent(int level, int len)
* logfile and/or syslog (so callers don't block writing messages to files)
*/
-void log_level(int space_id, int token_id, int level, const char *fmt, ...)
+void log_level(int space_id, int token_id, char *name_in, int level, const char *fmt, ...)
{
va_list ap;
char name[NAME_ID_SIZE + 1];
@@ -109,19 +109,19 @@ void log_level(int space_id, int token_id, int level, const char *fmt, ...)
memset(name, 0, sizeof(name));
- if (!space_id && !token_id)
- snprintf(name, NAME_ID_SIZE, "-");
- else if (space_id && !token_id)
- snprintf(name, NAME_ID_SIZE, "s%u", space_id);
+ if (space_id && !token_id)
+ snprintf(name, NAME_ID_SIZE, "s%u ", space_id);
else if (!space_id && token_id)
- snprintf(name, NAME_ID_SIZE, "t%u", token_id);
+ snprintf(name, NAME_ID_SIZE, "r%u ", token_id);
else if (space_id && token_id)
- snprintf(name, NAME_ID_SIZE, "s%u:t%u", space_id, token_id);
+ snprintf(name, NAME_ID_SIZE, "s%u:r%u ", space_id, token_id);
+ else if (name_in)
+ snprintf(name, NAME_ID_SIZE, "%.8s ", name_in);
pthread_mutex_lock(&log_mutex);
- ret = snprintf(log_str + pos, len - pos, "%ld %s ",
- time(NULL), name);
+ ret = snprintf(log_str + pos, len - pos, "%llu %s",
+ (unsigned long long)monotime(), name);
pos += ret;
va_start(ap, fmt);
diff --git a/src/log.h b/src/log.h
index 1aef3b2..eacb91b 100644
--- a/src/log.h
+++ b/src/log.h
@@ -9,22 +9,25 @@
#ifndef __LOG_H__
#define __LOG_H__
-void log_level(int space_id, int token_id, int level, const char *fmt, ...)
- __attribute__((format(printf, 4, 5)));
+void log_level(int space_id, int token_id, char *name_in, int level, const char *fmt, ...)
+ __attribute__((format(printf, 5, 6)));
int setup_logging(void);
void close_logging(void);
void write_log_dump(int fd);
-#define log_debug(fmt, args...) log_level(0, 0, LOG_DEBUG, fmt, ##args)
-#define log_space(space, fmt, args...) log_level(space->space_id, 0, LOG_DEBUG, fmt, ##args)
-#define log_token(token, fmt, args...) log_level(0, token->token_id, LOG_DEBUG, fmt, ##args)
-#define log_spoke(space, token, fmt, args...) log_level(space->space_id, token->token_id, LOG_DEBUG, fmt, ##args)
+#define log_debug(fmt, args...) log_level(0, 0, NULL, LOG_DEBUG, fmt, ##args)
+#define log_space(space, fmt, args...) log_level(space->space_id, 0, NULL, LOG_DEBUG, fmt, ##args)
+#define log_token(token, fmt, args...) log_level(0, token->token_id, NULL, LOG_DEBUG, fmt, ##args)
+#define log_spoke(space, token, fmt, args...) log_level(space->space_id, token->token_id, NULL, LOG_DEBUG, fmt, ##args)
-#define log_error(fmt, args...) log_level(0, 0, LOG_ERR, fmt, ##args)
-#define log_erros(space, fmt, args...) log_level(space->space_id, 0, LOG_ERR, fmt, ##args)
-#define log_errot(token, fmt, args...) log_level(0, token->token_id, LOG_ERR, fmt, ##args)
-#define log_errst(space, token, fmt, args...) log_level(space->space_id, token->token_id, LOG_ERR, fmt, ##args)
+#define log_error(fmt, args...) log_level(0, 0, NULL, LOG_ERR, fmt, ##args)
+#define log_erros(space, fmt, args...) log_level(space->space_id, 0, NULL, LOG_ERR, fmt, ##args)
+#define log_errot(token, fmt, args...) log_level(0, token->token_id, NULL, LOG_ERR, fmt, ##args)
+#define log_errst(space, token, fmt, args...) log_level(space->space_id, token->token_id, NULL, LOG_ERR, fmt, ##args)
+
+#define log_taske(task, fmt, args...) log_level(0, 0, task->name, LOG_ERR, fmt, ##args)
+#define log_taskd(task, fmt, args...) log_level(0, 0, task->name, LOG_DEBUG, fmt, ##args)
/* use log_tool for tool actions (non-daemon), and for daemon until
logging is set up */
diff --git a/src/main.c b/src/main.c
index 1ea5a36..c17b18f 100644
--- a/src/main.c
+++ b/src/main.c
@@ -30,6 +30,9 @@
#include <sys/time.h>
#include <sys/un.h>
#include <sys/mman.h>
+#include <sys/mman.h>
+#include <sys/utsname.h>
+#include <uuid/uuid.h>
#define EXTERN
#include "sanlock_internal.h"
@@ -48,9 +51,9 @@
#include "sanlock_admin.h"
/* priorities are LOG_* from syslog.h */
-int log_logfile_priority = LOG_ERR;
+int log_logfile_priority = LOG_WARNING;
int log_syslog_priority = LOG_ERR;
-int log_stderr_priority = LOG_ERR;
+int log_stderr_priority = -1; /* -D sets this to LOG_DEBUG */
struct client {
int used;
@@ -107,6 +110,10 @@ extern struct list_head spaces;
extern struct list_head spaces_rem;
extern pthread_mutex_t spaces_mutex;
+static struct random_data rand_data;
+static char rand_state[32];
+static pthread_mutex_t rand_mutex = PTHREAD_MUTEX_INITIALIZER;
+
/* FIXME: add a mutex for client array so we don't try to expand it
while a cmd thread is using it. Or, with a thread pool we know
when cmd threads are running and can expand when none are. */
@@ -853,16 +860,6 @@ static void cmd_acquire(struct task *task, struct cmd_args *ca)
token->token_id = token_id_counter++;
new_tokens[i] = token;
alloc_count++;
-
- /* We use the token_id in log messages because the combination
- * of full length space_name+resource_name in each log message
- * would make excessively long lines. */
-
- log_token(token, "cmd_acquire %d,%d,%d %.48s:%.48s:%s:%llu",
- cl_ci, cl_fd, cl_pid,
- token->r.lockspace_name, token->r.name,
- token->disks[0].path,
- (unsigned long long)token->disks[0].offset);
}
rv = recv(fd, &opt, sizeof(struct sanlk_options), MSG_WAITALL);
@@ -916,6 +913,15 @@ static void cmd_acquire(struct task *task, struct cmd_args *ca)
}
token->host_id = space.host_id;
token->host_generation = space.host_generation;
+
+ /* save a record of what this token_id is for later debugging */
+ log_level(space.space_id, token->token_id, NULL, LOG_WARNING,
+ "resource %.48s:%.48s:%.256s:%llu for %d,%d,%d",
+ token->r.lockspace_name,
+ token->r.name,
+ token->r.disks[0].path,
+ (unsigned long long)token->r.disks[0].offset,
+ cl_ci, cl_fd, cl_pid);
}
for (i = 0; i < new_tokens_count; i++) {
@@ -1416,6 +1422,7 @@ static void *thread_pool_worker(void *data)
struct task task;
struct cmd_args *ca;
+ memset(&task, 0, sizeof(struct task));
setup_task_timeouts(&task, main_task.io_timeout_seconds);
setup_task_aio(&task, main_task.use_aio, WORKER_AIO_CB_SIZE);
snprintf(task.name, NAME_ID_SIZE, "worker%ld", (long)data);
@@ -2118,6 +2125,56 @@ static void setup_priority(void)
}
}
+/* return a random int between a and b inclusive */
+
+int get_rand(int a, int b);
+
+int get_rand(int a, int b)
+{
+ int32_t val;
+ int rv;
+
+ pthread_mutex_lock(&rand_mutex);
+ rv = random_r(&rand_data, &val);
+ pthread_mutex_unlock(&rand_mutex);
+ if (rv < 0)
+ return rv;
+
+ return a + (int) (((float)(b - a + 1)) * val / (RAND_MAX+1.0));
+}
+
+static void setup_host_name(void)
+{
+ struct utsname name;
+ char uuid[37];
+ uuid_t uu;
+
+ memset(rand_state, 0, sizeof(rand_state));
+ memset(&rand_data, 0, sizeof(rand_data));
+
+ initstate_r(time(NULL), rand_state, sizeof(rand_state), &rand_data);
+
+ /* use host name from command line */
+
+ if (com.our_host_name[0]) {
+ memcpy(our_host_name_global, com.our_host_name, SANLK_NAME_LEN);
+ return;
+ }
+
+ /* make up something that's likely to be different among hosts */
+
+ memset(&our_host_name_global, 0, sizeof(our_host_name_global));
+ memset(&name, 0, sizeof(name));
+ memset(&uuid, 0, sizeof(uuid));
+
+ uname(&name);
+ uuid_generate(uu);
+ uuid_unparse_lower(uu, uuid);
+
+ snprintf(our_host_name_global, NAME_ID_SIZE, "%s.%s",
+ uuid, name.nodename);
+}
+
static int do_daemon(void)
{
struct sigaction act;
@@ -2157,9 +2214,13 @@ static int do_daemon(void)
setup_logging();
- log_error("sanlock daemon started aio %d %d renew %d %d",
+ setup_host_name();
+
+ log_error("sanlock daemon started aio %d %d renew %d %d host %s time %llu",
main_task.use_aio, main_task.io_timeout_seconds,
- main_task.id_renewal_seconds, main_task.id_renewal_fail_seconds);
+ main_task.id_renewal_seconds, main_task.id_renewal_fail_seconds,
+ our_host_name_global,
+ (unsigned long long)time(NULL));
setup_priority();
@@ -2601,6 +2662,8 @@ static int read_command_line(int argc, char *argv[])
break;
case 'a':
com.aio_arg = atoi(optionarg);
+ if (com.aio_arg && com.aio_arg != 1)
+ com.aio_arg = 1;
break;
case 't':
com.max_worker_threads = atoi(optionarg);
@@ -2895,8 +2958,10 @@ static int do_direct(void)
break;
case ACT_ACQUIRE_ID:
+ setup_host_name();
+
rv = direct_acquire_id(&main_task, &com.lockspace,
- com.our_host_name);
+ our_host_name_global);
log_tool("acquire_id done %d", rv);
break;
diff --git a/src/paxos_lease.c b/src/paxos_lease.c
index 25165f1..0816a64 100644
--- a/src/paxos_lease.c
+++ b/src/paxos_lease.c
@@ -30,6 +30,8 @@
#include "delta_lease.h"
#include "paxos_lease.h"
+int get_rand(int a, int b);
+
struct request_record {
uint64_t lver;
uint8_t force_mode;
diff --git a/src/sanlock_internal.h b/src/sanlock_internal.h
index 833822e..7f6f740 100644
--- a/src/sanlock_internal.h
+++ b/src/sanlock_internal.h
@@ -445,8 +445,9 @@ struct task {
int use_aio;
int cb_size;
+ char *iobuf;
io_context_t aio_ctx;
- struct aicb *read_timeout;
+ struct aicb *read_iobuf_timeout_aicb;
struct aicb *callbacks;
};
diff --git a/src/task.c b/src/task.c
index 5a04248..c0ab44b 100644
--- a/src/task.c
+++ b/src/task.c
@@ -114,14 +114,76 @@ void setup_task_aio(struct task *task, int use_aio, int cb_size)
task->use_aio = 0;
}
-/* TODO: do we need/want to go through all task->callbacks that are still used
- and wait to reap events for them before doing io_destroy? */
-
void close_task_aio(struct task *task)
{
- if (task->use_aio)
- io_destroy(task->aio_ctx);
+ struct timespec ts;
+ struct io_event event;
+ uint64_t last_warn;
+ int rv, i, used, warn;
+
+ if (!task->use_aio)
+ goto skip_aio;
+
+ memset(&ts, 0, sizeof(struct timespec));
+ ts.tv_sec = task->io_timeout_seconds;
+
+ last_warn = time(NULL);
+
+ /* wait for all outstanding aio to complete before
+ destroying aio context, freeing iocb and buffers */
+
+ while (1) {
+ warn = 0;
+
+ if (time(NULL) - last_warn >= task->io_timeout_seconds) {
+ last_warn = time(NULL);
+ warn = 1;
+ }
+
+ used = 0;
+
+ for (i = 0; i < task->cb_size; i++) {
+ if (!task->callbacks[i].used)
+ continue;
+ used++;
+
+ if (!warn)
+ continue;
+ log_taske(task, "close_task_aio %d %p busy",
+ i, &task->callbacks[i]);
+ }
+
+ if (!used)
+ break;
+
+ memset(&event, 0, sizeof(event));
+
+ rv = io_getevents(task->aio_ctx, 1, 1, &event, &ts);
+ if (rv == -EINTR)
+ continue;
+ if (rv < 0)
+ break;
+ if (rv == 1) {
+ struct iocb *ev_iocb = event.obj;
+ struct aicb *ev_aicb = container_of(ev_iocb, struct aicb, iocb);
+
+ if (ev_aicb->buf == task->iobuf)
+ task->iobuf = NULL;
+
+ log_taske(task, "aio collect %p:%p:%p result %ld:%ld close free",
+ ev_aicb, ev_iocb, ev_aicb->buf, event.res, event.res2);
+
+ ev_aicb->used = 0;
+ free(ev_aicb->buf);
+ ev_aicb->buf = NULL;
+ }
+ }
+ io_destroy(task->aio_ctx);
+
+ if (task->iobuf)
+ free(task->iobuf);
+ skip_aio:
if (task->callbacks)
free(task->callbacks);
task->callbacks = NULL;
diff --git a/src/token_manager.c b/src/token_manager.c
index 0c85197..df5d632 100644
--- a/src/token_manager.c
+++ b/src/token_manager.c
@@ -193,6 +193,7 @@ static void *async_release_thread(void *arg GNUC_UNUSED)
struct resource *r;
struct token *token;
+ memset(&task, 0, sizeof(struct task));
setup_task_timeouts(&task, main_task.io_timeout_seconds);
setup_task_aio(&task, main_task.use_aio, RELEASE_AIO_CB_SIZE);
sprintf(task.name, "%s", "release");