Pruner: A wapper to support DelPattern

Author: jsmorph

pruner/doc.go

@@ -0,0 +1,28 @@
+package pruner
+
+// Support for removing a pattern from a matcher
+//
+// The core quamina.Matcher doesn't currently support deleting
+// patterns.  Some of the contemplated implementations of would
+// probably be pretty difficult.  At least one approach is pretty
+// easy: Wrap the current matcher to filter removed patterns from
+// match results and periodically rebuild the matcher from scrach with
+// the live patterns.  More specifically:
+//
+// 1.  Remember patterns that have been added
+// 2. Remember patterns that have been removed (implicitly)
+// 3. Filter MatchedFor...() results to remove any removed patterns
+// 4. Support rebuilding the matcher state periodically with only the
+//    live patterns.
+// 5. Maintain some statistics to help decide when to rebuild
+//
+// The implementation of the set of live patterns is pluggable via a
+// Go interface (State).  The default implementation is a
+// `map[quamina.X]string (MemState)`.  Other implementations can
+// provide persistence.
+//
+// By default, rebuilding is triggered automatically (synchronously
+// currently) during mutations.  The code also supports pluggable
+// rebuilding policies, but those features are not currently exposed.
+// A Rebuild method is available for the application to force a
+// rebuild.

pruner/pruner.go

@@ -0,0 +1,327 @@
+package pruner
+
+import (
+	"sync"
+	"time"
+
+	quamina "quamina/lib"
+)
+
+// Stats reports basic counts to aid in deciding when to Rebuild.
+type Stats struct {
+	// Some of these values are ints instead of uints because Go
+	// likes to print uints in hex, and I'd like to see some
+	// temporary logging output in decimal.  ToDo: back to uints?
+
+	// Live the count of total live patterns.
+	Live int
+
+	// Added is the count of total patterns added.
+	Added int
+
+	// Deleted is the count of the patterns removed.
+	Deleted int
+
+	// Emitted is the count to total patterns found since the last
+	// rebuild.
+	Emitted int64
+
+	// Filtered is the count of pattners that have been removed
+	// from MatchFor results (since the last rebuild) because
+	// their patterns had been removed.
+	Filtered int64
+
+	// LastRebuilt is the time the last rebuild started.
+	LastRebuilt time.Time
+
+	// RebuildDuration is the duration of the last rebuild.
+	RebuildDuration time.Duration
+
+	// RebuildPurged is the count of patterns removed during
+	// rebuild.
+	RebuildPurged int
+}
+
+// Matcher provides DelPattern on top of quamina.Matcher.
+//
+// Matcher maintains the set of live patterns, and it will rebuild the
+// underlying matcher synchronously (currently) periodically sduring
+// standard operations (AddPattern, DelPattern, MatchesForJSONEvent).
+//
+// Roughly speaking, the current rebuild policy automatically rebuilds
+// the index when the ratio of filtered patterns to emitted patterns
+// exceeds 0.2 (and if there's been some traffic).
+//
+// An application can call Rebuild to force a rebuild at any any.  See
+// Stats() to obtain some useful statistics about the matcher.
+//
+// Eventually automatically-invoked rebuild policies might be
+// pluggable.
+type Matcher struct {
+	// Matcher is the underlying matcher that does the hard work.
+	//
+	// Matcher should maybe not be public.
+	*quamina.Matcher
+
+	// live is live set of patterns.
+	live State
+
+	stats Stats
+
+	// rebuildTrigger, if not nil, determines when a mutation
+	// triggers a rebuild.
+	//
+	// If nil, no automatic Rebuild is ever triggered.
+	rebuildTrigger rebuildTrigger
+
+	// lock protectes the pointer the underlying Matcher as well as stats.
+	//
+	// The Matcher pointer is updated after a successful Rebuild.
+	// Stats are updated by Add, Del, and Rebuild.
+	lock sync.RWMutex
+}
+
+var defaultRebuildTrigger = newTooMuchFiltering(0.2, 1000)
+
+// tooMuchFiltering is the standard rebuildTrigger, which will fire
+// when:
+//
+//   MinAction is less than the sum of counts of found and filtered
+//   patterns and
+//
+//   FilteredToEmitted is greater than the ratio of counts of filtered
+//   and found patterns.
+//
+// defaultRebuildTrigger provides the default trigger policy used by
+// NewMatcher.
+type tooMuchFiltering struct {
+	FilteredToEmitted float64
+	MinAction         int64
+}
+
+func newTooMuchFiltering(ratio float64, min int64) *tooMuchFiltering {
+	return &tooMuchFiltering{
+		FilteredToEmitted: ratio,
+		MinAction:         min,
+	}
+}
+
+func (t *tooMuchFiltering) Rebuild(added bool, s *Stats) bool {
+	if added {
+		// No need to think when we're adding a pattern since
+		// that operation cannot result in an increase of
+		// filtered patterns -- I think.
+		return false
+	}
+
+	if s.Emitted+s.Filtered < t.MinAction {
+		return false
+	}
+
+	if s.Emitted == 0 {
+		return true
+	}
+
+	var (
+		numerator   = float64(s.Filtered)
+		denominator = float64(s.Emitted)
+		ratio       = numerator / denominator
+	)
+
+	return t.FilteredToEmitted < ratio
+}
+
+// DisableRebuild will prevent any automatic rebuilds.
+func (m *Matcher) DisableRebuild() {
+	m.lock.Lock()
+	m.rebuildTrigger = nil
+	m.lock.Unlock()
+}
+
+// rebuildTrigger provides a way to control when rebuilds are
+// automatically triggered during standard operations.
+//
+// Currently an AddPattern, DelPattern, or MatchesForJSONEvent can
+// trigger a rebuild.  When a rebuild is triggered, it's executed
+// synchronous, the the Add/Del method doesn't return until the
+// rebuild is complete.
+type rebuildTrigger interface {
+	// Rebuild should return true to trigger a rebuild.
+	//
+	// This method is called by AddPattern and DelPattern.  added
+	// is true when called by AddPattern; false otherwise. These
+	// methods do not return until the rebuild is complete, so
+	// beware.
+	Rebuild(added bool, s *Stats) bool
+}
+
+// NewMatcher does what you'd expect.
+//
+// The State defaults to MemState.
+func NewMatcher(s State) *Matcher {
+	if s == nil {
+		s = NewMemState()
+	}
+	return &Matcher{
+		Matcher:        quamina.NewMatcher(),
+		live:           s,
+		rebuildTrigger: defaultRebuildTrigger,
+	}
+}
+
+// maybeRebuild calls rebuildTrigger (if not) and then calls rebuild()
+// if that trigger said to do that.  If rebuildTrigger is nil, no
+// rebuild is executed.
+//
+// This method assumes the caller has a write lock.
+func (m *Matcher) maybeRebuild(added bool) error {
+	if m.rebuildTrigger == nil {
+		return nil
+	}
+	if m.rebuildTrigger.Rebuild(added, &m.stats) {
+		return m.rebuild(added)
+	}
+
+	return nil
+}
+
+// AddPattern calls the underlying AddPattern method and then maybe
+// rebuilds the index (if the AddPattern succeeded).
+func (m *Matcher) AddPattern(x quamina.X, pat string) error {
+	var err error
+
+	// Do we m.live.Add first, do we m.Matcher.AddPattern first?
+	if err = m.Matcher.AddPattern(x, pat); err == nil {
+		m.lock.Lock()
+		m.stats.Added++
+		m.stats.Live++
+		m.maybeRebuild(true)
+		m.lock.Unlock()
+		err = m.live.Add(x, pat)
+		// ToDo: Contemplate what do to about an error here
+		// (or if we got an error from AddPattern after we did
+		// live.Add.
+	}
+
+	return err
+}
+
+// MatchesForJSONEvent calls the underlying MatchesForJSONEvent method
+// and then maybe rebuilds the index.
+func (m *Matcher) MatchesForJSONEvent(event []byte) ([]quamina.X, error) {
+	xs, err := m.Matcher.MatchesForJSONEvent(event)
+	if err != nil {
+		return nil, err
+	}
+
+	// Delove any X that isn't in the live set.
+
+	acc := make([]quamina.X, 0, len(xs))
+
+	// We're updating stats.Filtered, so we need a write lock.  If
+	// we forget about stats.Filtered, we can live with only a
+	// read lock here.
+	var emitted, filtered int64
+	for _, x := range xs {
+		var have string
+		if have, err = m.live.Get(x); err != nil {
+			break
+		}
+		if have == "" {
+			filtered++
+			continue
+		}
+		acc = append(acc, x)
+		emitted++
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	m.lock.Lock()
+	m.stats.Filtered += filtered
+	m.stats.Emitted += emitted
+	m.maybeRebuild(false)
+	m.lock.Unlock()
+
+	return acc, nil
+}
+
+// DelPattern "removes" the pattern from the index and maybe rebuilds
+// the index.
+func (m *Matcher) DelPattern(x quamina.X) (bool, error) {
+	had, err := m.live.Del(x)
+	if err == nil {
+		if had {
+			m.lock.Lock()
+			m.stats.Deleted++
+			m.stats.Live--
+			m.maybeRebuild(false)
+			m.lock.Unlock()
+		}
+	}
+
+	return had, err
+}
+
+// Rebuild rebuilds the matcher state based on only live patterns.
+//
+// If calling fearlessly, then the old matcher is released before
+// building the new one.
+//
+// This method resets the Stats.
+func (m *Matcher) Rebuild(fearlessly bool) error {
+	m.lock.Lock()
+	err := m.rebuild(fearlessly)
+	m.lock.Unlock()
+	return err
+}
+
+// rebuild is Rebuild but assumes having the lock.
+func (m *Matcher) rebuild(fearlessly bool) error {
+	// We assume we have the lock.
+
+	// Nothing fancy here now.
+
+	var (
+		then = time.Now()
+		m1   = quamina.NewMatcher()
+	)
+
+	if fearlessly {
+		// Let the GC reduce heap requirements?
+		m.Matcher = nil
+	}
+
+	count := 0
+	err := m.live.Iterate(func(x quamina.X, p string) error {
+		err := m1.AddPattern(x, p)
+		if err == nil {
+			count++
+		}
+		return err
+	})
+
+	if err == nil {
+		m.Matcher = m1
+		m.stats.RebuildPurged = m.stats.Deleted
+		m.stats.Live = count
+		m.stats.Added = 0
+		m.stats.Deleted = 0
+		m.stats.Filtered = 0
+		m.stats.LastRebuilt = then
+		m.stats.RebuildDuration = time.Now().Sub(then)
+	}
+
+	return err
+}
+
+// Stats returns some statistics that might be helpful to rebuild
+// policies.
+func (m *Matcher) Stats() Stats {
+	m.lock.RLock()
+	s := m.stats // Copies
+	m.lock.RUnlock()
+	return s
+}