adds key copying

server/chord.c

@@ -14,6 +14,7 @@
 #include <math.h>
 #include "chord.h"
 #include <pthread.h>
+#include "kvstore.h"
 
 #define MAXLOCATIONS 16
 #define MAXIDVALUE 65536
@@ -126,16 +127,24 @@ void setSuccessorAndPredecessor(Node allNodes[],int size) {
         if(chord.node.nodeId==allNodes[i].nodeId)
             break;
     }
+    Node oldPred = chord.predecessor;
     if(i==0){
         chord.predecessor = allNodes[size-1];
     }else{
         chord.predecessor = allNodes[i-1];
     }
+
     if(i == size -1) {
         chord.successor = allNodes[0];
     }else {
         chord.successor = allNodes[i+1];
     }
+
+    if (chord.predecessor.nodeId!=oldPred.nodeId)
+    {
+        TransferKeys(chord.predecessor.nodeId, chord.predecessor.ip, chord.predecessor.port); 
+    }
+    
 }
 
 void *periodicStabilize(void *arg) {

server/chord.h

@@ -35,3 +35,4 @@ void createNodeStruct(Node *givenNode,char ip[16],int port,int id);
 void stabilize();
 Node getSuccessor();
 void sendUDP(char * msg, char ip[], int port, bool recvResponse, char * buffer);
+int nodeToId(struct sockaddr_in *socketAddr)
\ No newline at end of file

server/kvserver.c

@@ -6,6 +6,7 @@
 #include <arpa/inet.h>
 #include <sys/types.h>
 #include <sys/socket.h>
+#include <sys/stat.h>
 #include <netinet/in.h>
 #include <sys/time.h>
 #include <stdbool.h>
@@ -66,6 +67,7 @@ int main(int argc, char* argv[])
     printf("NumSets = %d",numSets);
     char* cacheptr =  buildCache(numSets, setSize);
     initStore();
+  
     int sock_fd, addrlen, msglen, newConnection, client_socket[MAX_CLIENTS], max_sd;
     struct sockaddr_in socketAddr,joinServer,clientAddr;
     puts("getting socket");
@@ -91,6 +93,17 @@ int main(int argc, char* argv[])
         perror("bind failed");
         exit(1);
     }
+    
+    int id = nodeToId(&socketAddr);
+    sprintf(dir,"/%d/",id);
+
+    struct stat st = {0};
+
+    if (stat(dir, &st) == -1) {
+        mkdir(dir, 0700);
+    }
+
+
     initChordServer(socketAddr);
     // initChordStructure(&joinServer,socketAddr);
     printf("Listening on port %d \n", PORT);

server/kvstore.c

@@ -4,6 +4,9 @@
 #include<pthread.h> 
 #include<unistd.h> 
 #include<stdbool.h>
+#include "chord.h"
+#include "parsexml.h"
+#include "kvstore.h"
 
 #define NUMFILES 10
 typedef struct Store
@@ -47,11 +50,13 @@ bool dumpToFile(char* key, char* val){
 
     int setNo = getFileId(key);
 
-    char fileName[10];
-    char tempfileName[10];
+    char fileName[20];
+    char tempfileName[20];
     bool keyExists = false;
-    strcpy(tempfileName,"temp");
-    sprintf(fileName,"%d",setNo);
+    strcpy(tempfileName,dir);
+    strcat(tempfileName,"temp");
+    // strcpy(fileName,dir);
+    sprintf(fileName,"%s%d",dir,setNo);
     strcat(fileName,".csv");
     strcat(tempfileName,fileName);
 
@@ -68,7 +73,6 @@ bool dumpToFile(char* key, char* val){
     {
         /* code */
         while(!feof(fp)){
-            printf("ch2");
 
             if(fscanf(fp,"%[^,],%[^\n]\n",storedKey,storedVal) == EOF)
                 break;
@@ -84,7 +88,6 @@ bool dumpToFile(char* key, char* val){
     if(!keyExists)
         fprintf(newFp,"%s,%s\n",key,val);
 
-    printf("ch3");
 
     remove(fileName);
     // free(fp);
@@ -104,11 +107,14 @@ bool deleteStore(char* key) {
 
     int setNo = getFileId(key);
 
-    char fileName[10];
-    char tempfileName[10];
+    char fileName[20];
+    char tempfileName[20];
     bool keyExists = false;
-    strcpy(tempfileName,"temp");
-    sprintf(fileName,"%d",setNo);
+    // strcpy(tempfileName,"temp");
+    // sprintf(fileName,"%d",setNo);
+    strcpy(tempfileName,dir);
+    strcat(tempfileName,"temp");
+    sprintf(fileName,"%s%d",dir,setNo);
     strcat(fileName,".csv");
     strcat(tempfileName,fileName);
 
@@ -143,40 +149,49 @@ bool deleteStore(char* key) {
     return keyExists;
 }
 
-char* restoreFromFile(char* key) {
+void TransferKeys(int id, char ip[], int port) {
 
-    /////////////////add/////////////////
-    int setNo = getFileId(key);
-    /////////////////add end/////////////////
+    char * storedKey, *storedVal;
+    char *operation = (char *)malloc(sizeof(char)*7);
+    strcpy(operation,"putreq");
+
+    for (int i = 0; i < NUMFILES; i++)
+    {
+        /* code */
+        // int setNo = getFileId(key);
         
         char fileName[10];
-    sprintf(fileName,"%d",setNo);
+        // sprintf(fileName,"%d",i);
+        sprintf(fileName,"%s%d",dir,i);
         strcat(fileName,".csv");
 
         /////////////////add/////////////////
-    pthread_mutex_lock(&(store.locks[setNo]));
+        pthread_mutex_lock(&(store.locks[i]));
         /////////////////add end/////////////////
 
         FILE *fp = fopen(fileName,"r+");
         if(fp==NULL)
-        return NULL;
-    char* storedKey = malloc(sizeof(char)*256);
-    char* storedVal = malloc(sizeof(char)*256*1024+1);
+            continue;
+        storedKey = malloc(sizeof(char)*256);
+        storedVal = malloc(sizeof(char)*256*1024+1);
         while(1){
             if(fscanf(fp,"%[^,],%[^\n]\n",storedKey,storedVal) == EOF)
                 break;
-        if(!strcmp(key,storedKey)) {
-            free(storedKey);
-            fclose(fp);
-            pthread_mutex_unlock(&(store.locks[setNo]));
-            return storedVal;
+            int keyid = keyToId(storedKey);
+            if(keyid<=id) {
+                // free(storedKey);
+                // pthread_mutex_unlock(&(store.locks[i]));
+                char *msg = toXML(operation,storedKey,storedVal,NULL,0);
+                sendUDP(msg,ip,port,false,NULL);
             }
         }
 
         fclose(fp);
         /////////////////add/////////////////
-    pthread_mutex_unlock(&(store.locks[setNo]));
-    /////////////////add end/////////////////
+        pthread_mutex_unlock(&(store.locks[i]));
+    }
+    
+    ///////////////add end/////////////////
 
     free(storedKey);
     return NULL;
@@ -196,8 +211,9 @@ void dumpToFileXML(char* filename) {
 	fprintf(dumpPtr,"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<KVStore>\n");
 
 	for (int i = 0; i < 10; ++i) {
-		char fileSet[10];
-		sprintf(fileSet,"%d",i);
+		char fileSet[20];
+		// sprintf(fileSet,"%d",i);
+        sprintf(fileSet,"%s%d",dir,i);
 		strcat(fileSet,".csv");
 		FILE *fp = fopen(fileSet,"r+");
 		while(1){
@@ -228,3 +244,42 @@ int iniit() {
     return 0;
 }
 
+char * restoreFromFile(char* key) {
+
+    /////////////////add/////////////////
+    int setNo = getFileId(key);
+    /////////////////add end/////////////////
+    
+    char fileName[20];
+    sprintf(fileName,"%s%d",dir,setNo);
+    // sprintf(fileName,"%d",setNo);
+    strcat(fileName,".csv");
+
+    /////////////////add/////////////////
+    pthread_mutex_lock(&(store.locks[setNo]));
+    /////////////////add end/////////////////
+
+    FILE *fp = fopen(fileName,"r+");
+    if(fp==NULL)
+        return NULL;
+    char* storedKey = malloc(sizeof(char)*256);
+    char* storedVal = malloc(sizeof(char)*256*1024+1);
+    while(1){
+        if(fscanf(fp,"%[^,],%[^\n]\n",storedKey,storedVal) == EOF)
+            break;
+        if(!strcmp(key,storedKey)) {
+            free(storedKey);
+            fclose(fp);
+            pthread_mutex_unlock(&(store.locks[setNo]));
+            return storedVal;
+        }
+    }
+
+    fclose(fp);
+    /////////////////add/////////////////
+    pthread_mutex_unlock(&(store.locks[setNo]));
+    /////////////////add end/////////////////
+
+    free(storedKey);
+    return NULL;
+}

server/kvstore.h

@@ -2,4 +2,6 @@
 bool dumpToFile(char* key,char* val);
 bool deleteStore(char* key);
 char* restoreFromFile(char* key);
+char* TransferKeys(int id, char ip[], int port);
 void initStore();
+char * dir[15];