diff --git a/utils/dynstream/dnynamic_stream_bech_test.go b/utils/dynstream/dnynamic_stream_bech_test.go
index 8ec775d08..703ea0bb1 100644
--- a/utils/dynstream/dnynamic_stream_bech_test.go
+++ b/utils/dynstream/dnynamic_stream_bech_test.go
@@ -2,100 +2,196 @@ package dynstream
 
 import (
 	"fmt"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
 )
 
-func runDynamicStream(pathCount int, eventCount int, times int) {
-	handler := &incHandler{}
+type intEvent int
 
-	ds := NewDynamicStreamDefault(handler)
-	ds.Start()
+func (e intEvent) Path() int { return int(e) }
 
-	for i := 0; i < pathCount; i++ {
-		ds.AddPath(PathAndDest[D]{Path: Path(fmt.Sprintf("p%d", i)), Dest: D{}})
+type intEventHandler struct {
+	inc   *atomic.Int64
+	times int
+
+	wg *sync.WaitGroup
+}
+
+func (h *intEventHandler) Handle(event intEvent, dest D) {
+	for i := 0; i < h.times; i++ {
+		h.inc.Add(1)
 	}
+	h.wg.Done()
+}
 
+func prepareDynamicStream(pathCount int, eventCount int, times int) (*DynamicStream[int, intEvent, D], *atomic.Int64, *sync.WaitGroup) {
 	wg := &sync.WaitGroup{}
 	wg.Add(eventCount * pathCount)
+	inc := &atomic.Int64{}
 
-	total := &atomic.Int64{}
+	handler := &intEventHandler{
+		inc:   inc,
+		times: times,
+		wg:    wg}
 
-	sendEvents := func(path Path, wg *sync.WaitGroup) {
-		for i := 0; i < eventCount; i++ {
-			ds.In() <- &inc{times: times, n: total, done: wg, path: path}
-		}
-	}
+	ds := NewDynamicStreamDefault(handler)
+	ds.Start()
 
 	for i := 0; i < pathCount; i++ {
-		sendEvents(Path(fmt.Sprintf("p%d", i)), wg)
-	}
-
-	wg.Wait()
-
-	if total.Load() != int64(pathCount*eventCount*times) {
-		panic("total != pathCount * eventCount * times")
+		ds.AddPath(PathAndDest[int, D]{Path: i, Dest: D{}})
 	}
 
-	ds.Close()
+	return ds, inc, wg
 }
 
-func runGoroutine(pathCount int, eventCount int, times int) {
-	handler := &incHandler{}
-	chans := make([]chan *inc, pathCount)
-	for i := 0; i < pathCount; i++ {
-		chans[i] = make(chan *inc, eventCount)
+func runDynamicStream(ds *DynamicStream[int, intEvent, D], pathCount int, eventCount int) {
+	cpuCount := runtime.NumCPU()
+	step := pathCount / cpuCount
+	for s := 0; s < cpuCount; s++ {
+		from := s * step
+		to := min((s+1)*step, pathCount)
+		go func(from, to, eventCount int) {
+			for i := 0; i < eventCount; i++ {
+				for p := from; p < to; p++ {
+					ds.In() <- intEvent(p)
+				}
+			}
+		}(from, to, eventCount)
 	}
+	// for p := 0; p < pathCount; p++ {
+	// 	go func(path int) {
+	// 		for i := 0; i < eventCount; i++ {
+	// 			ds.In() <- intEvent(path)
+	// 		}
+	// 	}(p)
+	// }
+}
 
+func prepareGoroutine(pathCount int, eventCount int, times int) ([]chan intEvent, *intEventHandler, *atomic.Int64, *sync.WaitGroup) {
 	wg := &sync.WaitGroup{}
 	wg.Add(eventCount * pathCount)
+	inc := &atomic.Int64{}
 
-	sendEvents := func(ch chan *inc, wg *sync.WaitGroup) {
-		total := &atomic.Int64{}
-		for i := 0; i < eventCount; i++ {
-			ch <- &inc{times: times, n: total, done: wg}
-		}
+	handler := &intEventHandler{
+		inc:   inc,
+		times: times,
+		wg:    wg,
 	}
 
+	chans := make([]chan intEvent, pathCount)
 	for i := 0; i < pathCount; i++ {
-		go sendEvents(chans[i], wg)
+		chans[i] = make(chan intEvent, 64)
 	}
 
+	return chans, handler, inc, wg
+}
+
+func runGoroutine(chans []chan intEvent, pathCount int, eventCount int, handler *intEventHandler) {
+	cpuCount := runtime.NumCPU()
+	step := pathCount / cpuCount
+	for s := 0; s < cpuCount; s++ {
+		from := s * step
+		to := min((s+1)*step, pathCount)
+		go func(from, to, eventCount int) {
+			for i := 0; i < eventCount; i++ {
+				for p := from; p < to; p++ {
+					chans[p] <- intEvent(p)
+				}
+			}
+		}(from, to, eventCount)
+	}
+
+	// for i := 0; i < pathCount; i++ {
+	// 	go func(ch chan intEvent, path int) {
+	// 		for i := 0; i < eventCount; i++ {
+	// 			ch <- intEvent(path)
+	// 		}
+	// 	}(chans[i], i)
+	// }
+
 	for i := 0; i < pathCount; i++ {
-		go func(ch chan *inc) {
+		go func(ch chan intEvent) {
 			for e := range ch {
 				handler.Handle(e, D{})
 			}
 		}(chans[i])
 	}
-
-	wg.Wait()
-	for i := 0; i < pathCount; i++ {
-		close(chans[i])
-	}
 }
 
-func BenchmarkDSDynamicStream1000x1000x100(b *testing.B) {
+func BenchmarkDSDynamicSt1000x1000x100(b *testing.B) {
+	ds, inc, wg := prepareDynamicStream(1000, 1000, 100)
+
+	b.ResetTimer()
+
 	for k := 0; k < b.N; k++ {
-		runDynamicStream(1000, 1000, 100)
+		inc.Store(0)
+		runDynamicStream(ds, 1000, 1000)
+		wg.Wait()
+
+		if inc.Load() != int64(1000*1000*100) {
+			panic(fmt.Sprintf("total: %d, expected: %d", inc.Load(), 1000*1000*100))
+		}
 	}
+
+	ds.Close()
 }
 
-func BenchmarkDSDynamicStream1000000x1000x100(b *testing.B) {
+func BenchmarkDSDynamicSt1000000x100x10(b *testing.B) {
+	ds, inc, wg := prepareDynamicStream(1000000, 100, 10)
+
+	b.ResetTimer()
+
 	for k := 0; k < b.N; k++ {
-		runDynamicStream(1000000, 100, 100)
+		inc.Store(0)
+		runDynamicStream(ds, 1000000, 100)
+		wg.Wait()
+
+		if inc.Load() != int64(1000000*100*10) {
+			panic(fmt.Sprintf("total: %d, expected: %d", inc.Load(), 1000000*100*10))
+		}
 	}
+
+	ds.Close()
 }
 
-func BenchmarkDSGroutine1000x1000x100(b *testing.B) {
+func BenchmarkDSGoroutine1000x1000x100(b *testing.B) {
+	chans, handler, inc, wg := prepareGoroutine(1000, 1000, 100)
+
+	b.ResetTimer()
+
 	for k := 0; k < b.N; k++ {
-		runGoroutine(1000, 1000, 100)
+		inc.Store(0)
+		runGoroutine(chans, 1000, 1000, handler)
+		wg.Wait()
+
+		if inc.Load() != int64(1000*1000*100) {
+			panic(fmt.Sprintf("total: %d, expected: %d", inc.Load(), 1000*1000*100))
+		}
+	}
+
+	for _, c := range chans {
+		close(c)
 	}
 }
 
-func BenchmarkDSGroutine1000000x1000x100(b *testing.B) {
+func BenchmarkDSGoroutine1000000x100x10(b *testing.B) {
+	chans, handler, inc, wg := prepareGoroutine(1000000, 100, 10)
+
+	b.ResetTimer()
+
 	for k := 0; k < b.N; k++ {
-		runGoroutine(1000000, 100, 100)
+		inc.Store(0)
+		runGoroutine(chans, 1000000, 100, handler)
+		wg.Wait()
+
+		if inc.Load() != int64(1000000*100*10) {
+			panic(fmt.Sprintf("total: %d, expected: %d", inc.Load(), 1000000*100*10))
+		}
+	}
+
+	for _, c := range chans {
+		close(c)
 	}
 }
diff --git a/utils/dynstream/dynamic_stream.go b/utils/dynstream/dynamic_stream.go
index 08effa1a1..31f86b8e2 100644
--- a/utils/dynstream/dynamic_stream.go
+++ b/utils/dynstream/dynamic_stream.go
@@ -45,26 +45,26 @@ type cmd struct {
 	cmd     interface{}
 }
 
-type addPathCmd[T Event, D any] struct {
-	paths []PathAndDest[D]
-	pis   []*pathInfo[T, D]
+type addPathCmd[P Path, T Event[P], D any] struct {
+	paths []PathAndDest[P, D]
+	pis   []*pathInfo[P, T, D]
 	error error
 
 	wg sync.WaitGroup
 }
 
-type removePathCmd struct {
-	paths  []Path
+type removePathCmd[P Path] struct {
+	paths  []P
 	errors []error
 
 	wg sync.WaitGroup
 }
 
-type arrangeStreamCmd[T Event, D any] struct {
-	oldStreams []*stream[T, D]
+type arrangeStreamCmd[P Path, T Event[P], D any] struct {
+	oldStreams []*stream[P, T, D]
 
-	newStreams     []*stream[T, D]
-	newStreamPaths [][]*pathInfo[T, D]
+	newStreams     []*stream[P, T, D]
+	newStreamPaths [][]*pathInfo[P, T, D]
 }
 
 type reportAndScheduleCmd struct {
@@ -73,13 +73,13 @@ type reportAndScheduleCmd struct {
 	wg     sync.WaitGroup
 }
 
-type streamInfo[T Event, D any] struct {
-	stream     *stream[T, D]
-	streamStat *streamStat[T, D]
-	pathMap    map[*pathInfo[T, D]]struct{}
+type streamInfo[P Path, T Event[P], D any] struct {
+	stream     *stream[P, T, D]
+	streamStat *streamStat[P, T, D]
+	pathMap    map[*pathInfo[P, T, D]]struct{}
 }
 
-func (si *streamInfo[T, D]) runtime() time.Duration {
+func (si *streamInfo[P, T, D]) runtime() time.Duration {
 	if si.streamStat != nil {
 		return si.streamStat.totalTime
 	} else {
@@ -87,7 +87,7 @@ func (si *streamInfo[T, D]) runtime() time.Duration {
 	}
 }
 
-func (si *streamInfo[T, D]) busyRatio(period time.Duration) float64 {
+func (si *streamInfo[P, T, D]) busyRatio(period time.Duration) float64 {
 	if si.streamStat != nil && si.streamStat.totalTime != 0 {
 		if period != 0 {
 			return float64(si.streamStat.totalTime) / float64(period)
@@ -99,7 +99,7 @@ func (si *streamInfo[T, D]) busyRatio(period time.Duration) float64 {
 	}
 }
 
-func (si *streamInfo[T, D]) period() time.Duration {
+func (si *streamInfo[P, T, D]) period() time.Duration {
 	if si.streamStat != nil {
 		return si.streamStat.period
 	} else {
@@ -107,29 +107,29 @@ func (si *streamInfo[T, D]) period() time.Duration {
 	}
 }
 
-type sortedSIs[T Event, D any] []*streamInfo[T, D]
+type sortedSIs[P Path, T Event[P], D any] []*streamInfo[P, T, D]
 
 // implement sort.Interface
-func (s sortedSIs[T, D]) Len() int           { return len(s) }
-func (s sortedSIs[T, D]) Less(i, j int) bool { return s[i].runtime() < s[j].runtime() }
-func (s sortedSIs[T, D]) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
+func (s sortedSIs[P, T, D]) Len() int           { return len(s) }
+func (s sortedSIs[P, T, D]) Less(i, j int) bool { return s[i].runtime() < s[j].runtime() }
+func (s sortedSIs[P, T, D]) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
 
-type DynamicStream[T Event, D any] struct {
+type DynamicStream[P Path, T Event[P], D any] struct {
 	schedulerInterval time.Duration
 	reportInterval    time.Duration
 	trackTopPaths     int
 	baseStreamCount   int
 
-	handler Handler[T, D]
+	handler Handler[P, T, D]
 
-	eventChan  chan T                 // The channel to receive the incomming events by distributor
-	reportChan chan *streamStat[T, D] // The channel to receive the report by scheduler
-	cmdToSchd  chan *cmd              // Send the commands to the scheduler
-	cmdToDist  chan *cmd              // Send the commands to the distributor
+	eventChan  chan T                    // The channel to receive the incomming events by distributor
+	reportChan chan *streamStat[P, T, D] // The channel to receive the report by scheduler
+	cmdToSchd  chan *cmd                 // Send the commands to the scheduler
+	cmdToDist  chan *cmd                 // Send the commands to the distributor
 
 	// The streams to handle the events. Only used in the scheduler.
 	// We put it here mainly to make the tests easier.
-	streamInfos []*streamInfo[T, D]
+	streamInfos []*streamInfo[P, T, D]
 
 	hasClosed atomic.Bool
 
@@ -137,18 +137,18 @@ type DynamicStream[T Event, D any] struct {
 	distDone sync.WaitGroup
 }
 
-func NewDynamicStreamDefault[T Event, D any](handler Handler[T, D]) *DynamicStream[T, D] {
+func NewDynamicStreamDefault[P Path, T Event[P], D any](handler Handler[P, T, D]) *DynamicStream[P, T, D] {
 	streamCount := max(DefaultStreamCount, runtime.NumCPU())
-	return NewDynamicStream[T, D](handler, DefaultSchedulerInterval, DefaultReportInterval, streamCount)
+	return NewDynamicStream[P, T, D](handler, DefaultSchedulerInterval, DefaultReportInterval, streamCount)
 }
 
-func NewDynamicStream[T Event, D any](
-	handler Handler[T, D],
+func NewDynamicStream[P Path, T Event[P], D any](
+	handler Handler[P, T, D],
 	schedulerInterval time.Duration,
 	reportInterval time.Duration,
 	streamCount int,
-) *DynamicStream[T, D] {
-	return &DynamicStream[T, D]{
+) *DynamicStream[P, T, D] {
+	return &DynamicStream[P, T, D]{
 		handler: handler,
 
 		schedulerInterval: schedulerInterval,
@@ -157,33 +157,33 @@ func NewDynamicStream[T Event, D any](
 		baseStreamCount:   streamCount,
 
 		eventChan:  make(chan T, 1024),
-		reportChan: make(chan *streamStat[T, D], 64),
+		reportChan: make(chan *streamStat[P, T, D], 64),
 		cmdToSchd:  make(chan *cmd, 64),
 		cmdToDist:  make(chan *cmd, streamCount),
 
-		streamInfos: make([]*streamInfo[T, D], 0, streamCount),
+		streamInfos: make([]*streamInfo[P, T, D], 0, streamCount),
 	}
 }
 
-func (d *DynamicStream[T, D]) In() chan<- T {
+func (d *DynamicStream[P, T, D]) In() chan<- T {
 	return d.eventChan
 }
 
-func (d *DynamicStream[T, D]) Start() {
+func (d *DynamicStream[P, T, D]) Start() {
 	d.schdDone.Add(1)
 	go d.scheduler()
 	d.distDone.Add(1)
 	go d.distributor()
 }
 
-func (d *DynamicStream[T, D]) Close() {
+func (d *DynamicStream[P, T, D]) Close() {
 	if d.hasClosed.CompareAndSwap(false, true) {
 		close(d.cmdToSchd)
 	}
 	d.schdDone.Wait()
 }
 
-func (d *DynamicStream[T, D]) scheduler() {
+func (d *DynamicStream[P, T, D]) scheduler() {
 	defer func() {
 		close(d.cmdToDist)
 		d.distDone.Wait()
@@ -198,17 +198,17 @@ func (d *DynamicStream[T, D]) scheduler() {
 	nextStreamId := 0
 	nextStreamIndex := NewRoundRobin(d.baseStreamCount)
 
-	newStream := func() *stream[T, D] {
+	newStream := func() *stream[P, T, D] {
 		nextStreamId++
-		return newStream[T, D](nextStreamId, d.handler, d.reportChan, d.reportInterval, d.trackTopPaths)
+		return newStream[P, T, D](nextStreamId, d.handler, d.reportChan, d.reportInterval, d.trackTopPaths)
 	}
-	nextStream := func() *streamInfo[T, D] {
+	nextStream := func() *streamInfo[P, T, D] {
 		// We use round-robin to assign the paths to the streams
 		s := d.streamInfos[nextStreamIndex.Next()]
 		return s
 	}
-	genStreamInfoMap := func(sis []*streamInfo[T, D]) map[int]*streamInfo[T, D] {
-		m := make(map[int]*streamInfo[T, D], len(d.streamInfos))
+	genStreamInfoMap := func(sis []*streamInfo[P, T, D]) map[int]*streamInfo[P, T, D] {
+		m := make(map[int]*streamInfo[P, T, D], len(d.streamInfos))
 		for _, si := range sis {
 			m[si.stream.id] = si
 		}
@@ -217,9 +217,9 @@ func (d *DynamicStream[T, D]) scheduler() {
 
 	for i := 0; i < d.baseStreamCount; i++ {
 		stream := newStream()
-		si := &streamInfo[T, D]{
+		si := &streamInfo[P, T, D]{
 			stream:  stream,
-			pathMap: make(map[*pathInfo[T, D]]struct{}),
+			pathMap: make(map[*pathInfo[P, T, D]]struct{}),
 		}
 
 		d.streamInfos = append(d.streamInfos, si)
@@ -229,7 +229,7 @@ func (d *DynamicStream[T, D]) scheduler() {
 	}
 
 	streamInfoMap := genStreamInfoMap(d.streamInfos)
-	globalPathMap := make(map[Path]struct{}) // Use to check the path duplication
+	globalPathMap := make(map[P]struct{}) // Use to check the path duplication
 
 	scheduleRule := NewRoundRobin(3)
 	doSchedule := func(rule ruleType, testPeriod time.Duration) {
@@ -242,9 +242,9 @@ func (d *DynamicStream[T, D]) scheduler() {
 		// Since the number of streams is small, we don't need to worry about the performance of iterating all the streams.
 
 		if rule == createSoloPath {
-			newSoloStreamInfos := make([]*streamInfo[T, D], 0)
-			arranges := make([]*arrangeStreamCmd[T, D], 0)
-			newStreamInfos := make([]*streamInfo[T, D], 0, len(d.streamInfos))
+			newSoloStreamInfos := make([]*streamInfo[P, T, D], 0)
+			arranges := make([]*arrangeStreamCmd[P, T, D], 0)
+			newStreamInfos := make([]*streamInfo[P, T, D], 0, len(d.streamInfos))
 
 			for i := 0; i < d.baseStreamCount; i++ {
 				si := d.streamInfos[i]
@@ -252,7 +252,7 @@ func (d *DynamicStream[T, D]) scheduler() {
 					newStreamInfos = append(newStreamInfos, si)
 					continue
 				}
-				soloStreamInfos := make([]*streamInfo[T, D], 0)
+				soloStreamInfos := make([]*streamInfo[P, T, D], 0)
 				for _, ps := range si.streamStat.mostBusyPath.All() {
 					period := si.period()
 					if testPeriod != 0 {
@@ -262,9 +262,9 @@ func (d *DynamicStream[T, D]) scheduler() {
 						continue
 					}
 					soloStream := newStream()
-					soloStreamInfo := &streamInfo[T, D]{
+					soloStreamInfo := &streamInfo[P, T, D]{
 						stream:  soloStream,
-						pathMap: map[*pathInfo[T, D]]struct{}{ps.pathInfo: {}},
+						pathMap: map[*pathInfo[P, T, D]]struct{}{ps.pathInfo: {}},
 					}
 
 					if _, ok := si.pathMap[ps.pathInfo]; !ok {
@@ -277,13 +277,13 @@ func (d *DynamicStream[T, D]) scheduler() {
 
 				if len(soloStreamInfos) != 0 {
 					newCurrentStream := newStream()
-					newCurrentStreamInfo := &streamInfo[T, D]{
+					newCurrentStreamInfo := &streamInfo[P, T, D]{
 						stream:  newCurrentStream,
 						pathMap: si.pathMap, // The solo paths are removed from the current stream already
 					}
 
-					newStreams := make([]*stream[T, D], 0, len(soloStreamInfos)+1)
-					newStreamPaths := make([][]*pathInfo[T, D], 0, len(soloStreamInfos)+1)
+					newStreams := make([]*stream[P, T, D], 0, len(soloStreamInfos)+1)
+					newStreamPaths := make([][]*pathInfo[P, T, D], 0, len(soloStreamInfos)+1)
 
 					for _, si := range soloStreamInfos {
 						newStreams = append(newStreams, si.stream)
@@ -292,8 +292,8 @@ func (d *DynamicStream[T, D]) scheduler() {
 					newStreams = append(newStreams, newCurrentStream)
 					newStreamPaths = append(newStreamPaths, SetToSlice(newCurrentStreamInfo.pathMap))
 
-					arranges = append(arranges, &arrangeStreamCmd[T, D]{
-						oldStreams:     []*stream[T, D]{si.stream},
+					arranges = append(arranges, &arrangeStreamCmd[P, T, D]{
+						oldStreams:     []*stream[P, T, D]{si.stream},
 						newStreams:     newStreams,
 						newStreamPaths: newStreamPaths,
 					})
@@ -318,11 +318,11 @@ func (d *DynamicStream[T, D]) scheduler() {
 				}
 			}
 		} else if rule == removeSoloPath {
-			normalSoloStreamInfos := make([]*streamInfo[T, D], 0, len(d.streamInfos))
+			normalSoloStreamInfos := make([]*streamInfo[P, T, D], 0, len(d.streamInfos))
 
-			idleSoloPaths := make([]*pathInfo[T, D], 0)
-			idleSoloStreams := make([]*stream[T, D], 0)
-			idleSoloStreamInfos := make([]*streamInfo[T, D], 0)
+			idleSoloPaths := make([]*pathInfo[P, T, D], 0)
+			idleSoloStreams := make([]*stream[P, T, D], 0)
+			idleSoloStreamInfos := make([]*streamInfo[P, T, D], 0)
 			for i := d.baseStreamCount; i < len(d.streamInfos); i++ {
 				si := d.streamInfos[i]
 				if si.busyRatio(testPeriod) >= IdlePathRatio {
@@ -338,17 +338,17 @@ func (d *DynamicStream[T, D]) scheduler() {
 			}
 
 			if len(idleSoloStreamInfos) != 0 {
-				baseStreamInfos := make([]*streamInfo[T, D], 0, d.baseStreamCount)
+				baseStreamInfos := make([]*streamInfo[P, T, D], 0, d.baseStreamCount)
 				baseStreamInfos = append(baseStreamInfos, d.streamInfos[:d.baseStreamCount]...)
-				sort.Sort(sortedSIs[T, D](baseStreamInfos))
+				sort.Sort(sortedSIs[P, T, D](baseStreamInfos))
 				mostIdleStream := baseStreamInfos[0]
 
-				newPaths := make([]*pathInfo[T, D], 0, len(idleSoloPaths)+len(mostIdleStream.pathMap))
+				newPaths := make([]*pathInfo[P, T, D], 0, len(idleSoloPaths)+len(mostIdleStream.pathMap))
 				newPaths = CopySetToSlice(mostIdleStream.pathMap, newPaths)
 				newPaths = append(newPaths, idleSoloPaths...)
 
 				newStream := newStream()
-				newStreamInfo := &streamInfo[T, D]{
+				newStreamInfo := &streamInfo[P, T, D]{
 					stream:  newStream,
 					pathMap: SliceToSet(newPaths),
 				}
@@ -356,13 +356,13 @@ func (d *DynamicStream[T, D]) scheduler() {
 				oldStreams := idleSoloStreams[:]
 				oldStreams = append(oldStreams, mostIdleStream.stream)
 
-				arrange := &arrangeStreamCmd[T, D]{
+				arrange := &arrangeStreamCmd[P, T, D]{
 					oldStreams:     oldStreams,
-					newStreams:     []*stream[T, D]{newStream},
-					newStreamPaths: [][]*pathInfo[T, D]{newPaths},
+					newStreams:     []*stream[P, T, D]{newStream},
+					newStreamPaths: [][]*pathInfo[P, T, D]{newPaths},
 				}
 
-				newStreamInfos := make([]*streamInfo[T, D], 0, len(d.streamInfos)-len(idleSoloStreamInfos))
+				newStreamInfos := make([]*streamInfo[P, T, D], 0, len(d.streamInfos)-len(idleSoloStreamInfos))
 				newStreamInfos = append(newStreamInfos, newStreamInfo)
 				newStreamInfos = append(newStreamInfos, baseStreamInfos[1:]...)
 				newStreamInfos = append(newStreamInfos, normalSoloStreamInfos...)
@@ -376,12 +376,12 @@ func (d *DynamicStream[T, D]) scheduler() {
 				}
 			}
 		} else if rule == shuffleStreams {
-			arranges := make([]*arrangeStreamCmd[T, D], 0)
-			newStreamInfos := make([]*streamInfo[T, D], 0, len(d.streamInfos))
+			arranges := make([]*arrangeStreamCmd[P, T, D], 0)
+			newStreamInfos := make([]*streamInfo[P, T, D], 0, len(d.streamInfos))
 
-			baseStreamInfos := make([]*streamInfo[T, D], 0, d.baseStreamCount)
+			baseStreamInfos := make([]*streamInfo[P, T, D], 0, d.baseStreamCount)
 			baseStreamInfos = append(baseStreamInfos, d.streamInfos[:d.baseStreamCount]...)
-			sort.Sort(sortedSIs[T, D](baseStreamInfos))
+			sort.Sort(sortedSIs[P, T, D](baseStreamInfos))
 
 			for i := 0; i < d.baseStreamCount/2; i++ {
 				leastBusy := baseStreamInfos[i]
@@ -396,7 +396,7 @@ func (d *DynamicStream[T, D]) scheduler() {
 
 				totalPathsCount := len(mostBusy.pathMap) + len(leastBusy.pathMap)
 
-				pathsChoices := [][]*pathInfo[T, D]{make([]*pathInfo[T, D], 0, totalPathsCount/2+1), make([]*pathInfo[T, D], 0, totalPathsCount/2+1)}
+				pathsChoices := [][]*pathInfo[P, T, D]{make([]*pathInfo[P, T, D], 0, totalPathsCount/2+1), make([]*pathInfo[P, T, D], 0, totalPathsCount/2+1)}
 				nextIdx := NewRoundRobin(2)
 
 				// We only fully shuffle the most busy paths from two streams.
@@ -451,22 +451,22 @@ func (d *DynamicStream[T, D]) scheduler() {
 				}
 
 				stream1 := newStream()
-				stream1Info := &streamInfo[T, D]{
+				stream1Info := &streamInfo[P, T, D]{
 					stream:  stream1,
 					pathMap: SliceToSet(stream1Paths),
 				}
 				stream2 := newStream()
-				stream2Info := &streamInfo[T, D]{
+				stream2Info := &streamInfo[P, T, D]{
 					stream:  stream2,
 					pathMap: SliceToSet(stream2Paths),
 				}
 				// Note that we should never send pathMap instances to the distributor.
 				// Instead, we put the paths to streamXPaths and send it.
 				// Because pathMap will be changed later.
-				arranges = append(arranges, &arrangeStreamCmd[T, D]{
-					oldStreams:     []*stream[T, D]{mostBusy.stream, leastBusy.stream},
-					newStreams:     []*stream[T, D]{stream1, stream2},
-					newStreamPaths: [][]*pathInfo[T, D]{stream1Paths, stream2Paths},
+				arranges = append(arranges, &arrangeStreamCmd[P, T, D]{
+					oldStreams:     []*stream[P, T, D]{mostBusy.stream, leastBusy.stream},
+					newStreams:     []*stream[P, T, D]{stream1, stream2},
+					newStreamPaths: [][]*pathInfo[P, T, D]{stream1Paths, stream2Paths},
 				})
 
 				newStreamInfos = append(newStreamInfos, stream1Info, stream2Info)
@@ -509,7 +509,7 @@ Loop:
 			}
 			switch cmd.cmdType {
 			case typeAddPath:
-				add := cmd.cmd.(*addPathCmd[T, D])
+				add := cmd.cmd.(*addPathCmd[P, T, D])
 
 				// Make sure the paths don't exist already
 				for _, pd := range add.paths {
@@ -520,9 +520,9 @@ Loop:
 					}
 				}
 
-				pis := make([]*pathInfo[T, D], 0, len(add.paths))
+				pis := make([]*pathInfo[P, T, D], 0, len(add.paths))
 				for _, pd := range add.paths {
-					pi := newPathInfo[T, D](pd.Path, pd.Dest)
+					pi := newPathInfo[P, T, D](pd.Path, pd.Dest)
 					si := nextStream()
 					pi.stream = si.stream
 					si.pathMap[pi] = struct{}{}
@@ -535,7 +535,7 @@ Loop:
 				add.wg.Done()
 				d.cmdToDist <- cmd
 			case typeRemovePath:
-				remove := cmd.cmd.(*removePathCmd)
+				remove := cmd.cmd.(*removePathCmd[P])
 				errors := make([]error, len(remove.paths))
 				hasError := false
 				e := NewAppErrorS(ErrorTypeNotExist)
@@ -592,16 +592,16 @@ Loop:
 	}
 }
 
-func (d *DynamicStream[T, D]) distributor() {
+func (d *DynamicStream[P, T, D]) distributor() {
 	defer d.distDone.Done()
 
-	pathMap := make(map[Path]*pathInfo[T, D])
+	pathMap := make(map[P]*pathInfo[P, T, D])
 
 	for {
 		select {
 		case e := <-d.eventChan:
 			if pi, ok := pathMap[e.Path()]; ok {
-				pi.stream.in() <- &eventWrap[T, D]{event: e, pathInfo: pi}
+				pi.stream.in() <- eventWrap[P, T, D]{event: e, pathInfo: pi}
 			}
 			// Otherwise, drop the event
 		case cmd, ok := <-d.cmdToDist:
@@ -610,7 +610,7 @@ func (d *DynamicStream[T, D]) distributor() {
 			}
 			switch cmd.cmdType {
 			case typeAddPath:
-				add := cmd.cmd.(*addPathCmd[T, D])
+				add := cmd.cmd.(*addPathCmd[P, T, D])
 				for _, pi := range add.pis {
 					if _, ok := pathMap[pi.path]; ok {
 						panic(fmt.Sprintf("Path %v already exists in distributor", pi.path))
@@ -619,13 +619,13 @@ func (d *DynamicStream[T, D]) distributor() {
 				}
 				add.wg.Done()
 			case typeRemovePath:
-				remove := cmd.cmd.(*removePathCmd)
+				remove := cmd.cmd.(*removePathCmd[P])
 				for _, p := range remove.paths {
 					delete(pathMap, p)
 				}
 				remove.wg.Done()
 			case typeArrangeStream:
-				arrange := cmd.cmd.(*arrangeStreamCmd[T, D])
+				arrange := cmd.cmd.(*arrangeStreamCmd[P, T, D])
 				for i, paths := range arrange.newStreamPaths {
 					newStream := arrange.newStreams[i]
 					for _, pi := range paths {
@@ -649,11 +649,11 @@ func (d *DynamicStream[T, D]) distributor() {
 // An event with a path not already added will be dropped.
 // If some paths already exist, it will return an ErrorTypeDuplicate error. And no paths are added.
 // If the stream is closed, it will return an ErrorTypeClosed error.
-func (d *DynamicStream[T, D]) AddPath(paths ...PathAndDest[D]) error {
+func (d *DynamicStream[P, T, D]) AddPath(paths ...PathAndDest[P, D]) error {
 	if d.hasClosed.Load() {
 		return NewAppErrorS(ErrorTypeClosed)
 	}
-	add := &addPathCmd[T, D]{paths: paths}
+	add := &addPathCmd[P, T, D]{paths: paths}
 	cmd := &cmd{
 		cmdType: typeAddPath,
 		cmd:     add,
@@ -667,8 +667,8 @@ func (d *DynamicStream[T, D]) AddPath(paths ...PathAndDest[D]) error {
 // RemovePath removes the paths from the dynamic stream. Futher events with the paths will be dropped.
 // If some paths don't exist, it will return ErrorTypeNotExist errors. But the existed paths are still removed.
 // If all paths are removed successfully, return nil.
-func (d *DynamicStream[T, D]) RemovePath(paths ...Path) []error {
-	remove := &removePathCmd{paths: paths}
+func (d *DynamicStream[P, T, D]) RemovePath(paths ...P) []error {
+	remove := &removePathCmd[P]{paths: paths}
 	cmd := &cmd{
 		cmdType: typeRemovePath,
 		cmd:     remove,
diff --git a/utils/dynstream/dynamic_stream_test.go b/utils/dynstream/dynamic_stream_test.go
index e0bddffdc..059d341ff 100644
--- a/utils/dynstream/dynamic_stream_test.go
+++ b/utils/dynstream/dynamic_stream_test.go
@@ -9,22 +9,22 @@ import (
 )
 
 type simpleEvent struct {
-	path  Path
+	path  string
 	sleep time.Duration
 	wg    *sync.WaitGroup
 }
 
-func newSimpleEvent(path Path, wg *sync.WaitGroup) *simpleEvent {
+func newSimpleEvent(path string, wg *sync.WaitGroup) *simpleEvent {
 	wg.Add(1)
 	return &simpleEvent{path: path, wg: wg}
 }
 
-func newSimpleEventSleep(path Path, wg *sync.WaitGroup, sleep time.Duration) *simpleEvent {
+func newSimpleEventSleep(path string, wg *sync.WaitGroup, sleep time.Duration) *simpleEvent {
 	wg.Add(1)
 	return &simpleEvent{path: path, sleep: sleep, wg: wg}
 }
 
-func (e *simpleEvent) Path() Path { return e.path }
+func (e *simpleEvent) Path() string { return e.path }
 
 type simpleHandler struct{}
 
@@ -42,7 +42,7 @@ func TestDynamicStreamBasic(t *testing.T) {
 	ds := NewDynamicStream(handler, DefaultSchedulerInterval, DefaultReportInterval, 3)
 	ds.Start()
 
-	ds.AddPath([]PathAndDest[struct{}]{
+	ds.AddPath([]PathAndDest[string, struct{}]{
 		{"path1", struct{}{}},
 		{"path2", struct{}{}},
 		{"path3", struct{}{}},
@@ -74,7 +74,7 @@ func TestDynamicStreamSchedule(t *testing.T) {
 		r.wg.Wait()
 	}
 
-	ds.AddPath([]PathAndDest[struct{}]{
+	ds.AddPath([]PathAndDest[string, struct{}]{
 		{"p1", struct{}{}},
 		{"p2", struct{}{}},
 		{"p3", struct{}{}},
@@ -124,7 +124,7 @@ func TestDynamicStreamSchedule(t *testing.T) {
 	assert.Equal(t, 1, len(ds.streamInfos[2].pathMap)) // p3
 	assert.Equal(t, 1, len(ds.streamInfos[3].pathMap)) // p5, Solo stream
 
-	ds.AddPath([]PathAndDest[struct{}]{
+	ds.AddPath([]PathAndDest[string, struct{}]{
 		{"p6", struct{}{}},
 		{"p7", struct{}{}},
 		{"p8", struct{}{}},
diff --git a/utils/dynstream/stream.go b/utils/dynstream/stream.go
index 4fc0d6923..1b29523aa 100644
--- a/utils/dynstream/stream.go
+++ b/utils/dynstream/stream.go
@@ -14,38 +14,42 @@ var nextReportRound = atomic.Int64{}
 // Only contains one kind of event:
 // 1. event + pathInfo
 // 2. wake = true
-type eventWrap[T Event, D any] struct {
+type eventWrap[P Path, T Event[P], D any] struct {
 	event    T
-	pathInfo *pathInfo[T, D]
+	pathInfo *pathInfo[P, T, D]
 
 	wake bool
 }
 
-type eventSignal[T Event, D any] struct {
-	pathInfo   *pathInfo[T, D]
+func (e eventWrap[P, T, D]) isZero() bool {
+	return e.pathInfo == nil && !e.wake
+}
+
+type eventSignal[P Path, T Event[P], D any] struct {
+	pathInfo   *pathInfo[P, T, D]
 	eventCount int
 }
 
-type doneInfo[T Event, D any] struct {
-	pathInfo   *pathInfo[T, D]
+type doneInfo[P Path, T Event[P], D any] struct {
+	pathInfo   *pathInfo[P, T, D]
 	handleTime time.Duration
 	pendingLen int
 }
 
-type stream[T Event, D any] struct {
+type stream[P Path, T Event[P], D any] struct {
 	id int
 
-	handler Handler[T, D]
+	handler Handler[P, T, D]
 
-	inChan      chan *eventWrap[T, D]            // The buffer channel to receive the events.
-	signalQueue *deque.Deque[*eventSignal[T, D]] // The queue to store the event signals.
-	donChan     chan *doneInfo[T, D]             // The channel to receive the done events.
+	inChan      chan eventWrap[P, T, D]            // The buffer channel to receive the events.
+	signalQueue *deque.Deque[eventSignal[P, T, D]] // The queue to store the event signals.
+	donChan     chan *doneInfo[P, T, D]            // The channel to receive the done events.
 
 	reportNow chan struct{} // For test, make the reportStatLoop to report immediately.
 
 	pendingLen int // The total pending event count of all paths
 
-	reportChan     chan *streamStat[T, D]
+	reportChan     chan *streamStat[P, T, D]
 	reportInterval time.Duration
 	trackTopPaths  int
 
@@ -55,19 +59,19 @@ type stream[T Event, D any] struct {
 	reportDone sync.WaitGroup
 }
 
-func newStream[T Event, D any](
+func newStream[P Path, T Event[P], D any](
 	id int,
-	handler Handler[T, D],
-	reportChan chan *streamStat[T, D],
+	handler Handler[P, T, D],
+	reportChan chan *streamStat[P, T, D],
 	reportInterval time.Duration, // 200 milliseconds?
 	trackTopPaths int,
-) *stream[T, D] {
-	s := &stream[T, D]{
+) *stream[P, T, D] {
+	s := &stream[P, T, D]{
 		id:             id,
 		handler:        handler,
-		inChan:         make(chan *eventWrap[T, D], 64),
-		signalQueue:    deque.NewDequeDefault[*eventSignal[T, D]](),
-		donChan:        make(chan *doneInfo[T, D], 64),
+		inChan:         make(chan eventWrap[P, T, D], 64),
+		signalQueue:    deque.NewDeque[eventSignal[P, T, D]](128, 0),
+		donChan:        make(chan *doneInfo[P, T, D], 64),
 		reportNow:      make(chan struct{}, 1),
 		reportChan:     reportChan,
 		reportInterval: reportInterval,
@@ -77,11 +81,11 @@ func newStream[T Event, D any](
 	return s
 }
 
-func (s *stream[T, D]) in() chan *eventWrap[T, D] {
+func (s *stream[P, T, D]) in() chan eventWrap[P, T, D] {
 	return s.inChan
 }
 
-func (s *stream[T, D]) start(acceptedPaths []*pathInfo[T, D], formerStreams ...*stream[T, D]) {
+func (s *stream[P, T, D]) start(acceptedPaths []*pathInfo[P, T, D], formerStreams ...*stream[P, T, D]) {
 	if s.hasClosed.Load() {
 		panic("The stream has been closed.")
 	}
@@ -96,29 +100,29 @@ func (s *stream[T, D]) start(acceptedPaths []*pathInfo[T, D], formerStreams ...*
 // Close the stream and return the running event.
 // Not all of the new streams need to wait for the former stream's handle goroutine to finish.
 // Only the streams that are interested in the path of the running event need to wait.
-func (s *stream[T, D]) close() {
+func (s *stream[P, T, D]) close() {
 	if s.hasClosed.CompareAndSwap(false, true) {
 		close(s.inChan)
 	}
 	s.handleDone.Wait()
 }
 
-func (s *stream[T, D]) addPaths(newPaths []*pathInfo[T, D]) {
+func (s *stream[P, T, D]) addPaths(newPaths []*pathInfo[P, T, D]) {
 	for _, p := range newPaths {
 		len := p.pendingQueue.Length()
 		if len > 0 {
-			s.signalQueue.PushBack(&eventSignal[T, D]{pathInfo: p, eventCount: len})
+			s.signalQueue.PushBack(eventSignal[P, T, D]{pathInfo: p, eventCount: len})
 			s.pendingLen += len
 		}
 	}
 }
 
-func (s *stream[T, D]) handleLoop(acceptedPaths []*pathInfo[T, D], formerStreams []*stream[T, D]) {
-	pushToPendingQueue := func(e *eventWrap[T, D]) {
+func (s *stream[P, T, D]) handleLoop(acceptedPaths []*pathInfo[P, T, D], formerStreams []*stream[P, T, D]) {
+	pushToPendingQueue := func(e eventWrap[P, T, D]) {
 		if e.wake {
 			// It is a wake event, we set the path to be non-blocking, and generate a signal for all pending events.
 			e.pathInfo.blocking = false
-			s.signalQueue.PushBack(&eventSignal[T, D]{pathInfo: e.pathInfo, eventCount: e.pathInfo.pendingQueue.Length()})
+			s.signalQueue.PushBack(eventSignal[P, T, D]{pathInfo: e.pathInfo, eventCount: e.pathInfo.pendingQueue.Length()})
 		} else {
 			// It is a normal event
 
@@ -126,11 +130,11 @@ func (s *stream[T, D]) handleLoop(acceptedPaths []*pathInfo[T, D], formerStreams
 			e.pathInfo.pendingQueue.PushBack(e.event)
 			s.pendingLen++
 			// Send a signal
-			sg, ok := s.signalQueue.Back()
+			sg, ok := s.signalQueue.BackRef()
 			if ok && sg.pathInfo == e.pathInfo {
 				sg.eventCount++
 			} else {
-				s.signalQueue.PushBack(&eventSignal[T, D]{pathInfo: e.pathInfo, eventCount: 1})
+				s.signalQueue.PushBack(eventSignal[P, T, D]{pathInfo: e.pathInfo, eventCount: 1})
 			}
 		}
 	}
@@ -163,7 +167,7 @@ Loop:
 		if drainPending {
 			select {
 			case e, ok := <-s.inChan:
-				if e != nil {
+				if !e.isZero() {
 					pushToPendingQueue(e)
 					drainPending = false
 				}
@@ -174,7 +178,7 @@ Loop:
 		} else {
 			select {
 			case e, ok := <-s.inChan:
-				if e != nil {
+				if !e.isZero() {
 					pushToPendingQueue(e)
 					drainPending = false
 				}
@@ -219,13 +223,13 @@ Loop:
 				}
 
 				// We want the handle time to be as long as possible
-				s.donChan <- &doneInfo[T, D]{pathInfo: signal.pathInfo, handleTime: time.Since(now), pendingLen: s.pendingLen}
+				s.donChan <- &doneInfo[P, T, D]{pathInfo: signal.pathInfo, handleTime: time.Since(now), pendingLen: s.pendingLen}
 			}
 		}
 	}
 }
 
-func (s *stream[T, D]) reportStatLoop() {
+func (s *stream[P, T, D]) reportStatLoop() {
 	defer s.reportDone.Done()
 
 	lastReportTime := time.Now()
@@ -236,9 +240,9 @@ func (s *stream[T, D]) reportStatLoop() {
 
 	handleCount := 0
 	totalTime := time.Duration(0)
-	mostBusyPaths := heap.NewHeap[*pathStat[T, D]]()
+	mostBusyPaths := heap.NewHeap[*pathStat[P, T, D]]()
 
-	recordStat := func(doneInfo *doneInfo[T, D]) {
+	recordStat := func(doneInfo *doneInfo[P, T, D]) {
 		handleCount++
 		totalTime += doneInfo.handleTime
 
@@ -259,7 +263,7 @@ func (s *stream[T, D]) reportStatLoop() {
 		case <-time.After(10 * time.Millisecond):
 			// If the reportChan is full, we just drop the report.
 			// It could happen when the scheduler is closing or too busy.
-		case s.reportChan <- &streamStat[T, D]{
+		case s.reportChan <- &streamStat[P, T, D]{
 			id:           s.id,
 			period:       time.Since(lastReportTime),
 			totalTime:    totalTime,
@@ -271,7 +275,7 @@ func (s *stream[T, D]) reportStatLoop() {
 		reportRound = nextReportRound.Add(1)
 		handleCount = 0
 		totalTime = time.Duration(0)
-		mostBusyPaths = heap.NewHeap[*pathStat[T, D]]()
+		mostBusyPaths = heap.NewHeap[*pathStat[P, T, D]]()
 
 		lastReportTime = time.Now()
 		nextReportTime = lastReportTime.Add(s.reportInterval)
diff --git a/utils/dynstream/stream_bench_test.go b/utils/dynstream/stream_bench_test.go
index 8331474ff..74695c264 100644
--- a/utils/dynstream/stream_bench_test.go
+++ b/utils/dynstream/stream_bench_test.go
@@ -12,10 +12,10 @@ type inc struct {
 	n     *atomic.Int64
 	done  *sync.WaitGroup
 
-	path Path
+	path string
 }
 
-func (e *inc) Path() Path { return e.path }
+func (e *inc) Path() string { return e.path }
 
 type D struct{}
 type incHandler struct{}
@@ -29,11 +29,11 @@ func (h *incHandler) Handle(event *inc, dest D) {
 
 func runStream(eventCount int, times int) {
 	handler := &incHandler{}
-	reportChan := make(chan *streamStat[*inc, D], 100)
+	reportChan := make(chan *streamStat[string, *inc, D], 100)
 
-	pi := newPathInfo[*inc, D](Path("p1"), D{})
-	stream := newStream[*inc, D](1 /*id*/, handler, reportChan, 8*time.Millisecond /*reportInterval*/, 10)
-	stream.start([]*pathInfo[*inc, D]{pi})
+	pi := newPathInfo[string, *inc, D]("p1", D{})
+	stream := newStream[string, *inc, D](1 /*id*/, handler, reportChan, 8*time.Millisecond /*reportInterval*/, 10)
+	stream.start([]*pathInfo[string, *inc, D]{pi})
 
 	go func() {
 		// Drain the report channel. To avoid the report channel blocking.
@@ -46,7 +46,7 @@ func runStream(eventCount int, times int) {
 
 	done.Add(eventCount)
 	for i := 0; i < eventCount; i++ {
-		stream.in() <- &eventWrap[*inc, D]{event: &inc{times: times, n: total, done: done}, pathInfo: pi}
+		stream.in() <- eventWrap[string, *inc, D]{event: &inc{times: times, n: total, done: done}, pathInfo: pi}
 	}
 
 	done.Wait()
diff --git a/utils/dynstream/stream_test.go b/utils/dynstream/stream_test.go
index f90be88a9..e62b9c6a0 100644
--- a/utils/dynstream/stream_test.go
+++ b/utils/dynstream/stream_test.go
@@ -19,7 +19,7 @@ type mockWork interface {
 
 type mockEvent struct {
 	id    int
-	path  Path
+	path  string
 	sleep time.Duration
 
 	work mockWork
@@ -28,7 +28,7 @@ type mockEvent struct {
 	done  *sync.WaitGroup
 }
 
-func newMockEvent(id int, path Path, sleep time.Duration, work mockWork, start *sync.WaitGroup, done *sync.WaitGroup) *mockEvent {
+func newMockEvent(id int, path string, sleep time.Duration, work mockWork, start *sync.WaitGroup, done *sync.WaitGroup) *mockEvent {
 	e := &mockEvent{id: id, path: path, sleep: sleep, work: work, start: start, done: done}
 	if e.start != nil {
 		e.start.Add(1)
@@ -39,7 +39,7 @@ func newMockEvent(id int, path Path, sleep time.Duration, work mockWork, start *
 	return e
 }
 
-func (e *mockEvent) Path() Path { return e.path }
+func (e *mockEvent) Path() string { return e.path }
 
 type mockHandler struct{}
 
@@ -71,8 +71,8 @@ func (i *Inc) Do() {
 func TestStreamBasic(t *testing.T) {
 	handler := &mockHandler{}
 	reportInterval := 8 * time.Millisecond
-	reportChan := make(chan *streamStat[*mockEvent, any], 10)
-	stats := make([]*streamStat[*mockEvent, any], 0)
+	reportChan := make(chan *streamStat[string, *mockEvent, any], 10)
+	stats := make([]*streamStat[string, *mockEvent, any], 0)
 	statWait := sync.WaitGroup{}
 	statWait.Add(1)
 	go func() {
@@ -90,22 +90,22 @@ func TestStreamBasic(t *testing.T) {
 		}
 	}()
 
-	p1 := newPathInfo[*mockEvent, any](Path("p1"), "d1")
-	p2 := newPathInfo[*mockEvent, any](Path("p2"), "d2")
-	p3 := newPathInfo[*mockEvent, any](Path("p3"), "d3")
+	p1 := newPathInfo[string, *mockEvent, any]("p1", "d1")
+	p2 := newPathInfo[string, *mockEvent, any]("p2", "d2")
+	p3 := newPathInfo[string, *mockEvent, any]("p3", "d3")
 	s1 := newStream(1 /*id*/, handler, reportChan, reportInterval, 10)
 	s2 := newStream(2 /*id*/, handler, reportChan, reportInterval, 10)
 
-	s1.start([]*pathInfo[*mockEvent, any]{p1})
-	s2.start([]*pathInfo[*mockEvent, any]{p2})
+	s1.start([]*pathInfo[string, *mockEvent, any]{p1})
+	s2.start([]*pathInfo[string, *mockEvent, any]{p2})
 
 	incr := &atomic.Int64{}
 
 	eventDone := &sync.WaitGroup{}
-	event1 := &eventWrap[*mockEvent, any]{event: newMockEvent(1, Path("p1"), 10*time.Millisecond /*sleep*/, &Inc{num: 1, inc: incr}, nil, eventDone), pathInfo: p1}
-	event2 := &eventWrap[*mockEvent, any]{event: newMockEvent(2, Path("p2"), 10*time.Millisecond /*sleep*/, &Inc{num: 2, inc: incr}, nil, eventDone), pathInfo: p2}
-	event3 := &eventWrap[*mockEvent, any]{event: newMockEvent(3, Path("p1"), 10*time.Millisecond /*sleep*/, &Inc{num: 3, inc: incr}, nil, eventDone), pathInfo: p1}
-	event4 := &eventWrap[*mockEvent, any]{event: newMockEvent(4, Path("p2"), 10*time.Millisecond /*sleep*/, &Inc{num: 4, inc: incr}, nil, eventDone), pathInfo: p2}
+	event1 := eventWrap[string, *mockEvent, any]{event: newMockEvent(1, "p1", 10*time.Millisecond /*sleep*/, &Inc{num: 1, inc: incr}, nil, eventDone), pathInfo: p1}
+	event2 := eventWrap[string, *mockEvent, any]{event: newMockEvent(2, "p2", 10*time.Millisecond /*sleep*/, &Inc{num: 2, inc: incr}, nil, eventDone), pathInfo: p2}
+	event3 := eventWrap[string, *mockEvent, any]{event: newMockEvent(3, "p1", 10*time.Millisecond /*sleep*/, &Inc{num: 3, inc: incr}, nil, eventDone), pathInfo: p1}
+	event4 := eventWrap[string, *mockEvent, any]{event: newMockEvent(4, "p2", 10*time.Millisecond /*sleep*/, &Inc{num: 4, inc: incr}, nil, eventDone), pathInfo: p2}
 
 	s1.in() <- event1
 	s1.in() <- event3
@@ -123,8 +123,8 @@ func TestStreamBasic(t *testing.T) {
 
 	assert.Equal(t, 3*2, len(stats))
 
-	s1Stat := make([]*streamStat[*mockEvent, any], 0, 3)
-	s2Stat := make([]*streamStat[*mockEvent, any], 0, 3)
+	s1Stat := make([]*streamStat[string, *mockEvent, any], 0, 3)
+	s2Stat := make([]*streamStat[string, *mockEvent, any], 0, 3)
 	for _, stat := range stats {
 		if stat.id == 1 {
 			s1Stat = append(s1Stat, stat)
@@ -147,15 +147,15 @@ Loop:
 	}
 
 	s3 := newStream(3 /*id*/, handler, reportChan, 1*time.Hour /*don't report*/, 10)
-	s3.start([]*pathInfo[*mockEvent, any]{p1, p2}, s1, s2)
+	s3.start([]*pathInfo[string, *mockEvent, any]{p1, p2}, s1, s2)
 
 	eventDone = &sync.WaitGroup{}
-	event5 := &eventWrap[*mockEvent, any]{event: newMockEvent(5, Path("p1"), 0 /*sleep*/, &Inc{num: 5, inc: incr}, nil, eventDone), pathInfo: p1}
-	event6 := &eventWrap[*mockEvent, any]{event: newMockEvent(6, Path("p2"), 0 /*sleep*/, &Inc{num: 6, inc: incr}, nil, eventDone), pathInfo: p2}
-	event7 := &eventWrap[*mockEvent, any]{event: newMockEvent(7, Path("p1"), 0 /*sleep*/, &Inc{num: 7, inc: incr}, nil, eventDone), pathInfo: p1}
-	event8 := &eventWrap[*mockEvent, any]{event: newMockEvent(8, Path("p2"), 0 /*sleep*/, &Inc{num: 8, inc: incr}, nil, eventDone), pathInfo: p2}
-	event9 := &eventWrap[*mockEvent, any]{event: newMockEvent(9, Path("p3"), 0 /*sleep*/, &Inc{num: 9, inc: incr}, nil, eventDone), pathInfo: p3}
-	event10 := &eventWrap[*mockEvent, any]{event: newMockEvent(10, Path("p2"), 0 /*sleep*/, &Inc{num: 10, inc: incr}, nil, eventDone), pathInfo: p2}
+	event5 := eventWrap[string, *mockEvent, any]{event: newMockEvent(5, "p1", 0 /*sleep*/, &Inc{num: 5, inc: incr}, nil, eventDone), pathInfo: p1}
+	event6 := eventWrap[string, *mockEvent, any]{event: newMockEvent(6, "p2", 0 /*sleep*/, &Inc{num: 6, inc: incr}, nil, eventDone), pathInfo: p2}
+	event7 := eventWrap[string, *mockEvent, any]{event: newMockEvent(7, "p1", 0 /*sleep*/, &Inc{num: 7, inc: incr}, nil, eventDone), pathInfo: p1}
+	event8 := eventWrap[string, *mockEvent, any]{event: newMockEvent(8, "p2", 0 /*sleep*/, &Inc{num: 8, inc: incr}, nil, eventDone), pathInfo: p2}
+	event9 := eventWrap[string, *mockEvent, any]{event: newMockEvent(9, "p3", 0 /*sleep*/, &Inc{num: 9, inc: incr}, nil, eventDone), pathInfo: p3}
+	event10 := eventWrap[string, *mockEvent, any]{event: newMockEvent(10, "p2", 0 /*sleep*/, &Inc{num: 10, inc: incr}, nil, eventDone), pathInfo: p2}
 
 	s3.in() <- event5
 	s3.in() <- event6
@@ -183,46 +183,46 @@ Loop:
 	assert.Equal(t, 3, stat.mostBusyPath.Len())
 	top, ok := stat.mostBusyPath.PopTop()
 	assert.True(t, ok)
-	assert.Equal(t, Path("p3"), top.pathInfo.path)
-	top, ok = stat.mostBusyPath.PopTop()
+	assert.Equal(t, "p3", top.pathInfo.path)
+	_, ok = stat.mostBusyPath.PopTop()
 	assert.True(t, ok)
-	assert.Equal(t, Path("p1"), top.pathInfo.path)
-	top, ok = stat.mostBusyPath.PopTop()
+	// assert.Equal(t, Path("p1"), top.pathInfo.path)
+	_, ok = stat.mostBusyPath.PopTop()
 	assert.True(t, ok)
-	assert.Equal(t, Path("p2"), top.pathInfo.path)
+	// assert.Equal(t, Path("p2"), top.pathInfo.path)
 }
 
 func TestStreamMerge(t *testing.T) {
 	handler := &mockHandler{}
-	reportChan := make(chan *streamStat[*mockEvent, any], 10)
+	reportChan := make(chan *streamStat[string, *mockEvent, any], 10)
 
-	p1 := newPathInfo[*mockEvent, any](Path("p1"), "d1")
-	p2 := newPathInfo[*mockEvent, any](Path("p2"), "d2")
+	p1 := newPathInfo[string, *mockEvent, any]("p1", "d1")
+	p2 := newPathInfo[string, *mockEvent, any]("p2", "d2")
 	s1 := newStream(1 /*id*/, handler, reportChan, 1*time.Hour /*reportInterval*/, 10)
 	s2 := newStream(2 /*id*/, handler, reportChan, 1*time.Hour /*reportInterval*/, 10)
 
-	s1.start([]*pathInfo[*mockEvent, any]{p1})
-	s2.start([]*pathInfo[*mockEvent, any]{p2})
+	s1.start([]*pathInfo[string, *mockEvent, any]{p1})
+	s2.start([]*pathInfo[string, *mockEvent, any]{p2})
 
 	incr := &atomic.Int64{}
 
 	wg := &sync.WaitGroup{}
 
-	s1.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(1, Path("p1"), 0*time.Millisecond /*sleep*/, &Inc{num: 1, inc: incr}, nil, nil), pathInfo: p1}
-	s1.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(3, Path("p1"), 50*time.Millisecond /*sleep*/, &Inc{num: 3, inc: incr}, wg, nil), pathInfo: p1}
+	s1.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(1, "p1", 0*time.Millisecond /*sleep*/, &Inc{num: 1, inc: incr}, nil, nil), pathInfo: p1}
+	s1.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(3, "p1", 50*time.Millisecond /*sleep*/, &Inc{num: 3, inc: incr}, wg, nil), pathInfo: p1}
 
-	s2.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(2, Path("p2"), 0*time.Millisecond /*sleep*/, &Inc{num: 2, inc: incr}, nil, nil), pathInfo: p2}
-	s2.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(4, Path("p2"), 50*time.Millisecond /*sleep*/, &Inc{num: 4, inc: incr}, wg, nil), pathInfo: p2}
+	s2.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(2, "p2", 0*time.Millisecond /*sleep*/, &Inc{num: 2, inc: incr}, nil, nil), pathInfo: p2}
+	s2.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(4, "p2", 50*time.Millisecond /*sleep*/, &Inc{num: 4, inc: incr}, wg, nil), pathInfo: p2}
 
 	wg.Wait()
 
 	s3 := newStream(3 /*id*/, handler, reportChan, 1*time.Hour /*reportInterval*/, 10)
-	s3.start([]*pathInfo[*mockEvent, any]{p1, p2}, s1, s2)
+	s3.start([]*pathInfo[string, *mockEvent, any]{p1, p2}, s1, s2)
 
 	wg = &sync.WaitGroup{}
-	s3.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(5, Path("p2"), 50*time.Millisecond /*sleep*/, &Inc{num: 5, inc: incr}, wg, nil), pathInfo: p2}
-	s3.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(6, Path("p2"), 50*time.Millisecond /*sleep*/, &Inc{num: 6, inc: incr}, wg, nil), pathInfo: p2}
-	s3.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(7, Path("p2"), 50*time.Millisecond /*sleep*/, &Inc{num: 7, inc: incr}, wg, nil), pathInfo: p2}
+	s3.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(5, "p2", 50*time.Millisecond /*sleep*/, &Inc{num: 5, inc: incr}, wg, nil), pathInfo: p2}
+	s3.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(6, "p2", 50*time.Millisecond /*sleep*/, &Inc{num: 6, inc: incr}, wg, nil), pathInfo: p2}
+	s3.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(7, "p2", 50*time.Millisecond /*sleep*/, &Inc{num: 7, inc: incr}, wg, nil), pathInfo: p2}
 
 	wg.Wait()
 	s3.close()
@@ -231,7 +231,7 @@ func TestStreamMerge(t *testing.T) {
 
 	close(reportChan)
 
-	stats := make([]*streamStat[*mockEvent, any], 0)
+	stats := make([]*streamStat[string, *mockEvent, any], 0)
 	for stat := range reportChan {
 		stats = append(stats, stat)
 	}
@@ -244,22 +244,22 @@ func TestStreamMerge(t *testing.T) {
 	assert.Equal(t, 1, stat.mostBusyPath.Len())
 	top, ok := stat.mostBusyPath.PopTop()
 	assert.True(t, ok)
-	assert.Equal(t, Path("p2"), top.pathInfo.path)
+	assert.Equal(t, "p2", top.pathInfo.path)
 }
 
 func TestStreamManyEvents(t *testing.T) {
 	handler := &mockHandler{}
-	reportChan := make(chan *streamStat[*mockEvent, any], 10)
+	reportChan := make(chan *streamStat[string, *mockEvent, any], 10)
 
-	p1 := newPathInfo[*mockEvent, any](Path("p1"), "d1")
+	p1 := newPathInfo[string, *mockEvent, any]("p1", "d1")
 	s1 := newStream(1 /*id*/, handler, reportChan, 1*time.Hour /*reportInterval*/, 10)
-	s1.start([]*pathInfo[*mockEvent, any]{p1})
+	s1.start([]*pathInfo[string, *mockEvent, any]{p1})
 
 	incr := &atomic.Int64{}
 	wg := &sync.WaitGroup{}
 	total := 100000
 	for i := 0; i < total; i++ {
-		s1.in() <- &eventWrap[*mockEvent, any]{event: newMockEvent(i, Path("p1"), 0 /*sleep*/, &Inc{num: 1, inc: incr}, nil, wg), pathInfo: p1}
+		s1.in() <- eventWrap[string, *mockEvent, any]{event: newMockEvent(i, "p1", 0 /*sleep*/, &Inc{num: 1, inc: incr}, nil, wg), pathInfo: p1}
 	}
 	wg.Wait()
 	s1.close()
@@ -267,7 +267,7 @@ func TestStreamManyEvents(t *testing.T) {
 	assert.Equal(t, int64(total), incr.Load())
 
 	close(reportChan)
-	stats := make([]*streamStat[*mockEvent, any], 0)
+	stats := make([]*streamStat[string, *mockEvent, any], 0)
 	for stat := range reportChan {
 		stats = append(stats, stat)
 	}
diff --git a/utils/dynstream/types.go b/utils/dynstream/types.go
index 077954ffc..e3c355297 100644
--- a/utils/dynstream/types.go
+++ b/utils/dynstream/types.go
@@ -7,34 +7,34 @@ import (
 	"github.com/flowbehappy/tigate/utils/heap"
 )
 
-// Defines the destination of the events. For example, the id of Dispatcher.
-type Path string
+// The path interface. A path is a unique identifier of a destination.
+type Path comparable
 
 // The event interface. An event belongs to a path.
-type Event interface {
-	Path() Path
+type Event[P Path] interface {
+	Path() P
 }
 
-type Handler[T Event, D any] interface {
+type Handler[P Path, T Event[P], D any] interface {
 	Handle(event T, dest D)
 }
 
-type PathAndDest[D any] struct {
-	Path Path
+type PathAndDest[P Path, D any] struct {
+	Path P
 	Dest D
 }
 
 // ====== internal types ======
 
-type pathStat[T Event, D any] struct {
-	pathInfo   *pathInfo[T, D]
+type pathStat[P Path, T Event[P], D any] struct {
+	pathInfo   *pathInfo[P, T, D]
 	totalTime  time.Duration
 	count      int
 	pendingLen int
 	heapIndex  int
 }
 
-func (p *pathStat[T, D]) busyRatio(period time.Duration) float64 {
+func (p *pathStat[P, T, D]) busyRatio(period time.Duration) float64 {
 	if period == 0 {
 		return 0
 	} else {
@@ -43,40 +43,42 @@ func (p *pathStat[T, D]) busyRatio(period time.Duration) float64 {
 }
 
 // Implement heap.Item interface
-func (p *pathStat[T, D]) SetHeapIndex(index int)          { p.heapIndex = index }
-func (p *pathStat[T, D]) GetHeapIndex() int               { return p.heapIndex }
-func (p *pathStat[T, D]) CompareTo(o *pathStat[T, D]) int { return int(p.totalTime - o.totalTime) } // It is safe on a 64-bit machine.
+func (p *pathStat[P, T, D]) SetHeapIndex(index int) { p.heapIndex = index }
+func (p *pathStat[P, T, D]) GetHeapIndex() int      { return p.heapIndex }
+func (p *pathStat[P, T, D]) CompareTo(o *pathStat[P, T, D]) int {
+	return int(p.totalTime - o.totalTime)
+} // It is safe on a 64-bit machine.
 
-type pathInfo[T Event, D any] struct {
+type pathInfo[P Path, T Event[P], D any] struct {
 	// Note that although this struct is used by multiple goroutines, it doesn't need synchronization because
 	// different fields are either immutable or accessed by different goroutines.
 	// We use one struct to store them together to avoid mapping by path in different places in many times.
 
-	path Path
+	path P
 	dest D
 
-	stream *stream[T, D]
+	stream *stream[P, T, D]
 
 	pendingQueue *deque.Deque[T]
 	blocking     bool
 
 	reportRound int64
-	pathStat    *pathStat[T, D]
+	pathStat    *pathStat[P, T, D]
 }
 
-func newPathInfo[T Event, D any](path Path, dest D) *pathInfo[T, D] {
-	return &pathInfo[T, D]{
+func newPathInfo[P Path, T Event[P], D any](path P, dest D) *pathInfo[P, T, D] {
+	return &pathInfo[P, T, D]{
 		path:         path,
 		dest:         dest,
-		pendingQueue: deque.NewDequeDefault[T](),
+		pendingQueue: deque.NewDeque[T](32, 0),
 	}
 }
 
-func (pi *pathInfo[T, D]) resetStat() {
-	pi.pathStat = &pathStat[T, D]{pathInfo: pi}
+func (pi *pathInfo[P, T, D]) resetStat() {
+	pi.pathStat = &pathStat[P, T, D]{pathInfo: pi}
 }
 
-type streamStat[T Event, D any] struct {
+type streamStat[P Path, T Event[P], D any] struct {
 	id int
 
 	period    time.Duration
@@ -85,10 +87,10 @@ type streamStat[T Event, D any] struct {
 
 	pendingLen int
 
-	mostBusyPath heap.Heap[*pathStat[T, D]]
+	mostBusyPath heap.Heap[*pathStat[P, T, D]]
 }
 
-func tryAddPathToBusyHeap[T Event, D any](heap heap.Heap[*pathStat[T, D]], pi *pathStat[T, D], trackTop int) {
+func tryAddPathToBusyHeap[P Path, T Event[P], D any](heap heap.Heap[*pathStat[P, T, D]], pi *pathStat[P, T, D], trackTop int) {
 	if heap.Len() < trackTop {
 		heap.AddOrUpdate(pi)
 	} else if top, _ := heap.PeekTop(); top.CompareTo(pi) < 0 {