y/watermark.go
func (w *WaterMark) process(closer *Closer) {
	defer closer.Done()

	var indices uint64Heap
	// pending maps raft proposal index to the number of pending mutations for this proposal.
	pending := make(map[uint64]int)
	waiters := make(map[uint64][]chan struct{})

	heap.Init(&indices)
	var loop uint64

	processOne := func(index uint64, done bool) {
		// If not already done, then set. Otherwise, don't undo a done entry.
		prev, present := pending[index]
		if !present {
			heap.Push(&indices, index)
		}

		delta := 1
		if done {
			delta = -1
		}
		pending[index] = prev + delta

		loop++
		if len(indices) > 0 && loop%10000 == 0 {
			min := indices[0]
			w.elog.Printf("WaterMark %s: Done entry %4d. Size: %4d Watermark: %-4d Looking for: %-4d. Value: %d\n",
				w.Name, index, len(indices), w.DoneUntil(), min, pending[min])
		}

		// Update mark by going through all indices in order; and checking if they have
		// been done. Stop at the first index, which isn't done.
		doneUntil := w.DoneUntil()
		if doneUntil > index {
			AssertTruef(false, "Name: %s doneUntil: %d. Index: %d", w.Name, doneUntil, index)
		}

		until := doneUntil
		loops := 0

		for len(indices) > 0 {
			min := indices[0]
			if done := pending[min]; done > 0 {
				break // len(indices) will be > 0.
			}
			// Even if done is called multiple times causing it to become
			// negative, we should still pop the index.
			heap.Pop(&indices)
			delete(pending, min)
			until = min
			loops++
		}
		for i := doneUntil + 1; i <= until; i++ {
			toNotify := waiters[i]
			for _, ch := range toNotify {
				close(ch)
			}
			delete(waiters, i) // Release the memory back.
		}
		if until != doneUntil {
			AssertTrue(atomic.CompareAndSwapUint64(&w.doneUntil, doneUntil, until))
			w.elog.Printf("%s: Done until %d. Loops: %d\n", w.Name, until, loops)
		}
	}

	for {
		select {
		case <-closer.HasBeenClosed():
			return
		case mark := <-w.markCh:
			if mark.waiter != nil {
				doneUntil := atomic.LoadUint64(&w.doneUntil)
				if doneUntil >= mark.index {
					close(mark.waiter)
				} else {
					ws, ok := waiters[mark.index]
					if !ok {
						waiters[mark.index] = []chan struct{}{mark.waiter}
					} else {
						waiters[mark.index] = append(ws, mark.waiter)
					}
				}
			} else {
				if mark.index > 0 {
					processOne(mark.index, mark.done)
				}
				for _, index := range mark.indices {
					processOne(index, mark.done)
				}
			}
		}
	}
}
func (w *WaterMark) process(closer *Closer) {
	defer closer.Done()

	var indices uint64Heap
	// pending maps raft proposal index to the number of pending mutations for this proposal.
	pending := make(map[uint64]int)
	waiters := make(map[uint64][]chan struct{})

	heap.Init(&indices)
	var loop uint64

	processOne := func(index uint64, done bool) {
		// If not already done, then set. Otherwise, don't undo a done entry.
		prev, present := pending[index]
		if !present {
			heap.Push(&indices, index)
		}

		delta := 1
		if done {
			delta = -1
		}
		pending[index] = prev + delta

		loop++
		if len(indices) > 0 && loop%10000 == 0 {
			min := indices[0]
			w.elog.Printf("WaterMark %s: Done entry %4d. Size: %4d Watermark: %-4d Looking for: %-4d. Value: %d\n",
				w.Name, index, len(indices), w.DoneUntil(), min, pending[min])
		}

		// Update mark by going through all indices in order; and checking if they have
		// been done. Stop at the first index, which isn't done.
		doneUntil := w.DoneUntil()
		if doneUntil > index {
			AssertTruef(false, "Name: %s doneUntil: %d. Index: %d", w.Name, doneUntil, index)
		}

		until := doneUntil
		loops := 0

		for len(indices) > 0 {
			min := indices[0]
			if done := pending[min]; done > 0 {
				break // len(indices) will be > 0.
			}
			// Even if done is called multiple times causing it to become
			// negative, we should still pop the index.
			heap.Pop(&indices)
			delete(pending, min)
			until = min
			loops++
		}
		for i := doneUntil + 1; i <= until; i++ {
			toNotify := waiters[i]
			for _, ch := range toNotify {
				close(ch)
			}
			delete(waiters, i) // Release the memory back.
		}
		if until != doneUntil {
			AssertTrue(atomic.CompareAndSwapUint64(&w.doneUntil, doneUntil, until))
			w.elog.Printf("%s: Done until %d. Loops: %d\n", w.Name, until, loops)
		}
	}

	for {
		select {
		case <-closer.HasBeenClosed():
			return
		case mark := <-w.markCh:
			if mark.waiter != nil {
				doneUntil := atomic.LoadUint64(&w.doneUntil)
				if doneUntil >= mark.index {
					close(mark.waiter)
				} else {
					ws, ok := waiters[mark.index]
					if !ok {
						waiters[mark.index] = []chan struct{}{mark.waiter}
					} else {
						waiters[mark.index] = append(ws, mark.waiter)
					}
				}
			} else {
				if mark.index > 0 {
					processOne(mark.index, mark.done)
				}
				for _, index := range mark.indices {
					processOne(index, mark.done)
				}
			}
		}
	}
}
func ( w * WaterMark ) process ( closer * Closer ) { defer closer . Done ( ) var indices uint64Heap // pending maps raft proposal index to the number of pending mutations for this proposal. pending := make ( map [ uint64 ] int ) waiters := make ( map [ uint64 ] [ ] chan struct { } ) heap . Init ( & indices ) var loop uint64 processOne := func ( index uint64 , done bool ) { // If not already done, then set. Otherwise, don't undo a done entry. prev , present := pending [ index ] if ! present { heap . Push ( & indices , index ) } delta := 1 if done { delta = - 1 } pending [ index ] = prev + delta loop ++ if len ( indices ) > 0 && loop % 10000 == 0 { min := indices [ 0 ] w . elog . Printf ( " \n " , w . Name , index , len ( indices ) , w . DoneUntil ( ) , min , pending [ min ] ) } // Update mark by going through all indices in order; and checking if they have // been done. Stop at the first index, which isn't done. doneUntil := w . DoneUntil ( ) if doneUntil > index { AssertTruef ( false , " " , w . Name , doneUntil , index ) } until := doneUntil loops := 0 for len ( indices ) > 0 { min := indices [ 0 ] if done := pending [ min ] ; done > 0 { break // len(indices) will be > 0. } // Even if done is called multiple times causing it to become // negative, we should still pop the index. heap . Pop ( & indices ) delete ( pending , min ) until = min loops ++ } for i := doneUntil + 1 ; i <= until ; i ++ { toNotify := waiters [ i ] for _ , ch := range toNotify { close ( ch ) } delete ( waiters , i ) // Release the memory back. } if until != doneUntil { AssertTrue ( atomic . CompareAndSwapUint64 ( & w . doneUntil , doneUntil , until ) ) w . elog . Printf ( " \n " , w . Name , until , loops ) } } for { select { case <- closer . HasBeenClosed ( ) : return case mark := <- w . markCh : if mark . waiter != nil { doneUntil := atomic . LoadUint64 ( & w . doneUntil ) if doneUntil >= mark . index { close ( mark . waiter ) } else { ws , ok := waiters [ mark . index ] if ! ok { waiters [ mark . index ] = [ ] chan struct { } { mark . waiter } } else { waiters [ mark . index ] = append ( ws , mark . waiter ) } } } else { if mark . index > 0 { processOne ( mark . index , mark . done ) } for _ , index := range mark . indices { processOne ( index , mark . done ) } } } } }
// process is used to process the Mark channel. This is not thread-safe,
// so only run one goroutine for process. One is sufficient, because
// all goroutine ops use purely memory and cpu.
// Each index has to emit atleast one begin watermark in serial order otherwise waiters
// can get blocked idefinitely. Example: We had an watermark at 100 and a waiter at 101,
// if no watermark is emitted at index 101 then waiter would get stuck indefinitely as it
// can't decide whether the task at 101 has decided not to emit watermark or it didn't get
// scheduled yet.
process is used to process the Mark channel . This is not thread - safe so only run one goroutine for process . One is sufficient because all goroutine ops use purely memory and cpu . Each index has to emit atleast one begin watermark in serial order otherwise waiters can get blocked idefinitely . Example : We had an watermark at 100 and a waiter at 101 if no watermark is emitted at index 101 then waiter would get stuck indefinitely as it can t decide whether the task at 101 has decided not to emit watermark or it didn t get scheduled yet .