Skip to content

C
CS744 DECS-PA4-KEYVALUE-SERVER
Commit 9cee4e2d authored by mayankkakad's avatar mayankkakad
Browse files
Options

errors rectified

parent 731414cf
Hide whitespace changes
Inline Side-by-side
Showing with 980 additions and 314 deletions
LFU.h
View file @ 9cee4e2d
...@@ -330,6 +330,11 @@ public: ...@@ -330,6 +330,11 @@ public:
cout << endl; cout << endl;
} }
int hash(string s)
{
return (((int)s.at(0)) << 8) + ((int)s.at(1));
}
string getKeyValuePairs(int id) string getKeyValuePairs(int id)
{ {
unordered_map<string, string> flush; unordered_map<string, string> flush;
...@@ -379,15 +384,9 @@ public: ...@@ -379,15 +384,9 @@ public:
closedir(dirFile); closedir(dirFile);
} }
Node *temp = head->next; for(int i=0;i<=curr_pos;i++)
while (temp->next) if(hash(cacheHeap[i]->key)<=id)
{ flush[cacheHeap[i]->key]=cacheHeap[i]->value;
if (hash(temp->key) <= id)
{
flush[temp->key] = temp->payload;
}
temp = temp->next;
}
unordered_map<string, string>::iterator itr; unordered_map<string, string>::iterator itr;
for (itr = flush.begin(); itr != flush.end(); itr++) for (itr = flush.begin(); itr != flush.end(); itr++)
......
dns.cpp
View file @ 9cee4e2d
...@@ -186,7 +186,13 @@ private: ...@@ -186,7 +186,13 @@ private:
RequestType reqType; RequestType reqType;
}; };
void signalHandler(int signum) {
remove(SERVERS);
exit(0);
}
int main(int argc,char **argv) { int main(int argc,char **argv) {
signal(SIGINT, signalHandler);
srand(time(0)); srand(time(0));
string server_address("0.0.0.0:1234"); string server_address("0.0.0.0:1234");
builder.AddListeningPort(server_address, grpc::InsecureServerCredentials()); builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
......
server.cpp
View file @ 9cee4e2d
...@@ -104,13 +104,13 @@ public: ...@@ -104,13 +104,13 @@ public:
else if (status == PROCESS) { else if (status == PROCESS) {
new ServerData(service, cq, reqType); new ServerData(service, cq, reqType);
if (reqType == NEW) { if (reqType == NEW) {
//cout<<"New Server to join:"<<info.address()<<endl; cout<<"New Server to join:"<<info.address()<<endl;
//calculate id of node, return it's successor and predecessor //calculate id of node, return it's successor and predecessor
string address=info.address(); string address=info.address();
int id=stoi(address.substr(address.find(':')+1)); int id=stoi(address.substr(address.find(':')+1));
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -123,6 +123,7 @@ public: ...@@ -123,6 +123,7 @@ public:
fin.close(); fin.close();
int node=-1; int node=-1;
int next=-1; int next=-1;
bool fl=false;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<id&&my_id>=id) { if(nums>0&&fingers[nums-1]<id&&my_id>=id) {
node=fingers[nums-1]; node=fingers[nums-1];
...@@ -139,12 +140,13 @@ public: ...@@ -139,12 +140,13 @@ public:
next=fingers[i]; next=fingers[i];
break; break;
} }
else if(i>0&&fingers[i]<fingers[i-1]&&fingers[i]>id) { else if(i>0&&fingers[i]<fingers[i-1]) {
node=fingers[i-1]; node=fingers[i-1];
next=fingers[i]; next=fingers[i];
break; break;
} }
else if(i==nums-1) { else if(i==nums-1) {
fl=true;
node=fingers[i]; node=fingers[i];
next=my_id; next=my_id;
break; break;
...@@ -160,56 +162,175 @@ public: ...@@ -160,56 +162,175 @@ public:
Id x; Id x;
x.set_id(next); x.set_id(next);
Id y; Id y;
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
int mypred; int mypred,nextpred;
fin.open(NEIGHBOURS);
string temp;
getline(fin,temp);
getline(fin,temp);
fin.close();
mypred=stoi(temp.substr(temp.find(':')+1));
if(next!=my_id) { if(next!=my_id) {
stub->GETPREDECESSOR(&context,x,&y); stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id(); nextpred=y.id();
} }
else { if(fl==false&&next!=my_id&&nextpred==node) {
fin.open(NEIGHBOURS); cout<<"Yes it is. So we found the successor and the predecessor"<<endl;
string temp; cout<<"Successor: "<<x.id()<<endl;
getline(fin,temp); cout<<"Predecessor: "<<nextpred<<endl;
getline(fin,temp); successorInfo.set_succaddress("0.0.0.0:"+to_string(x.id()));
fin.close(); successorInfo.set_predaddress("0.0.0.0:"+to_string(nextpred));
mypred=stoi(temp.substr(temp.find(':')+1));
} }
if(mypred==node) { else if(fl==false&&next==my_id&&mypred==node) {
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; cout<<"Yes it is. So we found the successor and the predecessor"<<endl;
//cout<<"Successor: "<<x.id()<<endl; cout<<"Successor: "<<x.id()<<endl;
//cout<<"Predecessor: "<<mypred<<endl; cout<<"Predecessor: "<<mypred<<endl;
successorInfo.set_succaddress("0.0.0.0:"+to_string(x.id())); successorInfo.set_succaddress("0.0.0.0:"+to_string(x.id()));
successorInfo.set_predaddress("0.0.0.0:"+to_string(mypred)); successorInfo.set_predaddress("0.0.0.0:"+to_string(mypred));
} }
else if(fl==false){
cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl;
if(node==my_id) {
int mysucc;
fin.open(NEIGHBOURS);
string temp;
getline(fin,temp);
getline(fin,temp);
fin.close();
mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=id) {
successorInfo.set_succaddress("0.0.0.0:"+to_string(mysucc));
successorInfo.set_predaddress("0.0.0.0:"+to_string(my_id));
}
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
Id z;
string t_address("0.0.0.0:"+to_string(y.id()));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context2;
stub->GETPREDECESSOR(&context2,y,&z);
cout<<"Yes. We got the predecessor"<<endl;
cout<<"Successor: "<<y.id()<<endl;
cout<<"Predecessor: "<<z.id()<<endl;
successorInfo.set_succaddress("0.0.0.0:"+to_string(y.id()));
successorInfo.set_predaddress("0.0.0.0:"+to_string(z.id()));
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
Id z;
string t_address("0.0.0.0:"+to_string(y.id()));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context2;
stub->GETPREDECESSOR(&context2,y,&z);
cout<<"Yes. We got the predecessor"<<endl;
cout<<"Successor: "<<y.id()<<endl;
cout<<"Predecessor: "<<z.id()<<endl;
successorInfo.set_succaddress("0.0.0.0:"+to_string(y.id()));
successorInfo.set_predaddress("0.0.0.0:"+to_string(z.id()));
}
}
else { else {
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; if(node>next) {
string tar_address("0.0.0.0:"+to_string(node)); fin.open(NEIGHBOURS);
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); string xyz;
stub=KeyValueServices::NewStub(channel); getline(fin,xyz);
ClientContext context1; getline(fin,xyz);
x.set_id(id); fin.close();
stub->GETSUCCESSOR(&context1,x,&y); int xi=stoi(xyz.substr(xyz.find(':')+1));
//cout<<"Yes. We got the successor"<<endl; if(xi==node) {
Id z; successorInfo.set_succaddress("0.0.0.0:"+to_string(next));
string t_address("0.0.0.0:"+to_string(y.id())); successorInfo.set_predaddress("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); }
stub=KeyValueServices::NewStub(channel); else {
ClientContext context2; string tar_address("0.0.0.0:"+to_string(node));
stub->GETPREDECESSOR(&context2,y,&z); channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
//cout<<"Yes. We got the predecessor"<<endl; stub=KeyValueServices::NewStub(channel);
//cout<<"Successor: "<<y.id()<<endl; ClientContext context1;
//cout<<"Predecessor: "<<z.id()<<endl; x.set_id(id);
successorInfo.set_succaddress("0.0.0.0:"+to_string(y.id())); stub->GETSUCCESSOR(&context1,x,&y);
successorInfo.set_predaddress("0.0.0.0:"+to_string(z.id())); cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
Id z;
string t_address("0.0.0.0:"+to_string(y.id()));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context2;
stub->GETPREDECESSOR(&context2,y,&z);
cout<<"Yes. We got the predecessor"<<endl;
cout<<"Successor: "<<y.id()<<endl;
cout<<"Predecessor: "<<z.id()<<endl;
successorInfo.set_succaddress("0.0.0.0:"+to_string(y.id()));
successorInfo.set_predaddress("0.0.0.0:"+to_string(z.id()));
}
else {
successorInfo.set_succaddress("0.0.0.0:"+to_string(next));
successorInfo.set_predaddress("0.0.0.0:"+to_string(node));
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
successorInfo.set_succaddress("0.0.0.0:"+to_string(node));
successorInfo.set_predaddress("0.0.0.0:"+to_string(next));
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
Id z;
string t_address("0.0.0.0:"+to_string(y.id()));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context2;
stub->GETPREDECESSOR(&context2,y,&z);
cout<<"Yes. We got the predecessor"<<endl;
cout<<"Successor: "<<y.id()<<endl;
cout<<"Predecessor: "<<z.id()<<endl;
successorInfo.set_succaddress("0.0.0.0:"+to_string(y.id()));
successorInfo.set_predaddress("0.0.0.0:"+to_string(z.id()));
}
else {
successorInfo.set_succaddress("0.0.0.0:"+to_string(next));
successorInfo.set_predaddress("0.0.0.0:"+to_string(node));
}
}
}
} }
} }
else { else {
//cout<<"We got no node with id greater than new node"<<endl; cout<<"We got no node with id greater than new node"<<endl;
//cout<<"Only one node present right now, that is me"<<endl; cout<<"Only one node present right now, that is me"<<endl;
successorInfo.set_succaddress("0.0.0.0:"+to_string(my_id)); successorInfo.set_succaddress("0.0.0.0:"+to_string(my_id));
successorInfo.set_predaddress("0.0.0.0:"+to_string(my_id)); successorInfo.set_predaddress("0.0.0.0:"+to_string(my_id));
} }
//cout<<"Sending the successor and predecessor back to the new server"<<endl; cout<<"Sending the successor and predecessor back to the new server"<<endl;
newResponder.Finish(successorInfo,Status::OK,this); newResponder.Finish(successorInfo,Status::OK,this);
} }
else if(reqType==INFORMSUCCESSOR){ else if(reqType==INFORMSUCCESSOR){
...@@ -219,7 +340,7 @@ public: ...@@ -219,7 +340,7 @@ public:
string keyvalues=memManager->getKeyValuePairs(id); string keyvalues=memManager->getKeyValuePairs(id);
string keys=keyvalues.substr(0,keyvalues.find(";;")+1); string keys=keyvalues.substr(0,keyvalues.find(";;")+1);
string values=keyvalues.substr(keyvalues.find(";;")+2); string values=keyvalues.substr(keyvalues.find(";;")+2);
//cout<<"Okay, my new predecessor is: "<<info.address()<<endl; cout<<"Okay, my new predecessor is: "<<info.address()<<endl;
ifstream fin; ifstream fin;
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string successor,predecessor; string successor,predecessor;
...@@ -234,15 +355,15 @@ public: ...@@ -234,15 +355,15 @@ public:
fout.close(); fout.close();
keyValues.set_keys(keys); keyValues.set_keys(keys);
keyValues.set_values(values); keyValues.set_values(values);
//cout<<"Done making changes accordingly"<<endl; cout<<"Done making changes accordingly"<<endl;
informSuccessorResponder.Finish(keyValues,Status::OK,this); informSuccessorResponder.Finish(keyValues,Status::OK,this);
} }
else if(reqType==GETSUCCESSOR) { else if(reqType==GETSUCCESSOR) {
//cout<<"Some server asked me to find the successor of "<<idvar1.id()<<endl; cout<<"Some server asked me to find the successor of "<<idvar1.id()<<endl;
int idtofind=idvar1.id(); int id=idvar1.id();
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting the finger table"<<endl; cout<<"Getting the finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
for(int i=0;i<16;i++) { for(int i=0;i<16;i++) {
...@@ -254,64 +375,154 @@ public: ...@@ -254,64 +375,154 @@ public:
} }
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<idtofind&&my_id>=idtofind) { if(nums>0&&fingers[nums-1]<id&&my_id>=id) {
node=fingers[nums-1]; node=fingers[nums-1];
next=my_id; next=my_id;
} }
else if(nums>0&&my_id<idtofind&&fingers[0]>=idtofind) { else if(nums>0&&my_id<id&&fingers[0]>=id) {
node=my_id; node=my_id;
next=fingers[0]; next=fingers[0];
} }
else { else {
for(int i=0;i<nums;i++) { for(int i=0;i<nums;i++) {
if(i>0&&fingers[i-1]<idtofind&&fingers[i]>=idtofind) { if(i>0&&fingers[i-1]<id&&fingers[i]>=id) {
node=fingers[i-1]; node=fingers[i-1];
next=fingers[i]; next=fingers[i];
break; break;
} }
else if(i>0&&fingers[i]<fingers[i-1]&&fingers[i]>idtofind) { else if(i>0&&fingers[i]<fingers[i-1]&&fingers[i]>id) {
node=fingers[i-1]; node=fingers[i-1];
next=fingers[i]; next=fingers[i];
break; break;
} }
else if(i==nums-1) { else if(i==nums-1) {
fl=true;
node=fingers[i]; node=fingers[i];
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context;
Id x;
x.set_id(next);
Id y;
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
stub->GETPREDECESSOR(&context,x,&y);
if(y.id()==node) {
//cout<<"Yes it is. We found the successor"<<endl;
//cout<<"Successor: "<<y.id()<<endl;
idvar2.set_id(x.id());
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); stub=KeyValueServices::NewStub(channel);
ClientContext context1; ClientContext context;
x.set_id(idtofind); Id x;
//cout<<"No it is not. Asking the possible predecessor to find its successor"<<endl; x.set_id(next);
stub->GETSUCCESSOR(&context1,x,&y); Id y;
idvar2.set_id(y.id()); cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
int mypred,nextpred;
fin.open(NEIGHBOURS);
string temp;
getline(fin,temp);
getline(fin,temp);
fin.close();
mypred=stoi(temp.substr(temp.find(':')+1));
if(next!=my_id) {
stub->GETPREDECESSOR(&context,x,&y);
nextpred=y.id();
}
if(fl==false&&next!=my_id&&nextpred==node)
idvar2.set_id(x.id());
else if(fl==false&&next==my_id&&mypred==node)
idvar2.set_id(x.id());
else if(fl==false){
if(node==my_id) {
int mysucc;
fin.open(NEIGHBOURS);
string temp;
getline(fin,temp);
getline(fin,temp);
fin.close();
mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=id)
idvar2.set_id(mysucc);
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
idvar2.set_id(y.id());
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
idvar2.set_id(y.id());
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));;
if(xi==node) {
idvar2.set_id(next);
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
idvar2.set_id(y.id());
}
else {
idvar2.set_id(next);
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
cout<<"Here"<<endl;
idvar2.set_id(node);
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
idvar2.set_id(y.id());
}
else {
idvar2.set_id(next);
}
}
}
}
} }
getSuccessorResponder.Finish(idvar2,Status::OK,this); getSuccessorResponder.Finish(idvar2,Status::OK,this);
} }
else if(reqType==GETPREDECESSOR) { else if(reqType==GETPREDECESSOR) {
//cout<<"Someone asked me for my predecessor. Sending them"<<endl; cout<<"Someone asked me for my predecessor. Sending them"<<endl;
ifstream fin; ifstream fin;
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string successor,predecessor; string successor,predecessor;
...@@ -319,12 +530,12 @@ public: ...@@ -319,12 +530,12 @@ public:
getline(fin,predecessor); getline(fin,predecessor);
fin.close(); fin.close();
idvar2.set_id(stoi(predecessor.substr(predecessor.find(':')+1))); idvar2.set_id(stoi(predecessor.substr(predecessor.find(':')+1)));
//cout<<"Sent my predecessor"<<endl; cout<<"Sent my predecessor"<<endl;
getPredecessorResponder.Finish(idvar2,Status::OK,this); getPredecessorResponder.Finish(idvar2,Status::OK,this);
} }
else if(reqType==INFORMPREDECESSOR) { else if(reqType==INFORMPREDECESSOR) {
//cout<<"Okay, i got the information that my successor has changed"<<endl; cout<<"Okay, i got the information that my successor has changed"<<endl;
//cout<<"My new successor: "<<info.address()<<endl; cout<<"My new successor: "<<info.address()<<endl;
ifstream fin; ifstream fin;
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string successor,predecessor; string successor,predecessor;
...@@ -338,7 +549,7 @@ public: ...@@ -338,7 +549,7 @@ public:
fout<<predecessor<<endl; fout<<predecessor<<endl;
fout.close(); fout.close();
null.set_nothing(0); null.set_nothing(0);
//cout<<"Okay, i made the necessary changes"<<endl; cout<<"Okay, i made the necessary changes"<<endl;
informPredecessorResponder.Finish(null,Status::OK,this); informPredecessorResponder.Finish(null,Status::OK,this);
} }
else { else {
...@@ -463,7 +674,7 @@ public: ...@@ -463,7 +674,7 @@ public:
//transfer request //transfer request
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -475,6 +686,7 @@ public: ...@@ -475,6 +686,7 @@ public:
}while(fin); }while(fin);
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) { if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) {
...@@ -498,54 +710,128 @@ public: ...@@ -498,54 +710,128 @@ public:
break; break;
} }
else if(i==nums-1) { else if(i==nums-1) {
fl=true;
node=fingers[i]; node=fingers[i];
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context; stub=KeyValueServices::NewStub(channel);
Id x; ClientContext context;
x.set_id(next); Id x;
Id y; x.set_id(next);
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; Id y;
int mypred; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
if(next!=my_id) { int mypred,nextpred;
stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id();
}
else {
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string temp; string temp;
getline(fin,temp); getline(fin,temp);
getline(fin,temp); getline(fin,temp);
fin.close(); fin.close();
mypred=stoi(temp.substr(temp.find(':')+1)); mypred=stoi(temp.substr(temp.find(':')+1));
} if(next!=my_id) {
if(mypred==node) { stub->GETPREDECESSOR(&context,x,&y);
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; nextpred=y.id();
//cout<<"Successor: "<<x.id()<<endl; }
//cout<<"Predecessor: "<<mypred<<endl; if(fl==false&&next!=my_id&&nextpred==node)
succ=x.id(); succ=x.id();
} else if(fl==false&&next==my_id&&mypred==node)
else { succ=x.id();
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; else if(fl==false){
string tar_address("0.0.0.0:"+to_string(node)); if(node==my_id) {
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); int mysucc;
stub=KeyValueServices::NewStub(channel); fin.open(NEIGHBOURS);
ClientContext context1; string temp;
x.set_id(key_id); getline(fin,temp);
stub->GETSUCCESSOR(&context1,x,&y); getline(fin,temp);
//cout<<"Yes. We got the successor"<<endl; fin.close();
succ=x.id(); mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=key_id)
succ=mysucc;
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));
if(xi==node) {
succ=next;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
succ=node;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
}
} }
string t_address("0.0.0.0:"+to_string(succ)); string t_address("0.0.0.0:"+to_string(succ));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub=KeyValueServices::NewStub(channel);
ClientContext cont1; ClientContext cont1;
stub->GET(&cont1,key,&value); stub->GET(&cont1,key,&value);
} }
...@@ -586,7 +872,7 @@ public: ...@@ -586,7 +872,7 @@ public:
//transfer the request //transfer the request
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -598,6 +884,7 @@ public: ...@@ -598,6 +884,7 @@ public:
}while(fin); }while(fin);
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) { if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) {
...@@ -622,53 +909,127 @@ public: ...@@ -622,53 +909,127 @@ public:
} }
else if(i==nums-1) { else if(i==nums-1) {
node=fingers[i]; node=fingers[i];
fl=true;
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context; stub=KeyValueServices::NewStub(channel);
Id x; ClientContext context;
x.set_id(next); Id x;
Id y; x.set_id(next);
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; Id y;
int mypred; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
if(next!=my_id) { int mypred,nextpred;
stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id();
}
else {
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string temp; string temp;
getline(fin,temp); getline(fin,temp);
getline(fin,temp); getline(fin,temp);
fin.close(); fin.close();
mypred=stoi(temp.substr(temp.find(':')+1)); mypred=stoi(temp.substr(temp.find(':')+1));
} if(next!=my_id) {
if(mypred==node) { stub->GETPREDECESSOR(&context,x,&y);
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; nextpred=y.id();
//cout<<"Successor: "<<x.id()<<endl; }
//cout<<"Predecessor: "<<mypred<<endl; if(fl==false&&next!=my_id&&nextpred==node)
succ=x.id(); succ=x.id();
} else if(fl==false&&next==my_id&&mypred==node)
else { succ=x.id();
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; else if(fl==false){
string tar_address("0.0.0.0:"+to_string(node)); if(node==my_id) {
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); int mysucc;
stub=KeyValueServices::NewStub(channel); fin.open(NEIGHBOURS);
ClientContext context1; string temp;
x.set_id(key_id); getline(fin,temp);
stub->GETSUCCESSOR(&context1,x,&y); getline(fin,temp);
//cout<<"Yes. We got the successor"<<endl; fin.close();
succ=x.id(); mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=key_id)
succ=mysucc;
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));
if(xi==node) {
succ=next;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
succ=node;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
}
} }
string t_address("0.0.0.0:"+to_string(succ)); string t_address("0.0.0.0:"+to_string(succ));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub=KeyValueServices::NewStub(channel);
ClientContext cont1; ClientContext cont1;
stub->GET(&cont1,key,&value); stub->GET(&cont1,key,&value);
} }
...@@ -693,7 +1054,7 @@ public: ...@@ -693,7 +1054,7 @@ public:
//transfer request //transfer request
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -705,6 +1066,7 @@ public: ...@@ -705,6 +1066,7 @@ public:
}while(fin); }while(fin);
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) { if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) {
...@@ -729,53 +1091,127 @@ public: ...@@ -729,53 +1091,127 @@ public:
} }
else if(i==nums-1) { else if(i==nums-1) {
node=fingers[i]; node=fingers[i];
fl=true;
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context; stub=KeyValueServices::NewStub(channel);
Id x; ClientContext context;
x.set_id(next); Id x;
Id y; x.set_id(next);
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; Id y;
int mypred; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
if(next!=my_id) { int mypred,nextpred;
stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id();
}
else {
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string temp; string temp;
getline(fin,temp); getline(fin,temp);
getline(fin,temp); getline(fin,temp);
fin.close(); fin.close();
mypred=stoi(temp.substr(temp.find(':')+1)); mypred=stoi(temp.substr(temp.find(':')+1));
} if(next!=my_id) {
if(mypred==node) { stub->GETPREDECESSOR(&context,x,&y);
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; nextpred=y.id();
//cout<<"Successor: "<<x.id()<<endl; }
//cout<<"Predecessor: "<<mypred<<endl; if(fl==false&&next!=my_id&&nextpred==node)
succ=x.id(); succ=x.id();
} else if(fl==false&&next==my_id&&mypred==node)
else { succ=x.id();
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; else if(fl==false){
string tar_address("0.0.0.0:"+to_string(node)); if(node==my_id) {
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); int mysucc;
stub=KeyValueServices::NewStub(channel); fin.open(NEIGHBOURS);
ClientContext context1; string temp;
x.set_id(key_id); getline(fin,temp);
stub->GETSUCCESSOR(&context1,x,&y); getline(fin,temp);
//cout<<"Yes. We got the successor"<<endl; fin.close();
succ=x.id(); mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=key_id)
succ=mysucc;
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));
if(xi==node) {
succ=next;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
succ=node;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
}
} }
string t_address("0.0.0.0:"+to_string(succ)); string t_address("0.0.0.0:"+to_string(succ));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub=KeyValueServices::NewStub(channel);
ClientContext cont1; ClientContext cont1;
stub->PUT(&cont1,keyvalue,&stat); stub->PUT(&cont1,keyvalue,&stat);
} }
...@@ -804,7 +1240,7 @@ public: ...@@ -804,7 +1240,7 @@ public:
//transfer the request //transfer the request
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -816,6 +1252,7 @@ public: ...@@ -816,6 +1252,7 @@ public:
}while(fin); }while(fin);
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) { if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) {
...@@ -840,53 +1277,127 @@ public: ...@@ -840,53 +1277,127 @@ public:
} }
else if(i==nums-1) { else if(i==nums-1) {
node=fingers[i]; node=fingers[i];
fl=true;
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context; stub=KeyValueServices::NewStub(channel);
Id x; ClientContext context;
x.set_id(next); Id x;
Id y; x.set_id(next);
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; Id y;
int mypred; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
if(next!=my_id) { int mypred,nextpred;
stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id();
}
else {
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string temp; string temp;
getline(fin,temp); getline(fin,temp);
getline(fin,temp); getline(fin,temp);
fin.close(); fin.close();
mypred=stoi(temp.substr(temp.find(':')+1)); mypred=stoi(temp.substr(temp.find(':')+1));
} if(next!=my_id) {
if(mypred==node) { stub->GETPREDECESSOR(&context,x,&y);
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; nextpred=y.id();
//cout<<"Successor: "<<x.id()<<endl; }
//cout<<"Predecessor: "<<mypred<<endl; if(fl==false&&next!=my_id&&nextpred==node)
succ=x.id(); succ=x.id();
} else if(fl==false&&next==my_id&&mypred==node)
else { succ=x.id();
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; else if(fl==false){
string tar_address("0.0.0.0:"+to_string(node)); if(node==my_id) {
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); int mysucc;
stub=KeyValueServices::NewStub(channel); fin.open(NEIGHBOURS);
ClientContext context1; string temp;
x.set_id(key_id); getline(fin,temp);
stub->GETSUCCESSOR(&context1,x,&y); getline(fin,temp);
//cout<<"Yes. We got the successor"<<endl; fin.close();
succ=x.id(); mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=key_id)
succ=mysucc;
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));
if(xi==node) {
succ=next;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
succ=node;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
}
} }
string t_address("0.0.0.0:"+to_string(succ)); string t_address("0.0.0.0:"+to_string(succ));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub=KeyValueServices::NewStub(channel);
ClientContext cont1; ClientContext cont1;
stub->PUT(&cont1,keyvalue,&stat); stub->PUT(&cont1,keyvalue,&stat);
} }
...@@ -911,7 +1422,7 @@ public: ...@@ -911,7 +1422,7 @@ public:
//transfer request //transfer request
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -923,6 +1434,7 @@ public: ...@@ -923,6 +1434,7 @@ public:
}while(fin); }while(fin);
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) { if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) {
...@@ -947,53 +1459,127 @@ public: ...@@ -947,53 +1459,127 @@ public:
} }
else if(i==nums-1) { else if(i==nums-1) {
node=fingers[i]; node=fingers[i];
fl=true;
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context; stub=KeyValueServices::NewStub(channel);
Id x; ClientContext context;
x.set_id(next); Id x;
Id y; x.set_id(next);
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; Id y;
int mypred; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
if(next!=my_id) { int mypred,nextpred;
stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id();
}
else {
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string temp; string temp;
getline(fin,temp); getline(fin,temp);
getline(fin,temp); getline(fin,temp);
fin.close(); fin.close();
mypred=stoi(temp.substr(temp.find(':')+1)); mypred=stoi(temp.substr(temp.find(':')+1));
} if(next!=my_id) {
if(mypred==node) { stub->GETPREDECESSOR(&context,x,&y);
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; nextpred=y.id();
//cout<<"Successor: "<<x.id()<<endl; }
//cout<<"Predecessor: "<<mypred<<endl; if(fl==false&&next!=my_id&&nextpred==node)
succ=x.id(); succ=x.id();
} else if(fl==false&&next==my_id&&mypred==node)
else { succ=x.id();
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; else if(fl==false){
string tar_address("0.0.0.0:"+to_string(node)); if(node==my_id) {
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); int mysucc;
stub=KeyValueServices::NewStub(channel); fin.open(NEIGHBOURS);
ClientContext context1; string temp;
x.set_id(key_id); getline(fin,temp);
stub->GETSUCCESSOR(&context1,x,&y); getline(fin,temp);
//cout<<"Yes. We got the successor"<<endl; fin.close();
succ=x.id(); mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=key_id)
succ=mysucc;
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));
if(xi==node) {
succ=next;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
succ=node;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
}
} }
string t_address("0.0.0.0:"+to_string(succ)); string t_address("0.0.0.0:"+to_string(succ));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub=KeyValueServices::NewStub(channel);
ClientContext cont1; ClientContext cont1;
stub->DEL(&cont1,key,&stat); stub->DEL(&cont1,key,&stat);
} }
...@@ -1033,7 +1619,7 @@ public: ...@@ -1033,7 +1619,7 @@ public:
else { else {
int fingers[16]; int fingers[16];
ifstream fin; ifstream fin;
//cout<<"Getting my finger table"<<endl; cout<<"Getting my finger table"<<endl;
fin.open(FINGER_TABLE); fin.open(FINGER_TABLE);
int nums=0; int nums=0;
do { do {
...@@ -1045,6 +1631,7 @@ public: ...@@ -1045,6 +1631,7 @@ public:
}while(fin); }while(fin);
fin.close(); fin.close();
int node=-1; int node=-1;
bool fl=false;
int next=-1; int next=-1;
int my_id=stoi(params.find("LISTENING_PORT")->second); int my_id=stoi(params.find("LISTENING_PORT")->second);
if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) { if(nums>0&&fingers[nums-1]<key_id&&my_id>=key_id) {
...@@ -1069,53 +1656,127 @@ public: ...@@ -1069,53 +1656,127 @@ public:
} }
else if(i==nums-1) { else if(i==nums-1) {
node=fingers[i]; node=fingers[i];
fl=true;
next=my_id; next=my_id;
break; break;
} }
} }
} }
string target_address("0.0.0.0:"+to_string(next)); if(next!=-1) {
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); string target_address("0.0.0.0:"+to_string(next));
unique_ptr<KeyValueServices::Stub> stub; shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub;
ClientContext context; stub=KeyValueServices::NewStub(channel);
Id x; ClientContext context;
x.set_id(next); Id x;
Id y; x.set_id(next);
//cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl; Id y;
int mypred; cout<<"asking the possible successor whether its predecessor is less than the new node id"<<endl;
if(next!=my_id) { int mypred,nextpred;
stub->GETPREDECESSOR(&context,x,&y);
mypred=y.id();
}
else {
fin.open(NEIGHBOURS); fin.open(NEIGHBOURS);
string temp; string temp;
getline(fin,temp); getline(fin,temp);
getline(fin,temp); getline(fin,temp);
fin.close(); fin.close();
mypred=stoi(temp.substr(temp.find(':')+1)); mypred=stoi(temp.substr(temp.find(':')+1));
} if(next!=my_id) {
if(mypred==node) { stub->GETPREDECESSOR(&context,x,&y);
//cout<<"Yes it is. So we found the successor and the predecessor"<<endl; nextpred=y.id();
//cout<<"Successor: "<<x.id()<<endl; }
//cout<<"Predecessor: "<<mypred<<endl; if(fl==false&&next!=my_id&&nextpred==node)
succ=x.id(); succ=x.id();
} else if(fl==false&&next==my_id&&mypred==node)
else { succ=x.id();
//cout<<"No it is not. We will ask the possible predecessor to find the successor of new node"<<endl; else if(fl==false){
string tar_address("0.0.0.0:"+to_string(node)); if(node==my_id) {
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials()); int mysucc;
stub=KeyValueServices::NewStub(channel); fin.open(NEIGHBOURS);
ClientContext context1; string temp;
x.set_id(key_id); getline(fin,temp);
stub->GETSUCCESSOR(&context1,x,&y); getline(fin,temp);
//cout<<"Yes. We got the successor"<<endl; fin.close();
succ=x.id(); mysucc=stoi(temp.substr(temp.find(':')+1));
if(mysucc>=key_id)
succ=mysucc;
else {
string tar_address("0.0.0.0:"+to_string(mysucc));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
succ=y.id();
}
}
else {
if(node>next) {
fin.open(NEIGHBOURS);
string xyz;
getline(fin,xyz);
getline(fin,xyz);
fin.close();
int xi=stoi(xyz.substr(xyz.find(':')+1));
if(xi==node) {
succ=next;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
else {
ifstream fin1;
fin1.open(NEIGHBOURS);
string t1,t2;
getline(fin1,t1);
getline(fin1,t2);
fin1.close();
int t1i=stoi(t1.substr(t1.find(':')+1));
if(t1==t2||t1i==node) {
succ=node;
}
else {
string tar_address("0.0.0.0:"+to_string(node));
channel=grpc::CreateChannel(tar_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel);
ClientContext context1;
x.set_id(key_id);
stub->GETSUCCESSOR(&context1,x,&y);
cout<<"Yes. We got the successor"<<endl;
if(y.id()!=my_id) {
succ=y.id();
}
else {
succ=next;
}
}
}
}
} }
string t_address("0.0.0.0:"+to_string(succ)); string t_address("0.0.0.0:"+to_string(succ));
channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(t_address, grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); unique_ptr<KeyValueServices::Stub> stub=KeyValueServices::NewStub(channel);
ClientContext cont1; ClientContext cont1;
stub->DEL(&cont1,key,&stat); stub->DEL(&cont1,key,&stat);
} }
...@@ -1282,7 +1943,7 @@ void updateAllFingerTables() { ...@@ -1282,7 +1943,7 @@ void updateAllFingerTables() {
} }
void register_server_DNS(string my_address) { void register_server_DNS(string my_address) {
//cout<<"Registering to DNS"<<endl; cout<<"Registering to DNS"<<endl;
string target_address(DNS_SERVER); string target_address(DNS_SERVER);
shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials()); shared_ptr<Channel> channel=grpc::CreateChannel(target_address, grpc::InsecureChannelCredentials());
unique_ptr<KeyValueServices::Stub> stub; unique_ptr<KeyValueServices::Stub> stub;
...@@ -1292,29 +1953,29 @@ void register_server_DNS(string my_address) { ...@@ -1292,29 +1953,29 @@ void register_server_DNS(string my_address) {
Info info; Info info;
ClientContext context; ClientContext context;
Status status=stub->GETADDRESS(&context,null,&info); Status status=stub->GETADDRESS(&context,null,&info);
//cout<<"Address received:"<<info.address()<<endl; cout<<"Address received:"<<info.address()<<endl;
string old_server; string old_server;
if(status.ok()) { if(status.ok()) {
old_server=info.address(); old_server=info.address();
info.set_address(my_address); info.set_address(my_address);
ClientContext new_context; ClientContext new_context;
//cout<<"Adding address to DNS"<<endl; cout<<"Adding address to DNS"<<endl;
stub->ADDADDRESS(&new_context,info,&null); stub->ADDADDRESS(&new_context,info,&null);
//cout<<"Address added to DNS"<<endl; cout<<"Address added to DNS"<<endl;
ofstream fout; ofstream fout;
//cout<<"Generating initial finger table"<<endl; cout<<"Generating initial finger table"<<endl;
fout.open(FINGER_TABLE); fout.open(FINGER_TABLE);
for(int i=0;i<16;i++) for(int i=0;i<16;i++)
fout<<"null"<<endl; fout<<"null"<<endl;
fout.close(); fout.close();
//cout<<"Initial finger table generated"<<endl; cout<<"Initial finger table generated"<<endl;
if(old_server=="null") { if(old_server=="null") {
//cout<<"Initializing initial neighbours"<<endl; cout<<"Initializing initial neighbours"<<endl;
fout.open(NEIGHBOURS); fout.open(NEIGHBOURS);
fout<<"-1"<<endl; fout<<"-1"<<endl;
fout<<"-1"<<endl; fout<<"-1"<<endl;
fout.close(); fout.close();
//cout<<"Initialized initial neighbours"<<endl; cout<<"Initialized initial neighbours"<<endl;
return; return;
} }
channel=grpc::CreateChannel(old_server,grpc::InsecureChannelCredentials()); channel=grpc::CreateChannel(old_server,grpc::InsecureChannelCredentials());
...@@ -1322,27 +1983,27 @@ void register_server_DNS(string my_address) { ...@@ -1322,27 +1983,27 @@ void register_server_DNS(string my_address) {
info.set_address(my_address); info.set_address(my_address);
SuccessorInfo successorInfo; SuccessorInfo successorInfo;
ClientContext context1; ClientContext context1;
//cout<<"Sending request to server: "<<old_server<<endl; cout<<"Sending request to server: "<<old_server<<endl;
status=stub->NEW(&context1,info,&successorInfo); status=stub->NEW(&context1,info,&successorInfo);
//cout<<"Request sent. Successor and predecessor info received"<<endl; cout<<"Request sent. Successor and predecessor info received"<<endl;
if(status.ok()) { if(status.ok()) {
string successor=successorInfo.succaddress(); string successor=successorInfo.succaddress();
string predecessor=successorInfo.predaddress(); string predecessor=successorInfo.predaddress();
ofstream fout; ofstream fout;
//cout<<"Storing neighbours info"<<endl; cout<<"Storing neighbours info"<<endl;
fout.open(NEIGHBOURS); fout.open(NEIGHBOURS);
fout<<successor<<endl; fout<<successor<<endl;
fout<<predecessor<<endl; fout<<predecessor<<endl;
fout.close(); fout.close();
//cout<<"Successor: "<<successor<<endl; cout<<"Successor: "<<successor<<endl;
//cout<<"Predecessor: "<<predecessor<<endl; cout<<"Predecessor: "<<predecessor<<endl;
//cout<<"Stored neighbours info"<<endl; cout<<"Stored neighbours info"<<endl;
channel=grpc::CreateChannel(successor,grpc::InsecureChannelCredentials()); channel=grpc::CreateChannel(successor,grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); stub=KeyValueServices::NewStub(channel);
info.set_address(my_address); info.set_address(my_address);
KeyValues keyValues; KeyValues keyValues;
ClientContext context2; ClientContext context2;
//cout<<"Informing successor about my presence"<<endl; cout<<"Informing successor about my presence"<<endl;
status=stub->INFORMSUCCESSOR(&context2,info,&keyValues); status=stub->INFORMSUCCESSOR(&context2,info,&keyValues);
string keys=keyValues.keys(); string keys=keyValues.keys();
string values=keyValues.values(); string values=keyValues.values();
...@@ -1357,14 +2018,14 @@ void register_server_DNS(string my_address) { ...@@ -1357,14 +2018,14 @@ void register_server_DNS(string my_address) {
values=values.substr(values.find(';')+1); values=values.substr(values.find(';')+1);
memManager->put(key,value); memManager->put(key,value);
} }
//cout<<"Informed succesor"<<endl; cout<<"Informed succesor"<<endl;
channel=grpc::CreateChannel(predecessor,grpc::InsecureChannelCredentials()); channel=grpc::CreateChannel(predecessor,grpc::InsecureChannelCredentials());
stub=KeyValueServices::NewStub(channel); stub=KeyValueServices::NewStub(channel);
info.set_address(my_address); info.set_address(my_address);
ClientContext context3; ClientContext context3;
//cout<<"Informing predecessor about my presence"<<endl; cout<<"Informing predecessor about my presence"<<endl;
status=stub->INFORMPREDECESSOR(&context3,info,&null); status=stub->INFORMPREDECESSOR(&context3,info,&null);
//cout<<"Informed predecessor"<<endl; cout<<"Informed predecessor"<<endl;
updateAllFingerTables(); updateAllFingerTables();
} }
} }
...@@ -1391,7 +2052,7 @@ int main(int agrc, char **argv) { ...@@ -1391,7 +2052,7 @@ int main(int agrc, char **argv) {
setupServer(server_address); setupServer(server_address);
assignThreads(num_threads); assignThreads(num_threads);
sleep(1); sleep(2);
signal(SIGINT, signalHandler); signal(SIGINT, signalHandler);
server = builder.BuildAndStart(); server = builder.BuildAndStart();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment