EXTENSION = raftable

DATA = raftable--1.0.sql

raftable.so: raft/lib/libraft.a

raft/lib/libraft.a:

make -C raft

EXTRA_INSTALL = contrib/raftable

RAFT_PREFIX =$(HOME)/raft

RAFT_PREFIX = raft

overrideLDFLAGS += -L$(RAFT_PREFIX)/lib -Wl,-whole-archive -lraft -Wl,-no-whole-archive

overrideCFLAGS += -Wfatal-errors

overrideCPPFLAGS += -I$(RAFT_PREFIX)/include

Copyright (c) 2015-2016, Constantin S. Pan

All rights reserved.

Redistribution and use in source and binary forms, with or without

modification, are permitted provided that the following conditions are met:

1. Redistributions of source code must retain the above copyright notice, this

list of conditions and the following disclaimer.

2. Redistributions in binary form must reproduce the above copyright notice,

this list of conditions and the following disclaimer in the documentation

and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND

ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE

FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL

DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR

SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER

CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,

OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

overrideCFLAGS += -fpic -Wall -Wfatal-errors -O0 -g -pedantic -std=c99

overrideCPPFLAGS += -I. -Iinclude -DDEBUG

overrideHEART_LDFLAGS += -Llib -lraft -ljansson

AR = ar

ARFLAGS = -cru

.PHONY: all clean bindir objdir libdir

lib/libraft.a: obj/raft.o obj/util.o | libdir objdir

$(AR)$(ARFLAGS) lib/libraft.a obj/raft.o obj/util.o

all: lib/libraft.a bin/heart

@echo Done.

bin/heart: obj/heart.o lib/libraft.a | bindir objdir

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

obj/heart.o$(HEART_LDFLAGS)

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

$(CC)$(CFLAGS)$(CPPFLAGS) -c -o$@$<

obj/%.o: example/%.c | objdir

$(CC)$(CFLAGS)$(CPPFLAGS) -c -o$@$<

bindir:

mkdir -p bin

objdir:

mkdir -p obj

libdir:

mkdir -p lib

clean:

rm -rfv bin obj lib

libraft

=======

Raft protocol implementation in C.

Features

--------

+ Leader Election

+ Log Replication

+ Log Compaction

TODO

----

+ Membership Changes

#define_POSIX_C_SOURCE 2

staticvoidapplier(void*state,raft_update_tupdate,raft_bool_tsnapshot) {

json_error_terror;

json_t*patch=json_loadb(update.data,update.len,0,&error);

if (!patch) {

shout(

"error parsing json at position %d: %s\n",

error.column,

error.text

);

}

if (snapshot) {

json_object_clear(state);

}

if (json_object_update(state,patch)) {

shout("error updating state\n");

}

json_decref(patch);

char*encoded=json_dumps(state,JSON_INDENT(4) |JSON_SORT_KEYS);

if (encoded) {

debug(

"applied %s: new state is %s\n",

snapshot ?"a snapshot" :"an update",

);

}else {

shout(

"applied %s, but new state could not be encoded\n",

snapshot ?"a snapshot" :"an update"

);

}

free(encoded);

}

staticraft_update_tsnapshooter(void*state) {

raft_update_tshot;

shot.data=json_dumps(state,JSON_SORT_KEYS);

shot.len=strlen(shot.data);

if (shot.data) {

debug("snapshot taken: %.*s\n",shot.len,shot.data);

}else {

shout("failed to take a snapshot\n");

}

returnshot;

}

staticvoiddie(intsignum) {

debug("dying\n");

exit(signum);

}

staticvoidusage(char*prog) {

printf(

" -l : Run as a daemon and write output to LOGFILE.\n",

);

}

raft_traft;

staticvoidmain_loop(char*host,intport) {

mstimer_tt;

mstimer_reset(&t);

intr=raft_create_udp_socket(raft);

if (r==NOBODY) {

die(EXIT_FAILURE);

}

#defineEMIT_EVERY_MS 1000

intemit_ms=0;

while (true) {

intms;

raft_msg_tm;

ms=mstimer_reset(&t);

raft_tick(raft,ms);

m=raft_recv_message(raft);

if (m) {

raft_handle_message(raft,m);

}

if (raft_is_leader(raft)) {

emit_ms+=ms;

while (emit_ms>EMIT_EVERY_MS) {

emit_ms-=EMIT_EVERY_MS;

charbuf[1000];

charkey='a'+rand() %5;

intvalue=rand() %10000;

sprintf(buf,"{\"key-%c\":%d}",key,value);

shout("emit update: = %s\n",buf);

raft_update_tupdate= {strlen(buf),buf,NULL};

raft_emit(raft,update);

}

}

}

close(r);

}

intmain(intargc,char**argv) {

char*logfilename=NULL;

booldaemonize= false;

intmyid=NOBODY;

intid;

char*host;

char*str;

intport;

intopt;

char*myhost=NULL;

intmyport;

json_t*state=json_object();

raft_config_trc;

rc.peernum_max=64;

rc.heartbeat_ms=20;

rc.election_ms_min=150;

rc.election_ms_max=300;

rc.log_len=10;

rc.chunk_len=4;

rc.msg_len_max=500;

rc.userdata=state;

rc.applier=applier;

rc.snapshooter=snapshooter;

raft=raft_init(&rc);

intpeernum=0;

while ((opt=getopt(argc,argv,"hi:r:l:"))!=-1) {

switch (opt) {

case'i':

myid=atoi(optarg);

break;

case'r':

if (myid==NOBODY) {

usage(argv[0]);

returnEXIT_FAILURE;

}

str=strtok(optarg,":");

if (str) {

id=atoi(str);

}else {

usage(argv[0]);

returnEXIT_FAILURE;

}

host=strtok(NULL,":");

str=strtok(NULL,":");

if (str) {

port=atoi(str);

}else {

usage(argv[0]);

returnEXIT_FAILURE;

}

if (!raft_peer_up(raft,id,host,port,id==myid)) {

usage(argv[0]);

returnEXIT_FAILURE;

}

if (id==myid) {

myhost=host;

myport=port;

}

peernum++;

break;

case'l':

logfilename=optarg;

daemonize= true;

break;

case'h':

usage(argv[0]);

returnEXIT_SUCCESS;

default:

usage(argv[0]);

returnEXIT_FAILURE;

}

}

if (!myhost) {

usage(argv[0]);

returnEXIT_FAILURE;

}

if (logfilename) {

if (!freopen(logfilename,"a",stdout)) {

returnEXIT_FAILURE;

}

if (!freopen(logfilename,"a",stderr)) {

returnEXIT_FAILURE;

}

}

if (daemonize) {

if (daemon(true, true)==-1) {

shout("could not daemonize: %s\n",strerror(errno));

returnEXIT_FAILURE;

}

}

signal(SIGTERM,die);

signal(SIGINT,die);

main_loop(myhost,myport);

returnEXIT_SUCCESS;

}