Merge pull request #454 from google/store-expiration-performance

Switch the store RWMutex to a pair of categorical mutices.

Author: jaqx0r

internal/exporter/export.go

@@ -156,8 +156,8 @@ func formatLabels(name string, m map[string]string, ksep, sep, rep string) strin
 type formatter func(string, *metrics.Metric, *metrics.LabelSet, time.Duration) string
 
 func (e *Exporter) writeSocketMetrics(c io.Writer, f formatter, exportTotal *expvar.Int, exportSuccess *expvar.Int) error {
-	e.store.RLock()
-	defer e.store.RUnlock()
+	e.store.SearchMu.RLock()
+	defer e.store.SearchMu.RUnlock()
 
 	for _, ml := range e.store.Metrics {
 		for _, m := range ml {

internal/exporter/prometheus.go

@@ -30,8 +30,8 @@ func (e *Exporter) Describe(c chan<- *prometheus.Desc) {
 
 // Collect implements the prometheus.Collector interface.
 func (e *Exporter) Collect(c chan<- prometheus.Metric) {
-	e.store.RLock()
-	defer e.store.RUnlock()
+	e.store.SearchMu.RLock()
+	defer e.store.SearchMu.RUnlock()
 
 	for _, ml := range e.store.Metrics {
 		lastSource := ""

internal/exporter/varz.go

@@ -21,8 +21,8 @@ const varzFormat = "%s{%s} %s\n"
 
 // HandleVarz exports the metrics in Varz format via HTTP.
 func (e *Exporter) HandleVarz(w http.ResponseWriter, r *http.Request) {
-	e.store.RLock()
-	defer e.store.RUnlock()
+	e.store.SearchMu.RLock()
+	defer e.store.SearchMu.RUnlock()
 
 	w.Header().Add("Content-type", "text/plain")
 

internal/metrics/store.go

@@ -17,8 +17,9 @@ import (
 
 // Store contains Metrics.
 type Store struct {
-	sync.RWMutex
-	Metrics map[string][]*Metric
+	SearchMu sync.RWMutex // read for iterate and insert, write for delete
+	InsertMu sync.Mutex   // locked for insert and delete, unlocked for iterate
+	Metrics  map[string][]*Metric
 }
 
 // NewStore returns a new metric Store.
@@ -30,13 +31,15 @@ func NewStore() (s *Store) {
 
 // Add is used to add one metric to the Store.
 func (s *Store) Add(m *Metric) error {
-	s.Lock()
-	defer s.Unlock()
+	s.InsertMu.Lock()
+	defer s.InsertMu.Unlock()
+	s.SearchMu.RLock()
 	glog.V(1).Infof("Adding a new metric %v", m)
 	dupeIndex := -1
 	if len(s.Metrics[m.Name]) > 0 {
 		t := s.Metrics[m.Name][0].Kind
 		if m.Kind != t {
+			s.SearchMu.RUnlock()
 			return errors.Errorf("Metric %s has different kind %v to existing %v.", m.Name, m.Kind, t)
 		}
 
@@ -77,25 +80,31 @@ func (s *Store) Add(m *Metric) error {
 			}
 		}
 	}
+	s.SearchMu.RUnlock()
 
+	// We're in modify mode now so lock out search
+	s.SearchMu.Lock()
 	s.Metrics[m.Name] = append(s.Metrics[m.Name], m)
 	if dupeIndex >= 0 {
 		s.Metrics[m.Name] = append(s.Metrics[m.Name][0:dupeIndex], s.Metrics[m.Name][dupeIndex+1:]...)
 	}
+	s.SearchMu.Unlock()
 	return nil
 }
 
 // ClearMetrics empties the store of all metrics.
 func (s *Store) ClearMetrics() {
-	s.Lock()
-	defer s.Unlock()
+	s.InsertMu.Lock()
+	defer s.InsertMu.Unlock()
+	s.SearchMu.Lock()
+	defer s.SearchMu.Unlock()
 	s.Metrics = make(map[string][]*Metric)
 }
 
 // MarshalJSON returns a JSON byte string representing the Store.
 func (s *Store) MarshalJSON() (b []byte, err error) {
-	s.Lock()
-	defer s.Unlock()
+	s.SearchMu.RLock()
+	defer s.SearchMu.RUnlock()
 	ms := make([]*Metric, 0)
 	for _, ml := range s.Metrics {
 		ms = append(ms, ml...)
@@ -107,8 +116,8 @@ func (s *Store) MarshalJSON() (b []byte, err error) {
 // for expiry, and removing them if their expiration time has passed.
 func (s *Store) Gc() error {
 	glog.Info("Running Store.Expire()")
-	s.Lock()
-	defer s.Unlock()
+	s.SearchMu.RLock()
+	defer s.SearchMu.RUnlock()
 	now := time.Now()
 	for _, ml := range s.Metrics {
 		for _, m := range ml {
@@ -154,9 +163,9 @@ func (s *Store) StartGcLoop(ctx context.Context, duration time.Duration) {
 // WriteMetrics dumps the current state of the metrics store in JSON format to
 // the io.Writer.
 func (s *Store) WriteMetrics(w io.Writer) error {
-	s.RLock()
+	s.SearchMu.RLock()
 	b, err := json.MarshalIndent(s.Metrics, "", "  ")
-	s.RUnlock()
+	s.SearchMu.RUnlock()
 	if err != nil {
 		return errors.Wrap(err, "failed to marshal metrics into json")
 	}

internal/metrics/store_bench_test.go

@@ -75,10 +75,10 @@ func BenchmarkStore(b *testing.B) {
 				addToStore(b, items, *m, s)
 			},
 			b: func(b *testing.B, items int, m []*Metric, s *Store) {
-				s.RLock()
+				s.SearchMu.RLock()
 				for _ = range s.Metrics {
 				}
-				s.RUnlock()
+				s.SearchMu.RUnlock()
 			},
 		},
 	}

internal/mtail/examples_integration_test.go

@@ -178,7 +178,7 @@ func TestExamplePrograms(t *testing.T) {
 			goldenStore := metrics.NewStore()
 			golden.ReadTestData(g, tc.programfile, goldenStore)
 
-			testutil.ExpectNoDiff(t, goldenStore, store, testutil.IgnoreUnexported(sync.RWMutex{}, datum.String{}))
+			testutil.ExpectNoDiff(t, goldenStore, store, testutil.IgnoreUnexported(sync.RWMutex{}, sync.Mutex{}, datum.String{}))
 		})
 	}
 }
@@ -301,7 +301,7 @@ func TestFilePipeStreamComparison(t *testing.T) {
 			cancel()
 
 			// Ignore the usual field and the datum.Time field as well, as the results will be unstable otherwise.
-			testutil.ExpectNoDiff(t, fileStore, pipeStore, testutil.IgnoreUnexported(sync.RWMutex{}, datum.String{}), testutil.IgnoreFields(datum.BaseDatum{}, "Time"))
+			testutil.ExpectNoDiff(t, fileStore, pipeStore, testutil.IgnoreUnexported(sync.RWMutex{}, sync.Mutex{}, datum.String{}), testutil.IgnoreFields(datum.BaseDatum{}, "Time"))
 		})
 	}
 }

internal/mtail/golden/reader.go

@@ -21,8 +21,8 @@ var varRe = regexp.MustCompile(`^(counter|gauge|timer|text|histogram) ([^ ]+)(?:
 
 // FindMetricOrNil returns a metric in a store, or returns nil if not found.
 func FindMetricOrNil(store *metrics.Store, name string) *metrics.Metric {
-	store.RLock()
-	defer store.RUnlock()
+	store.SearchMu.RLock()
+	defer store.SearchMu.RUnlock()
 	for n, ml := range store.Metrics {
 		if n == name {
 			return ml[0]