The commands:

'h': hello()

The first message.

The arbiter replies with:

[RES_OK] if ready

[RES_FAILED] (or disconnection) if not ready

'r': reserve(minxid, minsize)

Claims a sequence ≥ minsize of xids ≥ minxid for local usage. This will

prevent the arbiter from using those values for global transactions.

The reply and 'wait' logic is the same as for the 'status' command.

'h': snapshot(xid)

't': snapshot(xid)

Tells the arbiter to generate a snapshot for the global transaction

identified by the given 'xid'. The arbiter will create a snapshot for

every participant, so when each of them asks for the snapshot it will

CFLAGS=-g -O2 -Wall -Iinclude -D_LARGEFILE64_SOURCE# -DDEBUG

SOCKHUB_PREFIX=../sockhub

SOCKHUB_CFLAGS=-I"$(SOCKHUB_PREFIX)"

SOCKHUB_LDFLAGS=-lsockhub -L"$(SOCKHUB_PREFIX)"

SYSTEM=$(shell uname -s)

ifeq ($(SYSTEM),Darwin)

.PHONY: all clean check bindir objdir

all: bin/dtmd

all: bin/dtmd bin/heart

@echo Done.

@echo Feel free to run the tests with\'make check\'.

bin/dtmd: obj/server.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o obj/ddd.o | bindir objdir

bin/dtmd: obj/server.o obj/raft.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o obj/ddd.o | bindir objdir

$(CC) -o bin/dtmd$(CFLAGS)\

obj/server.o obj/main.o\

obj/server.o obj/raft.o obj/main.o\

obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o\

obj/snapshot.o obj/ddd.o\

obj/snapshot.o obj/ddd.o

bin/heart: obj/heart.o obj/raft.o obj/util.o | bindir objdir

$(CC) -o bin/heart$(CFLAGS)\

obj/heart.o obj/raft.o obj/util.o\

obj/snapshot.o\

obj/server.o: src/server.c | objdir

#defineXIDS_PER_TERM 1000000

#defineNEW_TERM_THRESHOLD 100000

#defineMAX_TRANSACTIONS 4096

#defineBUFFER_SIZE (64 * 1024)

#defineLISTEN_QUEUE_SIZE 100

#defineMAX_STREAMS 4096

#defineMAX_SERVERS 16

#defineHEARTBEAT_TIMEOUT_MS 20

#defineELECTION_TIMEOUT_MS_MIN 150

#defineELECTION_TIMEOUT_MS_MAX 300

#defineRAFT_LOGLEN 1024

#defineRAFT_KEEP_APPLIED 512 // how many applied entries to keep during compaction

#defineCMD_HELLO 'h'

#defineCMD_RESERVE 'r'

#defineCMD_BEGIN 'b'

#defineCMD_FOR 'y'

#defineCMD_AGAINST 'n'

#defineCMD_SNAPSHOT 'h'

#defineCMD_SNAPSHOT 't'

#defineCMD_STATUS 's'

#defineCMD_DEADLOCK 'd'

#defineRES_FAILED 0xDEADBEEF

#defineRES_OK 0xC0FFEE

#defineRES_REDIRECT 404

#defineRES_DEADLOCK 0xDEADDEED

#defineRES_TRANSACTION_COMMITTED 1

#defineRES_TRANSACTION_ABORTED 2

#defineNOBODY -1

#defineDEFAULT_LISTENHOST "0.0.0.0"

#defineDEFAULT_LISTENPORT 5431

#defineROLE_FOLLOWER 0

#defineROLE_CANDIDATE 1

#defineROLE_LEADER 2

#ifRAFT_KEEP_APPLIED >=RAFT_LOGLEN

#error please ensure RAFT_KEEP_APPLIED < RAFT_LOGLEN

#ifHEARTBEAT_TIMEOUT_MS >=ELECTION_TIMEOUT_MS_MIN

#error please ensure HEARTBEAT_TIMEOUT_MS < ELECTION_TIMEOUT_MS_MIN (considerably)

#ifELECTION_TIMEOUT_MS_MIN >=ELECTION_TIMEOUT_MS_MAX

#error please ensure ELECTION_TIMEOUT_MS_MIN < ELECTION_TIMEOUT_MS_MAX

typedefstructraft_entry_t {

intterm;

boolsnapshot;// true if this is a snapshot entry

union {

struct {// snapshot == false

intaction;

intargument;

};

struct {// snapshot == true

intminarg;

intmaxarg;

};

};

}raft_entry_t;

typedefvoid (*raft_applier_t)(intaction,intargument);

typedefstructraft_log_t {

intfirst;

intsize;// number of entries past first

intacked;// number of entries replicated to the majority of servers

intapplied;// number of entries applied to the state machine

raft_entry_tentries[RAFT_LOGLEN];// wraps around

}raft_log_t;

typedefstructraft_server_t {

intseqno;// the rpc sequence number

inttosend;// index of the next entry to send

intacked;// index of the highest entry known to be replicated

char*host;

intport;

structsockaddr_inaddr;

}raft_server_t;

typedefstructraft_t {

intterm;// current term (latest term we have seen)

intvote;// who received our vote in current term

introle;

intme;// my id

intvotes;// how many votes are for me (if candidate)

intleader;// the id of the leader

raft_log_tlog;

intsock;

intservernum;

raft_server_tservers[MAX_SERVERS];

inttimer;

raft_applier_tapplier;

}raft_t;

#defineRAFT_LOG(RAFT,INDEX) ((RAFT)->log.entries[(INDEX) % (RAFT_LOGLEN)])

#defineRAFT_MSG_UPDATE 0 // append entry

#defineRAFT_MSG_DONE 1 // entry appended

#defineRAFT_MSG_CLAIM 2 // vote for me

#defineRAFT_MSG_VOTE 3 // my vote

typedefstructraft_msg_t {

intmsgtype;

intterm;

intfrom;

intseqno;

}raft_msg_t;

typedefstructraft_msg_update_t {

raft_msg_tmsg;

intprevindex;// the index of the preceding log entry

intprevterm;// the term of the preceding log entry

boolempty;// the message is just a heartbeat if empty

raft_entry_tentry;

intacked;// the leader's acked number

}raft_msg_update_t;

typedefstructraft_msg_done_t {

raft_msg_tmsg;

intindex;// the index of the appended entry

intterm;// the term of the appended entry

boolsuccess;

}raft_msg_done_t;

typedefstructraft_msg_claim_t {

raft_msg_tmsg;

intindex;// the index of my last entry

intterm;// the term of my last entry

}raft_msg_claim_t;

typedefstructraft_msg_vote_t {

raft_msg_tmsg;

boolgranted;

}raft_msg_vote_t;

voidraft_init(raft_t*r);

boolraft_add_server(raft_t*r,char*host,intport);

boolraft_set_myid(raft_t*r,intmyid);

boolraft_emit(raft_t*r,intaction,intargument);

intraft_apply(raft_t*r,raft_applier_tapplier);

voidraft_tick(raft_t*r,intmsec);

voidraft_handle_message(raft_t*r,raft_msg_t*m);

raft_msg_t*raft_recv_message(raft_t*r);

intraft_create_udp_socket(raft_t*r);

voidraft_start_next_term(raft_t*r);

);

voidserver_set_raft_socket(server_tserver,intsock);

boolserver_start(server_tserver);

boolserver_tick(server_tserver,inttimeout_ms);

voidserver_disable(server_tserver);

voidserver_loop(server_tserver);

voidserver_enable(server_tserver);

voidserver_set_enabled(server_tserver,boolenable);

boolclient_message_shortcut(client_tclient,xid_targ);

boolclient_redirect(client_tclient,unsignedaddr,intport);

unsignedclient_get_ip_addr(client_tclient);

typedefstructSnapshot {

xid_txmin;

#defineMAX_SNAPSHOTS_PER_TRANS 8

voidtransaction_clear(Transaction*t);

voidtransaction_push_listener(Transaction*t,charcmd,void*listener);

void*transaction_pop_listener(Transaction*t,charcmd);

booltransaction_participate(Transaction*t,intclientid);

booltransaction_remove_listener(Transaction*t,charcmd,void*listener);

char*join_path(constchar*dir,constchar*file);

boolinrange(xid_tmin,xid_tx,xid_tmax);

intfalloc(intfd,off64_tsize);

staticinlineintmin(inta,intb) {

returna<b ?a :b;

}

staticinlineintmax(inta,intb) {

returna>b ?a :b;

}

staticinlineintrand_between(intmin,intmax) {

returnrand() % (max-min+1)+min;

}

typedefstructmstimer_t {

structtimevaltv;

}mstimer_t;

intmstimer_reset(mstimer_t*t);

structtimevalms2tv(intms);

#definedebug(...)