<commit_before>package conntrack

import (
	"fmt"
	"io"
	"sync"
	"text/tabwriter"
	"time"
)

type reason = string

type Instance struct {
	maxEntries int
	Timeout    func(Entry) time.Duration

	mu              sync.Mutex
	entries         map[Entry]handles
	waitersByReason map[reason]handles
	waitersByEntry  map[Entry][]*EntryHandle
}

type handles = map[*EntryHandle]struct{}

func NewInstance() *Instance {
	i := &Instance{
		maxEntries: 200,
		Timeout: func(e Entry) time.Duration {
			// udp is the main offender, and the default is allegedly 30s.
			return 30 * time.Second
		},
		entries:         make(map[Entry]handles),
		waitersByReason: make(map[reason]handles),
		waitersByEntry:  make(map[Entry][]*EntryHandle),
	}
	return i
}

func (i *Instance) SetMaxEntries(max int) {
	i.mu.Lock()
	defer i.mu.Unlock()
	prev := i.maxEntries
	i.maxEntries = max
	for j := prev; j < max; j++ {
		i.wakeAny()
	}
}

func (i *Instance) remove(eh *EntryHandle) {
	i.mu.Lock()
	defer i.mu.Unlock()
	hs := i.entries[eh.e]
	delete(hs, eh)
	if len(hs) == 0 {
		delete(i.entries, eh.e)
		i.wakeWaiter(eh.reason)
	}
}

func (i *Instance) chooseWakeReason(avoid reason) reason {
	for k := range i.waitersByReason {
		if k == avoid {
			continue
		}
		return k
	}
	return avoid
}

func (i *Instance) wakeWaiter(avoid reason) {
	r := i.chooseWakeReason(avoid)
	i.wakeReason(r)
}

func (i *Instance) wakeAny() {
	for r := range i.waitersByReason {
		i.wakeReason(r)
		break
	}
}

func (i *Instance) wakeReason(r reason) {
	for k := range i.waitersByReason[r] {
		i.wakeEntry(k.e)
		break
	}
}

func (i *Instance) wakeEntry(e Entry) {
	i.entries[e] = make(handles)
	for _, eh := range i.waitersByEntry[e] {
		i.entries[e][eh] = struct{}{}
		delete(i.waitersByReason[eh.reason], eh)
		eh.added.Unlock()
	}
	delete(i.waitersByEntry, e)
}

func (i *Instance) Wait(e Entry, reason string) (eh *EntryHandle) {
	eh = &EntryHandle{
		reason: reason,
		e:      e,
		i:      i,
	}
	i.mu.Lock()
	hs, ok := i.entries[eh.e]
	if ok {
		hs[eh] = struct{}{}
		i.mu.Unlock()
		return
	}
	if len(i.entries) < i.maxEntries {
		i.entries[eh.e] = handles{
			eh: struct{}{},<commit_msg>conntrack: Fix spurious wake ups on empty reason wait lists<commit_after>package conntrack

import (
	"fmt"
	"io"
	"sync"
	"text/tabwriter"
	"time"
)

type reason = string

type Instance struct {
	maxEntries int
	Timeout    func(Entry) time.Duration

	mu              sync.Mutex
	entries         map[Entry]handles
	waitersByReason map[reason]handles
	waitersByEntry  map[Entry][]*EntryHandle
}

type handles = map[*EntryHandle]struct{}

func NewInstance() *Instance {
	i := &Instance{
		maxEntries: 200,
		Timeout: func(e Entry) time.Duration {
			// udp is the main offender, and the default is allegedly 30s.
			return 30 * time.Second
		},
		entries:         make(map[Entry]handles),
		waitersByReason: make(map[reason]handles),
		waitersByEntry:  make(map[Entry][]*EntryHandle),
	}
	return i
}

func (i *Instance) SetMaxEntries(max int) {
	i.mu.Lock()
	defer i.mu.Unlock()
	prev := i.maxEntries
	i.maxEntries = max
	for j := prev; j < max; j++ {
		i.wakeAny()
	}
}

func (i *Instance) remove(eh *EntryHandle) {
	i.mu.Lock()
	defer i.mu.Unlock()
	hs := i.entries[eh.e]
	delete(hs, eh)
	if len(hs) == 0 {
		delete(i.entries, eh.e)
		i.wakeWaiter(eh.reason)
	}
}

func (i *Instance) chooseWakeReason(avoid reason) reason {
	for k := range i.waitersByReason {
		if k == avoid {
			continue
		}
		return k
	}
	return avoid
}

func (i *Instance) wakeWaiter(avoid reason) {
	r := i.chooseWakeReason(avoid)
	i.wakeReason(r)
}

func (i *Instance) wakeAny() {
	for r := range i.waitersByReason {
		i.wakeReason(r)
		break
	}
}

func (i *Instance) wakeReason(r reason) {
	for k := range i.waitersByReason[r] {
		i.wakeEntry(k.e)
		break
	}
}

func (i *Instance) wakeEntry(e Entry) {
	i.entries[e] = make(handles)
	for _, eh := range i.waitersByEntry[e] {
		i.entries[e][eh] = struct{}{}
		delete(i.waitersByReason[eh.reason], eh)
		if len(i.waitersByReason[eh.reason]) == 0 {
			delete(i.waitersByReason, eh.reason)
		}
		eh.added.Unlock()
	}
	delete(i.waitersByEntry, e)
}

func (i *Instance) Wait(e Entry, reason string) (eh *EntryHandle) {
	eh = &EntryHandle{
		reason: reason,
		e:      e,
		i:      i,
	}
	i.mu.Lock()
	hs, ok := i.entries[eh.e]
	if ok {
		hs[eh] = struct{}{}
		i.mu.Unlock()
		return
	}
	if len(i.entries) < i.maxEntries {
		i.entries[eh.e] = handles{
			eh: struct{}{},
{'fraction_non_alphanumeric': 0.15155615696887687, 'fraction_numerical': 0.0038340099233198015, 'mean_word_length': 3.4483450351053158, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1639937', 'n_tokens_mistral': 1871, 'n_tokens_neox': 1768, 'n_words': 411}