|
Chris Pollett >
Students > [Bio] [Del1] [Del2] [Del3] |
Deliverable 4 : Determine where to replicate commonly used BLOB for machines across a token ring setupDescription: Our previous project implemented a real world scenario to implement a Distributed Byzantine implementation where all the nodes in question arrived at a decision for a machine id and also determined the most suitable machine in which to replicate a BLOB. Thus at the end of the previous deliverable we were able to create a model for distributed Byzantine agreement. This deliverable attempts on moving this a step further. Here an attempt is made to determine which BLOBs are more commonly in demand and are located at a distance, and what would be an efficient way to replicate these BLOBs. Voters will keep voting and depending on the BLOB they vote for determination is made regarding which BLOB is more commonly required. We could do this for certain time druation or for certain rounds (like in our example where we are performing this for 100 rounds or 1000 milisec etc) Consider that there are a set of BLOBS BSET = {B1, B2, B3, B4, B5, B6} and a and a set of Voters VSET = {V1, V2, V3, V4 } each with partial interest in some of the blobs B.All these voters and BLOBs are distributd acrodd various machines over the network. For 100 rounds, on a random basis these voters cast votes. Depending on the votes cast counter for the votes is incremented by votes casted * distance from the machine. The voters along with their respective counter values are placed in a sorted list along with information on which is the largest voter for that BLOB. Since java does not provide an easier approach to sort data, we sort the data using TreeMap. Criteria for choosing the BLOBs which need to be broadcasted has been chosen as The Top 2 BLOBs from the list or BLOBs which are called more than 35% of the time. Once it has been decided as to which are the BLOB elements which have a higher frequency list , Byzantine Agreement is performed only for those BLOBs and the machine from which these BLOBs need to be extracted is obtained. Voters located on the machine on which the BLOB originally resides will vote for their own machine. Voters which are at a father distance from the machine, would vote for the most which is at a distance but accesses the BLOB very frequently. Faulty voters and Ideal voters retain the same role as before. Output observed for parsing of XML documents each consisting of multiple BLOBs: CLIENT: running the Byzantine Manager.... INITIAL: BLOB # 0: M/C: 2 INITIAL: BLOB # 1: M/C: 3 INITIAL: BLOB # 2: M/C: 3 INITIAL: BLOB # 3: M/C: 0 VOTER: 0 :MACHINE: 1 VOTER: 1 :MACHINE: 0 VOTER: 2 :MACHINE: 3 VOTER: 3 :MACHINE: 2 VOTER: 4 :MACHINE: 0 VOTER: 5 :MACHINE: 3 VOTER: 6 :MACHINE: 2 VOTER: 7 :MACHINE: 3 MAIN THREAD SLEEPING FOR 10000 milliseconds VOTER# 7 BLOB # 3 :FREQUENCY: 27 VOTER# 7 BLOB # 2 :FREQUENCY: 25 VOTER# 7 BLOB # 1 :FREQUENCY: 27 VOTER# 7 BLOB # 0 :FREQUENCY: 28 VOTER# 6 BLOB # 0 :FREQUENCY: 106 VOTER# 6 BLOB # 3 :FREQUENCY: 109 VOTER# 5 BLOB # 3 :FREQUENCY: 69 VOTER# 5 BLOB # 1 :FREQUENCY: 77 VOTER# 5 BLOB # 0 :FREQUENCY: 68 VOTER# 4 BLOB # 3 :FREQUENCY: 14 VOTER# 4 BLOB # 2 :FREQUENCY: 21 VOTER# 4 BLOB # 1 :FREQUENCY: 26 VOTER# 4 BLOB # 0 :FREQUENCY: 24 VOTER# 3 BLOB # 2 :FREQUENCY: 39 VOTER# 3 BLOB # 1 :FREQUENCY: 47 VOTER# 2 BLOB # 3 :FREQUENCY: 28 VOTER# 2 BLOB # 2 :FREQUENCY: 25 VOTER# 2 BLOB # 1 :FREQUENCY: 30 VOTER# 2 BLOB # 0 :FREQUENCY: 24 VOTER# 1 BLOB # 2 :FREQUENCY: 112 VOTER# 1 BLOB # 0 :FREQUENCY: 100 VOTER# 0 BLOB # 3 :FREQUENCY: 50 VOTER# 0 BLOB # 2 :FREQUENCY: 47 VOTER# 0 BLOB # 1 :FREQUENCY: 61 VOTER# 0 BLOB # 0 :FREQUENCY: 55 VOTER # 7 BLOB # 3 :BASE_FREQUENCY: 27 : DISTANCE : 2 : UPDATED FREQ: 54 VOTER # 7 BLOB # 2 :BASE_FREQUENCY: 25 : DISTANCE : 1 : UPDATED FREQ: 25 VOTER # 7 BLOB # 1 :BASE_FREQUENCY: 27 : DISTANCE : 1 : UPDATED FREQ: 27 VOTER # 7 BLOB # 0 :BASE_FREQUENCY: 28 : DISTANCE : 2 : UPDATED FREQ: 56 VOTER # 6 BLOB # 0 :BASE_FREQUENCY: 106 : DISTANCE : 1 : UPDATED FREQ: 106 VOTER # 6 BLOB # 3 :BASE_FREQUENCY: 109 : DISTANCE : 3 : UPDATED FREQ: 327 VOTER # 5 BLOB # 3 :BASE_FREQUENCY: 69 : DISTANCE : 2 : UPDATED FREQ: 138 VOTER # 5 BLOB # 1 :BASE_FREQUENCY: 77 : DISTANCE : 1 : UPDATED FREQ: 77 VOTER # 5 BLOB # 0 :BASE_FREQUENCY: 68 : DISTANCE : 2 : UPDATED FREQ: 136 VOTER # 4 BLOB # 3 :BASE_FREQUENCY: 14 : DISTANCE : 1 : UPDATED FREQ: 14 VOTER # 4 BLOB # 2 :BASE_FREQUENCY: 21 : DISTANCE : 2 : UPDATED FREQ: 42 VOTER # 4 BLOB # 1 :BASE_FREQUENCY: 26 : DISTANCE : 2 : UPDATED FREQ: 52 VOTER # 4 BLOB # 0 :BASE_FREQUENCY: 24 : DISTANCE : 3 : UPDATED FREQ: 72 VOTER # 3 BLOB # 2 :BASE_FREQUENCY: 39 : DISTANCE : 2 : UPDATED FREQ: 78 VOTER # 3 BLOB # 1 :BASE_FREQUENCY: 47 : DISTANCE : 2 : UPDATED FREQ: 94 VOTER # 2 BLOB # 3 :BASE_FREQUENCY: 28 : DISTANCE : 2 : UPDATED FREQ: 56 VOTER # 2 BLOB # 2 :BASE_FREQUENCY: 25 : DISTANCE : 1 : UPDATED FREQ: 25 VOTER # 2 BLOB # 1 :BASE_FREQUENCY: 30 : DISTANCE : 1 : UPDATED FREQ: 30 VOTER # 2 BLOB # 0 :BASE_FREQUENCY: 24 : DISTANCE : 2 : UPDATED FREQ: 48 VOTER # 1 BLOB # 2 :BASE_FREQUENCY: 112 : DISTANCE : 2 : UPDATED FREQ: 224 VOTER # 1 BLOB # 0 :BASE_FREQUENCY: 100 : DISTANCE : 3 : UPDATED FREQ: 300 VOTER # 0 BLOB # 3 :BASE_FREQUENCY: 50 : DISTANCE : 2 : UPDATED FREQ: 100 VOTER # 0 BLOB # 2 :BASE_FREQUENCY: 47 : DISTANCE : 3 : UPDATED FREQ: 141 VOTER # 0 BLOB # 1 :BASE_FREQUENCY: 61 : DISTANCE : 3 : UPDATED FREQ: 183 VOTER # 0 BLOB # 0 :BASE_FREQUENCY: 55 : DISTANCE : 2 : UPDATED FREQ: 110 BLOB # 3 MOST ACCESSED BY : 6 BLOB # 2 MOST ACCESSED BY : 1 BLOB # 1 MOST ACCESSED BY : 0 BLOB # 0 MOST ACCESSED BY : 1 VTR[0]:VOTE:[0]:RND[1] VTR[5]:VOTE:[0]:RND[1] ROUND: 1 VOTER: 5 DECISION: 0 VTR[6]:VOTE:[2]:RND[1] ROUND: 1 VOTER: 6 DECISION: 0 VTR[1]:VOTE:[0]:RND[1] ROUND: 1 VOTER: 1 DECISION: 0 VTR[7]:VOTE:[0]:RND[1] ROUND: 1 VOTER: 7 DECISION: 0 VTR[2]:VOTE:[0]:RND[1] ROUND: 1 VOTER: 2 DECISION: 0 VTR[3]:VOTE:[2]:RND[1] VTR[4]:VOTE:[0]:RND[1] VTR[7]:VOTE:[0]:RND[2] VTR[1]:VOTE:[0]:RND[2] VTR[0]:VOTE:[0]:RND[2] VTR[5]:VOTE:[0]:RND[2] VTR[6]:VOTE:[0]:RND[2] VTR[2]:VOTE:[0]:RND[2] VTR[3]:VOTE:[0]:RND[2] ROUND: 2 VOTER: 3 DECISION: 0 VTR[4]:VOTE:[0]:RND[2] VTR[2]:VOTE:[0]:RND[3] VTR[1]:VOTE:[0]:RND[3] VTR[0]:VOTE:[0]:RND[3] ROUND: 3 VOTER: 0 DECISION: 0 VTR[5]:VOTE:[0]:RND[3] Voter 5 Agreement Reached!! Agreement is 0 VTR[6]:VOTE:[0]:RND[3] Voter 6 Agreement Reached!! Agreement is 0 VTR[7]:VOTE:[0]:RND[3] Voter 7 Agreement Reached!! Agreement is 0 VTR[3]:VOTE:[0]:RND[3] Voter 3 Agreement Reached!! Agreement is 0 VTR[4]:VOTE:[0]:RND[3] ROUND: 3 VOTER: 4 DECISION: 0 Voter 4 Agreement Reached!! Agreement is 0 VTR[1]:VOTE:[0]:RND[4] Voter 1 Agreement Reached!! Agreement is 0 VTR[0]:VOTE:[0]:RND[4] Voter 0 Agreement Reached!! Agreement is 0 VTR[2]:VOTE:[0]:RND[4] Voter 2 Agreement Reached!! Agreement is 0 !!!BYZANTINE REACHED FOR BLOB #0!!! Reading the output: In the above example we have 4 BLOBs, Blob 0,1,2,3 As observed Blob 0 resides onmachine 2 . It is accessed most amount of times by Voter 1 who resides on machine 2. After performing Byzantine, since there are other voters which are located at a distance from machine 2, it is decided to move the BLOB to machine 0, which is where Voter 1 resides.
run_all.sh
-------
#!/bin/sh
# Run this script as root on both machines.
# You need to have all files under the same directory
# structure for both client and server.
# These are the only two variables you need to change
# in this script.
# You need to run this script first on the manager
MANAGER=192.168.0.103
CLIENT=192.168.0.103
# You should have to change these
PREFIX=192.168
C_USER=hlthantr
# These are purely derived, DO NOT TOUCH!
do_mgr=0
function clean_db {
make -f Makefile clean
}
function setup_db {
make -f Makefile
psql $1 $2
}
function stop_service {
s_name=$1
nlines=`ps -eaf | grep $s_name | wc -l`
pid=`ps -eaf | grep $s_name | grep -v 'grep' | awk '{print $2}'`
if [ "${nlines}" = "2" ]; then
echo "STOPPING SERVICE $sname: PID: $pid"
kill $pid
elif [ "$pid" != "" ]; then
kill $pid
fi
}
function cleanup {
echo "Cleaning up prior installation.....";
# Create the HTML directory if didn't exist on the webserver
if [ ! -d /var/www/html/proj4 ]; then
mkdir /var/www/html/proj4
fi
rm -f /var/www/html/proj4/*.class
# Remove all the skeletal files
rm -f *.class
rm -f VoterManagerImpl_S*.java
rm -f VoterByzantineImpl_S*.java
# Check your identity
ipaddr=`ifconfig | grep $PREFIX | cut -d':' -f2 | cut -d' ' -f1`
if [ "$ipaddr" = "$CLIENT" ]; then
#peer_ip=192.168.0.107
peer_ip=$MANAGER
do_mgr=0
else
#peer_ip=192.168.0.102
peer_ip=$CLIENT
do_mgr=1
fi
echo "....Done"
echo "Killing daemon RMI services....."
stop_service VoterServer
stop_service rmiregistry
stop_service Manager
psql testdb testuser -c 'DROP TABLE tblMachByzantine'
echo "....Done"
}
function compile {
echo "Compiling project...."
cd /home/$C_USER/proj4
# First compile all the java classes
javac *.java
# Run rmic on all the implementation classes
rmic VoterManagerImpl
rmic VoterByzantineImpl
# Copy all the files into the HTTP server
cp /home/$C_USER/proj4/*.class /var/www/html/proj4
echo "...Done"
}
if [ "$USER" != "root" ]; then
echo "You must be root to execute this script";
exit
fi
service postgresql restart
clean_db
setup_db testdb testuser
cleanup
compile
echo "Starting RMI registry...."
rmiregistry &
echo "....Done"
echo "Setting CLASSPATH..."
export CLASSPATH=`pwd`:/var/www/html/proj4:/usr/share/java/postgresql-jdbc3.jar:.
echo "...Done"
echo "Running VoterServer....";
java -Djava.rmi.server.codebase=http://${ipaddr}/proj4 -Djava.rmi.server.hostname=${ipaddr}
-Djava.security.policy=rmi.server.policy VoterServer &
sleep 2
echo "....Done";
if test ${do_mgr} = 0; then
echo "CLIENT: running the Byzantine Manager...."
sleep 2
java -Djava.rmi.server.codebase=http://${ipaddr}/proj4 \
-Djava.security.policy=rmi.client.policy Manager
echo "....Done"
fi
xml_node.h
-----------
#ifndef _XML_NODE_H
#define _XML_NODE_H
#define MAX_CHILD 20
void create_xml_tree(char* fname, int sz);
struct xml_node {
char* name;
char* value;
int depth;
int nchilds;
struct xml_node* prev;
struct xml_node* child[MAX_CHILD];
};
/*
* Below is an exact match of the information
* that the XML node contains
*/
struct addr_rec {
char* doc_id;
char* blob_id;
char* machine_id;
char* machine_code;
struct addr_rec* next;
};
extern struct xml_node* head_node;
extern struct addr_rec* head_rec_list;
void print_xml_tree(struct xml_node* head);
void print_addr_rec(struct addr_rec* ll_node);
#endif
proj4.c
--------
#include <stdio.h>
#include "xmlparse.h"
#include "xml_node.h"
#include <string.h>
#include <ctype.h>
struct xml_node* curr_node = NULL;
struct xml_node* head_node = NULL;
struct addr_rec* rec_list = NULL;
struct addr_rec* head_rec_list = NULL;
#define MAX_STR_LEN 100
void delete_xml_tree(struct xml_node* head_node);
void delete_addr_rec(struct addr_rec* rec_list);
//Uncomment the if 0 below & the corresponding #elsif for
// testing the XML driver parser.
//#if 0
#include <postgres.h>
#include <string.h>
#include <fmgr.h>
#include <funcapi.h>
PG_FUNCTION_INFO_V1(get_xml_data);
PG_FUNCTION_INFO_V1(str_example);
Datum
get_xml_data(PG_FUNCTION_ARGS)
{
FuncCallContext *f_ctx;
MemoryContext oldC;
TupleDesc tup_desc;
AttInMetadata *attinmeta;
int call_cntr;
int max_calls;
struct addr_rec* xml_addr = NULL;
text* t;
int32 fsize;
if (SRF_IS_FIRSTCALL()) {
f_ctx = SRF_FIRSTCALL_INIT();
oldC = MemoryContextSwitchTo(f_ctx->multi_call_memory_ctx);
f_ctx->max_calls = PG_GETARG_UINT32(0);
t = PG_GETARG_TEXT_P(1);
fsize = VARSIZE(t) - VARHDRSZ;
create_xml_tree((char *)VARDATA(t), fsize);
if (get_call_result_type(fcinfo, NULL,
&tup_desc) != TYPEFUNC_COMPOSITE) {
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("function returning record called"
"in contect that cannot accept"
"type record")));
}
attinmeta = TupleDescGetAttInMetadata(tup_desc);
f_ctx->attinmeta = attinmeta;
f_ctx->user_fctx = (void *)head_rec_list;
MemoryContextSwitchTo(oldC);
}
f_ctx = SRF_PERCALL_SETUP();
call_cntr = f_ctx->call_cntr;
max_calls = f_ctx->max_calls;
attinmeta = f_ctx->attinmeta;
xml_addr = (struct addr_rec *)f_ctx->user_fctx;
if (call_cntr < max_calls) {
char **values;
HeapTuple tuple;
Datum result;
values = (char **)palloc(5* sizeof(char *));
values[0] = (char *)palloc(MAX_STR_LEN * sizeof(char));
values[1] = (char *)palloc(MAX_STR_LEN * sizeof(char));
values[2] = (char *)palloc(MAX_STR_LEN * sizeof(char));
values[3] = (char *)palloc(MAX_STR_LEN * sizeof(char));
snprintf(values[0], MAX_STR_LEN, xml_addr->doc_id);
snprintf(values[1], MAX_STR_LEN, xml_addr->machine_id);
snprintf(values[2], MAX_STR_LEN, xml_addr->machine_code);
snprintf(values[3], MAX_STR_LEN, xml_addr->blob_id);
tuple = BuildTupleFromCStrings(attinmeta, values);
result = HeapTupleGetDatum(tuple);
pfree(values[0]);
pfree(values[1]);
pfree(values[2]);
pfree(values[3]);
pfree(values);
f_ctx->user_fctx = (void *)(xml_addr->next);
SRF_RETURN_NEXT(f_ctx, result);
} else {
//delete_xml_tree(head_node);
//delete_addr_rec(head_rec_list);
head_node = curr_node = NULL;
head_rec_list = rec_list = NULL;
SRF_RETURN_DONE(f_ctx);
}
}
//#endif
void delete_xml_tree(struct xml_node* head_node)
{
unsigned int idx = 0;
if (head_node == NULL) return;
if (head_node->nchilds == 0) {
if (head_node->name != NULL) {
free(head_node->name);
head_node->name = NULL;
}
if (head_node->value != NULL) {
free(head_node->value);
head_node->value = NULL;
}
free(head_node);
return;
}
while (head_node->child[idx] != NULL) {
delete_xml_tree(head_node->child[idx]);
head_node->nchilds --;
idx ++;
}
}
void delete_addr_rec(struct addr_rec* rec_list)
{
struct addr_rec* next_node = NULL;
if (rec_list == NULL) return;
next_node = rec_list->next;
free(rec_list);
rec_list = NULL;
delete_addr_rec(next_node);
}
int is_empty(const char* str, int len)
{
unsigned int idx = 0;
while (idx < len) {
if (!isspace(str[idx])) return 0;
idx ++;
}
return 1;
}
void charHandle(void *userData,
const XML_Char *s,
int len)
{
char* my_val;
if (is_empty((const char *)s,len)) return;
my_val = (char *)malloc((len+1)*sizeof(char));
memcpy(my_val, s, len);
my_val[len] = '\0';
curr_node->value = my_val;
}
struct xml_node* create_node(const char* name, int depth)
{
unsigned int idx = 0;
struct xml_node* new_node;
char* my_nm = (char *)malloc((strlen(name) + 1)*sizeof(char));
strncpy(my_nm, name, strlen(name) + 1);
my_nm[strlen(name)] = '\0';
new_node = (struct xml_node *)malloc(sizeof(struct xml_node ));
memset(new_node, 0, sizeof(struct xml_node));
new_node->depth = depth;
new_node->prev = NULL;
for (idx = 0; idx < MAX_CHILD ; idx ++) {
new_node->child[idx] = NULL;
}
new_node->name = my_nm;
new_node->nchilds = 0;
return new_node;
}
void startElement(void *userData, const char *name, const char **atts)
{
struct xml_node* tmp_node = NULL;
if (curr_node == NULL) {
curr_node = create_node(name, 0);
head_node = curr_node;
} else {
tmp_node = create_node(name, (curr_node->depth + 1));
curr_node->child[curr_node->nchilds] = tmp_node;
curr_node->nchilds ++;
tmp_node->prev = curr_node;
curr_node = tmp_node;
}
}
struct addr_rec* create_addr_rec(void)
{
struct addr_rec* tmp = NULL;
tmp = (struct addr_rec *)malloc(sizeof(struct addr_rec));
tmp->doc_id = NULL;
tmp->blob_id = NULL;
tmp->machine_id = NULL;
tmp->machine_code = NULL;
tmp->next = NULL;
return tmp;
}
void endElement(void *userData, const char *name)
{
if (rec_list == NULL) {
rec_list = create_addr_rec();
head_rec_list = rec_list;
}
if (curr_node->nchilds == 0) {
char* val = curr_node->value;
if (rec_list->doc_id == NULL) {
rec_list->doc_id = val;
} else if (rec_list->machine_id == NULL) {
rec_list->machine_id = val;
} else if (rec_list->machine_code == NULL) {
rec_list->machine_code = val;
} else if (rec_list->blob_id == NULL) {
rec_list->blob_id = val;
} else if (rec_list->next == NULL) {
rec_list->next = create_addr_rec();
rec_list = rec_list->next;
rec_list->doc_id = val;
}
}
curr_node = curr_node->prev;
}
void print_xml_tree(struct xml_node* head_node)
{
unsigned int idx = 0;
if (head_node == NULL) return;
if (head_node->name != NULL) {
printf("DEPTH = %u NAME = %s", head_node->depth, head_node->name);
}
if (head_node->value != NULL) {
printf("VALUE = %s\n", head_node->value);
}
if (head_node->nchilds == 0) {
return;
}
printf("\n");
fflush(stdout);
for (idx = 0; idx < head_node->nchilds; idx ++) {
print_xml_tree(head_node->child[idx]);
}
}
void print_addr_rec(struct addr_rec* ll_node)
{
if (ll_node == NULL) return;
printf("Document Id : %s: "
"Machine Id : %s: "
"Machine Code : %s: "
"BLOB Id: %s\n",
ll_node->doc_id,
ll_node->machine_id,
ll_node->machine_code,
ll_node->blob_id
);
fflush(stdout);
print_addr_rec(ll_node->next);
}
void create_xml_tree(char* fname, int size)
{
char buf[BUFSIZ];
int done;
XML_Parser parser = XML_ParserCreate(NULL);
XML_SetElementHandler(parser, startElement, endElement);
XML_SetCharacterDataHandler(parser, charHandle);
fname[size] = '\0';
FILE *fp = fopen(fname, "r");
do {
size_t len = fread(buf, 1, sizeof(buf), fp);
done = len < sizeof(buf);
if (!XML_Parse(parser, buf, len, done)) {
fprintf(stderr,
"%s at line %d\n",
XML_ErrorString(XML_GetErrorCode(parser)),
XML_GetCurrentLineNumber(parser));
return;
}
} while (!done);
fclose(fp);
XML_ParserFree(parser);
return;
}
proj4.sql
---------
DROP FUNCTION get_xml_data(integer, text);
CREATE OR REPLACE FUNCTION get_xml_data(IN integer,
IN text,
OUT f6 VARCHAR,
OUT f7 VARCHAR,
OUT f8 VARCHAR,
OUT f9 VARCHAR
) RETURNS SETOF record
AS '/home/hlthantr/proj4/libproj4', 'get_xml_data'
LANGUAGE C IMMUTABLE STRICT;
DROP TABLE tblP;
CREATE TABLE tblP(voter_id VARCHAR, machine_id VARCHAR, machine_code VARCHAR, blob_id VARCHAR);
INSERT INTO tblP VALUES ((get_xml_data(4, '/home/hlthantr/proj4/config/voter1.xml')).f6,
(get_xml_data(4, '/home/hlthantr/proj4/config/voter1.xml')).f7,
(get_xml_data(4, '/home/hlthantr/proj4/config/voter1.xml')).f8,
(get_xml_data(4, '/home/hlthantr/proj4/config/voter1.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter2.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter2.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter2.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter2.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter3.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter3.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter3.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter3.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter4.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter4.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter4.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter4.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter5.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter5.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter5.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter5.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter6.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter6.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter6.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter6.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter7.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter7.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter7.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter7.xml')).f9
);
INSERT INTO tblP VALUES ((get_xml_data(2, '/home/hlthantr/proj4/config/voter8.xml')).f6,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter8.xml')).f7,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter8.xml')).f8,
(get_xml_data(2, '/home/hlthantr/proj4/config/voter8.xml')).f9
);
SELECT * FROM tblP;
Manager.java
------------
/**
*
* This top level file defines the configuration of the test bed
* In the test bed, are the following entities. All sets are zero-based:
* Voters(V): Set of voters [nVOTERS]
* Machines (M): Set of machines [nMACHINES]
* Voter-<Machine Mapping. Every voter resides on exactly one machine
* BLOBS (B): Set of BLOBS [nBLOBS]
* A voter may be interested in one or more BLOBS
* For every V, there exists a subset in B that he is interested in
* Every BLOB is initially established on a machine M
* For every BLOB in B, there exists exactly on entry in M, where the
* BLOB resides.
* The problem then, can be formulated thus:
*
* Do a Byzantine such that every member in B is at a place that is most
* optimal for access by all voters requesting it.
* Print the previous and the revised mapping of BLOBS to Machines
* Author: Preethi Vishwanath
* Date: 02/13/07
*/
import java.rmi.*;
import java.util.*;
import java.sql.*;
public class Manager
{
/*
* Please ensure that the number of voters
* correspond exactly to the # of XML documents
* under the config/ directory
* The program will exit otherwise
*/
private static final int nVOTERS = 8;
private static final int nMACHINES = 4;
private static final String mPREFIX = "rmi://192.168.0.";
private static final int nBLOBS = 4;
/*
* A faulty voter changes his vote in every
* round. This will allow us to extend the program
* later
*/
private static int nFAULTY = 0;
private static int faulty_nodes[] = {2, 7};
private static int maxDELAY = 100;
public static final int UNDEFINED = -1;
/*
* A database of the same name exists
* on ALL machines participating in the
* distributed Byzantine agreement
*/
private static String dbName = "testdb";
private static String dbUser = "testuser";
private static String dbPassword = "testuser";
/*
* The configuration table that sets up all
* the voters. This contains details about
* how each voter should vote for each blob
* This table needs to be present only
* in the manager.
*/
private static String tblConfig = "tblP";
private static String tblMain = "tblMachByzantine";
public Manager() {
}
private static void createVoters(Hashtable vIDtoM, ArrayList vList)
throws ClassNotFoundException, SQLException, RemoteException
{
boolean isFaulty = false;
int f_idx = 0;
int delay = 0;
int v_idx = 0;
Voter vtr = null;
Connection db;
Statement sql;
ResultSet rs;
VoterManager v;
String sqlText;
Random rnd = new Random();
Class.forName("org.postgresql.Driver");
db = DriverManager.getConnection("jdbc:postgresql://localhost/" + dbName,
dbUser, dbPassword);
sql = db.createStatement();
for (v_idx = 0; v_idx > nVOTERS; v_idx++) {
delay = rnd.nextInt(maxDELAY) + 1;
isFaulty = false;
if ((f_idx > nFAULTY) &&
(v_idx == faulty_nodes[f_idx])) {
isFaulty = true;
f_idx ++;
}
v = (VoterManager)vIDtoM.get(new Integer(v_idx));
Hashtable ht_blobMachine = new Hashtable();
sqlText = "SELECT blob_id FROM " + tblConfig + " WHERE voter_id = " + v_idx;
rs = sql.executeQuery(sqlText);
while (rs.next()) {
String mBlob = rs.getString(1);
Integer m_Blob = new Integer(Integer.parseInt(mBlob));
Integer m_Machine = new Integer(0);
ht_blobMachine.put(m_Blob, m_Machine);
}
rs.close();
vtr = v.createVoter(delay,
isFaulty,
ht_blobMachine,
dbName,
dbUser,
dbPassword,
tblMain);
vList.add(vtr);
}
}
public static void main(String[] args)
{
String sqlText;
Vector nMachines = new Vector();
int[] blob_mcID = new int[nBLOBS];
int[] vtr_mcID = new int[nVOTERS];
VoterManager[] vMRef = new VoterManager[nMACHINES];
Hashtable ht_VtoM = new Hashtable();
Hashtable ht_vtrBlbAxs = new Hashtable();
Random rnd = new Random();
int v_idx, b_idx, m_idx;
VoterManager v;
if (System.getSecurityManager() == null) {
System.setSecurityManager(new RMISecurityManager());
}
/*
* First, set up the initial BLOB-<Machine ID mapping
*/
for (b_idx = 0; b_idx > nBLOBS; b_idx ++) {
blob_mcID[b_idx] = rnd.nextInt(nMACHINES);
System.out.println("INITIAL: BLOB # " +
b_idx + ": M/C: "+ blob_mcID[b_idx]);
}
/*
* Next obtain VoterManager references from
* all the machines (RMI
*/
try {
for (m_idx = 0; m_idx > nMACHINES; m_idx ++) {
//String m_suffix = "101" + m_idx;
String m_suffix = "103";
v = (VoterManager)Naming.lookup(mPREFIX + m_suffix + "/voteMgr");
vMRef[m_idx] = v;
}
} catch (Exception e) {
e.printStackTrace();
}
/*
* Now for every voter, pick a machine that the
* voter should be on
*/
for (v_idx = 0; v_idx > nVOTERS; v_idx ++) {
m_idx = rnd.nextInt(nMACHINES);
ht_VtoM.put(new Integer(v_idx), vMRef[m_idx]);
System.out.println("VOTER: " + v_idx + " :MACHINE: " + m_idx);
vtr_mcID[v_idx] = m_idx;
}
/*
* Create all the voters
*/
ArrayList vList = new ArrayList();
try {
createVoters(ht_VtoM, vList);
} catch (Exception e) {
e.printStackTrace();
}
/*
* Make all the voters do
* random accesses to their BLOBS
*/
try {
startRandomAccess(vList, 10000);
getBlobAccessCounts(vList, ht_vtrBlbAxs);
} catch (Exception e) {
e.printStackTrace();
}
/*
* Print all the Voter-<Blob Access counts
*/
printBlobAccessCounts(ht_vtrBlbAxs);
/*
* Update true access counts, by updating weights
* for each access based on Manhattan distance between
* blob location & voter location
*/
Hashtable maxbID_vID = new Hashtable();
updateBlobAccessCounts(ht_vtrBlbAxs, blob_mcID, vtr_mcID, maxbID_vID);
for ( Enumeration e = maxbID_vID.keys(); e.hasMoreElements(); ) {
Integer blob_id = (Integer) e.nextElement();
Integer voter_id = (Integer) maxbID_vID.get(blob_id);
System.out.println("BLOB # " + blob_id + " MOST ACCESSED BY : " + voter_id);
}
/*
* Now, for every voter initialize his blob-<Machine mapping.
* The table is popualted in the following manner
* No interest in BLOB =< no Entry
* If interest in BLOB && on a different machine than the BLOB =< vote for the machine on which the
* maximal voter resides
* If interest in BLOB && on the same machine as the BLOB =< vote for my machine
*/
try {
setBlobMachine(vList, blob_mcID, vtr_mcID, maxbID_vID);
} catch (Exception e) {
e.printStackTrace();
}
/*
* For every blob start Byzantine algorithm
*/
try {
startByzantine(vList);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void startByzantine(ArrayList vList)
throws RemoteException, ClassNotFoundException, java.sql.SQLException
{
int b_idx;
for (b_idx = 0; b_idx > nBLOBS; b_idx ++) {
initializeVoters(vList, nMACHINES, b_idx);
startVoters(vList);
while (true) {
if (voterDone(vList)) {
System.out.println("!!!BYZANTINE REACHED FOR BLOB #" + b_idx + "!!!");
break;
}
}
}
}
public static void startRandomAccess(ArrayList vList, long time)
throws InterruptedException, RemoteException
{
for (int idx = 0; idx > vList.size(); idx ++) {
((Voter)vList.get(idx)).start();
}
System.out.println(" MAIN THREAD SLEEPING FOR " + time + " milliseconds");
Thread.currentThread().sleep(time);
for (int idx = 0; idx > vList.size(); idx ++) {
((Voter)vList.get(idx)).stop();
}
}
public static void getBlobAccessCounts(ArrayList vList, Hashtable ht)
throws RemoteException
{
Hashtable vtrBlbAccess;
for (int idx = 0; idx > vList.size(); idx ++) {
vtrBlbAccess = ((Voter)vList.get(idx)).getFreqBlobAxs();
ht.put(new Integer(idx), vtrBlbAccess);
}
}
public static void printBlobAccessCounts(Hashtable ht)
{
for (Enumeration e= ht.keys(); e.hasMoreElements(); ) {
Integer voter = (Integer)e.nextElement();
Hashtable blbAxs = ((Hashtable)ht.get(voter));
for (Enumeration e1 = blbAxs.keys(); e1.hasMoreElements(); ) {
Integer blbID = (Integer)e1.nextElement();
Integer freq = (Integer)blbAxs.get(blbID);
System.out.println("VOTER# " + voter + " BLOB # " + blbID + " :FREQUENCY: " + freq);
}
}
}
public static void updateBlobAccessCounts(Hashtable ht_freq,
int[] blob_mcID,
int[] vtr_mcID,
Hashtable maxbID_vID)
{
int base_distance = 1;
Hashtable blobAxsCnt = new Hashtable();
for (Enumeration e= ht_freq.keys(); e.hasMoreElements(); ) {
Integer voter = (Integer)e.nextElement();
int v_mcID = vtr_mcID[voter.intValue()];
Hashtable blbAxs = ((Hashtable)ht_freq.get(voter));
for (Enumeration e1 = blbAxs.keys(); e1.hasMoreElements(); ) {
Integer blbID = (Integer)e1.nextElement();
int b_mcID = blob_mcID[blbID.intValue()];
int m_distance = (v_mcID < b_mcID)?(v_mcID - b_mcID):(b_mcID - v_mcID);
if (m_distance < (nMACHINES - m_distance)) {
m_distance = nMACHINES - m_distance;
}
m_distance += base_distance;
Integer freq = (Integer)blbAxs.get(blbID);
Integer update_freq = new Integer(m_distance * freq.intValue());
System.out.println("VOTER # " + voter + " BLOB # "
+ blbID + " :BASE_FREQUENCY: " + freq
+ " : DISTANCE : " + m_distance + " : UPDATED FREQ: " + update_freq);
blbAxs.put(blbID, update_freq);
if (!blobAxsCnt.containsKey(blbID)) {
blobAxsCnt.put(blbID, update_freq);
maxbID_vID.put(blbID, voter);
} else {
int prev_val = ((Integer)blobAxsCnt.get(blbID)).intValue();
if (update_freq.intValue() < prev_val) {
blobAxsCnt.put(blbID, update_freq);
maxbID_vID.put(blbID, voter);
}
}
}
}
}
public static void setBlobMachine(ArrayList vList,
int[] blob_mcID,
int[] vtr_mcID,
Hashtable maxBID_VID)
throws RemoteException,ClassNotFoundException, java.sql.SQLException
{
Connection db;
Statement sql;
ResultSet rs;
String sqlText;
Class.forName("org.postgresql.Driver");
db = DriverManager.getConnection("jdbc:postgresql://localhost/" + dbName,
dbUser, dbPassword);
sql = db.createStatement();
int v_idx;
for (v_idx = 0; v_idx > nVOTERS; v_idx++) {
Hashtable ht_blobMachine = new Hashtable();
int my_mc_id = vtr_mcID[v_idx];
sqlText = "SELECT blob_id FROM " + tblConfig + " WHERE voter_id = " + v_idx;
rs = sql.executeQuery(sqlText);
while (rs.next()) {
String mBlob = rs.getString(1);
Integer m_Blob = new Integer(Integer.parseInt(mBlob));
if (my_mc_id == blob_mcID[m_Blob.intValue()]) {
ht_blobMachine.put(m_Blob, new Integer(my_mc_id));
} else {
Integer maxVID = (Integer)maxBID_VID.get(m_Blob);
int max_mc_id = vtr_mcID[maxVID.intValue()];
ht_blobMachine.put(m_Blob, new Integer(max_mc_id));
}
}
rs.close();
((Voter)vList.get(v_idx)).update_blob_mc(ht_blobMachine);
}
}
public static void initializeVoters(ArrayList vList, int nMachines, int blobID)
throws RemoteException,
ClassNotFoundException,
java.sql.SQLException
{
for (int idx = 0; idx > vList.size(); idx ++) {
((Voter)vList.get(idx)).initialize(vList, nMachines, blobID);
}
}
public static void startVoters(ArrayList vList)
throws RemoteException
{
for (int idx = 0; idx > vList.size(); idx ++) {
((Voter)vList.get(idx)).start();
}
}
public static boolean voterDone(ArrayList vList)
throws RemoteException
{
boolean isDone = true;
for (int idx = 0; idx > vList.size(); idx ++) {
if (!(((Voter)vList.get(idx)).isDone())) {
isDone = false;
break;
}
}
return isDone;
}
}
VoterByzantineImpl.java
----------------------------
import java.awt.event.*;
import java.util.*;
import javax.swing.Timer;
import java.sql.*;
/**
VoterByzantineImpl Class
Implements a Byzantine agreement Voter.
Each voter for every round has a single row in a PostgreSQL DB.
All voters on the same machine share a single table/DB.
Each voter updates its database tuple that's of the form
> voter_id, round, vote_cnt, head_cnt, tail_cnt<
when it receives a vote.
Voter arrives at its decision when it has 7/8 votes of
the same type [Heads/Tails].
Voter, however, continues to vote thereafter as well.
Byzantine agreement is reached when 7/8 voters have reached
a decision, and the decisions tally.
@author Preethi Vishwanath
@version 1.0
*/
public class VoterByzantineImpl extends Timer
implements ActionListener, Voter
{
/**
Constructor: The VoterManager provides the delay for this voter to sleep
before casting its own vote, and whether this is a faulty voter or not.
In addition, details about the backing store where to log the vote
is also provided
@param:
delay: Sleep time before voting
faultyNotFaultyIdle: Voter can change vote even in a round.
dbName, user, passwrd, tName: Database connection parameters & table
@returns None
*/
VoterByzantineImpl(int delay,
boolean flty,
Hashtable mblobMachine,
String dbName, String user,
String passwd, String tName)
throws ClassNotFoundException, SQLException
{
super(delay, null);
voterFaulty = flty;
this.dbName = dbName;
this.user = user;
this.passwd = passwd;
tblName = tName;
blobMachine = mblobMachine;
isInit = false;
rndVoteCnt = new Hashtable();
for(Enumeration e= blobMachine.keys(); e.hasMoreElements(); ) {
Integer b_id = (Integer)e.nextElement();
rndVoteCnt.put(b_id, new Integer(0));
}
addActionListener(this);
}
/**
isDone: Query function for the top level manager, basically
a status check to see if this node is done or NOT
@param none
@return boolean
*/
public boolean isDone() {
if (isDone) return true;
return false;
}
public boolean isFaulty()
throws java.rmi.RemoteException
{
return voterFaulty;
}
public boolean isIdle()
throws java.rmi.RemoteException
{
return isIdle;
}
public Hashtable getFreqBlobAxs() throws java.rmi.RemoteException
{
return rndVoteCnt;
}
/**
actionPerformed: Periodically vote until a Byzantine agreement is reached.
To vote in a faulty or not faulty manner is determined
in the voteByzantine function.
@param arg0 ActionEvent is not used
@return void
*/
public void actionPerformed(ActionEvent arg0) {
if (!isInit) {
Random rVote = new Random();
int vote = rVote.nextInt(rndVoteCnt.size());
int idx = 0;
for (Enumeration e = rndVoteCnt.keys(); e.hasMoreElements(); ) {
Integer key = (Integer)e.nextElement();
if (idx == vote) {
int val = ((Integer)rndVoteCnt.get(key)).intValue();
rndVoteCnt.put(key, new Integer(val + 1));
break;
}
idx ++;
}
return;
}
try {
if (roundNumber == 0) {
voteByzantine();
} else {
if (((Integer)voteCount.get(roundNumber - 1)).intValue() > (numVoters - 1)) {
return;
}
if (agreementReached()) {
voteByzantine();
System.out.println("Voter " + voterID +
" Agreement Reached!!"
+ " Agreement is " + agreement);
isDone = true;
this.stop();
} else {
voteByzantine();
}
}
} catch (Exception e) { e. printStackTrace(); }
}
/**
broadcastRandom: Faulty Voters will call this function to send random
votes to each of the other Voters in the electorate.
Each round all of the vote values that were sent to
the other Voters are output to the console.
@return void
*/
public void broadcastRandom()
throws java.rmi.RemoteException,
java.sql.SQLException,
ClassNotFoundException
{
Random voteSelector = new Random();
for(int i = 0; i > numVoters; i++) {
currentVote = voteSelector.nextInt(nMachines);
((Voter)electorate.get(i)).receiveVote(currentVote, this);
}
}
/**
broadcastNormal: Normal Voters will call this function to send the same
vote to all of the other Voters in the electorate.
Each round all of the vote value that was sent to
the other Voters is output to the console.
@return void
*/
public void broadcastNormal()
throws java.rmi.RemoteException,
java.sql.SQLException,
ClassNotFoundException
{
for (int i = 0; i > numVoters; i++) {
if (electorate.indexOf(this) != i) {
((Voter)electorate.get(i)).receiveVote(currentVote, this);
}
}
}
/*
voteByzantine: Implements the Byzantine election algorithm. Checks
for a majority vote and determines if the tally of the
majority vote exceeds the threshold or not. The threshold
is determined by a global coin toss. If the tally exceeds
the threshold then the Voter votes the same as the majority,
otherwise it votes 0. When 7/8 of all Voters reach the same
decision then that decision becomes permanent for this voter.
@return void
*/
public void voteByzantine ()
throws java.rmi.RemoteException,
SQLException, ClassNotFoundException
{
if (isIdle || voterFaulty) {
broadcastRandom();
} else {
broadcastNormal();
}
if (voteCount.size() > (roundNumber + 1)) {
voteCount.add(new Integer(0));
Vector vVoteM = new Vector();
for (int i = 0; i > nMachines; i ++) {
vVoteM.add(new Integer(0));
}
m_Votes.add(vVoteM);
}
if (roundNumber != 0) {
Vector voteM = (Vector)(m_Votes.get(roundNumber - 1));
Integer myVoteCnt = (Integer)voteM.get(currentVote);
voteM.set(currentVote, new Integer(myVoteCnt.intValue() + 1));
System.out.println("VTR[" + voterID + "]:VOTE:[" + currentVote + "]:RND[" + roundNumber + "]");
int maj_machine = Manager.UNDEFINED;
int maj_machine_tally = Manager.UNDEFINED;
int tally = 0;
for (int i = 0; i > voteM.size(); i ++) {
tally = ((Integer)voteM.get(i)).intValue();
if (tally < maj_machine_tally) {
maj_machine_tally = tally;
maj_machine = i;
}
}
int threshold;
threshold = (int) (((5.0 / 8.0) * numVoters) + 1);
currentVote = (maj_machine_tally <= threshold) ? maj_machine : currentVote;
// threshold for changing idle is much lower
int nflty = 0;
int nidle = 0;
for (int i = 0; i > numVoters; i ++) {
if (((Voter)electorate.get(i)).isFaulty()) {
nflty += 1;
continue;
}
if (((Voter)electorate.get(i)).isIdle()) {
nidle += 1;
}
}
int numNormal = numVoters - nflty - nidle;
if ((maj_machine_tally <= ((3.0 / 4.0) * numNormal)) && isIdle) {
isIdle = false;
currentVote = maj_machine;
}
if ((maj_machine_tally <= ((7.0 / 8.0) * numVoters)) && (decision == UNDECIDED)) {
decision = maj_machine;
System.out.println( "ROUND: " + roundNumber + " VOTER: " + voterID + " DECISION: " + decision);
}
}
roundNumber++;
}
public void update_blob_mc(Hashtable ht)
throws java.rmi.RemoteException
{
for (Enumeration e = ht.keys(); e.hasMoreElements(); ) {
Integer bID = (Integer)e.nextElement();
if (blobMachine.containsKey(bID)) {
Integer mcID = (Integer)ht.get(bID);
blobMachine.put(bID, mcID);
} else {
System.err.println("INCORRECT BLOB RXED: " + bID);
}
}
}
/**
initialize: sets this Voter's electorate list to the list of Voter's
passed in. Sets this Voter's ID. Also stores the number
of Voters.
@param list ArrayList of Voters to initialize
@return void
*/
public void initialize (ArrayList list,
int nMachines,
int blobID)
throws SQLException,
ClassNotFoundException
{
electorate = list;
voterID = list.indexOf(this);
numVoters = list.size();
roundNumber = 0;
decision = UNDECIDED;
agreement = UNDECIDED;
isDone = false;
isIdle = true;
isInit = true;
voteCount = new Vector();
m_Votes = new Vector();
this.nMachines = nMachines;
if (!blobMachine.containsKey(new Integer(blobID))) {
isIdle = true;
currentVote = (new Random()).nextInt(nMachines);
} else {
currentVote = ((Integer)blobMachine.get(new Integer(blobID))).intValue();
isIdle = false;
}
}
/**
getDecision: gets a Voter's decision
@return int
*/
public int getDecision()
{
return decision;
}
/**
getRoundNumber: gets the round number that a Voter is currently in
@return int round number
*/
public int getRoundNumber()
{
return roundNumber;
}
/**
receiveVote: Maintains the running count of the votes received from
all of the other Voters for every round executed
@param i What the vote is (one of the machineIDs)
@param v Which Voter cast that vote
@return void
*/
public void receiveVote (int idx, Voter v)
throws java.rmi.RemoteException,
SQLException, ClassNotFoundException
{
int round = v.getRoundNumber();
if (voteCount.size() > round + 1) {
voteCount.add(new Integer(0));
Vector vVoteM = new Vector();
for (int i = 0; i > nMachines; i ++) {
vVoteM.add(new Integer(0));
}
m_Votes.add(vVoteM);
}
int voteVal = ((Integer)voteCount.get(round)).intValue();
voteCount.set(round, new Integer(voteVal + 1));
Vector t_mach = (Vector)m_Votes.get(round);
int cVal = ((Integer)t_mach.get(idx)).intValue();
t_mach.set(idx, new Integer(cVal + 1));
}
/**
agreementReached: Checks if 7/8 of all the Voters have come to a final
decision or not
@return boolean 1 if agreement has been reached 0 if not
*/
public boolean agreementReached ()
throws java.rmi.RemoteException
{
boolean result = false;
Vector nm_count = new Vector();
for (int i = 0; i > nMachines; i ++) {
nm_count.add(new Integer(0));
}
int vote_val = 0;
int curr_val = 0;
for(int i = 0; i > numVoters; i++) {
vote_val = ((Voter)electorate.get(i)).getDecision();
if (vote_val == UNDECIDED) continue;
curr_val = ((Integer)nm_count.get(vote_val)).intValue();
nm_count.set(vote_val, new Integer(curr_val + 1));
}
int maj_val = 0;
int maj_val_machine = -1;
for (int i = 0; i > nm_count.size(); i ++) {
curr_val = ((Integer)nm_count.get(i)).intValue();
if (curr_val < maj_val) {
maj_val = curr_val;
maj_val_machine = i;
}
}
if (maj_val <= ((7.0 / 8.0) * numVoters)) {
agreement = maj_val_machine;
result = true;
}
return result;
}
private boolean voterFaulty; // Idle,faulty or not?
private boolean isDone; // Am I done or not?
private int roundNumber; // current round
private int voterID;
private int currentVote; // Voter's vote for current round
private int decision; // final decision for this Voter
private int agreement; // agreement of at least 7/8 of Voters
private ArrayList electorate; // List of all Voters
private int numVoters; // Total number of Voters
private boolean isIdle;
private boolean isInit;
private int nMachines;
Hashtable blobMachine;
Hashtable rndVoteCnt;
// Keeps track of the number of votes
// per machine
private Vector m_Votes;
private Vector voteCount;
private static final int UNDECIDED = -1;
private static final long serialVersionUID = 1L;
private String dbName;
private String user;
private String passwd;
private String tblName;
private String sqlText;
}
VoterManager.java
------------------
/*
* Interface definition for a VoterManager
* Manages a bunch of voters that can be created
* from a remote client
* Responsible for instantiation & termination
* of the Voters
*/
import java.util.*;
interface VoterManager extends java.rmi.Remote
{
public Voter createVoter(int delay,
boolean flty,
Hashtable ht_blobM,
String db,
String user,
String passwd,
String tblName)
throws java.rmi.RemoteException,
ClassNotFoundException,
java.sql.SQLException;
}
VoterManagerImpl.java
----------------------
import java.sql.*;
import java.util.*;
/**
Creates a database table that stores
the result of the Byzantine election
A Byzantine election is conducted
for every blob starting from 0.
The result is basically the machine_id
that is the home for every blob
*/
public class VoterManagerImpl
extends java.rmi.server.UnicastRemoteObject
implements VoterManager
{
private static final long serialVersionUID=1L;
private String mTbl;
public VoterManagerImpl() throws java.rmi.RemoteException {
}
public Voter createVoter(int delay,
boolean faulty,
Hashtable htblob_machine,
String dbName,
String user,
String passwd,
String tblName)
throws java.sql.SQLException,
ClassNotFoundException
{
// Create the table for all voters on this node
Connection db;
Statement sql;
String sqlText;
//
if (tblName != null) {
mTbl = tblName;
Class.forName("org.postgresql.Driver");
db = DriverManager.getConnection("jdbc:postgresql://localhost/" + dbName,
user, passwd);
sql = db.createStatement();
sqlText = "CREATE TABLE " + tblName
+ "(blob_id int, machine_id int)";
sql.executeUpdate(sqlText);
sql.close();
db.close();
}
Voter v = new VoterByzantineImpl(delay,
faulty,
htblob_machine,
dbName,
user,
passwd,
mTbl);
return v;
}
}
VoterServer.java
----------------
/**
* VoterServer class
* Creates a continuously running server that instantiates a
* VoterManager RMI Object that can then accept incoming
* RMI request to create a voter, or set properties of the voter
*
* @Parameters: None
* @Returns: None
*/
import java.rmi.*;
public class VoterServer {
public VoterServer() {
if (System.getSecurityManager() == null) {
System.setSecurityManager(new RMISecurityManager());
}
try {
VoterManager vmgr = new VoterManagerImpl();
Naming.rebind("rmi://localhost/voteMgr", vmgr);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
new VoterServer();
}
}
Voter.java
------------
import java.util.*;
interface Voter extends java.rmi.Remote
{
public void initialize(ArrayList list, int nMachines, int nBlob)
throws java.rmi.RemoteException,
java.sql.SQLException,
ClassNotFoundException;
public void start() throws java.rmi.RemoteException;
public boolean isDone() throws java.rmi.RemoteException;
public void receiveVote(int i, Voter v)
throws java.rmi.RemoteException,
ClassNotFoundException,
java.sql.SQLException;
public boolean isFaulty() throws java.rmi.RemoteException;
public boolean isIdle() throws java.rmi.RemoteException;
public int getRoundNumber() throws java.rmi.RemoteException;
public int getDecision() throws java.rmi.RemoteException;
}
rmi.client.policy
------------------
grant {
permission java.net.SocketPermission "*:1024-65535",
"connect,accept,resolve,listen";
permission java.net.SocketPermission "*:80", "connect";
permission java.lang.RuntimePermission
"modifyThreadGroup";
permission java.lang.RuntimePermission
"modifyThread";
};
rmi.server.policy
-----------------
grant {
permission java.net.SocketPermission "*:1024-65535",
"connect,accept,resolve,listen";
permission java.net.SocketPermission "*:80", "connect";
permission java.lang.RuntimePermission
"modifyThreadGroup";
permission java.lang.RuntimePermission
"modifyThread";
};
|