Commit e85133be authored by Sushant Mahajan's avatar Sushant Mahajan

modified kvstore to work with raft

parent dffef763
...@@ -144,6 +144,7 @@ func HandleClient(conn net.Conn, rft *raft.Raft) { ...@@ -144,6 +144,7 @@ func HandleClient(conn net.Conn, rft *raft.Raft) {
if flag { if flag {
if v, err := readValue(ch, nr); err { if v, err := readValue(ch, nr); err {
Write(conn, "ERR_CMD_ERR") Write(conn, "ERR_CMD_ERR")
continue
} else { } else {
command.Val = v command.Val = v
//command.isVal = true //command.isVal = true
......
...@@ -10,6 +10,7 @@ import ( ...@@ -10,6 +10,7 @@ import (
"strings" "strings"
"sync" "sync"
"time" "time"
"utils"
) )
/*Constants used throughout the program to identify commands, request, response, and error messages*/ /*Constants used throughout the program to identify commands, request, response, and error messages*/
...@@ -35,7 +36,7 @@ const ( ...@@ -35,7 +36,7 @@ const (
ERR_INTERNAL = "ERR_INTERNAL" ERR_INTERNAL = "ERR_INTERNAL"
//constant //constant
MAX_CMD_ARGS = 6 MAX_CMD_ARGS = 5
MIN_CMD_ARGS = 2 MIN_CMD_ARGS = 2
READ_TIMEOUT = 5 READ_TIMEOUT = 5
) )
...@@ -78,7 +79,7 @@ func write(conn net.Conn, msg string) { ...@@ -78,7 +79,7 @@ func write(conn net.Conn, msg string) {
func isValid(cmd string, tokens []string, conn net.Conn) int { func isValid(cmd string, tokens []string, conn net.Conn) int {
switch cmd { switch cmd {
case SET: case SET:
if len(tokens) > 5 || len(tokens) < 4 { if len(tokens) != 4 {
logger.Println(cmd, ":Invalid no. of tokens") logger.Println(cmd, ":Invalid no. of tokens")
write(conn, ERR_CMD_ERR) write(conn, ERR_CMD_ERR)
return 1 return 1
...@@ -88,11 +89,6 @@ func isValid(cmd string, tokens []string, conn net.Conn) int { ...@@ -88,11 +89,6 @@ func isValid(cmd string, tokens []string, conn net.Conn) int {
write(conn, ERR_CMD_ERR) write(conn, ERR_CMD_ERR)
return 1 return 1
} }
if len(tokens) == 5 && tokens[4] != NOREPLY {
logger.Println(cmd, ":optional arg incorrect")
write(conn, ERR_CMD_ERR)
return 1
}
if _, err := strconv.ParseUint(tokens[2], 10, 64); err != nil { if _, err := strconv.ParseUint(tokens[2], 10, 64); err != nil {
logger.Println(cmd, ":expiry time invalid") logger.Println(cmd, ":expiry time invalid")
write(conn, ERR_CMD_ERR) write(conn, ERR_CMD_ERR)
...@@ -129,7 +125,7 @@ func isValid(cmd string, tokens []string, conn net.Conn) int { ...@@ -129,7 +125,7 @@ func isValid(cmd string, tokens []string, conn net.Conn) int {
} }
case CAS: case CAS:
if len(tokens) > 6 || len(tokens) < 5 { if len(tokens) != 5 {
logger.Println(cmd, ":Invalid number of tokens") logger.Println(cmd, ":Invalid number of tokens")
write(conn, ERR_CMD_ERR) write(conn, ERR_CMD_ERR)
return 1 return 1
...@@ -139,11 +135,6 @@ func isValid(cmd string, tokens []string, conn net.Conn) int { ...@@ -139,11 +135,6 @@ func isValid(cmd string, tokens []string, conn net.Conn) int {
write(conn, ERR_CMD_ERR) write(conn, ERR_CMD_ERR)
return 1 return 1
} }
if len(tokens) == 6 && tokens[5] != NOREPLY {
logger.Println(cmd, ":optional arg incorrect")
write(conn, ERR_CMD_ERR)
return 1
}
if _, err := strconv.ParseUint(tokens[2], 10, 64); err != nil { if _, err := strconv.ParseUint(tokens[2], 10, 64); err != nil {
logger.Println(cmd, ":expiry time invalid") logger.Println(cmd, ":expiry time invalid")
write(conn, ERR_CMD_ERR) write(conn, ERR_CMD_ERR)
...@@ -179,12 +170,26 @@ func isValid(cmd string, tokens []string, conn net.Conn) int { ...@@ -179,12 +170,26 @@ func isValid(cmd string, tokens []string, conn net.Conn) int {
return 0 return 0
} }
func MonitorCommitChannel(ch chan LogEntry) {
for {
buffer := new(bytes.Buffer)
temp := <-ch
conn := temp.(*LogEntryData).conn
cmd := new(utils.Command)
if err := cmd.GobDecode(temp.Data()); err != nil {
log.Fatal("Error decoding command!")
}
ParseInput(conn, cmd)
}
}
/*Function parses the command provided by the client and delegates further action to command specific functions. /*Function parses the command provided by the client and delegates further action to command specific functions.
*Based on the return values of those functions, send appropriate messages to the client. *Based on the return values of those functions, send appropriate messages to the client.
*arguments: client connection, message from client, channel shared with myRead function *arguments: client connection, message from client, channel shared with myRead function
*return: none *return: none
*/ */
func ParseInput(conn net.Conn, msg string, ch chan []byte) { func ParseInput(conn net.Conn, cmd *utils.Command) {
msg := string(cmd.Cmd)
tokens := strings.Fields(msg) tokens := strings.Fields(msg)
//general error, don't check for commands, avoid the pain ;) //general error, don't check for commands, avoid the pain ;)
if len(tokens) > MAX_CMD_ARGS || len(tokens) < MIN_CMD_ARGS { if len(tokens) > MAX_CMD_ARGS || len(tokens) < MIN_CMD_ARGS {
...@@ -201,7 +206,7 @@ func ParseInput(conn net.Conn, msg string, ch chan []byte) { ...@@ -201,7 +206,7 @@ func ParseInput(conn net.Conn, msg string, ch chan []byte) {
if isValid(SET, tokens, conn) != 0 { if isValid(SET, tokens, conn) != 0 {
return return
} }
if ver, ok, r := performSet(conn, tokens[1:len(tokens)], ch); ok { if ver, ok, r := performSet(conn, tokens[1:len(tokens)], cmd); ok {
//debug(table) //debug(table)
logger.Println(ver) logger.Println(ver)
if r { if r {
...@@ -267,7 +272,7 @@ func ParseInput(conn net.Conn, msg string, ch chan []byte) { ...@@ -267,7 +272,7 @@ func ParseInput(conn net.Conn, msg string, ch chan []byte) {
if isValid(CAS, tokens, conn) != 0 { if isValid(CAS, tokens, conn) != 0 {
return return
} }
if ver, ok, r := performCas(conn, tokens[1:len(tokens)], ch); r { if ver, ok, r := performCas(conn, tokens[1:len(tokens)], cmd); r {
if r { if r {
switch ok { switch ok {
case 0: case 0:
...@@ -313,50 +318,40 @@ func ParseInput(conn net.Conn, msg string, ch chan []byte) { ...@@ -313,50 +318,40 @@ func ParseInput(conn net.Conn, msg string, ch chan []byte) {
} }
/*Delegate function responsible for all parsing and hashtable interactions for the SET command sent by client /*Delegate function responsible for all parsing and hashtable interactions for the SET command sent by client
*arguments: client connection, tokenized command sent by the client, channel shared with myRead *arguments: client connection, tokenized command sent by the client, command structure @utils.Command
*return: version of inserted key (if successful, 0 otherwise), success or failure, whether to send reply to client *return: version of inserted key (if successful, 0 otherwise), success or failure, whether to send reply to client
*/ */
func performSet(conn net.Conn, tokens []string, ch chan []byte) (uint64, bool, bool) { func performSet(conn net.Conn, tokens []string, cmd *utils.Command) (uint64, bool, bool) {
//-k := tokens[0] k := tokens[0]
//expiry time offset //expiry time offset
//-e, _ := strconv.ParseUint(tokens[1], 10, 64) e, _ := strconv.ParseUint(tokens[1], 10, 64)
//numbytes //numbytes
//-n, _ := strconv.ParseUint(tokens[2], 10, 64) n, _ := strconv.ParseUint(tokens[2], 10, 64)
r := true r := true
if len(tokens) == 4 && tokens[3] == NOREPLY {
r = false
}
logger.Println(r) logger.Println(r)
//if v, err := readValue(ch, n); err { defer table.Unlock()
// write(conn, ERR_CMD_ERR) table.Lock()
// return 0, false, r //critical section start
//} else { var val *Data
// defer table.Unlock() if _, ok := table.dictionary[k]; ok {
// table.Lock() val = table.dictionary[k]
// //critical section start } else {
// var val *Data val = new(Data)
// if _, ok := table.dictionary[k]; ok { table.dictionary[k] = val
// val = table.dictionary[k] }
// } else { val.numBytes = n
// val = new(Data) val.version++
// table.dictionary[k] = val if e == 0 {
// } val.isPerpetual = true
// val.numBytes = n val.expTime = 0
// val.version++ } else {
// if e == 0 { val.isPerpetual = false
// val.isPerpetual = true val.expTime = e + uint64(time.Now().Unix())
// val.expTime = 0 }
// } else { val.value = cmd.Val
// val.isPerpetual = false return val.version, true, r
// val.expTime = e + uint64(time.Now().Unix())
// }
// val.value = v
// return val.version, true, r
//}
return 2, true, true
} }
/*Delegate function reponsible for activities related to the GET command sent by the client. /*Delegate function reponsible for activities related to the GET command sent by the client.
...@@ -413,11 +408,11 @@ func performGetm(conn net.Conn, tokens []string) (*Data, bool) { ...@@ -413,11 +408,11 @@ func performGetm(conn net.Conn, tokens []string) (*Data, bool) {
} }
/*Delegate function reponsible for activities related to the CAS command sent by the client. /*Delegate function reponsible for activities related to the CAS command sent by the client.
*arguments: client connection, tokenized command sent by the client, channel shared with myRead *arguments: client connection, tokenized command sent by the client, cmd pointer @utils.Command
*return: new version of updated key (if it is updated), error status {0: error while reading new value, 1: key found and changed, *return: new version of updated key (if it is updated), error status {0: error while reading new value, 1: key found and changed,
*2: version mismatch with key, 3: key not found}, whether to reply to client *2: version mismatch with key, 3: key not found}, whether to reply to client
*/ */
func performCas(conn net.Conn, tokens []string, ch chan []byte) (uint64, int, bool) { func performCas(conn net.Conn, tokens []string, cmd *utils.Command) (uint64, int, bool) {
k := tokens[0] k := tokens[0]
e, _ := strconv.ParseUint(tokens[1], 10, 64) e, _ := strconv.ParseUint(tokens[1], 10, 64)
ve, _ := strconv.ParseUint(tokens[2], 10, 64) ve, _ := strconv.ParseUint(tokens[2], 10, 64)
...@@ -425,46 +420,37 @@ func performCas(conn net.Conn, tokens []string, ch chan []byte) (uint64, int, bo ...@@ -425,46 +420,37 @@ func performCas(conn net.Conn, tokens []string, ch chan []byte) (uint64, int, bo
r := true r := true
logger.Println(k, e, ve, n, r) logger.Println(k, e, ve, n, r)
if len(tokens) == 5 && tokens[4] == NOREPLY {
r = false
}
//read value defer table.Unlock()
//if v, err := readValue(ch, n); err { table.Lock()
// return 0, 1, r if val, ok := table.dictionary[k]; ok {
//} else { if val.version == ve {
// defer table.Unlock() if val.isPerpetual || val.expTime >= uint64(time.Now().Unix()) {
// table.Lock() //if expiry time is zero, key should not be deleted
// if val, ok := table.dictionary[k]; ok { if e == 0 {
// if val.version == ve { val.isPerpetual = true
// if val.isPerpetual || val.expTime >= uint64(time.Now().Unix()) { val.expTime = 0
// //if expiry time is zero, key should not be deleted } else {
// if e == 0 { val.isPerpetual = false
// val.isPerpetual = true val.expTime = e + uint64(time.Now().Unix())
// val.expTime = 0 }
// } else { val.numBytes = n
// val.isPerpetual = false val.version++
// val.expTime = e + uint64(time.Now().Unix()) val.value = cmd.Val
// } //key found and changed
// val.numBytes = n return val.version, 0, r
// val.version++ } else {
// val.value = v logger.Println("expired key found!")
// //key found and changed //version found but key expired, can delete key safely and tell client that it does not exist
// return val.version, 0, r delete(table.dictionary, k)
// } else { return 0, 3, r
// logger.Println("expired key found!") }
// //version found but key expired, can delete key safely and tell client that it does not exist }
// delete(table.dictionary, k) //version mismatch
// return 0, 3, r return 0, 2, r
// } }
// } //key not found
// //version mismatch return 0, 3, r
// return 0, 2, r
// }
// //key not found
// return 0, 3, r
//}
return 1, 1, true
} }
/*Delegate function reponsible for activities related to the DELETE command sent by the client. /*Delegate function reponsible for activities related to the DELETE command sent by the client.
...@@ -503,32 +489,13 @@ func debug() { ...@@ -503,32 +489,13 @@ func debug() {
logger.Println("----end debug----") logger.Println("----end debug----")
} }
func InitKVStore() { func InitKVStore(log *log.Logger) {
toLog := "" logger := log
if len(os.Args) > 1 {
toLog = os.Args[1]
}
//toLog = "s"
if toLog != "" {
logf, _ := os.OpenFile("serverlog.log", os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0666)
defer logf.Close()
logger = log.New(logf, "SERVER: ", log.Ltime|log.Lshortfile)
//logger = log.New(os.Stdout, "SERVER: ", log.Ltime|log.Lshortfile)
} else {
logger = log.New(ioutil.Discard, "SERVER: ", log.Ldate)
}
//initialize key value store //initialize key value store
table = &KeyValueStore{dictionary: make(map[string]*Data)} table = &KeyValueStore{dictionary: make(map[string]*Data)}
} }
func MonitorCommitChannel(ch chan LogEntry) {
for {
//temp := <-ch
}
}
//server will not call this, we'll call it from test cases to clear the map //server will not call this, we'll call it from test cases to clear the map
func ReInitServer() { func ReInitServer() {
defer table.Unlock() defer table.Unlock()
......
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