rename temp.c top StorageHandler

KVServer.c

@@ -8,6 +8,7 @@
 #include <sys/epoll.h>
 #include <string.h> 
 #include "LRU.h"
+#include "StorageHandler.h"
 #include "KVMessageFormat.h"
 #define MAX_EVENTS 10
 #define DEBUG (0)
@@ -33,7 +34,7 @@ void *worker(void *args) {
     int status;
 
 
-    // Generate name for each thread for debugging
+    /* Generate name for each thread for debugging */
     char *name = (char *) malloc(5*sizeof(char));
     gen_random(name, 5);
     if DEBUG printf("[%s] Thread started!\n", name);
@@ -51,7 +52,7 @@ void *worker(void *args) {
         perror("epoll_ctl: read_pipe");
         exit(EXIT_FAILURE);
     }
-    //if DEBUG printf("[%s] Added read pipe to epoll fd set!\n", name);
+
     while (1) {
 
         if DEBUG printf("[%s] waiting for epoll event!\n", name);
@@ -91,8 +92,7 @@ void *worker(void *args) {
                 if (flag & EPOLLIN) {
                     /* Parse the actual message from client */
                     struct message *requestMessage= malloc(sizeof(struct message));
-                    int readlength=read(events[i].data.fd  , requestMessage, sizeof(struct message)); 
-                    //printf("[Message Received from client] (%d, %s, %s)\n",requestMessage->status,requestMessage->key,requestMessage->value);               
+                    int readlength=read(events[i].data.fd  , requestMessage, sizeof(struct message));              
                     
                     if DEBUG printf("[%s][EVENT][EPOLLIN] \n", name);
                     switch(requestMessage->status) {
@@ -175,6 +175,7 @@ int main (int argc, int argv) {
     clilen = sizeof(cli_addr);
 
     init_cache();
+    init_storage()
 
     while(1) {
 

StorageHandler.c

+#include<semaphore.h>
+#include<stdio.h>
+#include<stdlib.h>
+#include<unistd.h>
+#include<pthread.h>
+#include <stdint.h>
+#include <limits.h>
+#include<fcntl.h> 
+#include<unistd.h>
+#include<string.h>
+#include<sys/stat.h>
+#include <sys/types.h>
+
+int *fds;
+int setSize=2;
+int *readCounters;
+sem_t *mutex;
+sem_t *readerLocks;
+sem_t x,y;
+pthread_t tid;
+pthread_t writerthreads[100],readerthreads[100];
+int readercount = 0;
+
+int modulus(char *num, int size, int divisor) {
+  int rem = 0;
+ 
+  while (size-- > 0) {
+    rem = ((UCHAR_MAX + 1)*rem + *num) % divisor;
+    num++;
+  }
+  return rem;
+}
+void file_del(off_t offset, char *key)
+{   
+    int index=modulus(key,256,setSize);
+    fflush(stdout);
+    int length=0;
+    sem_wait(&mutex[index]);
+    char blankspace[512];
+     char ch;
+    int k=-1;
+    lseek(fds[index], offset, SEEK_SET);
+    while(read(fds[index], &ch,sizeof(ch))!=-1 && ch!='\n')
+    {  
+        length++;
+    }
+    printf("length %d",length);
+    memset(blankspace, 0, length);
+    lseek(fds[index], offset, SEEK_SET);
+    write(fds[index], blankspace, length);
+    sem_post(&mutex[index]);
+    
+}
+void file_get(off_t offset, char *key, char *value)
+{   
+    /* Gets the value stored at offset */
+    /* Does not depend on key argument */
+
+    int index=modulus(key,256,setSize);
+    sem_wait(&readerLocks[index]);
+    readCounters[index]+=1;
+    if(readCounters[index]==1)
+        sem_wait(&mutex[index]);
+    sem_post(&readerLocks[index]);
+    
+    lseek(fds[index], offset, SEEK_SET);
+    char line[10];
+    //FILE *fp =fdopen(fds[index],"r+");
+   // fseek(fp, offset,SEEK_SET);
+    size_t len=0;
+    
+    char ch;
+    int k=-1;
+    while(read(fds[index], &ch,sizeof(ch))!=-1 && ch!='\n')
+    {  
+        if(k>=0)
+        {
+        value[k++]=ch;
+        }
+        if(ch==':')
+        {
+            k=0;  
+        }
+    }
+    sem_wait(&readerLocks[index]);
+    readCounters[index]-=1;
+    if(readCounters[index]==0)
+    {
+        sem_post(&mutex[index]);
+    }
+    sem_post(&readerLocks[index]);
+}
+
+off_t file_put(char *key,char *value) {
+
+    int index=modulus(key,256,setSize);
+    int bytes, rembytes, i;
+    off_t position;
+
+    sem_wait(&mutex[index]);
+    printf("[Write to File: %d]\n",index);
+    position = lseek(fds[index], 0, SEEK_END);
+    char line [514];
+    snprintf(line, sizeof(line),"%s:%s\n",key,value);
+    if(write(fds[index], line, strlen(line))<0)
+        printf("\n[unable to perform file_put]\n");
+    sem_post(&mutex[index]);
+
+    return position;
+}
+
+int init_storage() {
+    /*
+        define the array of file descriptors  depending on the prefix 
+        define the array of readCount as well as the semaphore (read x and write y) for the same
+        PUT,DEL would use write lock 
+        GET would use read lock
+        each write should return the line number
+    */
+    fds=(int *)malloc(sizeof(int)*setSize);
+    readCounters=(int *)malloc(sizeof(int)*setSize);
+    readerLocks=(sem_t *)malloc(sizeof(sem_t)*setSize);
+    mutex=(sem_t *)malloc(sizeof(sem_t)*setSize);
+    int i=0;
+    char fileName[setSize+20];
+    for(i=0;i<setSize;i++)
+    {
+        snprintf(fileName,sizeof(fileName),"File%d.txt",i);
+        fds[i]=open(fileName, O_CREAT|O_RDWR,S_IRWXU);
+        if(fds[i]<0)
+        {
+            printf("\n[Unable to Open File%d.txt]\n",i);
+        }
+        sem_init(&readerLocks[i],0,1);
+        sem_init(&mutex[i],0,1);
+        readCounters[i]=0;
+    }
+    return 0;
+    
+}
\ No newline at end of file

StorageHandler.h

+#include <sys/types.h>
+
+void * file_del(off_t offset, char *key);
+
+void file_get(off_t offset, char *key, char *value);
+
+off_t file_put(char *key,char *value);
+
+int init_storage();
+