diff --git a/src/pacemaker.c b/src/pacemaker.c index d7c16ca..a560c7e 100644 --- a/src/pacemaker.c +++ b/src/pacemaker.c @@ -1,141 +1,169 @@ /* * 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 "log.h" #include "pacemaker.h" #define COMMAND_MAX 256 static void pcmk_grant_ticket(const void *ticket) { FILE *p; char cmd[COMMAND_MAX]; snprintf(cmd, COMMAND_MAX, "crm_ticket -t %s -v true", (char *)ticket); log_info("command: '%s' was executed", cmd); p = popen(cmd, "r"); if (p == NULL) { log_error("popen error: %s", cmd); return; } pclose(p); return; } static void pcmk_revoke_ticket(const void *ticket) { FILE *p; char cmd[COMMAND_MAX]; snprintf(cmd, COMMAND_MAX, "crm_ticket -t %s -v false", (char *)ticket); log_info("command: '%s' was executed", cmd); p = popen(cmd, "r"); if (p == NULL) { log_error("popen error: %s", cmd); return; } pclose(p); return; } -static void pcmk_store_ticket(const void *ticket, int owner, +static void pcmk_store_ticket(const void *ticket, int owner, int ballot, unsigned long long expires) { FILE *p; char cmd[COMMAND_MAX]; snprintf(cmd, COMMAND_MAX, "crm_attribute -t tickets -n owner-%s -v %d", (char *)ticket, owner); log_info("command: '%s' was executed", cmd); p = popen(cmd, "r"); if (p == NULL) { log_error("popen error: %s", cmd); return; } pclose(p); snprintf(cmd, COMMAND_MAX, "crm_attribute -t tickets -n expires-%s -v %llu", (char *)ticket, expires); log_info("command: '%s' was executed", cmd); p = popen(cmd, "r"); if (p == NULL) { log_error("popen error: %s", cmd); return; } pclose(p); + snprintf(cmd, COMMAND_MAX, + "crm_attribute -t tickets -n ballot-%s -v %d", + (char *)ticket, ballot); + log_info("command: '%s' was executed", cmd); + p = popen(cmd, "r"); + if (p == NULL) { + log_error("popen error: %s", cmd); + return; + } + pclose(p); + return; } -static void pcmk_load_ticket(const void *ticket, int *owner, +static void pcmk_load_ticket(const void *ticket, int *owner, int *ballot, unsigned long long *expires) { FILE *p; char cmd[COMMAND_MAX]; char line[256]; - int ow; + int ow, ba; unsigned long long ex; snprintf(cmd, COMMAND_MAX, "crm_attribute -t tickets -n owner-%s -G --quiet", (char *)ticket); log_info("command: '%s' was executed", cmd); p = popen(cmd, "r"); if (p == NULL) { log_error("popen error: %s", cmd); return; } if (fgets(line, sizeof(line) - 1, p) == NULL) { pclose(p); return; } if (sscanf(line, "%d", &ow) == 1) *owner = ow; pclose(p); snprintf(cmd, COMMAND_MAX, "crm_attribute -t tickets -n expires-%s -G --quiet", (char *)ticket); log_info("command: '%s' was executed", cmd); p = popen(cmd, "r"); if (p == NULL) { log_error("popen error: %s", cmd); return; } if (fgets(line, sizeof(line) - 1, p) == NULL) { pclose(p); return; } if (sscanf(line, "%llu", &ex) == 1) *expires = ex; pclose(p); + snprintf(cmd, COMMAND_MAX, + "crm_attribute -t tickets -n ballot-%s -G --quiet", + (char *)ticket); + log_info("command: '%s' was executed", cmd); + p = popen(cmd, "r"); + if (p == NULL) { + log_error("popen error: %s", cmd); + return; + } + if (fgets(line, sizeof(line) - 1, p) == NULL) { + pclose(p); + return; + } + if (sscanf(line, "%d", &ba) == 1) + *ballot = ba; + pclose(p); + return; } struct ticket_handler pcmk_handler = { .grant_ticket = pcmk_grant_ticket, .revoke_ticket = pcmk_revoke_ticket, .store_ticket = pcmk_store_ticket, .load_ticket = pcmk_load_ticket, }; diff --git a/src/pacemaker.h b/src/pacemaker.h index c3eb9d2..90c19c7 100644 --- a/src/pacemaker.h +++ b/src/pacemaker.h @@ -1,31 +1,31 @@ /* * 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 */ #ifndef _PACEMAKER_H #define _PACEMAKER_H struct ticket_handler { void (*grant_ticket) (const void *); void (*revoke_ticket) (const void *); - void (*store_ticket) (const void *, int, unsigned long long); - void (*load_ticket) (const void *, int *, unsigned long long *); + void (*store_ticket) (const void *, int, int, unsigned long long); + void (*load_ticket) (const void *, int *, int *, unsigned long long *); }; struct ticket_handler pcmk_handler; #endif /* _PACEMAKER_H */ diff --git a/src/paxos.c b/src/paxos.c index 5da909a..e2a4ae1 100644 --- a/src/paxos.c +++ b/src/paxos.c @@ -1,859 +1,868 @@ /* * 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 "list.h" #include "paxos.h" #include "log.h" typedef enum { INIT = 1, PREPARING, PROMISING, PROPOSING, ACCEPTING, RECOVERY, COMMITTED, } paxos_state_t; struct proposal { int ballot_number; char value[0]; }; struct learned { int ballot; int number; }; struct paxos_msghdr { paxos_state_t state; int from; char psname[PAXOS_NAME_LEN+1]; char piname[PAXOS_NAME_LEN+1]; int ballot_number; int reject; int proposer_id; unsigned int extralen; unsigned int valuelen; }; struct proposer { int state; int ballot; int open_number; int accepted_number; int proposed; struct proposal *proposal; }; struct acceptor { int state; int highest_promised; struct proposal *accepted_proposal; }; struct learner { int state; int learned_max; int learned_ballot; struct learned learned[0]; }; struct paxos_space; struct paxos_instance; struct proposer_operations { void (*prepare) (struct paxos_instance *, int *); void (*propose) (struct paxos_space *, struct paxos_instance *, void *, int); void (*commit) (struct paxos_space *, struct paxos_instance *, void *, int); }; struct acceptor_operations { void (*promise) (struct paxos_space *, struct paxos_instance *, void *, int); void (*accepted) (struct paxos_space *, struct paxos_instance *, void *, int); }; struct learner_operations { void (*response) (struct paxos_space *, struct paxos_instance *, void *, int); }; struct paxos_space { char name[PAXOS_NAME_LEN+1]; unsigned int number; unsigned int extralen; unsigned int valuelen; const unsigned char *role; const struct paxos_operations *p_op; const struct proposer_operations *r_op; const struct acceptor_operations *a_op; const struct learner_operations *l_op; struct list_head list; struct list_head pi_head; }; struct paxos_instance { char name[PAXOS_NAME_LEN+1]; int round; int *prio; struct proposer *proposer; struct acceptor *acceptor; struct learner *learner; void (*end) (pi_handle_t pih, int round, int result); struct list_head list; struct paxos_space *ps; }; static LIST_HEAD(ps_head); static int have_quorum(struct paxos_space *ps, int member) { int i, sum = 0; for (i = 0; i < ps->number; i++) { if (ps->role[i] & ACCEPTOR) sum++; } if (member * 2 > sum) return 1; else return 0; } static int next_ballot_number(struct paxos_instance *pi) { int ballot; int myid = pi->ps->p_op->get_myid(); if (pi->prio) ballot = pi->prio[myid]; else ballot = myid; while (ballot <= pi->round) ballot += pi->ps->number; return ballot; } static void proposer_prepare(struct paxos_instance *pi, int *round) { struct paxos_msghdr *hdr; void *msg; int msglen = sizeof(struct paxos_msghdr) + pi->ps->extralen; int ballot; log_debug("preposer prepare ..."); msg = malloc(msglen); if (!msg) { log_error("no mem for msg"); *round = -ENOMEM; return; } memset(msg, 0, msglen); hdr = msg; if (*round > pi->round) pi->round = *round; ballot = next_ballot_number(pi); pi->proposer->ballot = ballot; hdr->state = htonl(PREPARING); hdr->from = htonl(pi->ps->p_op->get_myid()); hdr->proposer_id = hdr->from; strcpy(hdr->psname, pi->ps->name); strcpy(hdr->piname, pi->name); hdr->ballot_number = htonl(ballot); hdr->extralen = htonl(pi->ps->extralen); if (pi->ps->p_op->broadcast) pi->ps->p_op->broadcast(msg, msglen); else { int i; for (i = 0; i < pi->ps->number; i++) { if (pi->ps->role[i] & ACCEPTOR) pi->ps->p_op->send(i, msg, msglen); } } free(msg); *round = ballot; } static void proposer_propose(struct paxos_space *ps, struct paxos_instance *pi, void *msg, int msglen) { struct paxos_msghdr *hdr; pi_handle_t pih = (pi_handle_t)pi; void *extra, *value, *message; int ballot; log_debug("proposer propose ..."); if (msglen != sizeof(struct paxos_msghdr) + ps->extralen) { log_error("message length incorrect, " "msglen: %d, msghdr len: %lu, extralen: %u", msglen, (long)sizeof(struct paxos_msghdr), ps->extralen); return; } hdr = msg; ballot = ntohl(hdr->ballot_number); if (pi->proposer->ballot != ballot) { log_debug("not the same ballot, proposer ballot: %d, " "received ballot: %d", pi->proposer->ballot, ballot); return; } if (ntohl(hdr->reject)) { log_debug("proposal was rejected"); pi->round = ballot; pi->proposer->state = INIT; pi->end(pih, pi->round, -EAGAIN); return; } extra = (char *)msg + sizeof(struct paxos_msghdr); if (ps->p_op->prepare) { if (ps->p_op->prepare(pih, extra)) pi->proposer->open_number++; } else pi->proposer->open_number++; if (!have_quorum(ps, pi->proposer->open_number)) return; if (pi->proposer->proposed) return; pi->proposer->proposed = 1; value = pi->proposer->proposal->value; if (ps->p_op->propose) ps->p_op->propose(pih, extra, ballot, value); hdr->valuelen = htonl(ps->valuelen); message = malloc(msglen + ps->valuelen); if (!message) { log_error("no mem for value"); return; } memset(message, 0, msglen + ps->valuelen); memcpy(message, msg, msglen); memcpy((char *)message + msglen, value, ps->valuelen); pi->proposer->state = PROPOSING; hdr = message; hdr->from = htonl(ps->p_op->get_myid()); hdr->state = htonl(PROPOSING); if (ps->p_op->broadcast) ps->p_op->broadcast(message, msglen + ps->valuelen); else { int i; for (i = 0; i < ps->number; i++) { if (ps->role[i] & ACCEPTOR) ps->p_op->send(i, message, msglen + ps->valuelen); } } } static void proposer_commit(struct paxos_space *ps, struct paxos_instance *pi, void *msg, int msglen) { struct paxos_msghdr *hdr; pi_handle_t pih = (pi_handle_t)pi; void *extra; int ballot; log_debug("proposer commit ..."); if (msglen != sizeof(struct paxos_msghdr) + ps->extralen) { log_error("message length incorrect, " "msglen: %d, msghdr len: %lu, extralen: %u", msglen, (long)sizeof(struct paxos_msghdr), ps->extralen); return; } extra = (char *)msg + sizeof(struct paxos_msghdr); hdr = msg; ballot = ntohl(hdr->ballot_number); if (pi->proposer->ballot != ballot) { log_debug("not the same ballot, proposer ballot: %d, " "received ballot: %d", pi->proposer->ballot, ballot); return; } pi->proposer->accepted_number++; if (!have_quorum(ps, pi->proposer->accepted_number)) return; if (pi->proposer->state == COMMITTED) return; pi->round = ballot; if (ps->p_op->commit) ps->p_op->commit(pih, extra, pi->round); pi->proposer->state = COMMITTED; - pi->end(pih, pi->round, 0); + if (pi->end) + pi->end(pih, pi->round, 0); } static void acceptor_promise(struct paxos_space *ps, struct paxos_instance *pi, void *msg, int msglen) { struct paxos_msghdr *hdr; unsigned long to; pi_handle_t pih = (pi_handle_t)pi; void *extra; log_debug("acceptor promise ..."); if (pi->acceptor->state == RECOVERY) { log_debug("still in recovery"); return; } if (msglen != sizeof(struct paxos_msghdr) + ps->extralen) { log_error("message length incorrect, " "msglen: %d, msghdr len: %lu, extralen: %u", msglen, (long)sizeof(struct paxos_msghdr), ps->extralen); return; } hdr = msg; extra = (char *)msg + sizeof(struct paxos_msghdr); if (ntohl(hdr->ballot_number) < pi->acceptor->highest_promised) { log_debug("ballot number: %d, highest promised: %d", ntohl(hdr->ballot_number), pi->acceptor->highest_promised); to = ntohl(hdr->from); hdr->from = htonl(ps->p_op->get_myid()); hdr->state = htonl(PROMISING); hdr->reject = htonl(1); memset(extra, 0, ps->extralen); ps->p_op->send(to, msg, msglen); return; } pi->acceptor->highest_promised = ntohl(hdr->ballot_number); if (ps->p_op->promise) ps->p_op->promise(pih, extra); pi->acceptor->state = PROMISING; to = ntohl(hdr->from); hdr->from = htonl(ps->p_op->get_myid()); hdr->state = htonl(PROMISING); ps->p_op->send(to, msg, msglen); } static void acceptor_accepted(struct paxos_space *ps, struct paxos_instance *pi, void *msg, int msglen) { struct paxos_msghdr *hdr; unsigned long to; pi_handle_t pih = (pi_handle_t)pi; void *extra, *value; int myid = ps->p_op->get_myid(); int ballot; log_debug("acceptor accepted ..."); if (pi->acceptor->state == RECOVERY) { log_debug("still in recovery"); return; } if (msglen != sizeof(struct paxos_msghdr) + ps->extralen + ps->valuelen) { log_error("message length incorrect, msglen: " "%d, msghdr len: %lu, extralen: %u, valuelen: %u", msglen, (long)sizeof(struct paxos_msghdr), ps->extralen, ps->valuelen); return; } hdr = msg; extra = (char *)msg + sizeof(struct paxos_msghdr); ballot = ntohl(hdr->ballot_number); if (ballot < pi->acceptor->highest_promised) { log_debug("ballot: %d, highest promised: %d", ballot, pi->acceptor->highest_promised); to = ntohl(hdr->from); hdr->from = htonl(myid); hdr->state = htonl(ACCEPTING); hdr->reject = htonl(1); ps->p_op->send(to, hdr, sizeof(struct paxos_msghdr)); return; } value = pi->acceptor->accepted_proposal->value; memcpy(value, (char *)msg + sizeof(struct paxos_msghdr) + ps->extralen, ps->valuelen); if (ps->p_op->accepted) ps->p_op->accepted(pih, extra, ballot, value); pi->acceptor->state = ACCEPTING; to = ntohl(hdr->from); hdr->from = htonl(myid); hdr->state = htonl(ACCEPTING); if (ps->p_op->broadcast) ps->p_op->broadcast(msg, sizeof(struct paxos_msghdr) + ps->extralen); else { int i; for (i = 0; i < ps->number; i++) { if (ps->role[i] & LEARNER) ps->p_op->send(i, msg, sizeof(struct paxos_msghdr) + ps->extralen); } if (!(ps->role[to] & LEARNER)) ps->p_op->send(to, msg, sizeof(struct paxos_msghdr) + ps->extralen); } } static void learner_response(struct paxos_space *ps, struct paxos_instance *pi, void *msg, int msglen) { struct paxos_msghdr *hdr; pi_handle_t pih = (pi_handle_t)pi; void *extra; int i, unused = 0, found = 0; int ballot; log_debug("learner response ..."); if (msglen != sizeof(struct paxos_msghdr) + ps->extralen) { log_error("message length incorrect, " "msglen: %d, msghdr len: %lu, extralen: %u", msglen, (long)sizeof(struct paxos_msghdr), ps->extralen); return; } hdr = msg; extra = (char *)msg + sizeof(struct paxos_msghdr); ballot = ntohl(hdr->ballot_number); for (i = 0; i < ps->number; i++) { if (!pi->learner->learned[i].ballot) { unused = i; break; } if (pi->learner->learned[i].ballot == ballot) { pi->learner->learned[i].number++; if (pi->learner->learned[i].number > pi->learner->learned_max) pi->learner->learned_max = pi->learner->learned[i].number; found = 1; break; } } if (!found) { pi->learner->learned[unused].ballot = ntohl(hdr->ballot_number); pi->learner->learned[unused].number = 1; } if (!have_quorum(ps, pi->learner->learned_max)) return; if (ps->p_op->learned) ps->p_op->learned(pih, extra, ballot); } const struct proposer_operations generic_proposer_operations = { .prepare = proposer_prepare, .propose = proposer_propose, .commit = proposer_commit, }; const struct acceptor_operations generic_acceptor_operations = { .promise = acceptor_promise, .accepted = acceptor_accepted, }; const struct learner_operations generic_learner_operations = { .response = learner_response, }; ps_handle_t paxos_space_init(const void *name, unsigned int number, unsigned int extralen, unsigned int valuelen, const unsigned char *role, const struct paxos_operations *p_op) { struct paxos_space *ps; list_for_each_entry(ps, &ps_head, list) { if (!strcmp(ps->name, name)) { log_info("paxos space (%s) has already been " "initialized", (char *)name); return -EEXIST; } } if (!number || !valuelen || !p_op || !p_op->get_myid || !p_op->send) { log_error("invalid agruments"); return -EINVAL; } ps = malloc(sizeof(struct paxos_space)); if (!ps) { log_error("no mem for paxos space"); return -ENOMEM; } memset(ps, 0, sizeof(struct paxos_space)); strncpy(ps->name, name, PAXOS_NAME_LEN + 1); ps->number = number; ps->extralen = extralen; ps->valuelen = valuelen; ps->role = role; ps->p_op = p_op; ps->r_op = &generic_proposer_operations; ps->a_op = &generic_acceptor_operations; ps->l_op = &generic_learner_operations; list_add_tail(&ps->list, &ps_head); INIT_LIST_HEAD(&ps->pi_head); return (ps_handle_t)ps; } pi_handle_t paxos_instance_init(ps_handle_t handle, const void *name, int *prio) { struct paxos_space *ps = (struct paxos_space *)handle; struct paxos_instance *pi; struct proposer *proposer = NULL; struct acceptor *acceptor = NULL; struct learner *learner = NULL; int myid, valuelen, rv; list_for_each_entry(pi, &ps->pi_head, list) { if (!strcmp(pi->name, name)) return (pi_handle_t)pi; } if (handle <= 0 || !ps->p_op || !ps->p_op->get_myid) { log_error("invalid agruments"); rv = -EINVAL; goto out; } myid = ps->p_op->get_myid(); valuelen = ps->valuelen; pi = malloc(sizeof(struct paxos_instance)); if (!pi) { log_error("no mem for paxos instance"); rv = -ENOMEM; goto out; } memset(pi, 0, sizeof(struct paxos_instance)); strncpy(pi->name, name, PAXOS_NAME_LEN + 1); if (prio) { pi->prio = malloc(ps->number * sizeof(int)); if (!pi->prio) { log_error("no mem for prio"); rv = -ENOMEM; goto out_pi; } memcpy(pi->prio, prio, ps->number * sizeof(int)); } if (ps->role[myid] & PROPOSER) { proposer = malloc(sizeof(struct proposer)); if (!proposer) { log_error("no mem for proposer"); rv = -ENOMEM; goto out_prio; } memset(proposer, 0, sizeof(struct proposer)); proposer->state = INIT; proposer->proposal = malloc(sizeof(struct proposal) + valuelen); if (!proposer->proposal) { log_error("no mem for proposal"); rv = -ENOMEM; goto out_proposer; } memset(proposer->proposal, 0, sizeof(struct proposal) + valuelen); pi->proposer = proposer; } if (ps->role[myid] & ACCEPTOR) { acceptor = malloc(sizeof(struct acceptor)); if (!acceptor) { log_error("no mem for acceptor"); rv = -ENOMEM; goto out_proposal; } memset(acceptor, 0, sizeof(struct acceptor)); acceptor->state = INIT; acceptor->accepted_proposal = malloc(sizeof(struct proposal) + valuelen); if (!acceptor->accepted_proposal) { log_error("no mem for accepted proposal"); rv = -ENOMEM; goto out_acceptor; } memset(acceptor->accepted_proposal, 0, sizeof(struct proposal) + valuelen); pi->acceptor = acceptor; - if (ps->p_op->catchup) { + if (ps->p_op->catchup) pi->acceptor->state = RECOVERY; - ps->p_op->catchup(name); + else pi->acceptor->state = INIT; - } } if (ps->role[myid] & LEARNER) { learner = malloc(sizeof(struct learner) + ps->number * sizeof(struct learned)); if (!learner) { log_error("no mem for learner"); rv = -ENOMEM; goto out_accepted_proposal; } memset(learner, 0, sizeof(struct learner) + ps->number * sizeof(struct learned)); learner->state = INIT; pi->learner = learner; } pi->ps = ps; list_add_tail(&pi->list, &ps->pi_head); return (pi_handle_t)pi; out_accepted_proposal: if (ps->role[myid] & ACCEPTOR) free(acceptor->accepted_proposal); out_acceptor: if (ps->role[myid] & ACCEPTOR) free(acceptor); out_proposal: if (ps->role[myid] & PROPOSER) free(proposer->proposal); out_proposer: if (ps->role[myid] & PROPOSER) free(proposer); out_prio: if (pi->prio) free(pi->prio); out_pi: free(pi); out: return rv; } int paxos_round_request(pi_handle_t handle, void *value, int *round, void (*end_request) (pi_handle_t handle, int round, int result)) { struct paxos_instance *pi = (struct paxos_instance *)handle; int myid = pi->ps->p_op->get_myid(); int rv = *round; if (!(pi->ps->role[myid] & PROPOSER)) { log_debug("only proposer can do this"); return -EOPNOTSUPP; } pi->proposer->state = PREPARING; pi->proposer->open_number = 0; pi->proposer->accepted_number = 0; pi->proposer->proposed = 0; memcpy(pi->proposer->proposal->value, value, pi->ps->valuelen); pi->end = end_request; pi->ps->r_op->prepare(pi, &rv); return rv; } int paxos_recovery_status_get(pi_handle_t handle) { struct paxos_instance *pi = (struct paxos_instance *)handle; int myid = pi->ps->p_op->get_myid(); if (!(pi->ps->role[myid] & ACCEPTOR)) return -EOPNOTSUPP; if (pi->acceptor->state == RECOVERY) return 1; else return 0; } int paxos_recovery_status_set(pi_handle_t handle, int recovery) { struct paxos_instance *pi = (struct paxos_instance *)handle; int myid = pi->ps->p_op->get_myid(); if (!(pi->ps->role[myid] & ACCEPTOR)) return -EOPNOTSUPP; if (recovery) pi->acceptor->state = RECOVERY; else pi->acceptor->state = INIT; return 0; } int paxos_propose(pi_handle_t handle, void *value, int round) { struct paxos_instance *pi = (struct paxos_instance *)handle; struct paxos_msghdr *hdr; void *extra, *msg; int len = sizeof(struct paxos_msghdr) + pi->ps->extralen + pi->ps->valuelen; + if (!pi->proposer->ballot) + pi->proposer->ballot = round; if (round != pi->proposer->ballot) { log_debug("round: %d, proposer ballot: %d", round, pi->proposer->ballot); return -EINVAL; } msg = malloc(len); if (!msg) { log_error("no mem for msg"); return -ENOMEM; } pi->proposer->state = PROPOSING; strcpy(pi->proposer->proposal->value, value); pi->proposer->accepted_number = 0; pi->round = round; memset(msg, 0, len); hdr = msg; hdr->state = htonl(PROPOSING); hdr->from = htonl(pi->ps->p_op->get_myid()); hdr->proposer_id = hdr->from; strcpy(hdr->psname, pi->ps->name); strcpy(hdr->piname, pi->name); hdr->ballot_number = htonl(pi->round); hdr->extralen = htonl(pi->ps->extralen); extra = (char *)msg + sizeof(struct paxos_msghdr); memcpy((char *)msg + sizeof(struct paxos_msghdr) + pi->ps->extralen, value, pi->ps->valuelen); if (pi->ps->p_op->propose) pi->ps->p_op->propose(handle, extra, round, value); if (pi->ps->p_op->broadcast) pi->ps->p_op->broadcast(msg, len); else { int i; for (i = 0; i < pi->ps->number; i++) { if (pi->ps->role[i] & ACCEPTOR) pi->ps->p_op->send(i, msg, len); } } return 0; } +int paxos_catchup(pi_handle_t handle) +{ + struct paxos_instance *pi = (struct paxos_instance *)handle; + + return pi->ps->p_op->catchup(handle); +} + int paxos_recvmsg(void *msg, int msglen) { struct paxos_msghdr *hdr = msg; struct paxos_space *ps; struct paxos_instance *pi; int found = 0; int myid; list_for_each_entry(ps, &ps_head, list) { if (!strcmp(ps->name, hdr->psname)) { found = 1; break; } } if (!found) { log_error("could not find the received ps name (%s) " "in registered list", hdr->psname); return -EINVAL; } myid = ps->p_op->get_myid(); found = 0; list_for_each_entry(pi, &ps->pi_head, list) { if (!strcmp(pi->name, hdr->piname)) { found = 1; break; } } if (!found) paxos_instance_init((ps_handle_t)ps, hdr->piname, NULL); switch (ntohl(hdr->state)) { case PREPARING: if (ps->role[myid] & ACCEPTOR) ps->a_op->promise(ps, pi, msg, msglen); break; case PROMISING: ps->r_op->propose(ps, pi, msg, msglen); break; case PROPOSING: if (ps->role[myid] & ACCEPTOR) ps->a_op->accepted(ps, pi, msg, msglen); break; case ACCEPTING: if (ntohl(hdr->proposer_id) == myid) ps->r_op->commit(ps, pi, msg, msglen); else if (ps->role[myid] & LEARNER) ps->l_op->response(ps, pi, msg, msglen); break; default: log_debug("invalid message type: %d", ntohl(hdr->state)); break; }; return 0; } diff --git a/src/paxos.h b/src/paxos.h index b91d9a9..4a8ba79 100644 --- a/src/paxos.h +++ b/src/paxos.h @@ -1,80 +1,82 @@ /* * 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 */ #ifndef _PAXOS_H #define _PAXOS_H #define PAXOS_NAME_LEN 63 #define PROPOSER 0x4 #define ACCEPTOR 0x2 #define LEARNER 0x1 typedef long ps_handle_t; typedef long pi_handle_t; struct paxos_operations { int (*get_myid) (void); int (*send) (unsigned long id, void *value, int len); int (*broadcast) (void *value, int len); - int (*catchup) (const void *name); + int (*catchup) (pi_handle_t handle); int (*prepare) (pi_handle_t handle, void *extra); int (*promise) (pi_handle_t handle, void *extra); int (*propose) (pi_handle_t handle, void *extra, int round, void *value); int (*accepted) (pi_handle_t handle, void *extra, int round, void *value); int (*commit) (pi_handle_t handle, void *extra, int round); int (*learned) (pi_handle_t handle, void *extra, int round); }; int paxos_recvmsg(void *msg, int msglen); ps_handle_t paxos_space_init(const void *name, unsigned int number, unsigned int extralen, unsigned int valuelen, const unsigned char *role, const struct paxos_operations *p_op); pi_handle_t paxos_instance_init(ps_handle_t handle, const void *name, int *prio); int paxos_round_request(pi_handle_t handle, void *value, int *round, void (*end_request) (pi_handle_t handle, int round, int result)); int paxos_round_discard(pi_handle_t handle, int round); int paxos_leader_get(pi_handle_t handle, int *round); int paxos_recovery_status_get(pi_handle_t handle); int paxos_recovery_status_set(pi_handle_t handle, int recovery); +int paxos_catchup(pi_handle_t handle); + int paxos_propose(pi_handle_t handle, void *value, int round); int paxos_instance_exit(pi_handle_t handle); int paxos_space_exit(ps_handle_t handle); #endif /* _PAXOS_H */ diff --git a/src/paxos_lease.c b/src/paxos_lease.c index 8cdaf9c..1d6d4ec 100644 --- a/src/paxos_lease.c +++ b/src/paxos_lease.c @@ -1,583 +1,599 @@ /* * 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 "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 struct paxos_lease_msghdr { int leased; }; struct paxos_lease_value { char name[PAXOS_NAME_LEN+1]; int owner; int expiry; }; 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_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 void end_paxos_request(pi_handle_t handle, int round, int result) { struct paxos_lease *pl; int found = 0; list_for_each_entry(pl, &lease_head, list) { if (pl->pih == handle) { found = 1; break; } } if (!found) { log_error("cound not found the handle for paxos lease: %ld", handle); 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 ..."); if (!pl->release) { 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, 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; } strncpy(value.name, pl->name, PAXOS_NAME_LEN + 1); value.owner = myid; value.expiry = pl->expiry; 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 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; 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->release = 0; 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) { struct paxos_lease *pl = (struct paxos_lease *)handle; pl->release = 1; return 0; } -static int lease_catchup(const void *name) +static int lease_catchup(pi_handle_t handle) { struct paxos_lease *pl; struct paxos_lease_result plr; int found = 0; list_for_each_entry(pl, &lease_head, list) { - if (!strcmp(pl->name, name)) { + if (pl->pih == handle) { found = 1; break; } } if (!found) { - log_error("could not found the lease name (%s) " - "in registered list", (char *)name); + log_error("could not find the lease handle: %ld", handle); return -1; } - p_l_op->catchup(name, &pl->owner, &pl->expires); - log_debug("catchup result: name: %s, owner: %d, expires: %llu", - (char *)name, pl->owner, pl->expires); + 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); if (pl->owner == -1) return 0; 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_prepared(pi_handle_t handle __attribute__((unused)), void *header) { struct paxos_lease_msghdr *hdr = header; if (hdr->leased) { log_debug("already leased"); return 0; } else { log_debug("not leased"); return 1; } } static int handle_lease_request(pi_handle_t handle, void *header) { struct paxos_lease_msghdr *hdr; struct paxos_lease *pl; int found = 0; hdr = header; list_for_each_entry(pl, &lease_head, list) { if (pl->pih == handle) { found = 1; break; } } if (!found) { log_error("could not find the lease handle: %ld", handle); return -1; } if (pl->owner == -1) { log_debug("has not been leased"); hdr->leased = 0; } else { log_debug("has been leased"); hdr->leased = 1; } return 0; } static int lease_propose(pi_handle_t handle, void *extra __attribute__((unused)), int round, void *value) { struct paxos_lease *pl; int found = 0; list_for_each_entry(pl, &lease_head, list) { if (pl->pih == handle) { found = 1; break; } } if (!found) { log_error("could not find the lease handle: %ld", handle); 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; } return 0; } static int lease_accepted(pi_handle_t handle, void *extra __attribute__((unused)), int round, void *value) { struct paxos_lease *pl; int found = 0; list_for_each_entry(pl, &lease_head, list) { if (pl->pih == handle) { found = 1; break; } } if (!found) { log_error("could not find the lease handle: %ld", handle); 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)); 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; return 0; } static int lease_commit(pi_handle_t handle, void *extra __attribute__((unused)), int round) { struct paxos_lease *pl; struct paxos_lease_result plr; int found = 0; list_for_each_entry(pl, &lease_head, list) { if (pl->pih == handle) { found = 1; break; } } if (!found) { log_error("could not find the lease handle: %ld", handle); 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->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); return 0; } static int lease_learned(pi_handle_t handle, void *extra __attribute__((unused)), int round) { struct paxos_lease *pl; struct paxos_lease_result plr; int found = 0; list_for_each_entry(pl, &lease_head, list) { if (pl->pih == handle) { found = 1; break; } } if (!found) { log_error("could not find the lease handle: %ld", handle); 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; } 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); return 0; } 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_prepared; px_op->promise = handle_lease_request; 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/paxos_lease.h b/src/paxos_lease.h index 7c369f9..bb46ec1 100644 --- a/src/paxos_lease.h +++ b/src/paxos_lease.h @@ -1,66 +1,69 @@ /* * 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 */ #ifndef _PAXOS_LEASE_H #define _PAXOS_LEASE_H #define PLEASE_NAME_LEN 63 typedef long pl_handle_t; struct paxos_lease_result { char name[PLEASE_NAME_LEN+1]; int owner; + int ballot; unsigned long long expires; }; struct paxos_lease_operations { int (*get_myid) (void); int (*send) (unsigned long id, void *value, int len); int (*broadcast) (void *value, int len); - int (*catchup) (const void *name, int *owner, + int (*catchup) (const void *name, int *owner, int *ballot, unsigned long long *expires); int (*notify) (pl_handle_t handle, struct paxos_lease_result *result); }; 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); int paxos_lease_on_receive(void *msg, int msglen); int paxos_lease_acquire(pl_handle_t handle, int renew, void (*end_acquire) (pl_handle_t handle, int result)); /* int paxos_lease_owner_get(const void *name); int paxos_lease_epoch_get(const void *name); int paxos_lease_timeout(const void *name); */ +int paxos_lease_status_recovery(pl_handle_t handle); + int paxos_lease_release(pl_handle_t handle); int paxos_lease_exit(pl_handle_t handle); #endif /* _PAXOS_LEASE_H */ diff --git a/src/ticket.c b/src/ticket.c index fa09fed..2550b02 100644 --- a/src/ticket.c +++ b/src/ticket.c @@ -1,397 +1,415 @@ /* * 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 "ticket.h" #include "config.h" #include "pacemaker.h" #include "list.h" #include "log.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 { 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 result) { struct ticket *tk; int found = 0; if (result == 0) { list_for_each_entry(tk, &ticket_list, list) { if (tk->handle == handle) { tk->owner = ticket_get_myid(); found = 1; break; } } if (!found) log_error("BUG: ticket handle %ld does not exist", handle); log_info("ticket %s acquired", tk->id); log_info("ticket %s granted to local (id %d)", tk->id, ticket_get_myid()); } } 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; } -static int ticket_read(const void *name, int *owner, +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->expires); + pcmk_handler.load_ticket(tk->id, &tk->owner, &tk->ballot, &tk->expires); *owner = tk->owner; *expires = tk->expires; + *ballot = tk->ballot; return 0; } 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->expires); + 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->expires); + 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->expires); + 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, int force) { struct ticket *tk; int found = 0; if (force) { - pcmk_handler.store_ticket(ticket, ticket_get_myid(), -1); + pcmk_handler.store_ticket(ticket, ticket_get_myid(), 0, -1); pcmk_handler.grant_ticket(ticket); return BOOTHC_RLT_SYNC_SUCC; } 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, 1, end_acquire); return BOOTHC_RLT_ASYNC; } } int revoke_ticket(char *ticket, int force) { 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 (force) { - pcmk_handler.store_ticket(tk->id, -1, 0); + pcmk_handler.store_ticket(tk->id, -1, 0, 0); pcmk_handler.revoke_ticket(tk->id); } if (tk->owner == -1) return BOOTHC_RLT_SYNC_SUCC; else { paxos_lease_release(tk->handle); return BOOTHC_RLT_ASYNC; } } int list_ticket(char **pdata, unsigned int *len) { struct ticket *tk; char tmp[TK_LINE]; *pdata = NULL; *len = 0; list_for_each_entry(tk, &ticket_list, list) { memset(tmp, 0, TK_LINE); snprintf(tmp, TK_LINE, "ticket: %s, owner: %d, expires: %llu\n", tk->id, tk->owner, tk->expires); *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; } const struct paxos_lease_operations ticket_operations = { .get_myid = ticket_get_myid, .send = ticket_send, .broadcast = ticket_broadcast, .catchup = ticket_read, .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; if (!tk->expiry) tk->expiry = DEFAULT_TICKET_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; }