feat(dot/network): implement streamManager to cleanup not recently used streams

[noot]

Changes

• implement streamManager to cleanup not recently used streams, when a new inbound stream is opened the stream is logged with the streamManager. if data is received on the stream, then the time is logged with the streamManager. tthe streamManager runs a cleanup goroutine every minute to cleanup streams that we haven't received any data on in that amount of time

Tests

go test ./dot/network -short

Issues

• related to Node Optimizations #1494

[arijitAD]

Why are we using two maps?
We can have a single map with the following value

type val struct {
  s network.Stream
  t time.time.
}

[noot]

updated!

[arijitAD]

Rename streamData to streamMap since we already have streamData.

[arijitAD]

Change this to pointer.
data := &streamData{}

[noot]

just wondering why?

[timwu20]

any reason why this spawns it's own goroutine instead of calling it via go s.streamManager.start()

[timwu20]

I see that there's a ctx created on NewService. Do we need both the closeCh and ctx?

[noot]

no particular reason, I like it like this since the main network service doesn't need to be concerned with what the streamManager does

[timwu20]

never used a sync.Map before. Are there multiple goroutines trying to read this map concurrently? Potentially you may get better performance from using a standard map and sync.RWMutex.

[noot]

yep there are multiple goroutines that'll access it, I think sync.Map is optimized for this us case so would be preferred over a standard map and a mutex

[noot]

https://golang.org/pkg/sync/#Map this case would be use case 2

[timwu20]

Suggested change

	for {
	select {
	case <-sm.ctx.Done():
	return
	case <-time.After(cleanupStreamInterval):
	sm.cleanupStreams()
	}
	}
	ticker := time.NewTicker(cleanupStreamInterval)
	for {
	select {
	case <-sm.ctx.Done():
	return
	case <-ticker.C:
	sm.cleanupStreams()
	}
	}

[noot]

very cool, can definitely use this in other places too

[EclesioMeloJunior]

Suggested change

	ticker := time.NewTicker(cleanupStreamInterval)
	ticker := time.NewTicker(cleanupStreamInterval)
	defer ticker.Stop()

[noot]

added!

[github-actions]

🎉 This PR is included in version 0.6.0 🎉

The release is available on GitHub release

Your semantic-release bot 📦🚀