diff --git a/src/paxos_lease.c b/src/paxos_lease.c index 872a82d..b120330 100644 --- a/src/paxos_lease.c +++ b/src/paxos_lease.c @@ -1,922 +1,928 @@ /* * Copyright (C) 2011 Jiaju Zhang * * 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 the Free Software Foundation; either * version 2.1 of the License, or (at your option) any later version. * * This software is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * General Public License for more details. * * You should have received a copy of the GNU General Public * License along with this library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include "paxos.h" #include "paxos_lease.h" #include "transport.h" #include "config.h" #include "timer.h" #include "list.h" #include "log.h" #define PAXOS_LEASE_SPACE "paxoslease" #define PLEASE_VALUE_LEN 1024 #define OP_START_LEASE 0 #define OP_STOP_LEASE 1 #define LEASE_STARTED 0 #define LEASE_STOPPED 1 struct paxos_lease_msghdr { int op; int clear; int leased; }; struct paxos_lease_value { char name[PAXOS_NAME_LEN+1]; int owner; int expiry; }; struct lease_action { int op; int clear; }; struct lease_state { int round; struct paxos_lease_value *plv; unsigned long long expires; struct timerlist *timer1; struct timerlist *timer2; }; struct paxos_lease { char name[PAXOS_NAME_LEN+1]; pi_handle_t pih; struct lease_action action; struct lease_state proposer; struct lease_state acceptor; int owner; int expiry; int renew; int failover; int release; unsigned long long expires; void (*end_lease) (pi_handle_t, int); struct timerlist *timer; struct list_head list; }; static LIST_HEAD(lease_head); static int myid = -1; static struct paxos_operations *px_op = NULL; const struct paxos_lease_operations *p_l_op; ps_handle_t ps_handle = 0; static int find_paxos_lease(pi_handle_t handle, struct paxos_lease **pl) { struct paxos_lease *lpl; int found = 0; list_for_each_entry(lpl, &lease_head, list) { if (lpl->pih == handle) { found = 1; break; } } if (!found) log_error("cound not found the handle for paxos lease: %ld", handle); else *pl = lpl; return found; } static void end_paxos_request(pi_handle_t handle, int round, int result) { struct paxos_lease *pl; if (!find_paxos_lease(handle, &pl)) return; if (round != pl->proposer.round) { log_error("current paxos round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return; } if (pl->end_lease) pl->end_lease((pl_handle_t)pl, result); return; } static void renew_expires(unsigned long data) { struct paxos_lease *pl = (struct paxos_lease *)data; struct paxos_lease_value value; log_debug("renew expires ..."); memset(&value, 0, sizeof(struct paxos_lease_value)); strncpy(value.name, pl->name, PAXOS_NAME_LEN + 1); value.owner = myid; value.expiry = pl->expiry; paxos_propose(pl->pih, &value, pl->proposer.round); } static void lease_expires(unsigned long data) { struct paxos_lease *pl = (struct paxos_lease *)data; pl_handle_t plh = (pl_handle_t)pl; struct paxos_lease_result plr; log_debug("lease expires ..."); pl->owner = -1; strcpy(plr.name, pl->name); plr.owner = -1; plr.expires = 0; plr.ballot = pl->acceptor.round; p_l_op->notify(plh, &plr); if (pl->proposer.timer1) del_timer(&pl->proposer.timer1); if (pl->proposer.timer2) del_timer(&pl->proposer.timer2); if (pl->acceptor.timer1) del_timer(&pl->acceptor.timer1); if (pl->acceptor.timer2) del_timer(&pl->acceptor.timer2); if (pl->failover) paxos_lease_acquire(plh, NOT_CLEAR_RELEASE, 1, NULL); } static void lease_retry(unsigned long data) { struct paxos_lease *pl = (struct paxos_lease *)data; struct paxos_lease_value value; int round; log_debug("lease_retry ..."); if (pl->proposer.timer2) del_timer(&pl->proposer.timer2); if (pl->owner == myid) { log_debug("already got the lease, no need to retry"); return; } memset(&value, 0, sizeof(struct paxos_lease_value)); strncpy(value.name, pl->name, PAXOS_NAME_LEN + 1); value.owner = myid; value.expiry = pl->expiry; pl->action.op = OP_START_LEASE; /** * We don't know whether the lease_retry after ticket grant * is manual or not, so set clear as NOT_CLEAR_RELEASE is * the only safe choice. **/ pl->action.clear = NOT_CLEAR_RELEASE; round = paxos_round_request(pl->pih, &value, &pl->acceptor.round, end_paxos_request); if (round > 0) pl->proposer.round = round; } int paxos_lease_acquire(pl_handle_t handle, int clear, int renew, void (*end_acquire) (pl_handle_t handle, int result)) { struct paxos_lease *pl = (struct paxos_lease *)handle; struct paxos_lease_value value; int round; memset(&value, 0, sizeof(struct paxos_lease_value)); strncpy(value.name, pl->name, PAXOS_NAME_LEN + 1); value.owner = myid; value.expiry = pl->expiry; pl->renew = renew; pl->end_lease = end_acquire; pl->action.op = OP_START_LEASE; pl->action.clear = clear; round = paxos_round_request(pl->pih, &value, &pl->acceptor.round, end_paxos_request); pl->proposer.timer2 = add_timer(1 * pl->expiry / 10, (unsigned long)pl, lease_retry); if (round <= 0) return -1; else { pl->proposer.round = round; return 0; } } int paxos_lease_release(pl_handle_t handle, void (*end_release) (pl_handle_t handle, int result)) { struct paxos_lease *pl = (struct paxos_lease *)handle; struct paxos_lease_value value; int round; log_debug("enter paxos_lease_release"); + if (pl->owner != myid) { + log_error("can not release the lease " + "because I'm not the lease owner"); + return -1; + } + memset(&value, 0, sizeof(struct paxos_lease_value)); strncpy(value.name, pl->name, PAXOS_NAME_LEN + 1); pl->end_lease = end_release; pl->action.op = OP_STOP_LEASE; round = paxos_round_request(pl->pih, &value, &pl->acceptor.round, end_paxos_request); if (round > 0) pl->proposer.round = round; log_debug("exit paxos_lease_release"); return (round > 0)? 0: -1; } static int lease_catchup(pi_handle_t handle) { struct paxos_lease *pl; struct paxos_lease_result plr; if (!find_paxos_lease(handle, &pl)) return -1; p_l_op->catchup(pl->name, &pl->owner, &pl->proposer.round, &pl->expires); log_debug("catchup result: name: %s, owner: %d, ballot: %d, expires: %llu", (char *)pl->name, pl->owner, pl->proposer.round, pl->expires); /** * 1. If no site hold the ticket, the relet will be set LEASE_STOPPED. * Grant commond will set the relet to LEASE_STARTED first, so we don't * need worry about it. * 2. If someone hold the ticket, the relet will be set LEASE_STARTED. * Because we must make sure that the site can renew, and relet also * must be set to LEASE_STARTED. **/ if (-1 == pl->owner) { pl->release = LEASE_STOPPED; return 0; } else pl->release = LEASE_STARTED; if (current_time() > pl->expires) { plr.owner = pl->owner = -1; plr.expires = pl->expires = 0; strcpy(plr.name, pl->name); p_l_op->notify((pl_handle_t)pl, &plr); return 0; } if (pl->owner == myid) { pl->acceptor.timer1 = add_timer(pl->expires - current_time(), (unsigned long)pl, lease_expires); if (current_time() < pl->expires - 1 * pl->expiry / 5) pl->proposer.timer1 = add_timer(pl->expires - 1 * pl->expiry / 5 - current_time(), (unsigned long)pl, renew_expires); } else pl->acceptor.timer1 = add_timer(pl->expires - current_time(), (unsigned long)pl, lease_expires); plr.owner = pl->owner; plr.expires = pl->expires; plr.ballot = pl->proposer.round; strcpy(plr.name, pl->name); p_l_op->notify((pl_handle_t)pl, &plr); return 0; } static int lease_prepare(pi_handle_t handle, void *header) { struct paxos_lease_msghdr *msghdr = header; struct paxos_lease *pl; log_debug("enter lease_prepare"); if (!find_paxos_lease(handle, &pl)) return -1; msghdr->op = htonl(pl->action.op); msghdr->clear = htonl(pl->action.clear); /** * Action of paxos_lease is only used to pass args, * so clear it now **/ memset(&pl->action, 0, sizeof(struct lease_action)); log_debug("exit lease_prepare"); return 0; } static inline int start_lease_is_prepared(pi_handle_t handle __attribute__((unused)), void *header) { struct paxos_lease_msghdr *hdr = header; log_debug("enter start_lease_is_prepared"); if (hdr->leased) { log_debug("already leased"); return 0; } else { log_debug("not leased"); return 1; } } static inline int stop_lease_is_prepared(pi_handle_t handle __attribute__((unused)), void *header __attribute__((unused))) { log_debug("enter stop_lease_is_prepared"); return 1; } static int lease_is_prepared(pi_handle_t handle, void *header) { struct paxos_lease_msghdr *hdr = header; int ret = 0; int op = ntohl(hdr->op); log_debug("enter lease_is_prepared"); assert(OP_START_LEASE == op || OP_STOP_LEASE == op); switch (op) { case OP_START_LEASE: ret = start_lease_is_prepared(handle, header); break; case OP_STOP_LEASE: ret = stop_lease_is_prepared(handle, header); break; } log_debug("exit lease_is_prepared"); return ret; } static int start_lease_promise(pi_handle_t handle, void *header) { struct paxos_lease_msghdr *hdr = header; struct paxos_lease *pl; int clear = ntohl(hdr->clear); log_debug("enter start_lease_promise"); if (!find_paxos_lease(handle, &pl)) return -1; if (NOT_CLEAR_RELEASE == clear && LEASE_STOPPED == pl->release) { log_debug("could not be leased"); hdr->leased = 1; } else if (-1 == pl->owner) { log_debug("has not been leased"); hdr->leased = 0; } else { log_debug("has been leased"); hdr->leased = 1; } log_debug("exit start_lease_promise"); return 0; } static int stop_lease_promise(pi_handle_t handle, void *header __attribute__((unused))) { struct paxos_lease *pl; log_debug("enter stop_lease_promise"); if (!find_paxos_lease(handle, &pl)) return -1; log_debug("exit stop_lease_promise"); return 0; } static int lease_promise(pi_handle_t handle, void *header) { struct paxos_lease_msghdr *hdr = header; int ret = 0; int op = ntohl(hdr->op); log_debug("enter lease_promise"); assert(OP_START_LEASE == op || OP_STOP_LEASE == op); switch (op) { case OP_START_LEASE: ret = start_lease_promise(handle, header); break; case OP_STOP_LEASE: ret = stop_lease_promise(handle, header); break; } log_debug("exit lease_promise"); return ret; } static int start_lease_propose(pi_handle_t handle, void *extra, int round, void *value) { struct paxos_lease *pl; log_debug("enter start_lease_propose"); if (!find_paxos_lease(handle, &pl)) return -1; if (round != pl->proposer.round) { log_error("current round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return -1; } if (!pl->proposer.plv) { pl->proposer.plv = malloc(sizeof(struct paxos_lease_value)); if (!pl->proposer.plv) { log_error("could not alloc mem for propsoer plv"); return -ENOMEM; } } memcpy(pl->proposer.plv, value, sizeof(struct paxos_lease_value)); if (pl->proposer.timer1) del_timer(&pl->proposer.timer1); if (pl->renew) { pl->proposer.timer1 = add_timer(4 * pl->expiry / 5, (unsigned long)pl, renew_expires); pl->proposer.expires = current_time() + 4 * pl->expiry / 5; } else { pl->proposer.timer1 = add_timer(pl->expiry, (unsigned long)pl, lease_expires); pl->proposer.expires = current_time() + pl->expiry; } log_debug("exit start_lease_propose"); return 0; } static int stop_lease_propose(pi_handle_t handle, void *extra __attribute__((unused)), int round, void *value) { struct paxos_lease *pl; log_debug("enter stop_lease_propose"); if (!find_paxos_lease(handle, &pl)) return -1; if (round != pl->proposer.round) { log_error("current round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return -1; } if (!pl->proposer.plv) { pl->proposer.plv = malloc(sizeof(struct paxos_lease_value)); if (!pl->proposer.plv) { log_error("could not alloc mem for propsoer plv"); return -ENOMEM; } } memcpy(pl->proposer.plv, value, sizeof(struct paxos_lease_value)); log_debug("exit stop_lease_propose"); return 0; } static int lease_propose(pi_handle_t handle, void *extra, int round, void *value) { struct paxos_lease_msghdr *hdr = extra; int ret = 0; int op = ntohl(hdr->op); log_debug("enter lease_propose"); assert(OP_START_LEASE == op || OP_STOP_LEASE == op); switch (op) { case OP_START_LEASE: ret = start_lease_propose(handle, extra, round, value); break; case OP_STOP_LEASE: ret = stop_lease_propose(handle, extra, round, value); break; } log_debug("exit lease_propose"); return ret; } static int start_lease_accepted(pi_handle_t handle, void *extra, int round, void *value) { struct paxos_lease_msghdr *hdr = extra; struct paxos_lease *pl; log_debug("enter start_lease_accepted"); if (!find_paxos_lease(handle, &pl)) return -1; pl->acceptor.round = round; if (NOT_CLEAR_RELEASE == hdr->clear && LEASE_STOPPED == pl->release) { log_debug("could not be leased"); return -1; } if (!pl->acceptor.plv) { pl->acceptor.plv = malloc(sizeof(struct paxos_lease_value)); if (!pl->acceptor.plv) { log_error("could not alloc mem for acceptor plv"); return -ENOMEM; } } memcpy(pl->acceptor.plv, value, sizeof(struct paxos_lease_value)); if (pl->acceptor.timer1 && pl->acceptor.timer2 != pl->acceptor.timer1) del_timer(&pl->acceptor.timer1); pl->acceptor.timer1 = add_timer(pl->expiry, (unsigned long)pl, lease_expires); pl->acceptor.expires = current_time() + pl->expiry; log_debug("exit start_lease_accepted"); return 0; } static int stop_lease_accepted(pi_handle_t handle, void *extra __attribute__((unused)), int round, void *value) { struct paxos_lease *pl; log_debug("enter stop_lease_accepted"); if (!find_paxos_lease(handle, &pl)) return -1; pl->acceptor.round = round; if (!pl->acceptor.plv) { pl->acceptor.plv = malloc(sizeof(struct paxos_lease_value)); if (!pl->acceptor.plv) { log_error("could not alloc mem for acceptor plv"); return -ENOMEM; } } memcpy(pl->acceptor.plv, value, sizeof(struct paxos_lease_value)); log_debug("exit stop_lease_accepted"); return 0; } static int lease_accepted(pi_handle_t handle, void *extra, int round, void *value) { struct paxos_lease_msghdr *hdr = extra; int ret = 0; int op = ntohl(hdr->op); log_debug("enter lease_accepted"); assert(OP_START_LEASE == op || OP_STOP_LEASE == op); switch (op) { case OP_START_LEASE: ret = start_lease_accepted(handle, extra, round, value); break; case OP_STOP_LEASE: ret = stop_lease_accepted(handle, extra, round, value); break; } log_debug("exit lease_accepted"); return ret; } static int start_lease_commit(pi_handle_t handle, void *extra, int round) { struct paxos_lease *pl; struct paxos_lease_result plr; log_debug("enter start_lease_commit"); if (!find_paxos_lease(handle, &pl)) return -1; if (round != pl->proposer.round) { log_error("current round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return -1; } pl->release = LEASE_STARTED; pl->owner = pl->proposer.plv->owner; pl->expiry = pl->proposer.plv->expiry; if (pl->acceptor.timer2 != pl->acceptor.timer1) { if (pl->acceptor.timer2) del_timer(&pl->acceptor.timer2); pl->acceptor.timer2 = pl->acceptor.timer1; } strcpy(plr.name, pl->proposer.plv->name); plr.owner = pl->proposer.plv->owner; plr.expires = current_time() + pl->proposer.plv->expiry; plr.ballot = round; p_l_op->notify((pl_handle_t)pl, &plr); log_debug("exit start_lease_commit"); return 0; } static int stop_lease_commit(pi_handle_t handle, void *extra __attribute__((unused)), int round) { struct paxos_lease *pl; struct paxos_lease_result plr; log_debug("enter stop_lease_commit"); if (!find_paxos_lease(handle, &pl)) return -1; if (round != pl->proposer.round) { log_error("current round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return -1; } if (pl->acceptor.timer2) del_timer(&pl->acceptor.timer2); if (pl->acceptor.timer1) del_timer(&pl->acceptor.timer1); if (pl->proposer.timer2) del_timer(&pl->proposer.timer2); if (pl->proposer.timer1) del_timer(&pl->proposer.timer1); pl->release = LEASE_STOPPED; strcpy(plr.name, pl->proposer.plv->name); plr.owner = pl->owner = -1; plr.ballot = round; plr.expires = 0; p_l_op->notify((pl_handle_t)pl, &plr); log_debug("exit stop_lease_commit"); return 0; } static int lease_commit(pi_handle_t handle, void *extra, int round) { struct paxos_lease_msghdr *hdr = extra; int ret = 0; int op = ntohl(hdr->op); log_debug("enter lease_commit"); assert(OP_START_LEASE == op || OP_STOP_LEASE == op); switch (op) { case OP_START_LEASE: ret = start_lease_commit(handle, extra, round); break; case OP_STOP_LEASE: ret = stop_lease_commit(handle, extra, round); break; } log_debug("exit lease_commit"); return ret; } static int start_lease_learned(pi_handle_t handle, void *extra, int round) { struct paxos_lease *pl; struct paxos_lease_result plr; log_debug("enter start_lease_learned"); if (!find_paxos_lease(handle, &pl)) return -1; if (round != pl->acceptor.round) { log_error("current round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return -1; } if (!pl->acceptor.plv) return -1; pl->release = LEASE_STARTED; pl->owner = pl->acceptor.plv->owner; pl->expiry = pl->acceptor.plv->expiry; if (pl->acceptor.timer2 != pl->acceptor.timer1) { if (pl->acceptor.timer2) del_timer(&pl->acceptor.timer2); pl->acceptor.timer2 = pl->acceptor.timer1; } strcpy(plr.name, pl->acceptor.plv->name); plr.owner = pl->acceptor.plv->owner; plr.expires = current_time() + pl->acceptor.plv->expiry; plr.ballot = round; p_l_op->notify((pl_handle_t)pl, &plr); log_debug("exit start_lease_learned"); return 0; } static int stop_lease_learned(pi_handle_t handle, void *extra __attribute__((unused)), int round) { struct paxos_lease *pl; struct paxos_lease_result plr; log_debug("enter stop_lease_learned"); if (!find_paxos_lease(handle, &pl)) return -1; if (round != pl->acceptor.round) { log_error("current round is not the proposer round, " "current round: %d, proposer round: %d", round, pl->proposer.round); return -1; } if (!pl->acceptor.plv) return -1; if (pl->acceptor.timer2) del_timer(&pl->acceptor.timer2); if (pl->acceptor.timer1) del_timer(&pl->acceptor.timer1); pl->release = LEASE_STOPPED; strcpy(plr.name, pl->acceptor.plv->name); plr.owner = pl->owner = -1; plr.ballot = round; plr.expires = 0; p_l_op->notify((pl_handle_t)pl, &plr); log_debug("exit stop_lease_learned"); return 0; } static int lease_learned(pi_handle_t handle, void *extra, int round) { struct paxos_lease_msghdr *hdr = extra; int ret = 0; int op = ntohl(hdr->op); log_debug("enter lease_learned"); assert(OP_START_LEASE == op || OP_STOP_LEASE == op); switch (op) { case OP_START_LEASE: ret = start_lease_learned(handle, extra, round); break; case OP_STOP_LEASE: ret = stop_lease_learned(handle, extra, round); break; } log_debug("exit lease_learned"); return ret; } pl_handle_t paxos_lease_init(const void *name, unsigned int namelen, int expiry, int number, int failover, unsigned char *role, int *prio, const struct paxos_lease_operations *pl_op) { ps_handle_t psh; pi_handle_t pih; struct paxos_lease *lease; if (namelen > PAXOS_NAME_LEN) { log_error("length of paxos name is too long (%u)", namelen); return -EINVAL; } if (myid == -1) myid = pl_op->get_myid(); if (!ps_handle) { px_op = malloc(sizeof(struct paxos_operations)); if (!px_op) { log_error("could not alloc for paxos operations"); return -ENOMEM; } memset(px_op, 0, sizeof(struct paxos_operations)); px_op->get_myid = pl_op->get_myid; px_op->send = pl_op->send; px_op->broadcast = pl_op->broadcast; px_op->catchup = lease_catchup; px_op->prepare = lease_prepare; px_op->is_prepared = lease_is_prepared; px_op->promise = lease_promise; px_op->propose = lease_propose; px_op->accepted = lease_accepted; px_op->commit = lease_commit; px_op->learned = lease_learned; p_l_op = pl_op; psh = paxos_space_init(PAXOS_LEASE_SPACE, number, sizeof(struct paxos_lease_msghdr), PLEASE_VALUE_LEN, role, px_op); if (psh <= 0) { log_error("failed to initialize paxos space: %ld", psh); free(px_op); px_op = NULL; return psh; } ps_handle = psh; } lease = malloc(sizeof(struct paxos_lease)); if (!lease) { log_error("cound not alloc for paxos lease"); return -ENOMEM; } memset(lease, 0, sizeof(struct paxos_lease)); strncpy(lease->name, name, PAXOS_NAME_LEN + 1); lease->owner = -1; lease->expiry = expiry; lease->failover = failover; list_add_tail(&lease->list, &lease_head); pih = paxos_instance_init(ps_handle, name, prio); if (pih <= 0) { log_error("failed to initialize paxos instance: %ld", pih); free(lease); return pih; } lease->pih = pih; return (pl_handle_t)lease; } int paxos_lease_status_recovery(pl_handle_t handle) { struct paxos_lease *pl = (struct paxos_lease *)handle; if (paxos_recovery_status_get(pl->pih) == 1) { pl->renew = 1; if (paxos_catchup(pl->pih) == 0) paxos_recovery_status_set(pl->pih, 0); } return 0; } int paxos_lease_on_receive(void *msg, int msglen) { return paxos_recvmsg(msg, msglen); } int paxos_lease_exit(pl_handle_t handle) { struct paxos_lease *pl = (struct paxos_lease *)handle; if (px_op) free(px_op); if (pl->proposer.plv) free(pl->proposer.plv); if (pl->proposer.timer1) del_timer(&pl->proposer.timer1); if (pl->proposer.timer2) del_timer(&pl->proposer.timer2); if (pl->acceptor.plv) free(pl->acceptor.plv); if (pl->acceptor.timer1) del_timer(&pl->acceptor.timer1); if (pl->acceptor.timer2) del_timer(&pl->acceptor.timer2); return 0; } diff --git a/src/ticket.c b/src/ticket.c index 79f7404..b63020d 100644 --- a/src/ticket.c +++ b/src/ticket.c @@ -1,615 +1,615 @@ /* * Copyright (C) 2011 Jiaju Zhang * * 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 the Free Software Foundation; either * version 2.1 of the License, or (at your option) any later version. * * This software is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * General Public License for more details. * * You should have received a copy of the GNU General Public * License along with this library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include #include #include "ticket.h" #include "config.h" #include "pacemaker.h" #include "list.h" #include "log.h" #include "booth.h" #include "timer.h" #include "paxos_lease.h" #include "paxos.h" #define PAXOS_MAGIC 0xDB12 #define TK_LINE 256 struct booth_msghdr { uint16_t magic; uint16_t checksum; uint32_t len; } __attribute__((packed)); struct ticket_msg { char id[BOOTH_NAME_LEN+1]; uint32_t owner; uint32_t expiry; uint32_t ballot; uint32_t result; } __attribute__((packed)); struct ticket { char id[BOOTH_NAME_LEN+1]; pl_handle_t handle; int owner; int expiry; int ballot; unsigned long long expires; struct list_head list; }; static LIST_HEAD(ticket_list); static unsigned char *role; int check_ticket(char *ticket) { int i; if (!booth_conf) return 0; for (i = 0; i < booth_conf->ticket_count; i++) { if (!strcmp(booth_conf->ticket[i].name, ticket)) return 1; } return 0; } int check_site(char *site, int *local) { int i; if (!booth_conf) return 0; for (i = 0; i < booth_conf->node_count; i++) { if (booth_conf->node[i].type == SITE && !strcmp(booth_conf->node[i].addr, site)) { *local = booth_conf->node[i].local; return 1; } } return 0; } static int * ticket_priority(int i) { int j; /* TODO: need more precise check */ for (j = 0; j < booth_conf->node_count; j++) { if (booth_conf->ticket[i].weight[j] == 0) return NULL; } return booth_conf->ticket[i].weight; } static int ticket_get_myid(void) { return booth_transport[booth_conf->proto].get_myid(); } static void end_acquire(pl_handle_t handle, int error) { struct ticket *tk; int found = 0; log_debug("enter end_acquire"); list_for_each_entry(tk, &ticket_list, list) { if (tk->handle == handle) { found = 1; break; } } if (!found) { log_error("BUG: ticket handle %ld does not exist", handle); return; } if (error) log_info("ticket %s was granted failed (site %d), error:%s", tk->id, ticket_get_myid(), strerror(error)); else log_info("ticket %s was granted successfully (site %d)", tk->id, ticket_get_myid()); log_debug("exit end_acquire"); } static void end_release(pl_handle_t handle, int error) { struct ticket *tk; int found = 0; log_debug("enter end_release"); list_for_each_entry(tk, &ticket_list, list) { if (tk->handle == handle) { found = 1; break; } } if (!found) { log_error("BUG: ticket handle %ld does not exist", handle); return; } if (error) log_info("ticket %s was reovked failed (site %d), error:%s", tk->id, ticket_get_myid(), strerror(error)); else log_info("ticket %s was reovked successfully (site %d)", tk->id, ticket_get_myid()); log_debug("exit end_release"); } static int ticket_send(unsigned long id, void *value, int len) { int i, rv = -1; struct booth_node *to = NULL; struct booth_msghdr *hdr; void *buf; for (i = 0; i < booth_conf->node_count; i++) { if (booth_conf->node[i].nodeid == id) to = &booth_conf->node[i]; } if (!to) return rv; buf = malloc(sizeof(struct booth_msghdr) + len); if (!buf) return -ENOMEM; memset(buf, 0, sizeof(struct booth_msghdr) + len); hdr = buf; hdr->magic = htons(PAXOS_MAGIC); hdr->len = htonl(sizeof(struct booth_msghdr) + len); memcpy((char *)buf + sizeof(struct booth_msghdr), value, len); rv = booth_transport[booth_conf->proto].send( (unsigned long)to, buf, sizeof(struct booth_msghdr) + len); free(buf); return rv; } static int ticket_broadcast(void *value, int len) { void *buf; struct booth_msghdr *hdr; int rv; buf = malloc(sizeof(struct booth_msghdr) + len); if (!buf) return -ENOMEM; memset(buf, 0, sizeof(struct booth_msghdr) + len); hdr = buf; hdr->magic = htons(PAXOS_MAGIC); hdr->len = htonl(sizeof(struct booth_msghdr) + len); memcpy((char *)buf + sizeof(struct booth_msghdr), value, len); rv = booth_transport[booth_conf->proto].broadcast( buf, sizeof(struct booth_msghdr) + len); free(buf); return rv; } #if 0 static int ticket_read(const void *name, int *owner, int *ballot, unsigned long long *expires) { struct ticket *tk; int found = 0; list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, name)) { found = 1; break; } } if (!found) { log_error("BUG: ticket_read failed (ticket %s does not exist)", (char *)name); return -1; } pcmk_handler.load_ticket(tk->id, &tk->owner, &tk->ballot, &tk->expires); *owner = tk->owner; *expires = tk->expires; *ballot = tk->ballot; return 0; } #endif static int ticket_parse(struct ticket_msg *tmsg) { struct ticket *tk; int found = 0; if (!tmsg->result) return -1; list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, tmsg->id)) { tk->owner = tmsg->owner; tk->expires = current_time() + tmsg->expiry; tk->ballot = tmsg->ballot; found = 1; break; } } if (!found) return -1; else return 0; } static int ticket_catchup(const void *name, int *owner, int *ballot, unsigned long long *expires) { struct ticket *tk; int i, s, buflen, rv = 0; char *buf = NULL; struct boothc_header *h; struct ticket_msg *tmsg; int myid = ticket_get_myid(); if (booth_conf->node[myid].type != ARBITRATOR) { list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, name)) { pcmk_handler.load_ticket(tk->id, &tk->owner, &tk->ballot, &tk->expires); if (current_time() >= tk->expires) { tk->owner = -1; tk->expires = 0; } } } } buflen = sizeof(struct boothc_header) + sizeof(struct ticket_msg); buf = malloc(buflen); if (!buf) return -ENOMEM; memset(buf, 0, buflen); h = (struct boothc_header *)buf; h->magic = BOOTHC_MAGIC; h->version = BOOTHC_VERSION; h->cmd = BOOTHC_CMD_CATCHUP; h->len = sizeof(struct ticket_msg); tmsg = (struct ticket_msg *)(buf + sizeof(struct boothc_header)); for (i = 0; i < booth_conf->node_count; i++) { if (booth_conf->node[i].type == SITE && !(booth_conf->node[i].local)) { strncpy(tmsg->id, name, BOOTH_NAME_LEN + 1); log_debug("attempting catchup from %s", booth_conf->node[i].addr); s = booth_transport[TCP].open(&booth_conf->node[i]); if (s < 0) continue; log_debug("connected to %s", booth_conf->node[i].addr); rv = booth_transport[TCP].send(s, buf, buflen); if (rv < 0) { booth_transport[TCP].close(s); continue; } log_debug("sent catchup command to %s", booth_conf->node[i].addr); memset(tmsg, 0, sizeof(struct ticket_msg)); rv = booth_transport[TCP].recv(s, buf, buflen); if (rv < 0) { booth_transport[TCP].close(s); continue; } booth_transport[TCP].close(s); ticket_parse(tmsg); memset(tmsg, 0, sizeof(struct ticket_msg)); } } list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, name)) { if (booth_conf->node[myid].type != ARBITRATOR) { if (current_time() >= tk->expires) { tk->owner = -1; tk->expires = 0; } pcmk_handler.store_ticket(tk->id, tk->owner, tk->ballot, tk->expires); if (tk->owner == myid) pcmk_handler.grant_ticket(tk->id); else pcmk_handler.revoke_ticket(tk->id); } *owner = tk->owner; *expires = tk->expires; *ballot = tk->ballot; } } free(buf); return rv; } static int ticket_write(pl_handle_t handle, struct paxos_lease_result *result) { struct ticket *tk; int found = 0; list_for_each_entry(tk, &ticket_list, list) { if (tk->handle == handle) { found = 1; break; } } if (!found) { log_error("BUG: ticket_write failed " "(ticket handle %ld does not exist)", handle); return -1; } tk->owner = result->owner; tk->expires = result->expires; tk->ballot = result->ballot; if (tk->owner == ticket_get_myid()) { pcmk_handler.store_ticket(tk->id, tk->owner, tk->ballot, tk->expires); pcmk_handler.grant_ticket(tk->id); } else if (tk->owner == -1) { pcmk_handler.store_ticket(tk->id, tk->owner, tk->ballot, tk->expires); pcmk_handler.revoke_ticket(tk->id); } else pcmk_handler.store_ticket(tk->id, tk->owner, tk->ballot, tk->expires); return 0; } static void ticket_status_recovery(pl_handle_t handle) { paxos_lease_status_recovery(handle); } int ticket_recv(void *msg, int msglen) { struct booth_msghdr *hdr; char *data; hdr = msg; if (ntohs(hdr->magic) != PAXOS_MAGIC || ntohl(hdr->len) != msglen) { log_error("message received error"); return -1; } data = (char *)msg + sizeof(struct booth_msghdr); return paxos_lease_on_receive(data, msglen - sizeof(struct booth_msghdr)); } int grant_ticket(char *ticket) { struct ticket *tk; int found = 0; list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, ticket)) { found = 1; break; } } if (!found) { log_error("ticket %s does not exist", ticket); return BOOTHC_RLT_SYNC_FAIL; } if (tk->owner == ticket_get_myid()) return BOOTHC_RLT_SYNC_SUCC; else { paxos_lease_acquire(tk->handle, CLEAR_RELEASE, 1, end_acquire); return BOOTHC_RLT_ASYNC; } } int revoke_ticket(char *ticket) { struct ticket *tk; int found = 0; list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, ticket)) { found = 1; break; } } if (!found) { log_error("ticket %s does not exist", ticket); return BOOTHC_RLT_SYNC_FAIL; } if (tk->owner == -1) return BOOTHC_RLT_SYNC_SUCC; else { - paxos_lease_release(tk->handle, end_release); - return BOOTHC_RLT_ASYNC; + int ret = paxos_lease_release(tk->handle, end_release); + return (ret < 0)? BOOTHC_RLT_SYNC_FAIL: BOOTHC_RLT_ASYNC; } } int get_ticket_info(char *name, int *owner, int *expires) { struct ticket *tk; list_for_each_entry(tk, &ticket_list, list) { if (!strncmp(tk->id, name, BOOTH_NAME_LEN + 1)) { if(owner) *owner = tk->owner; if(expires) *expires = tk->expires; return 0; } } return -1; } int list_ticket(char **pdata, unsigned int *len) { struct ticket *tk; char timeout_str[100]; char node_name[BOOTH_NAME_LEN]; char tmp[TK_LINE]; *pdata = NULL; *len = 0; list_for_each_entry(tk, &ticket_list, list) { memset(tmp, 0, TK_LINE); strncpy(timeout_str, "INF", sizeof(timeout_str)); strncpy(node_name, "None", sizeof(node_name)); if (tk->owner < MAX_NODES && tk->owner > -1) strncpy(node_name, booth_conf->node[tk->owner].addr, sizeof(node_name)); if (tk->expires != 0) strftime(timeout_str, sizeof(timeout_str), "%Y/%m/%d %H:%M:%S", localtime((time_t *)&tk->expires)); snprintf(tmp, TK_LINE, "ticket: %s, owner: %s, expires: %s\n", tk->id, node_name, timeout_str); *pdata = realloc(*pdata, *len + TK_LINE); if (*pdata == NULL) return -ENOMEM; memset(*pdata + *len, 0, TK_LINE); memcpy(*pdata + *len, tmp, TK_LINE); *len += TK_LINE; } return 0; } int catchup_ticket(char **pdata, unsigned int len) { struct ticket_msg *tmsg; struct ticket *tk; assert(len == sizeof(struct ticket_msg)); tmsg = (struct ticket_msg *)(*pdata); list_for_each_entry(tk, &ticket_list, list) { if (!strcmp(tk->id, tmsg->id) && tk->owner == ticket_get_myid() && current_time() < tk->expires) { tmsg->owner = tk->owner; tmsg->expiry = tk->expires - current_time(); tmsg->ballot = tk->ballot; tmsg->result = 1; break; } } if (!tmsg->result) memset(*pdata, 0, len); return 0; } const struct paxos_lease_operations ticket_operations = { .get_myid = ticket_get_myid, .send = ticket_send, .broadcast = ticket_broadcast, .catchup = ticket_catchup, .notify = ticket_write, }; int setup_ticket(void) { struct ticket *tk, *tmp; int i, rv; pl_handle_t plh; int myid; role = malloc(booth_conf->node_count * sizeof(unsigned char)); if (!role) return -ENOMEM; memset(role, 0, booth_conf->node_count * sizeof(unsigned char)); for (i = 0; i < booth_conf->node_count; i++) { if (booth_conf->node[i].type == SITE) role[i] = PROPOSER | ACCEPTOR | LEARNER; else if (booth_conf->node[i].type == ARBITRATOR) role[i] = ACCEPTOR | LEARNER; } for (i = 0; i < booth_conf->ticket_count; i++) { tk = malloc(sizeof(struct ticket)); if (!tk) { rv = -ENOMEM; goto out; } memset(tk, 0, sizeof(struct ticket)); strcpy(tk->id, booth_conf->ticket[i].name); tk->owner = -1; tk->expiry = booth_conf->ticket[i].expiry; list_add_tail(&tk->list, &ticket_list); plh = paxos_lease_init(tk->id, BOOTH_NAME_LEN, tk->expiry, booth_conf->node_count, 1, role, ticket_priority(i), &ticket_operations); if (plh <= 0) { log_error("paxos lease initialization failed"); rv = plh; goto out; } tk->handle = plh; } myid = ticket_get_myid(); assert(myid < booth_conf->node_count); if (role[myid] & ACCEPTOR) { list_for_each_entry(tk, &ticket_list, list) { ticket_status_recovery(tk->handle); } } return 0; out: list_for_each_entry_safe(tk, tmp, &ticket_list, list) { list_del(&tk->list); } free(role); return rv; }