realistic exp

Author: Xm0onh

config/config.go

@@ -19,23 +19,23 @@ type NodeData struct {
 
 // Must change respected to the codign method
 const (
-	ExpectedChunks = 10
+	ExpectedChunks = 1
 )
 
 // Reed-Solomon parameters
 const (
-	DataShards   = 10
-	ParityShards = 2
+	DataShards   = 30
+	ParityShards = 5
 )
 
 // Luby-Transform parameters
 const (
-	LTSourceBlocks = 10
+	LTSourceBlocks = 1
 	RandomSeed     = 42
 )
 
 var (
-	c                   = 2.0
+	c                   = 1.0
 	LTEncodedBlockCount = int(c*math.Sqrt(float64(LTSourceBlocks))) + LTSourceBlocks
 )
 
@@ -48,7 +48,7 @@ var (
 	NodeID         string
 	CodingMethod   string
 	Mode           string
-	Nodes          = 11
+	Nodes          = 2
 	BazantineNodes = 3
 	ReceivedChunks = sync.Map{}
 	SentChunks     = sync.Map{}
@@ -60,12 +60,12 @@ var (
 	// Must be changed to the coding method
 	// if LT then it should be LTEncodedBlockCount
 	// if RS then it should be DataShards + ParityShards
-	ChunksRecByNode = make([][]byte, DataShards+ParityShards)
+	ChunksRecByNode = make([][]byte, LTEncodedBlockCount)
 	ReadyCounter    = 0
 	StartTime       time.Time
 	OriginalLength  = 18876679
 )
 
 var (
-	K = 10
+	K = 1
 )

experimentbash.sh

@@ -8,11 +8,11 @@ rm -rf output
 mkdir output
 
 # Number of regular nodes to start
-NUM_NODES=10
+NUM_NODES=2
 PORT_BASE=4007
 BOOTSTRAP_PORT_BASE=4001
 BOOTSTRAP_NODES=1
-CODING_METHOD="RS"
+CODING_METHOD="LT"
 MODE="download"
 IP_BASE="127.0.0."
 
@@ -90,6 +90,7 @@ ip="${IP_BASE}7"
 tmux new-window -t nodes -n $session
 tmux send-keys -t nodes:$session "cd '$(pwd)' && cpulimit -l 50 -- go run ./cmd/main.go -node=Node1 -port=4006 -ip=$ip -mode=$MODE -coding=$CODING_METHOD" C-m
 
+sleep 2
 # Add network latency to Node 1
 # add_network_latency lo0 "50ms"
 

handlers/stream_handler.go

@@ -26,7 +26,7 @@ var log = logrus.New()
 
 var flag = true
 
-const subChunkSize = 512 * 1024 // 256 KB
+const subChunkSize = 512 * 1024 // 512 KB
 const maxRetries = 5
 
 // Helper function to write sub-chunks to the stream
@@ -310,56 +310,59 @@ func StoreReceivedChunk(nodeID string, chunkIndex int, chunk []byte, h host.Host
 		nodeData.Received[chunkIndex] = chunk
 		fmt.Printf("Node %s received chunk %d\n", nodeID, chunkIndex)
 		fmt.Println("Length of received chunks:", len(nodeData.Received))
-
-		if config.Counter == config.ExpectedChunks {
-			log.WithFields(logrus.Fields{"nodeID": nodeID}).Info("Node complete received data")
-
-			var _ string
-			var err error
-			log.WithField("codingMethod", config.CodingMethod).Info("Node decoding data")
-			droplets := make([][]byte, 0, config.ExpectedChunks)
-			if config.CodingMethod == "LT" {
-				for _, droplet := range config.ChunksRecByNode {
-					if len(droplet) > 0 {
-						droplets = append(droplets, droplet)
+		if config.K != 1 {
+			if config.Counter == config.ExpectedChunks {
+				log.WithFields(logrus.Fields{"nodeID": nodeID}).Info("Node complete received data")
+
+				var _ string
+				var err error
+				log.WithField("codingMethod", config.CodingMethod).Info("Node decoding data")
+				droplets := make([][]byte, 0, config.ExpectedChunks)
+				if config.CodingMethod == "LT" {
+					for _, droplet := range config.ChunksRecByNode {
+						if len(droplet) > 0 {
+							droplets = append(droplets, droplet)
+						}
 					}
+					_, err = lt.LTDecode(droplets)
+				} else if config.CodingMethod == "RS" {
+					_, err = rs.RSDecode(config.ChunksRecByNode)
 				}
-				_, err = lt.LTDecode(droplets)
-			} else if config.CodingMethod == "RS" {
-				_, err = rs.RSDecode(config.ChunksRecByNode)
-			}
 
-			if (err != nil) && (config.CodingMethod == "LT") {
-				log.WithFields(logrus.Fields{"nodeID": nodeID, "Error": err, "length of valid chunks:": len(droplets)}).Error("Node failed to decode data")
-				flag = false
-				return
-			} else if (err != nil) && (config.CodingMethod == "RS") {
-				log.WithFields(logrus.Fields{"nodeID": nodeID, "Error": err, "length of valid chunks:": len(config.ChunksRecByNode)}).Error("Node failed to decode data")
-				// flag = false
-				return
-			}
+				if (err != nil) && (config.CodingMethod == "LT") {
+					log.WithFields(logrus.Fields{"nodeID": nodeID, "Error": err, "length of valid chunks:": len(droplets)}).Error("Node failed to decode data")
+					flag = false
+					return
+				} else if (err != nil) && (config.CodingMethod == "RS") {
+					log.WithFields(logrus.Fields{"nodeID": nodeID, "Error": err, "length of valid chunks:": len(config.ChunksRecByNode)}).Error("Node failed to decode data")
+					// flag = false
+					return
+				}
 
-			log.WithFields(logrus.Fields{"nodeID": nodeID}).Info("Node reconstructed data")
-
-			// outputFilePath := fmt.Sprintf("output/%s_out.txt", config.NodeID)
-			// if err := os.WriteFile(outputFilePath, []byte(decodedData), 0644); err != nil {
-			// 	log.WithFields(logrus.Fields{"nodeID": nodeID, "Error": err}).Error("Node failed to write reconstructed data to file")
-			// 	return
-			// }
-			if config.Mode == "upload" {
-				for _, peerInfo := range config.ConnectedPeers {
-					if peerInfo.ID.String() != nodeID {
-						readyKey := fmt.Sprintf("%s-ready", peerInfo.ID.String())
-						if _, ok := config.SentChunks.Load(readyKey); !ok {
-							SendReady(context.Background(), h, peerInfo, nodeID)
-							config.SentChunks.Store(readyKey, struct{}{})
+				log.WithFields(logrus.Fields{"nodeID": nodeID}).Info("Node reconstructed data")
+
+				// outputFilePath := fmt.Sprintf("output/%s_out.txt", config.NodeID)
+				// if err := os.WriteFile(outputFilePath, []byte(decodedData), 0644); err != nil {
+				// 	log.WithFields(logrus.Fields{"nodeID": nodeID, "Error": err}).Error("Node failed to write reconstructed data to file")
+				// 	return
+				// }
+				if config.Mode == "upload" {
+					for _, peerInfo := range config.ConnectedPeers {
+						if peerInfo.ID.String() != nodeID {
+							readyKey := fmt.Sprintf("%s-ready", peerInfo.ID.String())
+							if _, ok := config.SentChunks.Load(readyKey); !ok {
+								SendReady(context.Background(), h, peerInfo, nodeID)
+								config.SentChunks.Store(readyKey, struct{}{})
+							}
 						}
 					}
+					time.Sleep(10 * time.Second)
+				} else if config.Mode == "download" {
+					logrus.WithField("Total time", time.Since(config.StartTime)).Info("Total time")
 				}
-				time.Sleep(10 * time.Second)
-			} else if config.Mode == "download" {
-				logrus.WithField("Total time", time.Since(config.StartTime)).Info("Total time")
 			}
+		} else {
+			logrus.WithField("Total time", time.Since(config.StartTime)).Info("Total time")
 		}
 	} else {
 		return
@@ -384,7 +387,11 @@ func HandleDownloadStream(s network.Stream, h host.Host, wg *sync.WaitGroup) {
 
 	originalFilePath := "eth_transactions.json"
 	originalData, _ := os.ReadFile(originalFilePath)
-
+	peerInfo := peer.AddrInfo{ID: s.Conn().RemotePeer()}
+	if config.K == 1 {
+		SendChunk(context.Background(), h, peerInfo, 0, originalData)
+		return
+	}
 	var chunks [][]byte
 
 	if config.CodingMethod == "RS" {
@@ -397,8 +404,6 @@ func HandleDownloadStream(s network.Stream, h host.Host, wg *sync.WaitGroup) {
 		panic("Invalid coding method")
 	}
 
-	peerInfo := peer.AddrInfo{ID: s.Conn().RemotePeer()}
-
 	SendChunk(context.Background(), h, peerInfo, int(chunkIndex), chunks[chunkIndex])
 }