Init repo with basic code structure.

sdfs/datanode.go

+/* Datanode.go
+Data: {filename, version} sdfs/localfile_version
+Func:
+	Handle put/get/delete
+*/
+
+package sdfs
+
+import (
+	"encoding/json"
+	"log"
+	"net"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type Datanode struct {
+	IP    string
+	Port  string
+	Files map[string]int // filename, latest version
+}
+
+func InitDatanode(ip string) *Datanode {
+	d := new(Datanode)
+	d.IP = ip
+	d.Port = "5005"
+	d.Files = make(map[string]int)
+	return d
+}
+
+func (n *Datanode) Run() {
+	//TODO
+	go n.Listener()
+}
+
+func (n *Datanode) Listener() {
+	// Create a TCP listener for Datanodes.
+	address := "0.0.0.0" + ":" + n.Port
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		log.Fatalf("Datanode: Failed to start Datanode listener: %v", err)
+	}
+	defer listener.Close()
+
+	log.Println("Datanode: datanode is listening...")
+
+	// Listen for incoming messages via network communication
+	for {
+		// Accept incoming connections
+		conn, err := listener.Accept()
+		if err != nil {
+			log.Println("Datanode: Error accepting connection:", err)
+			continue
+		}
+
+		receivedMessage, err := n.readMessage(conn)
+		if err != nil {
+			log.Printf("Datanode: Error received message: %v\n", err)
+			continue
+		}
+		log.Printf("Datanode: Received message %+v\n", receivedMessage)
+
+		go n.handleMessage(receivedMessage, conn)
+	}
+}
+
+func (dn *Datanode) handleMessage(receivedMessage Message, conn net.Conn) {
+	// Listen for Datanode requests and process them (e.g., put/get/delete operations)
+	sender := receivedMessage.Sender
+	messageType := receivedMessage.Type
+
+	switch messageType {
+	case RECOVER:
+		dn.HandleRecover(conn, sender)
+	default:
+		log.Println("Datanode: wrong message type")
+	}
+}
+
+func (n *Datanode) HandleRecover(conn net.Conn, sender string) {
+	defer conn.Close()
+	// Step 1: find the latest version for each file in /sdfs
+	replicaMap, err := generateReplicaMap("/sdfs")
+	if err != nil {
+		log.Println("Datanode: Failed to find replicas for files in /sdfs", err)
+	}
+	// Step 2: write back the latest version
+	n.writeMessage(conn, RECOVER, replicaMap, time.Now().Unix())
+}
+
+/* Helper */
+func (c *Datanode) writeMessage(conn net.Conn, msgType MsgType, payload interface{}, id int64) error {
+	message := Message{
+		ID:      id,
+		Sender:  c.IP,
+		Type:    msgType,
+		Payload: payload,
+	}
+	request, err := json.Marshal(message)
+	if err != nil {
+		return err
+	}
+	conn.Write([]byte(request))
+	return nil
+}
+
+func (c *Datanode) readMessage(conn net.Conn) (Message, error) {
+	var message Message
+	buffer := make([]byte, 1024)
+	numBytes, err := conn.Read(buffer)
+	if err != nil {
+		return message, err
+	}
+	data := buffer[:numBytes]
+	// Decode the incoming data into a Message struct
+	err = json.Unmarshal(data, &message)
+	if err != nil {
+		return message, err
+	}
+	return message, nil
+}
+
+func generateReplicaMap(directoryPath string) (map[string]Replica, error) {
+	replicaMap := make(map[string]Replica)
+
+	files, err := os.ReadDir(directoryPath)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, file := range files {
+		if !file.IsDir() {
+			filename := file.Name()
+			parts := strings.Split(filename, "-")
+			if len(parts) != 2 {
+				continue // Skip files with invalid filenames
+			}
+
+			address := parts[0]
+			versionStr := parts[1]
+			version, err := strconv.Atoi(versionStr)
+			if err != nil {
+				continue // Skip files with invalid version numbers
+			}
+
+			// Assuming you already have a replicaMap with existing entries
+			if existingReplica, exists := replicaMap[filename]; exists {
+				if version > existingReplica.Version {
+					// Update the version only if it's greater
+					replicaMap[filename] = Replica{
+						Address: address,
+						Version: version,
+					}
+				}
+			} else {
+				// If the filename doesn't exist in the map, create a new entry
+				replicaMap[filename] = Replica{
+					Address: address,
+					Version: version,
+				}
+			}
+		}
+	}
+
+	return replicaMap, nil
+}

sdfs/go.mod

+module cs425mp4/sdfs
+
+go 1.21.0

sdfs/server.go

+// server.go
+// sdfs node that can join the sdfs, keep track of current leader/leader re-election, switch between datanode and Namenode
+
+package sdfs
+
+import (
+	"context"
+	"log"
+	"strings"
+)
+
+type Server struct {
+	ID       string // hostname : ip address : port : timestamp
+	Namenode string // id of current Namenode
+	Role     string // Namenode or datanode
+}
+
+func InitializeServer(ID, Namenode string) *Server {
+	s := new(Server)
+	s.ID = ID
+	s.Namenode = Namenode
+	s.Role = "datanode"
+	if Namenode == parseHostname(ID) {
+		s.Role = "Namenode"
+	}
+	return s
+}
+
+func (s *Server) SetNamenode(Namenode string) {
+	s.Namenode = Namenode
+	if Namenode == parseHostname(s.ID) {
+		log.Println("Server: set role as Namenode")
+		s.Role = "Namenode"
+	}
+}
+
+func (s *Server) Run(ctx context.Context) {
+	dn := InitDatanode(parseAddress(s.ID))
+	nn := InitNamenode()
+	dn.Run()
+	nn.Run()
+}
+
+/* Helper functions */
+func parseHostname(id string) string {
+	components := strings.Split(id, ":")
+	return components[0]
+}
+
+func parseAddress(id string) string {
+	components := strings.Split(id, ":")
+	return components[1] + components[2]
+}

sdfs/token.go

+package sdfs
+
+import (
+	"sync"
+)
+
+type File struct {
+	filename   string
+	rwMutex    sync.RWMutex
+	readSem    chan struct{}
+	writeSem   chan struct{}
+	readers    int
+	writers    int
+	readerCond *sync.Cond
+	writerCond *sync.Cond
+	mux        sync.Mutex
+}
+
+func GetFile(filename string) *File {
+	file := &File{
+		filename: filename,
+		rwMutex:  sync.RWMutex{},
+		readSem:  make(chan struct{}, 2),
+		writeSem: make(chan struct{}, 1),
+		readers:  0,
+		writers:  0,
+		mux:      sync.Mutex{},
+	}
+	file.readerCond = sync.NewCond(&file.mux)
+	file.writerCond = sync.NewCond(&file.mux)
+	return file
+}
+
+func (f *File) GetReaderToken() {
+	// increment semaphore
+	f.readSem <- struct{}{}
+	// limit at most 4 successive readers
+
+	f.mux.Lock()
+	for f.readers >= 4 && len(f.writeSem) > 0 {
+		f.readerCond.Wait()
+	}
+	f.readers++
+	f.mux.Unlock()
+	// get reader token, any write cannot get token
+	f.rwMutex.RLock()
+	f.writers = 0
+	// Signal waiting writer that they may be able to proceed.
+	f.writerCond.Broadcast()
+}
+
+func (f *File) ReleaseReaderToken() {
+	// unlock WRMutex so that write can get WRMutex
+	f.rwMutex.RUnlock()
+	// descrease the semaphore.
+	<-f.readSem
+}
+
+func (f *File) GetWriterToken() {
+	// increment semaphore
+	f.writeSem <- struct{}{}
+	for f.writers >= 4 && len(f.readSem) > 0 {
+		// Wait until the conditions are met for a writer token.
+		f.writerCond.Wait()
+	}
+	f.writers++
+	// get writer token, any other write and read cannot get lock
+	f.rwMutex.Lock()
+	f.readers = 0
+	// Signal waiting readers that they may be able to proceed.
+	f.readerCond.Broadcast()
+}
+
+func (f *File) ReleaseWriterToken() {
+	// unlock WRMutex so that reader and writer can get WRMutex
+	f.rwMutex.Unlock()
+	// descreses semophore
+	<-f.writeSem
+
+}

sdfs/types.go

+package sdfs
+
+import (
+	"fmt"
+	"log"
+)
+
+const Limit = 9000
+
+type MsgType int
+
+const (
+	PUT MsgType = iota
+	GET
+	DELETE
+	RECOVER
+	REPLICATION
+	REPCONFIRM
+	PUTCONFIRM
+	DELCONFIRM
+	GETCONFIRM
+	RECCONFIRM
+	PUTACK
+	DELETEACK
+	LISTREPLICA
+)
+
+type Assignment struct {
+	Filename  string
+	Version   int
+	Datanodes []string
+}
+
+type Message struct {
+	ID      int64
+	Sender  string
+	Type    MsgType
+	Payload interface{}
+}
+
+func payloadToAssignment(payload interface{}) (Assignment, error) {
+	var assignment Assignment
+
+	data, ok := payload.(map[string]interface{})
+	if !ok {
+		// Payload is not an Assignment
+		log.Println("CLIENT: Fail to parse payload to map")
+		return assignment, fmt.Errorf("failed to parse payload as map")
+	}
+
+	// Check if the required fields exist in the map
+	if filename, ok := data["Filename"].(string); ok {
+		assignment.Filename = filename
+	} else {
+		return assignment, fmt.Errorf("Missing or invalid 'Filename' field")
+	}
+
+	if version, ok := data["Version"].(float64); ok {
+		assignment.Version = int(version)
+	} else {
+		return assignment, fmt.Errorf("Missing or invalid 'Version' field")
+	}
+
+	if datanodes, ok := data["Datanodes"].([]interface{}); ok {
+		for _, node := range datanodes {
+			if strNode, ok := node.(string); ok {
+				assignment.Datanodes = append(assignment.Datanodes, strNode)
+			} else {
+				return assignment, fmt.Errorf("Invalid datanode value in 'Datanodes' field")
+			}
+		}
+	} else {
+		return assignment, fmt.Errorf("Missing or invalid 'Datanodes' field")
+	}
+
+	return assignment, nil
+}

utils/gen_scp_keys.sh

+#!/bin/bash
+
+# Generate an SSH key
+ssh-keygen -t rsa -b 4096 -C "cs425mp3scp"
+
+# List of remote VMs
+remote_vms=(
+   "fa23-cs425-0201.cs.illinois.edu"
+   "fa23-cs425-0202.cs.illinois.edu"
+   "fa23-cs425-0203.cs.illinois.edu"
+   "fa23-cs425-0204.cs.illinois.edu"
+   "fa23-cs425-0205.cs.illinois.edu"
+   "fa23-cs425-0206.cs.illinois.edu"
+   "fa23-cs425-0207.cs.illinois.edu"
+   "fa23-cs425-0208.cs.illinois.edu"
+   "fa23-cs425-0209.cs.illinois.edu"
+   "fa23-cs425-0210.cs.illinois.edu"
+)
+
+# Get the current user and hostname
+current_user=$(whoami)
+current_hostname=$(hostname)
+
+# Loop through the remote VMs and copy the SSH key if it's not the current VM
+for vm in "${remote_vms[@]}"; do
+    ssh-copy-id "$current_user@$vm"
+done

utils/go.mod

+module cs425mp4/utils
+
+go 1.21.0

utils/update_go_mod_version.sh

+#!/bin/bash
+
+# Specify the desired Go version
+desired_go_version=1.19  # Replace with your desired version (e.g., "1.19" or "1.23.0")
+
+# Find all go.mod files in the current directory and its subdirectories
+go_mod_files=$(find . -type f -name "go.mod")
+
+for go_mod_file in $go_mod_files; do
+    # Check if the file exists and is readable
+    if [ -r "$go_mod_file" ]; then
+        # Replace the Go version in the go.mod file
+        sed -i "s/go [0-9]\+\(\.[0-9]\+\)*\(\.0\)\?/go $desired_go_version/" "$go_mod_file"
+        echo "Modified $go_mod_file to use Go version $desired_go_version"
+    else
+        echo "Error: Cannot read $go_mod_file"
+    fi
+done

utils/utils.go

+package utils
+
+import (
+	"log"
+	"net"
+	"os"
+	"strconv"
+	"strings"
+)
+
+func parseHostname(id string) string {
+	components := strings.Split(id, ":")
+	return components[0]
+}
+
+func parseAddress(id string) string {
+	components := strings.Split(id, ":")
+	return components[1] + ":" + components[2]
+}
+
+func parseTimestamp(id string) int64 {
+	components := strings.Split(id, ":")
+	t, _ := strconv.ParseInt(components[3], 10, 64)
+	return t
+}
+
+// resolve hosts' name to IP address
+func getIP(hostname string) net.IP {
+	ips, err := net.LookupIP(hostname)
+	if err != nil {
+		log.Println("Fail to resolve hostname:", err)
+		os.Exit(1)
+	}
+	for _, ip := range ips {
+		log.Println("IP Address:", ip.String())
+	}
+	return ips[0]
+}
+
+func getLocalIP() net.IP {
+	hostname, err := os.Hostname()
+	if err != nil {
+		log.Println("Fail to get hostname:", err)
+		os.Exit(1)
+	}
+	return getIP(hostname)
+}