Merge branch 'dev' into feature/hash

Author: harshu4

.gitignore

@@ -9,7 +9,6 @@
 *.dylib
 
 openmesh-core
-core
 
 # Test binary, built with `go test -c`
 *.test
@@ -36,8 +35,13 @@ go.work
 .DS_Store
 executable-file
 /ignore
+/collector/.env
 log.txt
 
+# Default cometbft state (config.toml is important)
+/default-cometbft-home/data
+/default-cometbft-home/config/node_key.json
+/default-cometbft-home/config/priv_validator_key.json
 # Get rid of logs
 logs
 

collector/collector.go

@@ -4,8 +4,8 @@ import (
 	"context"
 
 	"github.com/ipfs/go-cid"
-	"github.com/multiformats/go-multicodec"
 	log "github.com/openmesh-network/core/internal/logger"
+	"github.com/openmesh-network/core/resourcepool"
 	"github.com/sourcegraph/conc"
 )
 
@@ -15,16 +15,14 @@ type Request struct {
 }
 
 type Summary struct {
-	// XXX: This might not be efficient, array of pointers means many cache misses.
-	// Not sure if the Go compiler will realize we want these sequentially in memory.
-	// But whatever man we're only doing a handfull of these a second.
 	DataHashes []cid.Cid
 }
 
 type CollectorWorker struct {
-	summary *Summary
-	request Request
-	message chan []byte
+	summary  *Summary
+	request  Request
+	message  chan []byte
+	rpStream *resourcepool.Stream
 
 	// Could make these into the same function.
 	pause  chan bool
@@ -43,7 +41,7 @@ type CollectorInstance struct {
 	subscriptionsCancel       context.CancelFunc
 }
 
-const WORKER_COUNT = 5
+const WORKER_COUNT = 1
 
 // const BUFFER_SIZE_MAX = 1024
 // const BUFFER_MAX = 1024
@@ -93,6 +91,7 @@ func (ci *CollectorInstance) SubmitRequests(requestsSortedByPriority []Request)
 
 		// Have to declare variable here otherwise go will pass i as value and cause problems.
 		index := i
+
 		subscribeFunc := func() {
 			r := requestsSortedByPriority[index]
 
@@ -102,13 +101,14 @@ func (ci *CollectorInstance) SubmitRequests(requestsSortedByPriority []Request)
 
 			log.Info("Subscribing ", index)
 			messageChannel, err := Subscribe(ci.subscriptionsContext, r.Source, r.Source.Topics[r.Topic])
+
+			// XXX: Handle this case by skipping this request.
+			// Worker will do nothing for this period. Maybe optimize this?
 			if err != nil {
-				// XXX: Handle this case by skipping this request.
-				// Panicking now to highlight this.
-				panic(err)
+				log.Warn("Couldnt connect to source: ", r.Source.Name, "-", r.Source.Topics[r.Topic], ", skipping instead. Reason: ", err)
+			} else {
+				ci.workers[index].message = messageChannel
 			}
-
-			ci.workers[index].message = messageChannel
 		}
 
 		subscribeWaitGroup.Go(subscribeFunc)
@@ -127,42 +127,9 @@ func (ci *CollectorInstance) SubmitRequests(requestsSortedByPriority []Request)
 	return ci.summariesOld[:maxSummaries]
 }
 
-func (cw *CollectorWorker) run(ctx context.Context, buffer []byte) {
+func (cw *CollectorWorker) run(ctx context.Context) {
 	log.Info("Started worker.")
 
-	if buffer == nil {
-		panic("Buffer is nil dummy.")
-	}
-	if len(buffer) < 100 {
-		panic("Buffer is too small, is this an error?")
-	}
-
-	// XXX: Maybe move this function to RP? Also it will crash if length == 0
-	// Also I could move this to another function.
-	summaryAppend := func(summary *Summary, buffer []byte, length int) {
-		// TODO: Consider adding:
-		//	- Timestamp.
-		//	- Fragmentation flag (Whether there is a half message or not).
-		//	- Message count.
-
-		cidBuilder := cid.V1Builder{
-			Codec:    uint64(multicodec.DagPb),
-			MhType:   uint64(multicodec.Sha2_256),
-			MhLength: -1,
-		}
-
-		c, err := cidBuilder.Sum(buffer[0:length])
-		if err != nil {
-			// If this fails to parse a buffer the input is invalid.
-			panic(err)
-		}
-
-		summary.DataHashes = append(summary.DataHashes, c)
-
-		log.Info("Added ", length, "  bytes, now: ", c.String())
-	}
-
-	bufferOffset := 0
 	printedDebug := false
 	paused := false
 	log.Info("Running for loop.")
@@ -177,6 +144,7 @@ func (cw *CollectorWorker) run(ctx context.Context, buffer []byte) {
 				case <-cw.resume:
 					log.Info("Worker resumed.")
 					// Clear the summary cid buffer.
+					cw.rpStream.Reset()
 					cw.summary.DataHashes = cw.summary.DataHashes[:0]
 					paused = false
 
@@ -189,16 +157,15 @@ func (cw *CollectorWorker) run(ctx context.Context, buffer []byte) {
 					log.Info("Channel stopped.")
 
 					// Flush the buffer!
-					if len(buffer) > 0 {
-						log.Info("Flushed")
-						summaryAppend(cw.summary, buffer, len(buffer))
-					}
+					cw.rpStream.Flush()
 
 					// Hopefully go will just call memset here...
-					for i := range buffer {
-						buffer[i] = 0
-					}
-					bufferOffset = 0
+					log.Debug("One: ", len(cw.rpStream.GetCids()))
+					log.Debug("Two: ", len(cw.summary.DataHashes))
+
+					cw.summary.DataHashes = make([]cid.Cid, len(cw.rpStream.GetCids()))
+
+					copy(cw.summary.DataHashes[:], cw.rpStream.GetCids())
 
 					log.Info("Worker paused until resume is called.")
 					paused = true
@@ -215,26 +182,7 @@ func (cw *CollectorWorker) run(ctx context.Context, buffer []byte) {
 						break
 					}
 
-					if bufferOffset+len(message) > len(buffer) {
-						// TODO: Add to Resource Pool from here?
-						summaryAppend(cw.summary, buffer, bufferOffset)
-						bufferOffset = 0
-					}
-
-					// If the message still doesn't fit, divide it into chunks and add it until it fits.
-					for len(message) > len(buffer) {
-						// XXX: Should the cids we post be capped at the length of the buffer?
-						// Or can they be any size? For now I assume they are capped at the size of the buffer.
-						// Do we do padding? Need a spreadsheet to "empirically" test this.
-						summaryAppend(cw.summary, message, len(buffer))
-						message = message[len(buffer):]
-					}
-
-					// Add message to buffer.
-					copy(buffer[bufferOffset:], message)
-					// log.Info("Done here.")
-
-					bufferOffset += len(message)
+					cw.rpStream.Append(message)
 				}
 			}
 		}
@@ -246,14 +194,14 @@ func (ci *CollectorInstance) Start(ctx context.Context) {
 	ci.ctx = ctx
 
 	for i := range ci.workers {
-		buffer := make([]byte, 4096)
 		ci.workers[i].pause = make(chan bool)
 		ci.workers[i].resume = make(chan bool)
 		ci.workers[i].message = make(chan []byte)
 		ci.workers[i].summary = &ci.summariesNew[i]
+		ci.workers[i].rpStream = resourcepool.NewStream()
 
 		index := i
-		runFunc := func() { ci.workers[index].run(ci.ctx, buffer) }
+		runFunc := func() { ci.workers[index].run(ci.ctx) }
 
 		log.Infof("Deploying worker for collector.")
 		ci.workerWaitGroup.Go(runFunc)

collector/collector_test.go

@@ -5,5 +5,5 @@ import (
 )
 
 func TestBasic(t *testing.T) {
-
+	New()
 }

collector/data_size1.csv

@@ -0,0 +1,22 @@
+Source,Symbol,Data Size (bytes),Throughput,"Busiest Time (Data collected for 60 seconds Btw 13:28:07, 29 Apr 2024 till 13:29:07, 29 Apr 2024)",Throughput in the interval,messages count received during interval
+bybit,publicTrade.BTCUSDT,151389,2523,2024-04-29T13:28:30+10:00,5542.90,217
+okx,sprd-books5,318,5,2024-04-29T13:28:00+10:00,10.70,2
+okx,sprd-public-trades,114,1,2024-04-29T13:28:00+10:00,11.40,1
+coinbase,BTC-USD,116172,1936,2024-04-29T13:28:30+10:00,2657.20,66
+binance,btcusdt,131467,2191,2024-04-29T13:28:30+10:00,3328.40,199
+okx,sprd-tickers,326,5,2024-04-29T13:28:00+10:00,10.80,2
+coinbase,ETH-USD,165348,2755,2024-04-29T13:28:30+10:00,7399.20,185
+binance,solusdt,33707,561,2024-04-29T13:28:30+10:00,1138.90,70
+binance,ethusdt,249488,4158,2024-04-29T13:28:30+10:00,11899.00,715
+coinbase,BTC-ETH,128,2,2024-04-29T13:28:00+10:00,9.00,2
+bybit,orderbook.50.BTCUSDT,1010316,16838,2024-04-29T13:28:30+10:00,21417.10,485
+polygon-ankr-rpc,,919856,15330,2024-04-29T13:28:20+10:00,17162.30,3
+ethereum-ankr-rpc,,218175,3636,2024-04-29T13:28:30+10:00,7719.40,1
+bybit,kline.M.BTCUSDT,15137,252,2024-04-29T13:28:40+10:00,300.70,10
+bybit,tickers.BTCUSDT,218856,3647,2024-04-29T13:28:30+10:00,7737.50,237
+dydx,MATIC-USD,11941,199,2024-04-29T13:28:00+10:00,1079.50,2
+dydx,ETH-USD,23918,398,2024-04-29T13:28:00+10:00,1255.10,9
+dydx,SOL-USD,10768,179,2024-04-29T13:28:00+10:00,9.00,2
+dydx,LINK-USD,10816,180,2024-04-29T13:28:00+10:00,1055.50,2
+okx,sprd-bbo-tbt,261,4,2024-04-29T13:28:00+10:00,10.80,2
+dydx,BTC-USD,14570,242,2024-04-29T13:28:00+10:00,1083.00,2