Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
C
cs733
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
Analytics
Analytics
Repository
Value Stream
Wiki
Wiki
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Commits
Issue Boards
Open sidebar
Sushant Mahajan
cs733
Commits
2099c0f1
Commit
2099c0f1
authored
Feb 09, 2015
by
Harshit Pande
Browse files
Options
Browse Files
Download
Plain Diff
Merge branch 'master' of git.cse.iitb.ac.in:smahajan/cs733
parents
c8890231
4dceaccc
Changes
4
Hide whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
55 additions
and
148 deletions
+55
-148
assignment2/src/connhandler/connhandler.go
assignment2/src/connhandler/connhandler.go
+5
-6
assignment2/src/raft/kvstore.go
assignment2/src/raft/kvstore.go
+5
-97
assignment2/src/raft/raft.go
assignment2/src/raft/raft.go
+32
-38
assignment2/src/server.go
assignment2/src/server.go
+13
-7
No files found.
assignment2/src/connhandler/connhandler.go
View file @
2099c0f1
...
...
@@ -4,7 +4,6 @@ import (
"bufio"
"bytes"
"encoding/gob"
"kvstore"
"net"
"raft"
"strconv"
...
...
@@ -26,7 +25,7 @@ func readValue(ch chan []byte, n uint64) ([]byte, bool) {
up
:=
make
(
chan
bool
,
1
)
//after 5 seconds passed reading value, we'll just send err to client
go
func
()
{
time
.
Sleep
(
kvstore
.
READ_TIMEOUT
*
time
.
Second
)
time
.
Sleep
(
5
*
time
.
Second
)
up
<-
true
}()
...
...
@@ -113,7 +112,7 @@ func MyRead(ch chan []byte, conn net.Conn) {
*/
func
Write
(
conn
net
.
Conn
,
msg
string
)
{
buf
:=
[]
byte
(
msg
)
buf
=
append
(
buf
,
[]
byte
(
kvstore
.
CRLF
)
...
)
buf
=
append
(
buf
,
[]
byte
(
"
\r\n
"
)
...
)
conn
.
Write
(
buf
)
}
...
...
@@ -133,18 +132,18 @@ func HandleClient(conn net.Conn, rft *raft.Raft) {
flag
:=
false
nr
:=
uint64
(
0
)
tokens
:=
strings
.
Fields
(
string
(
msg
))
if
kvstore
.
IsCas
(
tokens
[
0
])
{
if
tokens
[
0
]
==
"CAS"
{
n
,
_
:=
strconv
.
ParseUint
(
tokens
[
3
],
10
,
64
)
nr
=
n
flag
=
true
}
else
if
kvstore
.
IsSet
(
tokens
[
0
])
{
}
else
if
tokens
[
0
]
==
"SET"
{
n
,
_
:=
strconv
.
ParseUint
(
tokens
[
2
],
10
,
64
)
nr
=
n
flag
=
true
}
if
flag
{
if
v
,
err
:=
readValue
(
ch
,
nr
);
err
{
Write
(
conn
,
kvstore
.
ERR_CMD_ERR
)
Write
(
conn
,
"ERR_CMD_ERR"
)
}
else
{
command
.
Val
=
v
//command.isVal = true
...
...
assignment2/src/
kvstore
/kvstore.go
→
assignment2/src/
raft
/kvstore.go
View file @
2099c0f1
package
kvstore
package
raft
import
(
"bytes"
...
...
@@ -61,50 +61,6 @@ var logger *log.Logger
//cache
var
table
*
KeyValueStore
/*Function to start the server and accept connections.
*arguments: none
*return: none
*/
//func startServer() {
// logger.Println("Server started")
// listener, err := net.Listen("tcp", ":5000")
// if err != nil {
// logger.Println("Could not start server!")
// }
// //initialize key value store
// table = &KeyValueStore{dictionary: make(map[string]*Data)}
// //infinite loop
// for {
// conn, err := listener.Accept()
// if err != nil {
// logger.Println(err)
// continue
// }
// go handleClient(conn) //client connection handler
// }
//}
/*Function to read data from the connection and put it on the channel so it could be read in a systematic fashion.
*arguments: channel shared between this go routine and other functions performing actions based on the commands given, client connection
*return: none
*/
//func myRead(ch chan []byte, conn net.Conn) {
// scanner := bufio.NewScanner(conn)
// scanner.Split(CustomSplitter)
// for {
// if ok := scanner.Scan(); !ok {
// break
// } else {
// temp := scanner.Bytes()
// ch <- temp
// logger.Println(temp, "$$")
// }
// }
//}
/*Simple write function to send information to the client
*arguments: client connection, msg to send to the client
*return: none
...
...
@@ -112,30 +68,9 @@ var table *KeyValueStore
func
write
(
conn
net
.
Conn
,
msg
string
)
{
buf
:=
[]
byte
(
msg
)
buf
=
append
(
buf
,
[]
byte
(
CRLF
)
...
)
//logger.Println(buf, len(buf))
conn
.
Write
(
buf
)
}
/*After initial establishment of the connection with the client, this go routine handles further interaction
*arguments: client connection
*return: none
*/
//func handleClient(conn net.Conn) {
// defer conn.Close()
// //channel for every connection for every client
// ch := make(chan []byte)
// go myRead(ch, conn)
// for {
// msg := <-ch
// logger.Println("Channel: ", msg, string(msg))
// if len(msg) == 0 {
// continue
// }
// parseInput(conn, string(msg), table, ch)
// }
//}
/*Basic validations for various commands
*arguments: command to check against, other parmameters sent with the command (excluding the value), client connection
*return: integer representing error state
...
...
@@ -568,31 +503,6 @@ func debug() {
logger
.
Println
(
"----end debug----"
)
}
/*Entry point of this program. Initializes the start of ther server and sets up the logger.
*arguments: none
*return: none
*/
//func main() {
// toLog := ""
// 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)
// }
// go startServer()
// var input string
// fmt.Scanln(&input)
//}
func
InitKVStore
()
{
toLog
:=
""
if
len
(
os
.
Args
)
>
1
{
...
...
@@ -613,12 +523,10 @@ func InitKVStore() {
table
=
&
KeyValueStore
{
dictionary
:
make
(
map
[
string
]
*
Data
)}
}
func
IsCas
(
msg
string
)
bool
{
return
msg
==
CAS
}
func
IsSet
(
msg
string
)
bool
{
return
msg
==
SET
func
MonitorCommitChannel
(
ch
chan
LogEntry
)
{
for
{
temp
:=
<-
ch
}
}
//server will not call this, we'll call it from test cases to clear the map
...
...
assignment2/src/raft/raft.go
View file @
2099c0f1
...
...
@@ -10,6 +10,7 @@ import (
//constant values used
const
(
CLIENT_PORT
=
9000
LOG_PORT
=
8000
ACK_TIMEOUT
=
5
)
...
...
@@ -19,21 +20,9 @@ var Info *log.Logger
// Flag for enabling/disabling logging functionality
var
DEBUG
=
true
type
Lsn
uint64
//Log sequence number, unique for all time.
type
ErrRedirect
int
// See Log.Append. Implements Error interface.
type
LogEntry
interface
{
Lsn
()
Lsn
Data
()
[]
byte
Committed
()
bool
}
type
LogEntryData
struct
{
id
Lsn
data
[]
byte
committed
bool
}
type
Lsn
uint64
//Log sequence number, unique for all time.
type
ServerConfig
struct
{
Id
int
// Id of server. Must be unique
...
...
@@ -58,6 +47,19 @@ type Raft struct {
id
int
//this server id
}
type
LogEntry
interface
{
Lsn
()
Lsn
Data
()
[]
byte
Committed
()
bool
}
type
LogEntryData
struct
{
id
Lsn
data
[]
byte
committed
bool
conn
net
.
Conn
}
type
Args
struct
{
X
int
}
...
...
@@ -78,18 +80,8 @@ func NewRaft(config *ClusterConfig, thisServerId int, commitCh chan LogEntry) (*
return
rft
,
nil
}
//goroutine that monitors channel for commiting log entry
func
monitorCommitChannel
(
raft
*
Raft
)
{
//unidirectional -- can only read from the channel
for
{
//var temp LogEntry
temp
:=
<-
raft
.
commitCh
//receive from the channel
raft
.
log_array
[
temp
.
(
*
LogEntryData
)
.
id
]
.
committed
=
true
//commit the value
//update the kv store here
}
}
//goroutine that monitors channel to check if the majority of servers have replied
func
monitorAckChannel
(
rft
*
Raft
,
ack_ch
<-
chan
int
,
log_entry
LogEntry
,
maj
EventCh
chan
int
)
{
func
monitorAckChannel
(
rft
*
Raft
,
ack_ch
<-
chan
int
,
log_entry
LogEntry
,
maj
Ch
chan
bool
)
{
acks_received
:=
0
num_servers
:=
len
(
rft
.
cluster_config
.
Servers
)
required_acks
:=
num_servers
/
2
...
...
@@ -106,8 +98,9 @@ func monitorAckChannel(rft *Raft, ack_ch <-chan int, log_entry LogEntry, majEven
case
temp
:=
<-
ack_ch
:
acks_received
+=
temp
if
acks_received
==
required_acks
{
rft
.
log_array
[
log_entry
.
(
*
LogEntryData
)
.
id
]
.
committed
=
true
rft
.
commitCh
<-
log_entry
maj
EventCh
<-
1
maj
Ch
<-
true
err
=
true
break
}
...
...
@@ -136,20 +129,20 @@ func (entry *LogEntryData) Committed() bool {
}
//make raft implement the append function
func
(
r
aft
*
Raft
)
Append
(
data
[]
byte
)
(
LogEntry
,
error
)
{
if
r
a
ft
.
id
!=
1
{
func
(
r
ft
*
Raft
)
Append
(
data
[]
byte
,
conn
net
.
Conn
)
(
LogEntry
,
error
)
{
if
rft
.
id
!=
1
{
return
nil
,
ErrRedirect
(
1
)
}
temp
:=
new
(
LogEntryData
)
temp
.
id
=
1
temp
.
committed
=
false
temp
.
data
=
data
raft
.
log_array
=
append
(
raft
.
log_array
,
temp
)
temp
.
conn
=
conn
rft
.
log_array
=
append
(
rft
.
log_array
,
temp
)
ackChan
:=
make
(
chan
int
)
majEventCh
:=
make
(
chan
int
)
go
monitorAckChannel
(
raft
,
ackChan
,
temp
,
majEventCh
)
go
monitorCommitChannel
(
raft
)
majChan
:=
make
(
chan
bool
)
go
monitorAckChannel
(
rft
,
ackChan
,
temp
,
majChan
)
for
_
,
server
:=
range
cluster_config
.
Servers
[
1
:
]
{
go
func
(
ackChan
chan
int
)
{
...
...
@@ -164,19 +157,20 @@ func (raft *Raft) Append(data []byte) (LogEntry, error) {
ackChan
<-
reply
.
X
}(
ackChan
)
}
//channel will return 1 if majority
if
<-
maj
EventCh
==
1
{
raft
.
commitCh
<-
temp
if
<-
maj
Chan
{
//
}
return
temp
,
nil
}
func
NewServerConfig
(
server
_i
d
int
)
(
*
ServerConfig
,
error
)
{
func
NewServerConfig
(
server
I
d
int
)
(
*
ServerConfig
,
error
)
{
server
:=
new
(
ServerConfig
)
server
.
Id
=
server
_i
d
server
.
Id
=
server
I
d
server
.
Hostname
=
"127.0.0.1"
server
.
ClientPort
=
CLIENT_PORT
server
.
LogPort
=
CLIENT_PORT
+
server_i
d
server
.
ClientPort
=
CLIENT_PORT
+
serverId
server
.
LogPort
=
LOG_PORT
+
serverI
d
return
server
,
nil
}
...
...
assignment2/src/server.go
View file @
2099c0f1
...
...
@@ -3,7 +3,6 @@ package main
import
(
"connhandler"
"fmt"
"io/ioutil"
"log"
"net"
...
...
@@ -34,7 +33,7 @@ func (t *AppendEntries) AppendEntriesRPC(args *Args, reply *Reply) error {
return
nil
}
func
initInterServerCommunication
(
server
*
raft
.
ServerConfig
,
rft
*
raft
.
Raft
)
{
func
initInterServerCommunication
(
server
*
raft
.
ServerConfig
,
rft
*
raft
.
Raft
,
ch
chan
bool
)
{
appendRpc
:=
new
(
AppendEntries
)
rpc
.
Register
(
appendRpc
)
listener
,
e
:=
net
.
Listen
(
"tcp"
,
":"
+
strconv
.
Itoa
(
server
.
LogPort
))
...
...
@@ -49,6 +48,7 @@ func initInterServerCommunication(server *raft.ServerConfig, rft *raft.Raft) {
go
rpc
.
ServeConn
(
conn
)
}
}
ch
<-
true
}
// Initialize Logger
...
...
@@ -64,7 +64,7 @@ func initLogger(serverId int) {
Info
.
Println
(
"Initialized server"
)
}
func
initClientCommunication
(
server
*
raft
.
ServerConfig
,
rft
*
raft
.
Raft
)
{
func
initClientCommunication
(
server
*
raft
.
ServerConfig
,
rft
*
raft
.
Raft
,
ch
chan
bool
)
{
listener
,
e
:=
net
.
Listen
(
"tcp"
,
":"
+
strconv
.
Itoa
(
server
.
ClientPort
))
if
e
!=
nil
{
Info
.
Fatal
(
"client listen error:"
,
e
)
...
...
@@ -77,10 +77,13 @@ func initClientCommunication(server *raft.ServerConfig, rft *raft.Raft) {
go
connhandler
.
HandleClient
(
conn
,
rft
)
}
}
ch
<-
true
}
func
main
()
{
sid
,
err
:=
strconv
.
Atoi
(
os
.
Args
[
1
])
ch1
:=
make
(
chan
bool
)
ch2
:=
make
(
chan
bool
)
if
err
!=
nil
{
Info
.
Println
(
"argument "
,
os
.
Args
[
1
],
"is not string"
)
}
...
...
@@ -98,10 +101,13 @@ func main() {
commitCh
:=
make
(
chan
raft
.
LogEntry
)
rft
,
_
:=
raft
.
NewRaft
(
clusterConfig
,
sid
,
commitCh
)
raft
.
InitKVStore
()
initClientCommunication
(
server
,
rft
)
initInterServerCommunication
(
server
,
rft
)
go
raft
.
MonitorCommitChannel
(
commitCh
)
go
initClientCommunication
(
server
,
rft
,
ch1
)
go
initInterServerCommunication
(
server
,
rft
,
ch2
)
var
dummy
string
fmt
.
Scanln
(
&
dummy
)
for
<-
ch1
&&
<-
ch2
{
}
}
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment