Check for duplicate attributes under read lock on insert.

Changes the behavior of `AttrStore.SetAttrs()` to first verify
that the attributes haven't changed using a read-only lock before
obtaining a write lock to update the attributes.

Since the write lock serializes access, the previous insert time
sufferred lock contention when inserting a lot of duplicate
attributes. With the `RWMutex`, reads can be done in parallel so
multiple requests don't block each other.

Below is a simple benchmark showing the performance at 1, 5, & 10
goroutines:

	BenchmarkAttrStore_Duplicate          100000      137120 ns/op      8428 B/op       76 allocs/op
	BenchmarkAttrStore_Duplicate-5        100000      417459 ns/op      8431 B/op       76 allocs/op
	BenchmarkAttrStore_Duplicate-10       100000      139466 ns/op      8433 B/op       76 allocs/op

	BenchmarkAttrStore_Duplicate        20000000         703 ns/op       368 B/op        5 allocs/op
	BenchmarkAttrStore_Duplicate-5      100000000        213 ns/op       368 B/op        5 allocs/op
	BenchmarkAttrStore_Duplicate-10     100000000        223 ns/op       368 B/op        5 allocs/op

attr.go

@@ -41,7 +41,7 @@ const (
 
 // AttrStore represents a storage layer for attributes.
 type AttrStore struct {
-	mu   sync.Mutex
+	mu   sync.RWMutex
 	path string
 	db   *bolt.DB
 
@@ -92,8 +92,8 @@ func (s *AttrStore) Close() error {
 
 // Attrs returns a set of attributes by ID.
 func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 
 	// Check cache for map.
 	if m = s.attrs[id]; m != nil {
@@ -119,6 +119,19 @@ func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
 
 // SetAttrs sets attribute values for a given ID.
 func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
+	// Ignore empty maps.
+	if len(m) == 0 {
+		return nil
+	}
+
+	// Check if the attributes already exist under a read-only lock.
+	if attr, err := s.Attrs(id); err != nil {
+		return err
+	} else if attr != nil && mapContains(attr, m) {
+		return nil
+	}
+
+	// Obtain write lock.
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -493,3 +506,14 @@ func (cur *blockCursor) next() (key, value []byte) {
 
 	return key, value
 }
+
+// mapContains returns true if all keys & values of subset are in m.
+func mapContains(m, subset map[string]interface{}) bool {
+	for k, v := range subset {
+		value, ok := m[k]
+		if !ok || value != v {
+			return false
+		}
+	}
+	return true
+}

attr_test.go

@@ -18,6 +18,8 @@ import (
 	"io/ioutil"
 	"os"
 	"reflect"
+	"runtime"
+	"sync"
 	"testing"
 
 	"github.com/pilosa/pilosa"
@@ -143,6 +145,38 @@ func NewAttrStore() *AttrStore {
 	return &AttrStore{AttrStore: pilosa.NewAttrStore(f.Name())}
 }
 
+func BenchmarkAttrStore_Duplicate(b *testing.B) {
+	s := MustOpenAttrStore()
+	defer s.Close()
+
+	// Set attributes.
+	const n = 5
+	for i := 0; i < n; i++ {
+		if err := s.SetAttrs(uint64(i), map[string]interface{}{"A": 100, "B": "foo", "C": true, "D": 100.2}); err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	b.ReportAllocs()
+	b.ResetTimer()
+
+	// Update attributes with an existing subset.
+	cpuN := runtime.GOMAXPROCS(0)
+	var wg sync.WaitGroup
+	for i := 0; i < cpuN; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < b.N/cpuN; j++ {
+				if err := s.SetAttrs(uint64(j%n), map[string]interface{}{"A": int64(100), "B": "foo", "D": 100.2}); err != nil {
+					b.Fatal(err)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
+
 // MustOpenAttrStore returns a new, opened attribute store at a temporary path. Panic on error.
 func MustOpenAttrStore() *AttrStore {
 	s := NewAttrStore()