Repository: numaproj/numaflow
Pull request #617 (2023-03-17 14:31:44)

feat: track and expose oldest work yet to be done to the reduce loop
Track the oldest work yet to be completed in the reduce-loop so that both read-loop and the process-and-forward are in agreement with the watermark.


Patch:
diff --git a/pkg/reduce/pbq/partition/partition.go b/pkg/reduce/pbq/partition/partition.go
index d6ac343e..822c3aa2 100644
--- a/pkg/reduce/pbq/partition/partition.go
+++ b/pkg/reduce/pbq/partition/partition.go
@@ -30,5 +30,5 @@ type ID struct {
 }
 
 func (p ID) String() string {
-	return fmt.Sprintf("%v-%v", p.Start.UnixMilli(), p.End.UnixMilli())
+	return fmt.Sprintf("%v-%v-%s", p.Start.UnixMilli(), p.End.UnixMilli(), p.Slot)
 }
diff --git a/pkg/reduce/pbq/pbq.go b/pkg/reduce/pbq/pbq.go
index 5bfcf67f..d8c83af7 100644
--- a/pkg/reduce/pbq/pbq.go
+++ b/pkg/reduce/pbq/pbq.go
@@ -25,6 +25,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
+	"github.com/numaproj/numaflow/pkg/window"
 
 	"go.uber.org/zap"
 
@@ -46,6 +47,7 @@ type PBQ struct {
 	options       *options
 	manager       *Manager
 	log           *zap.SugaredLogger
+	kw            window.AlignedKeyedWindower
 	mu            sync.Mutex
 }
 
diff --git a/pkg/reduce/pbq/pbqmanager.go b/pkg/reduce/pbq/pbqmanager.go
index dc120d22..42215f31 100644
--- a/pkg/reduce/pbq/pbqmanager.go
+++ b/pkg/reduce/pbq/pbqmanager.go
@@ -24,17 +24,24 @@ import (
 	"sync"
 	"time"
 
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/wait"
+
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/store"
-
-	"go.uber.org/zap"
-	"k8s.io/apimachinery/pkg/util/wait"
+	"github.com/numaproj/numaflow/pkg/window"
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
 
+// RegisteredWindow to track the number of windows for a start and end time.
+type RegisteredWindow struct {
+	window.AlignedKeyedWindower
+	count int
+}
+
 // Manager helps in managing the lifecycle of PBQ instances
 type Manager struct {
 	vertexName    string
@@ -44,6 +51,7 @@ type Manager struct {
 	pbqOptions    *options
 	pbqMap        map[string]*PBQ
 	log           *zap.SugaredLogger
+	yetToBeClosed *window.SortedWindowList[*RegisteredWindow]
 	// we need lock to access pbqMap, since deregister will be called inside pbq
 	// and each pbq will be inside a go routine, and also entire PBQ could be managed
 	// through a go routine (depends on the orchestrator)
@@ -70,13 +78,14 @@ func NewManager(ctx context.Context, vertexName string, pipelineName string, vr
 		pbqMap:        make(map[string]*PBQ),
 		pbqOptions:    pbqOpts,
 		log:           logging.FromContext(ctx),
+		yetToBeClosed: window.NewSortedWindowList[*RegisteredWindow](),
 	}
 
 	return pbqManager, nil
 }
 
 // CreateNewPBQ creates new pbq for a partition
-func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (ReadWriteCloser, error) {
+func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID, win window.AlignedKeyedWindower) (ReadWriteCloser, error) {
 	persistentStore, err := m.storeProvider.CreateStore(ctx, partitionID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create a PBQ store, %w", err)
@@ -93,6 +102,7 @@ func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (R
 		PartitionID:   partitionID,
 		options:       m.pbqOptions,
 		manager:       m,
+		kw:            win,
 		log:           logging.FromContext(ctx).With("PBQ", partitionID),
 	}
 	m.register(partitionID, p)
@@ -197,6 +207,13 @@ func (m *Manager) ShutDown(ctx context.Context) {
 
 // register is intended to be used by PBQ to register itself with the manager.
 func (m *Manager) register(partitionID partition.ID, p *PBQ) {
+	ww := &RegisteredWindow{
+		AlignedKeyedWindower: p.kw,
+	}
+
+	ww, _ = m.yetToBeClosed.InsertIfNotPresent(ww)
+	ww.count += 1
+
 	m.Lock()
 	defer m.Unlock()
 
@@ -216,6 +233,18 @@ func (m *Manager) register(partitionID partition.ID, p *PBQ) {
 func (m *Manager) deregister(partitionID partition.ID) error {
 	m.Lock()
 	defer m.Unlock()
+
+	ww := &RegisteredWindow{
+		AlignedKeyedWindower: m.pbqMap[partitionID.String()].kw,
+	}
+
+	ww, _ = m.yetToBeClosed.InsertIfNotPresent(ww)
+	ww.count -= 1
+
+	if ww.count == 0 {
+		m.yetToBeClosed.DeleteWindow(ww)
+	}
+
 	delete(m.pbqMap, partitionID.String())
 	activePartitionCount.With(map[string]string{
 		metrics.LabelVertex:             m.vertexName,
@@ -254,3 +283,11 @@ func (m *Manager) Replay(ctx context.Context) {
 	wg.Wait()
 	m.log.Infow("Finished replaying records from store", zap.Duration("took", time.Since(tm)), zap.Any("partitions", partitionsIds))
 }
+
+// NextWindowToBeClosed returns the next keyed window that is yet to be closed
+func (m *Manager) NextWindowToBeClosed() window.AlignedKeyedWindower {
+	if m.yetToBeClosed.Len() == 0 {
+		return nil
+	}
+	return m.yetToBeClosed.Front()
+}
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index 697e6ade..6fb965a4 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -126,7 +126,7 @@ func (rl *ReadLoop) Startup(ctx context.Context) error {
 		keyedWindow.AddSlot(p.Slot)
 
 		// create and invoke process and forward for the partition
-		rl.associatePBQAndPnF(ctx, p)
+		rl.associatePBQAndPnF(ctx, p, keyedWindow)
 	}
 
 	// replays the data (replay internally writes the data from persistent store on to the PBQ)
@@ -224,22 +224,20 @@ messagesLoop:
 
 		// identify and add window for the message
 		windows := rl.upsertWindowsAndKeys(message)
+
 		// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
 		// We need to write the messages to these PBQs.
 		for _, kw := range windows {
-			// identify partition for message
-			partitionID := partition.ID{
-				Start: kw.StartTime(),
-				End:   kw.EndTime(),
-				Slot:  "slot-0", // FIXME: revisit this later. for now hard coded to slot-0
-			}
 
-			err := rl.writeToPBQ(ctx, partitionID, message)
-			// there is no point continuing because we are seeing an error.
-			// this error will ONLY BE set if we are in a erroring loop and ctx.Done() has been invoked.
-			if err != nil {
-				rl.log.Errorw("Failed to write message, asked to stop trying", zap.Any("msgOffSet", message.ReadOffset.String()), zap.String("partitionID", partitionID.String()), zap.Error(err))
-				break messagesLoop
+			for _, partitionID := range kw.Partitions() {
+
+				err := rl.writeToPBQ(ctx, message, partitionID, kw)
+				// there is no point continuing because we are seeing an error.
+				// this error will ONLY BE set if we are in a erroring loop and ctx.Done() has been invoked.
+				if err != nil {
+					rl.log.Errorw("Failed to write message, asked to stop trying", zap.Any("msgOffSet", message.ReadOffset.String()), zap.String("partitionID", partitionID.String()), zap.Error(err))
+					break messagesLoop
+				}
 			}
 		}
 
@@ -251,7 +249,7 @@ messagesLoop:
 
 // writeToPBQ writes to the PBQ. It will return error only if it is not failing to write to PBQ and is in a continuous
 // error loop, and we have received ctx.Done() via SIGTERM.
-func (rl *ReadLoop) writeToPBQ(ctx context.Context, p partition.ID, m *isb.ReadMessage) error {
+func (rl *ReadLoop) writeToPBQ(ctx context.Context, m *isb.ReadMessage, p partition.ID, kw window.AlignedKeyedWindower) error {
 	startTime := time.Now()
 	defer pbqWriteTime.With(map[string]string{
 		metrics.LabelVertex:             rl.vertexName,
@@ -266,7 +264,7 @@ func (rl *ReadLoop) writeToPBQ(ctx context.Context, p partition.ID, m *isb.ReadM
 		Jitter:   0.1,
 	}
 
-	q := rl.associatePBQAndPnF(ctx, p)
+	q := rl.associatePBQAndPnF(ctx, p, kw)
 
 	err := wait.ExponentialBackoff(pbqWriteBackoff, func() (done bool, err error) {
 		rErr := q.Write(context.Background(), m)
@@ -360,7 +358,7 @@ func (rl *ReadLoop) ackMessages(ctx context.Context, messages []*isb.ReadMessage
 
 // associatePBQAndPnF associates a PBQ with the partition if a PBQ exists, else creates a new one and then associates
 // it to the partition.
-func (rl *ReadLoop) associatePBQAndPnF(ctx context.Context, partitionID partition.ID) pbq.ReadWriteCloser {
+func (rl *ReadLoop) associatePBQAndPnF(ctx context.Context, partitionID partition.ID, kw window.AlignedKeyedWindower) pbq.ReadWriteCloser {
 	// look for existing pbq
 	q := rl.pbqManager.GetPBQ(partitionID)
 
@@ -375,7 +373,7 @@ func (rl *ReadLoop) associatePBQAndPnF(ctx context.Context, partitionID partitio
 		}
 		pbqErr = wait.ExponentialBackoffWithContext(ctx, infiniteBackoff, func() (done bool, err error) {
 			var attempt int
-			q, pbqErr = rl.pbqManager.CreateNewPBQ(ctx, partitionID)
+			q, pbqErr = rl.pbqManager.CreateNewPBQ(ctx, partitionID, kw)
 			if pbqErr != nil {
 				attempt += 1
 				rl.log.Warnw("Failed to create pbq during startup, retrying", zap.Any("attempt", attempt), zap.String("partitionID", partitionID.String()), zap.Error(pbqErr))
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 7f1ea122..6a8fcd9b 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -275,10 +275,8 @@ func (mg *memgen) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	if len(msgs) <= 0 {
 		return
 	}
-	// use the first event time as watermark to make it conservative
-	nanos, _ := msgs[0].ReadOffset.Sequence()
-	// remove the nanosecond precision
-	mg.sourcePublishWM.PublishWatermark(wmb.Watermark(time.Unix(0, nanos)), nil) // Source publisher does not care about the offset
+	// use the first event time of the message as watermark to make it conservative
+	mg.sourcePublishWM.PublishWatermark(wmb.Watermark(msgs[0].EventTime), nil) // Source publisher does not care about the offset
 }
 
 // Ack acknowledges an array of offset.
diff --git a/pkg/window/keyed/keyed.go b/pkg/window/keyed/keyed.go
index 154cafbc..3a1bf822 100644
--- a/pkg/window/keyed/keyed.go
+++ b/pkg/window/keyed/keyed.go
@@ -26,16 +26,16 @@ import (
 	"github.com/numaproj/numaflow/pkg/window"
 )
 
-// AlignedKeyedWindow maintains association between keys and a window.
+// AlignedKeyedWindow maintains association between slots and a window.
 // In a keyed stream, we need to close all the partitions when the watermark is past the window.
 type AlignedKeyedWindow struct {
 	// Start time of the window
 	Start time.Time
 	// End time of the window
 	End time.Time
-	// keys map of keys
-	keys map[string]struct{}
-	lock sync.RWMutex
+	// slots map of slots
+	slots map[string]struct{}
+	lock  sync.RWMutex
 }
 
 var _ window.AlignedKeyedWindower = (*AlignedKeyedWindow)(nil)
@@ -45,7 +45,7 @@ func NewKeyedWindow(start time.Time, end time.Time) *AlignedKeyedWindow {
 	kw := &AlignedKeyedWindow{
 		Start: start,
 		End:   end,
-		keys:  make(map[string]struct{}),
+		slots: make(map[string]struct{}),
 		lock:  sync.RWMutex{},
 	}
 	return kw
@@ -62,11 +62,11 @@ func (kw *AlignedKeyedWindow) EndTime() time.Time {
 }
 
 // AddSlot adds a slot to an existing window
-func (kw *AlignedKeyedWindow) AddSlot(key string) {
+func (kw *AlignedKeyedWindow) AddSlot(slot string) {
 	kw.lock.Lock()
 	defer kw.lock.Unlock()
-	if _, ok := kw.keys[key]; !ok {
-		kw.keys[key] = struct{}{}
+	if _, ok := kw.slots[slot]; !ok {
+		kw.slots[slot] = struct{}{}
 	}
 }
 
@@ -75,9 +75,9 @@ func (kw *AlignedKeyedWindow) Partitions() []partition.ID {
 	kw.lock.RLock()
 	defer kw.lock.RUnlock()
 
-	partitions := make([]partition.ID, len(kw.keys))
+	partitions := make([]partition.ID, len(kw.slots))
 	idx := 0
-	for k := range kw.keys {
+	for k := range kw.slots {
 		partitions[idx] = partition.ID{Start: kw.StartTime(), End: kw.EndTime(), Slot: k}
 		idx++
 	}
@@ -85,13 +85,13 @@ func (kw *AlignedKeyedWindow) Partitions() []partition.ID {
 	return partitions
 }
 
-func (kw *AlignedKeyedWindow) Keys() []string {
+func (kw *AlignedKeyedWindow) Slots() []string {
 	kw.lock.RLock()
 	defer kw.lock.RUnlock()
 
-	keys := make([]string, len(kw.keys))
+	keys := make([]string, len(kw.slots))
 	idx := 0
-	for k := range kw.keys {
+	for k := range kw.slots {
 		keys[idx] = k
 		idx++
 	}
diff --git a/pkg/window/sorted_window_list.go b/pkg/window/sorted_window_list.go
new file mode 100644
index 00000000..4c566fde
--- /dev/null
+++ b/pkg/window/sorted_window_list.go
@@ -0,0 +1,204 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package window
+
+import (
+	"container/list"
+	"sync"
+	"time"
+)
+
+// SortedWindowList is a thread safe list implementation, which is sorted by window start time
+// from lowest to highest
+type SortedWindowList[W AlignedWindower] struct {
+	windows *list.List
+	lock    *sync.RWMutex
+}
+
+// NewSortedWindowList implements a window list ordered by the start time. The Front/Head of the list will always have the smallest
+// element while the End/Tail will have the largest element (start time).
+func NewSortedWindowList[W AlignedWindower]() *SortedWindowList[W] {
+	return &SortedWindowList[W]{
+		windows: list.New(),
+		lock:    new(sync.RWMutex),
+	}
+}
+
+// InsertFront tries to insert the window to the Front of the list as long as the window is smaller than the current
+// Front/Head.
+func (s *SortedWindowList[W]) InsertFront(kw W) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	if s.windows.Len() == 0 {
+		s.windows.PushFront(kw)
+		return
+	}
+	if !s.windows.Front().Value.(W).StartTime().After(kw.StartTime()) {
+		return
+	}
+	s.windows.PushFront(kw)
+}
+
+// InsertBack tries to insert the window to the Back of the list as long as the window is larger than the current
+// Back/Tail.
+func (s *SortedWindowList[W]) InsertBack(kw W) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	if s.windows.Len() == 0 {
+		s.windows.PushBack(kw)
+		return
+	}
+	if !s.windows.Back().Value.(W).StartTime().Before(kw.StartTime()) {
+		return
+	}
+	s.windows.PushBack(kw)
+}
+
+// InsertIfNotPresent inserts a window to the list of active windows if not present and returns the window.
+func (s *SortedWindowList[W]) InsertIfNotPresent(kw W) (aw W, isPresent bool) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	// this could be the first window
+	if s.windows.Len() == 0 {
+		s.windows.PushFront(kw)
+		return kw, false
+	}
+
+	earliestWindow := s.windows.Front().Value.(W)
+	recentWindow := s.windows.Back().Value.(W)
+
+	// if there is only one window
+	if earliestWindow.StartTime().Equal(kw.StartTime()) && earliestWindow.EndTime().Equal(kw.EndTime()) {
+		aw = earliestWindow
+		isPresent = true
+	} else if earliestWindow.StartTime().After(kw.StartTime()) {
+		// late arrival
+		s.windows.PushFront(kw)
+		aw = kw
+	} else if recentWindow.StartTime().Before(kw.StartTime()) {
+		// early arrival
+		s.windows.PushBack(kw)
+		aw = kw
+	} else {
+		// a window in the middle
+		for e := s.windows.Back(); e.Prev() != nil; e = e.Prev() {
+			win := e.Value.(W)
+			prevWin := e.Prev().Value.(W)
+			if win.StartTime().Equal(kw.StartTime()) && win.EndTime().Equal(kw.EndTime()) {
+				aw = win
+				isPresent = true
+				break
+			}
+			if win.StartTime().After(kw.StartTime()) && prevWin.StartTime().Before(kw.StartTime()) {
+				s.windows.InsertBefore(kw, e)
+				aw = kw
+				break
+			}
+		}
+	}
+	return
+}
+
+// DeleteWindow deletes a window from the list.
+func (s *SortedWindowList[W]) DeleteWindow(kw W) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	for e := s.windows.Front(); e != nil; {
+		win := e.Value.(W)
+		next := e.Next()
+
+		if win.StartTime().Equal(kw.StartTime()) && win.EndTime().Equal(kw.EndTime()) {
+			s.windows.Remove(e)
+		}
+		e = next
+	}
+}
+
+// RemoveWindows removes a set of windows smaller than or equal to the given time.
+func (s *SortedWindowList[W]) RemoveWindows(tm time.Time) []W {
+
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	closedWindows := make([]W, 0)
+
+	if s.windows.Len() == 0 {
+		return closedWindows
+	}
+	// examine the earliest window
+	earliestWindow := s.windows.Front().Value.(W)
+	if earliestWindow.EndTime().After(tm) {
+		// no windows to close since the watermark is behind the earliest window
+		return closedWindows
+	}
+
+	// close the window if the window end time is equal to the watermark
+	// because window is right exclusive
+	for e := s.windows.Front(); e != nil; {
+		win := e.Value.(W)
+		next := e.Next()
+
+		// break, if we find a window with end time > watermark
+		if win.EndTime().After(tm) {
+			break
+		}
+
+		s.windows.Remove(e)
+		closedWindows = append(closedWindows, win)
+		e = next
+	}
+
+	return closedWindows
+}
+
+// Len returns the length of the window.
+func (s *SortedWindowList[W]) Len() int {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.windows.Len()
+}
+
+// Front returns the smallest element from the list.
+func (s *SortedWindowList[W]) Front() W {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	// FIXME: add len == 0 check
+	return s.windows.Front().Value.(W)
+}
+
+// Back returns the largest element from the list.
+func (s *SortedWindowList[W]) Back() W {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	// FIXME: add len == 0 check
+	return s.windows.Back().Value.(W)
+}
+
+// Items returns the entire window list.
+func (s *SortedWindowList[W]) Items() []W {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	items := make([]W, 0)
+
+	for e := s.windows.Front(); e != nil; {
+		items = append(items, e.Value.(W))
+		e = e.Next()
+	}
+
+	return items
+}
diff --git a/pkg/window/strategy/fixed/fixed.go b/pkg/window/strategy/fixed/fixed.go
index 3d6428c4..ddaaccf2 100644
--- a/pkg/window/strategy/fixed/fixed.go
+++ b/pkg/window/strategy/fixed/fixed.go
@@ -18,12 +18,11 @@ limitations under the License.
 // defined by a static window size, e.g. minutely windows or hourly windows. They are generally aligned, i.e. every
 // window applies across all the data for the corresponding period of time.
 // Package fixed also maintains the state of active keyed windows in a vertex.
-// Keyed Window maintains the association between set of keys and an interval window.
+// Keyed AlignedWindower maintains the association between set of keys and an interval window.
 // keyed also provides the lifecycle management of an interval window. Watermark is used to trigger the expiration of windows.
 package fixed
 
 import (
-	"container/list"
 	"sync"
 	"time"
 
@@ -48,7 +47,7 @@ type Fixed struct {
 	// the traversal from the tail of the list for Get and Create Operations. For Remove Operations, since
 	// the earlier windows are expected to be closed before the more recent ones, we start the traversal
 	// from the Head.
-	entries *list.List
+	entries *window.SortedWindowList[window.AlignedKeyedWindower]
 	lock    sync.RWMutex
 }
 
@@ -58,7 +57,7 @@ var _ window.Windower = (*Fixed)(nil)
 func NewFixed(length time.Duration) window.Windower {
 	return &Fixed{
 		Length:  length,
-		entries: list.New(),
+		entries: window.NewSortedWindowList[window.AlignedKeyedWindower](),
 		lock:    sync.RWMutex{},
 	}
 }
@@ -79,82 +78,11 @@ func (f *Fixed) AssignWindow(eventTime time.Time) []window.AlignedKeyedWindower
 
 // InsertIfNotPresent inserts a window to the list of active windows if not present and returns the window
 func (f *Fixed) InsertIfNotPresent(kw window.AlignedKeyedWindower) (aw window.AlignedKeyedWindower, isPresent bool) {
-	f.lock.Lock()
-	defer f.lock.Unlock()
-	// this could be the first window
-	if f.entries.Len() == 0 {
-		f.entries.PushFront(kw)
-		return kw, false
-	}
-
-	earliestWindow := f.entries.Front().Value.(*keyed.AlignedKeyedWindow)
-	recentWindow := f.entries.Back().Value.(*keyed.AlignedKeyedWindow)
-
-	// if there is only one window
-	if earliestWindow.StartTime().Equal(kw.StartTime()) && earliestWindow.EndTime().Equal(kw.EndTime()) {
-		aw = earliestWindow
-		isPresent = true
-	} else if !earliestWindow.StartTime().Before(kw.EndTime()) {
-		// late arrival
-		f.entries.PushFront(kw)
-		aw = kw
-	} else if !recentWindow.EndTime().After(kw.StartTime()) {
-		// early arrival
-		f.entries.PushBack(kw)
-		aw = kw
-	} else {
-		// a window in the middle
-		for e := f.entries.Back(); e.Prev() != nil; e = e.Prev() {
-			win := e.Value.(*keyed.AlignedKeyedWindow)
-			prevWin := e.Prev().Value.(*keyed.AlignedKeyedWindow)
-			if win.StartTime().Equal(kw.StartTime()) && win.EndTime().Equal(kw.EndTime()) {
-				aw = win
-				isPresent = true
-				break
-			}
-			if !win.StartTime().Before(kw.EndTime()) && !prevWin.EndTime().After(kw.StartTime()) {
-				f.entries.InsertBefore(kw, e)
-				aw = kw
-				break
-			}
-		}
-	}
-	return
+	return f.entries.InsertIfNotPresent(kw)
 }
 
 // RemoveWindows returns an array of keyed windows that are before the current watermark.
 // So these windows can be closed.
 func (f *Fixed) RemoveWindows(wm time.Time) []window.AlignedKeyedWindower {
-	f.lock.Lock()
-	defer f.lock.Unlock()
-
-	closedWindows := make([]window.AlignedKeyedWindower, 0)
-
-	if f.entries.Len() == 0 {
-		return closedWindows
-	}
-	// examine the earliest window
-	earliestWindow := f.entries.Front().Value.(*keyed.AlignedKeyedWindow)
-	if earliestWindow.EndTime().After(wm) {
-		// no windows to close since the watermark is behind the earliest window
-		return closedWindows
-	}
-
-	// close the window if the window end time is equal to the watermark
-	// because window is right exclusive
-	for e := f.entries.Front(); e != nil; {
-		win := e.Value.(*keyed.AlignedKeyedWindow)
-		next := e.Next()
-
-		// break, if we find a window with end time > watermark
-		if win.EndTime().After(wm) {
-			break
-		}
-
-		f.entries.Remove(e)
-		closedWindows = append(closedWindows, win)
-		e = next
-	}
-
-	return closedWindows
+	return f.entries.RemoveWindows(wm)
 }
diff --git a/pkg/window/strategy/sliding/sliding.go b/pkg/window/strategy/sliding/sliding.go
index 604cff18..3a241589 100644
--- a/pkg/window/strategy/sliding/sliding.go
+++ b/pkg/window/strategy/sliding/sliding.go
@@ -22,7 +22,6 @@ limitations under the License.
 package sliding
 
 import (
-	"container/list"
 	"sync"
 	"time"
 
@@ -47,7 +46,7 @@ type Sliding struct {
 	// the traversal from the tail of the list for Get and Create Operations. For Remove Operations, since
 	// the earlier windows are expected to be closed before the more recent ones, we start the traversal
 	// from the Head.
-	entries *list.List
+	entries *window.SortedWindowList[window.AlignedKeyedWindower]
 	lock    sync.RWMutex
 }
 
@@ -58,7 +57,7 @@ func NewSliding(length time.Duration, slide time.Duration) *Sliding {
 	return &Sliding{
 		Length:  length,
 		Slide:   slide,
-		entries: list.New(),
+		entries: window.NewSortedWindowList[window.AlignedKeyedWindower](),
 		lock:    sync.RWMutex{},
 	}
 }
@@ -94,78 +93,10 @@ func (s *Sliding) AssignWindow(eventTime time.Time) []window.AlignedKeyedWindowe
 }
 
 // InsertIfNotPresent inserts a window to the list of active windows if not present and returns the window
-func (s *Sliding) InsertIfNotPresent(kw window.AlignedKeyedWindower) (aw window.AlignedKeyedWindower, isPresent bool) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-	// this could be the first window
-	if s.entries.Len() == 0 {
-		s.entries.PushFront(kw)
-		return kw, false
-	}
-
-	earliestWindow := s.entries.Front().Value.(*keyed.AlignedKeyedWindow)
-	recentWindow := s.entries.Back().Value.(*keyed.AlignedKeyedWindow)
-
-	// if there is only one window
-	if earliestWindow.StartTime().Equal(kw.StartTime()) && earliestWindow.EndTime().Equal(kw.EndTime()) {
-		aw = earliestWindow
-		isPresent = true
-	} else if earliestWindow.StartTime().After(kw.StartTime()) {
-		// late arrival
-		s.entries.PushFront(kw)
-		aw = kw
-	} else if recentWindow.StartTime().Before(kw.StartTime()) {
-		// early arrival
-		s.entries.PushBack(kw)
-		aw = kw
-	} else {
-		// a window in the middle
-		for e := s.entries.Back(); e.Prev() != nil; e = e.Prev() {
-			win := e.Value.(*keyed.AlignedKeyedWindow)
-			prevWin := e.Prev().Value.(*keyed.AlignedKeyedWindow)
-			if win.StartTime().Equal(kw.StartTime()) && win.EndTime().Equal(kw.EndTime()) {
-				aw = win
-				isPresent = true
-				break
-			}
-			if win.StartTime().After(kw.StartTime()) && prevWin.StartTime().Before(kw.StartTime()) {
-				s.entries.InsertBefore(kw, e)
-				aw = kw
-				break
-			}
-		}
-	}
-	return
+func (s *Sliding) InsertIfNotPresent(kw window.AlignedKeyedWindower) (window.AlignedKeyedWindower, bool) {
+	return s.entries.InsertIfNotPresent(kw)
 }
 
 func (s *Sliding) RemoveWindows(wm time.Time) []window.AlignedKeyedWindower {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-
-	closedWindows := make([]window.AlignedKeyedWindower, 0)
-
-	// examine the earliest window
-	earliestWindow := s.entries.Front().Value.(*keyed.AlignedKeyedWindow)
-	if earliestWindow.EndTime().After(wm) {
-		// no windows to close since the watermark is behind the earliest window
-		return closedWindows
-	}
-
-	// close the window if the window end time is equal to the watermark
-	// because window is right exclusive
-	for e := s.entries.Front(); e != nil; {
-		win := e.Value.(*keyed.AlignedKeyedWindow)
-		next := e.Next()
-
-		// break, if we find a window with end time > watermark
-		if win.EndTime().After(wm) {
-			break
-		}
-
-		s.entries.Remove(e)
-		closedWindows = append(closedWindows, win)
-		e = next
-	}
-
-	return closedWindows
+	return s.entries.RemoveWindows(wm)
 }
diff --git a/pkg/window/windower.go b/pkg/window/windower.go
index 3093e1c3..a757deee 100644
--- a/pkg/window/windower.go
+++ b/pkg/window/windower.go
@@ -22,19 +22,23 @@ import (
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 )
 
-// AlignedKeyedWindower represents a bounded window (i.e., it will have a definite start and end time), and the
-// keys that also fall into the same window.
-type AlignedKeyedWindower interface {
+// AlignedWindower defines a window that is aligned and bounded by start and end time.
+type AlignedWindower interface {
 	// StartTime returns the start time of the window
 	StartTime() time.Time
 	// EndTime returns the end time of the window
 	EndTime() time.Time
+}
+
+// AlignedKeyedWindower represents a keyed or non-keyed aligned window (bounded by start and end).
+type AlignedKeyedWindower interface {
+	AlignedWindower
 	// AddSlot adds a slot to the window. Slots are hash-ranges for keys.
 	AddSlot(string)
 	// Partitions returns an array of partition ids
 	Partitions() []partition.ID
-	// Keys returns an array of keys
-	Keys() []string
+	// Slots returns an array of keys
+	Slots() []string
 }
 
 // Windower manages AlignedKeyedWindower


Test patch:
diff --git a/pkg/reduce/pbq/pbq_test.go b/pkg/reduce/pbq/pbq_test.go
index 254899c2..2ccdd582 100644
--- a/pkg/reduce/pbq/pbq_test.go
+++ b/pkg/reduce/pbq/pbq_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/window/keyed"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -51,10 +52,13 @@ func TestPBQ_ReadWrite(t *testing.T) {
 	partitionID := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "new-partition",
+		Slot:  "slot-1",
 	}
 
-	pq, err := qManager.CreateNewPBQ(ctx, partitionID)
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
+	pq, err := qManager.CreateNewPBQ(ctx, partitionID, kwOne)
 	assert.NoError(t, err)
 
 	var readMessages []*isb.ReadMessage
@@ -114,10 +118,14 @@ func Test_PBQReadWithCanceledContext(t *testing.T) {
 	partitionID := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "new-partition",
+		Slot:  "slot-1",
 	}
+
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
 	var pq ReadWriteCloser
-	pq, err = qManager.CreateNewPBQ(ctx, partitionID)
+	pq, err = qManager.CreateNewPBQ(ctx, partitionID, kwOne)
 	assert.NoError(t, err)
 
 	var readMessages []*isb.ReadMessage
@@ -178,11 +186,14 @@ func TestPBQ_WriteWithStoreFull(t *testing.T) {
 	partitionID := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "new-partition",
+		Slot:  "slot-1",
 	}
 
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
 	var pq ReadWriteCloser
-	pq, err = qManager.CreateNewPBQ(ctx, partitionID)
+	pq, err = qManager.CreateNewPBQ(ctx, partitionID, kwOne)
 	assert.NoError(t, err)
 
 	for _, msg := range writeMessages {
diff --git a/pkg/reduce/pbq/pbqmanager_test.go b/pkg/reduce/pbq/pbqmanager_test.go
index cc8c6888..073fc25a 100644
--- a/pkg/reduce/pbq/pbqmanager_test.go
+++ b/pkg/reduce/pbq/pbqmanager_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/window/keyed"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -45,21 +46,28 @@ func TestManager_ListPartitions(t *testing.T) {
 	testPartition := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "partition-1",
+		Slot:  "slot-1",
 	}
 	partitionTwo := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "partition-2",
+		Slot:  "slot-2",
 	}
+
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
+	kwTwo := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-2")
+
 	var pq1, pq2 ReadWriteCloser
-	pq1, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+	pq1, err = pbqManager.CreateNewPBQ(ctx, testPartition, kwOne)
 	assert.NoError(t, err)
 
-	pq2, err = pbqManager.CreateNewPBQ(ctx, partitionTwo)
+	pq2, err = pbqManager.CreateNewPBQ(ctx, partitionTwo, kwTwo)
 	assert.NoError(t, err)
 
-	assert.Len(t, pbqManager.ListPartitions(), 1)
+	assert.Len(t, pbqManager.ListPartitions(), 2)
 
 	err = pq1.GC()
 	assert.NoError(t, err)
@@ -83,9 +91,13 @@ func TestManager_GetPBQ(t *testing.T) {
 	testPartition := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "partition-1",
+		Slot:  "slot-1",
 	}
-	pb1, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
+	pb1, err = pbqManager.CreateNewPBQ(ctx, testPartition, kwOne)
 	assert.NoError(t, err)
 
 	// get the created pbq
@@ -102,14 +114,18 @@ func TestPBQFlow(t *testing.T) {
 	pbqManager, err := NewManager(ctx, "reduce", "test-pipeline", 0, memory.NewMemoryStores(memory.WithStoreSize(size)),
 		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
 	assert.NoError(t, err)
+
 	testPartition := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "partition-1",
+		Slot:  "slot-1",
 	}
 
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
 	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition, kwOne)
 	assert.NoError(t, err)
 
 	msgsCount := 5
@@ -165,12 +181,15 @@ func TestPBQFlowWithNoOpStore(t *testing.T) {
 	testPartition := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "partition-1",
+		Slot:  "slot-1",
 	}
 
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
 	// create a pbq backed with no op store
 	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition, kwOne)
 	msgsCount := 50
 	var wg sync.WaitGroup
 
@@ -223,11 +242,14 @@ func TestManager_Replay(t *testing.T) {
 	testPartition := partition.ID{
 		Start: time.Unix(60, 0),
 		End:   time.Unix(120, 0),
-		Slot:  "partition-1",
+		Slot:  "slot-1",
 	}
 
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
 	var pq ReadWriteCloser
-	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+	pq, err = pbqManager.CreateNewPBQ(ctx, testPartition, kwOne)
 	assert.NoError(t, err)
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -298,3 +320,79 @@ func TestManager_StartUp(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Len(t, ids, 2)
 }
+
+func TestManager_NextWindowToBeClosed(t *testing.T) {
+	size := int64(100)
+
+	ctx := context.Background()
+	pbqManager, err := NewManager(ctx, "reduce", "test-pipeline", 0, memory.NewMemoryStores(memory.WithStoreSize(size)),
+		WithReadTimeout(1*time.Second), WithChannelBufferSize(10))
+	assert.NoError(t, err)
+
+	// create a new pbq using pbq manager
+	partitionOne := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Slot:  "slot-1",
+	}
+	partitionTwo := partition.ID{
+		Start: time.Unix(60, 0),
+		End:   time.Unix(120, 0),
+		Slot:  "slot-2",
+	}
+	partitionThree := partition.ID{
+		Start: time.Unix(120, 0),
+		End:   time.Unix(240, 0),
+		Slot:  "slot-1",
+	}
+	partitionFour := partition.ID{
+		Start: time.Unix(120, 0),
+		End:   time.Unix(240, 0),
+		Slot:  "slot-2",
+	}
+
+	kwOne := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-1")
+
+	kwTwo := keyed.NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
+	kwOne.AddSlot("slot-2")
+
+	kwThree := keyed.NewKeyedWindow(time.Unix(120, 0), time.Unix(240, 0))
+	kwOne.AddSlot("slot-1")
+
+	kwFour := keyed.NewKeyedWindow(time.Unix(120, 0), time.Unix(240, 0))
+	kwOne.AddSlot("slot-2")
+
+	var pq1, pq2, pq3, pq4 ReadWriteCloser
+	pq1, err = pbqManager.CreateNewPBQ(ctx, partitionOne, kwOne)
+	assert.NoError(t, err)
+
+	pq2, err = pbqManager.CreateNewPBQ(ctx, partitionTwo, kwTwo)
+	assert.NoError(t, err)
+
+	pq3, err = pbqManager.CreateNewPBQ(ctx, partitionThree, kwThree)
+	assert.NoError(t, err)
+
+	pq4, err = pbqManager.CreateNewPBQ(ctx, partitionFour, kwFour)
+	assert.NoError(t, err)
+
+	aw := pbqManager.NextWindowToBeClosed()
+	assert.Equal(t, aw.EndTime(), time.Unix(120, 0))
+
+	_ = pq1.GC()
+	aw = pbqManager.NextWindowToBeClosed()
+	assert.Equal(t, aw.EndTime(), time.Unix(120, 0))
+
+	_ = pq2.GC()
+	aw = pbqManager.NextWindowToBeClosed()
+	assert.Equal(t, aw.EndTime(), time.Unix(240, 0))
+
+	_ = pq3.GC()
+	aw = pbqManager.NextWindowToBeClosed()
+	assert.Equal(t, aw.EndTime(), time.Unix(240, 0))
+
+	_ = pq4.GC()
+	aw = pbqManager.NextWindowToBeClosed()
+	assert.Nil(t, aw)
+
+}
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 84899824..c7b0a824 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -36,6 +36,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
 	"github.com/numaproj/numaflow/pkg/watermark/wmb"
+	"github.com/numaproj/numaflow/pkg/window/keyed"
 
 	"github.com/golang/mock/gomock"
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
@@ -97,6 +98,9 @@ func TestProcessAndForward_Process(t *testing.T) {
 		End:   time.UnixMilli(120000),
 		Slot:  "partition-1",
 	}
+	kw := keyed.NewKeyedWindow(time.UnixMilli(60000), time.UnixMilli(120000))
+	kw.AddSlot("partition-1")
+
 	var err error
 	var pbqManager *pbq.Manager
 
@@ -106,7 +110,7 @@ func TestProcessAndForward_Process(t *testing.T) {
 
 	// create a pbq for a partition
 	var simplePbq pbq.ReadWriteCloser
-	simplePbq, err = pbqManager.CreateNewPBQ(ctx, testPartition)
+	simplePbq, err = pbqManager.CreateNewPBQ(ctx, testPartition, kw)
 	assert.NoError(t, err)
 
 	// write messages to pbq
@@ -310,11 +314,13 @@ func createProcessAndForwardAndOTStore(ctx context.Context, key string, pbqManag
 		End:   time.UnixMilli(120000),
 		Slot:  key,
 	}
+	kw := keyed.NewKeyedWindow(time.UnixMilli(60000), time.UnixMilli(120000))
+	kw.AddSlot(key)
 
 	// create a pbq for a partition
 	pw, otStore := buildPublisherMapAndOTStore(toBuffers)
 	var simplePbq pbq.Reader
-	simplePbq, _ = pbqManager.CreateNewPBQ(ctx, testPartition)
+	simplePbq, _ = pbqManager.CreateNewPBQ(ctx, testPartition, kw)
 
 	resultPayload, _ := json.Marshal(PayloadForTest{
 		Key:   "result_payload",
diff --git a/pkg/reduce/readloop/ordered_test.go b/pkg/reduce/readloop/ordered_test.go
index 5b32288b..04b6c642 100644
--- a/pkg/reduce/readloop/ordered_test.go
+++ b/pkg/reduce/readloop/ordered_test.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/window/keyed"
 	"github.com/stretchr/testify/assert"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -109,7 +110,10 @@ func TestOrderedProcessing(t *testing.T) {
 			cCtx, cancelFn := context.WithCancel(ctx)
 			defer cancelFn()
 			for _, _partition := range tt.partitions {
-				p, _ := pbqManager.CreateNewPBQ(ctx, _partition)
+				kw := keyed.NewKeyedWindow(_partition.Start, _partition.End)
+				kw.AddSlot(_partition.Slot)
+
+				p, _ := pbqManager.CreateNewPBQ(ctx, _partition, kw)
 				t := op.schedulePnF(cCtx, identityReducer, p, _partition, toSteps, myForwardTest{}, pw)
 				op.insertTask(t)
 			}
diff --git a/pkg/window/keyed/keyed_test.go b/pkg/window/keyed/keyed_test.go
index 54bfd096..2c9b86cb 100644
--- a/pkg/window/keyed/keyed_test.go
+++ b/pkg/window/keyed/keyed_test.go
@@ -40,7 +40,7 @@ func TestKeyedWindow_AddKey(t *testing.T) {
 		{
 			name: "with_some_existing_keys",
 			given: &AlignedKeyedWindow{
-				keys: map[string]struct{}{"key2": {}, "key3": {}},
+				slots: map[string]struct{}{"key2": {}, "key3": {}},
 			},
 			input:        "key4",
 			expectedKeys: map[string]struct{}{"key2": {}, "key3": {}, "key4": {}},
@@ -50,13 +50,13 @@ func TestKeyedWindow_AddKey(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			kw = NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
-			for k := range tt.given.keys {
+			for k := range tt.given.slots {
 				kw.AddSlot(k)
 			}
 			kw.AddSlot(tt.input)
-			assert.Equal(t, len(tt.expectedKeys), len(kw.keys))
+			assert.Equal(t, len(tt.expectedKeys), len(kw.slots))
 			for k := range tt.expectedKeys {
-				_, ok := kw.keys[k]
+				_, ok := kw.slots[k]
 				assert.True(t, ok)
 			}
 		})
diff --git a/pkg/window/sorted_window_list_test.go b/pkg/window/sorted_window_list_test.go
new file mode 100644
index 00000000..66854e3d
--- /dev/null
+++ b/pkg/window/sorted_window_list_test.go
@@ -0,0 +1,529 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package window
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type TestWindow struct {
+	Start time.Time
+	End   time.Time
+}
+
+func (t *TestWindow) StartTime() time.Time {
+	return t.Start
+}
+
+func (t *TestWindow) EndTime() time.Time {
+	return t.End
+}
+
+func TestSortedWindowList_InsertIfNotPresent(t *testing.T) {
+	tests := []struct {
+		name            string
+		given           []*TestWindow
+		input           *TestWindow
+		expectedWindows []*TestWindow
+		isPresent       bool
+	}{
+		{
+			name:  "FirstWindow",
+			given: []*TestWindow{},
+			input: &TestWindow{
+				Start: time.Unix(0, 0),
+				End:   time.Unix(60, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(0, 0),
+					End:   time.Unix(60, 0),
+				},
+			},
+			isPresent: false,
+		},
+		{
+			name: "late_window",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(60, 0),
+				End:   time.Unix(120, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(60, 0),
+					End:   time.Unix(120, 0),
+				},
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+			},
+			isPresent: false,
+		},
+		{
+			name: "early_window",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(240, 0),
+				End:   time.Unix(300, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			isPresent: false,
+		},
+		{
+			name: "insert_middle",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(180, 0),
+				End:   time.Unix(240, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			isPresent: false,
+		},
+		{
+			name: "already_present",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(180, 0),
+				End:   time.Unix(240, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			isPresent: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			windows := NewSortedWindowList[*TestWindow]()
+			setup(windows, tt.given)
+			ret, isPresent := windows.InsertIfNotPresent(tt.input)
+			assert.Equal(t, tt.isPresent, isPresent)
+			assert.Equal(t, tt.input.Start, ret.StartTime())
+			assert.Equal(t, tt.input.End, ret.EndTime())
+			assert.Equal(t, len(tt.expectedWindows), windows.Len())
+			nodes := windows.Items()
+			i := 0
+			for _, kw := range tt.expectedWindows {
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
+			}
+		})
+	}
+}
+
+func TestSortedWindowList_RemoveWindows(t *testing.T) {
+	tests := []struct {
+		name            string
+		given           []*TestWindow
+		input           time.Time
+		expectedWindows []*TestWindow
+	}{
+		{
+			name: "remove_windows",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: time.Unix(180, 0),
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			windows := NewSortedWindowList[*TestWindow]()
+			setup(windows, tt.given)
+			removedWindows := windows.RemoveWindows(tt.input)
+			assert.Equal(t, len(tt.expectedWindows), len(removedWindows))
+			nodes := removedWindows
+			i := 0
+			for _, kw := range tt.expectedWindows {
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
+			}
+		})
+	}
+}
+
+func TestSortedWindowList_DeleteWindow(t *testing.T) {
+	tests := []struct {
+		name            string
+		given           []*TestWindow
+		input           *TestWindow
+		expectedWindows []*TestWindow
+	}{
+		{
+			name: "delete_middle",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(180, 0),
+				End:   time.Unix(240, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+		{
+			name: "delete_first",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(120, 0),
+				End:   time.Unix(180, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+		{
+			name: "delete_last",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(240, 0),
+				End:   time.Unix(300, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(120, 0),
+					End:   time.Unix(180, 0),
+				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			windows := NewSortedWindowList[*TestWindow]()
+			setup(windows, tt.given)
+			windows.DeleteWindow(tt.input)
+			assert.Equal(t, len(tt.expectedWindows), windows.Len())
+			nodes := windows.Items()
+			i := 0
+			for _, kw := range tt.expectedWindows {
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
+			}
+		})
+	}
+}
+
+func TestSortedWindowList_InsertFront(t *testing.T) {
+	tests := []struct {
+		name            string
+		given           []*TestWindow
+		input           *TestWindow
+		expectedWindows []*TestWindow
+	}{
+		{
+			name:  "insert_empty",
+			given: []*TestWindow{},
+			input: &TestWindow{
+				Start: time.Unix(240, 0),
+				End:   time.Unix(300, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+		{
+			name: "insert_equal",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(240, 0),
+				End:   time.Unix(300, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+		{
+			name: "insert_greater",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(300, 0),
+				End:   time.Unix(360, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			windows := NewSortedWindowList[*TestWindow]()
+			setup(windows, tt.given)
+			windows.InsertFront(tt.input)
+			assert.Equal(t, len(tt.expectedWindows), windows.Len())
+			nodes := windows.Items()
+			i := 0
+			for _, kw := range tt.expectedWindows {
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
+			}
+		})
+	}
+}
+
+func TestSortedWindowList_InsertBack(t *testing.T) {
+	tests := []struct {
+		name            string
+		given           []*TestWindow
+		input           *TestWindow
+		expectedWindows []*TestWindow
+	}{
+		{
+			name:  "insert_empty",
+			given: []*TestWindow{},
+			input: &TestWindow{
+				Start: time.Unix(240, 0),
+				End:   time.Unix(300, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+		{
+			name: "insert_equal",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(240, 0),
+				End:   time.Unix(300, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+		{
+			name: "insert_smaller",
+			given: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+			input: &TestWindow{
+				Start: time.Unix(120, 0),
+				End:   time.Unix(180, 0),
+			},
+			expectedWindows: []*TestWindow{
+				{
+					Start: time.Unix(240, 0),
+					End:   time.Unix(300, 0),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			windows := NewSortedWindowList[*TestWindow]()
+			setup(windows, tt.given)
+			windows.InsertBack(tt.input)
+			assert.Equal(t, len(tt.expectedWindows), windows.Len())
+			nodes := windows.Items()
+			i := 0
+			for _, kw := range tt.expectedWindows {
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
+			}
+		})
+	}
+}
+
+func setup(windows *SortedWindowList[*TestWindow], wins []*TestWindow) {
+	for _, win := range wins {
+		windows.InsertBack(win)
+	}
+}
diff --git a/pkg/window/strategy/fixed/fixed_test.go b/pkg/window/strategy/fixed/fixed_test.go
index e8debe36..33b9c61a 100644
--- a/pkg/window/strategy/fixed/fixed_test.go
+++ b/pkg/window/strategy/fixed/fixed_test.go
@@ -17,13 +17,13 @@ limitations under the License.
 package fixed
 
 import (
-	"container/list"
 	"testing"
 	"time"
 
-	"github.com/numaproj/numaflow/pkg/window"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/numaproj/numaflow/pkg/window"
+
 	"github.com/numaproj/numaflow/pkg/window/keyed"
 )
 
@@ -240,11 +240,12 @@ func TestAligned_InsertIfNotPresent(t *testing.T) {
 			assert.Equal(t, tt.input.Start, ret.StartTime())
 			assert.Equal(t, tt.input.End, ret.EndTime())
 			assert.Equal(t, len(tt.expectedWindows), windows.entries.Len())
-			node := windows.entries.Front()
+			nodes := windows.entries.Items()
+			i := 0
 			for _, kw := range tt.expectedWindows {
-				assert.Equal(t, kw.StartTime(), node.Value.(*keyed.AlignedKeyedWindow).StartTime())
-				assert.Equal(t, kw.EndTime(), node.Value.(*keyed.AlignedKeyedWindow).EndTime())
-				node = node.Next()
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
 			}
 		})
 	}
@@ -385,8 +386,8 @@ func TestFixed_RemoveWindows(t *testing.T) {
 }
 
 func setup(windows *Fixed, wins []*keyed.AlignedKeyedWindow) {
-	windows.entries = list.New()
+	windows.entries = window.NewSortedWindowList[window.AlignedKeyedWindower]()
 	for _, win := range wins {
-		windows.entries.PushBack(win)
+		windows.entries.InsertBack(win)
 	}
 }
diff --git a/pkg/window/strategy/sliding/sliding_test.go b/pkg/window/strategy/sliding/sliding_test.go
index d92580d4..74df10f3 100644
--- a/pkg/window/strategy/sliding/sliding_test.go
+++ b/pkg/window/strategy/sliding/sliding_test.go
@@ -17,13 +17,13 @@ limitations under the License.
 package sliding
 
 import (
-	"container/list"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/numaproj/numaflow/pkg/window"
 	"github.com/numaproj/numaflow/pkg/window/keyed"
-	"github.com/stretchr/testify/assert"
 )
 
 // TestSliding_AssignWindow tests the assignment of element to a set of windows
@@ -287,11 +287,12 @@ func TestAligned_CreateWindow(t *testing.T) {
 			assert.Equal(t, tt.input.StartTime(), ret.StartTime())
 			assert.Equal(t, tt.input.EndTime(), ret.EndTime())
 			assert.Equal(t, len(tt.expectedWindows), windows.entries.Len())
-			node := windows.entries.Front()
+			nodes := windows.entries.Items()
+			i := 0
 			for _, kw := range tt.expectedWindows {
-				assert.Equal(t, kw.StartTime(), node.Value.(*keyed.AlignedKeyedWindow).Start)
-				assert.Equal(t, kw.EndTime(), node.Value.(*keyed.AlignedKeyedWindow).End)
-				node = node.Next()
+				assert.Equal(t, kw.StartTime(), nodes[i].StartTime())
+				assert.Equal(t, kw.EndTime(), nodes[i].EndTime())
+				i += 1
 			}
 		})
 	}
@@ -320,8 +321,8 @@ func TestSliding_RemoveWindows(t *testing.T) {
 }
 
 func setup(windows *Sliding, wins []*keyed.AlignedKeyedWindow) {
-	windows.entries = list.New()
+	windows.entries = window.NewSortedWindowList[window.AlignedKeyedWindower]()
 	for _, win := range wins {
-		windows.entries.PushBack(win)
+		windows.entries.InsertBack(win)
 	}
 }