Outputhost replica stream: resume from read-level instead of ack-level (

#186)

* resume replica connection from read-level instead of ack-level

* comments

* fix

* log

* resume replica connection from read-level instead of ack-level

* comments

* fix

* log

* resume replica connection from read-level instead of ack-level

* comments

* fix

* log

* resume replica connection from read-level instead of ack-level

* comments

* fix

* test

* set

* test

* fix lint

* fix lint

* glide

common/set/bench_test.go

@@ -0,0 +1,83 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package set
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+func BenchmarkSetInsert(b *testing.B) {
+
+	for n := 1; n < 1048576; n *= 4 {
+
+		var keys []string
+		for i := 0; i < n; i++ {
+			keys = append(keys, uuid.New().String())
+		}
+
+		for _, set := range []string{"SliceSet", "SortedSet", "MapSet"} {
+
+			s := newTestSet(set, n)
+
+			b.Run(fmt.Sprintf("%s/%d", set, n), func(b *testing.B) {
+
+				b.ReportAllocs()
+
+				for i := 0; i < b.N; i++ {
+					s.Insert(keys[rand.Intn(n)])
+				}
+			})
+		}
+	}
+}
+
+func BenchmarkSetContains(b *testing.B) {
+
+	for n := 1; n < 65536; n *= 2 {
+
+		var keys []string
+		for i := 0; i < n; i++ {
+			keys = append(keys, uuid.New().String())
+		}
+
+		for _, set := range []string{"SliceSet", "SortedSet", "MapSet"} {
+
+			s := newTestSet(set, n)
+
+			for i := 0; i < n/2; i++ {
+				s.Insert(keys[rand.Intn(n)])
+			}
+
+			b.Run(fmt.Sprintf("%s/%d", set, n), func(b *testing.B) {
+
+				b.ReportAllocs()
+
+				for i := 0; i < b.N; i++ {
+					s.Contains(keys[rand.Intn(n)])
+				}
+			})
+		}
+	}
+}

common/set/mapset.go

@@ -0,0 +1,124 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package set
+
+type mapset struct {
+	m map[string]struct{}
+}
+
+// NewMapSet initializes a map-based set
+func NewMapSet(cap int) Set {
+
+	return &mapset{
+		m: make(map[string]struct{}),
+	}
+}
+
+// Clear clears the set
+func (t *mapset) Clear() {
+	t.m = make(map[string]struct{})
+}
+
+// Empty checks if the set is empty
+func (t *mapset) Empty() bool {
+	return len(t.m) == 0
+}
+
+// Count returns the size of the set
+func (t *mapset) Count() int {
+	return len(t.m)
+}
+
+// Insert adds a key to the set
+func (t *mapset) Insert(key string) {
+	t.m[key] = struct{}{}
+}
+
+// Contains checks if the key exists in the set
+func (t *mapset) Contains(key string) bool {
+	_, ok := t.m[key]
+	return ok
+}
+
+// Remove removes the key from the set
+func (t *mapset) Remove(key string) {
+	delete(t.m, key)
+}
+
+// Keys returns the set of keys
+func (t *mapset) Keys() []string {
+
+	m := make([]string, 0, len(t.m))
+
+	for k := range t.m {
+		m = append(m, k)
+	}
+
+	return m
+}
+
+// Equals compares the set against another set
+func (t *mapset) Equals(s Set) bool {
+
+	if s.Count() != t.Count() {
+		return false
+	}
+
+	for _, k := range s.Keys() {
+		if !t.Contains(k) {
+			return false
+		}
+	}
+
+	return true
+}
+
+// Subset checks if this set is a subset of another
+func (t *mapset) Subset(s Set) bool {
+
+	if s.Count() < t.Count() {
+		return false
+	}
+
+	for _, k := range t.Keys() {
+		if !s.Contains(k) {
+			return false
+		}
+	}
+
+	return true
+}
+
+// Superset checks if this set is a superset of another
+func (t *mapset) Superset(s Set) bool {
+
+	if s.Count() > t.Count() {
+		return false
+	}
+
+	for _, k := range s.Keys() {
+		if !t.Contains(k) {
+			return false
+		}
+	}
+
+	return true
+}

common/set/set.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package set
+
+// Set defines the interface for a set
+type Set interface {
+	// Clear clears the set
+	Clear()
+	// Empty checks if the set is empty
+	Empty() bool
+	// Count returns the size of the set
+	Count() int
+	// Insert adds a key to the set
+	Insert(key string)
+	// Contains checks if the key exists in the set
+	Contains(key string) bool
+	// Remove removes the key from the set
+	Remove(key string)
+	// Keys returns the set of keys
+	Keys() []string
+	// Equals compares the set against another set
+	Equals(s Set) bool
+	// Subset checks if this set is a subset of another
+	Subset(s Set) bool
+	// Superset checks if this set is a superset of another
+	Superset(s Set) bool
+}
+
+// use a sliceset if 'cap' is greater than given constant
+const sliceSetMax = 16
+
+// New initializes a new set, picking the appropriate underlying
+// implementation based on the specified expected capacity
+func New(cap int) Set {
+
+	// pick the underlying set implementation based on the specified expected
+	// capacity; the thresholds are determined based on benchmarks.
+	switch {
+	case cap == 0:
+		fallthrough
+
+	case cap > sliceSetMax:
+		return NewMapSet(cap)
+
+	default:
+		return NewSliceSet(cap)
+	}
+}

common/set/set_test.go

@@ -0,0 +1,109 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package set
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func newTestSet(setType string, cap int) (s Set) {
+
+	switch setType {
+	case "SliceSet":
+		s = NewSliceSet(0)
+	case "SortedSet":
+		s = NewSortedSet(0)
+	case "MapSet":
+		s = NewMapSet(0)
+	}
+
+	return s
+}
+
+func TestSet(t *testing.T) {
+
+	testCases := []struct {
+		set0, set1 string // set type
+	}{
+		{"SliceSet", "SliceSet"},
+		{"SliceSet", "SortedSet"},
+		{"SliceSet", "MapSet"},
+		{"SortedSet", "SliceSet"},
+		{"SortedSet", "SortedSet"},
+		{"SortedSet", "MapSet"},
+		{"MapSet", "SliceSet"},
+		{"MapSet", "SortedSet"},
+		{"MapSet", "MapSet"},
+	}
+
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("{%v,%v}", tc.set0, tc.set1), func(t *testing.T) {
+
+			s0 := newTestSet(tc.set0, 0)
+			s1 := newTestSet(tc.set1, 0)
+
+			assert.False(t, s0.Contains("foo"))
+			assert.Equal(t, 0, s0.Count())
+
+			s0.Insert("foo")
+			assert.Equal(t, 1, s0.Count())
+			assert.True(t, s0.Contains("foo"))
+
+			assert.True(t, s1.Subset(s0))
+			assert.False(t, s0.Subset(s1))
+			assert.False(t, s1.Superset(s0))
+			assert.True(t, s0.Superset(s1))
+			assert.False(t, s1.Equals(s0))
+			assert.False(t, s0.Equals(s1))
+
+			s1.Insert("foo")
+			assert.True(t, s1.Subset(s0))
+			assert.True(t, s0.Subset(s1))
+			assert.True(t, s1.Superset(s0))
+			assert.True(t, s0.Superset(s1))
+			assert.True(t, s1.Equals(s0))
+			assert.True(t, s0.Equals(s1))
+
+			s1.Insert("bar")
+			assert.Equal(t, 2, s1.Count())
+			assert.False(t, s1.Subset(s0))
+			assert.True(t, s0.Subset(s1))
+			assert.True(t, s1.Superset(s0))
+			assert.False(t, s0.Superset(s1))
+			assert.False(t, s1.Equals(s0))
+			assert.False(t, s0.Equals(s1))
+
+			s0.Insert("bar")
+			assert.True(t, s1.Subset(s0))
+			assert.True(t, s0.Subset(s1))
+			assert.True(t, s1.Superset(s0))
+			assert.True(t, s0.Superset(s1))
+			assert.True(t, s1.Equals(s0))
+			assert.True(t, s0.Equals(s1))
+
+			s0.Clear()
+			assert.Equal(t, 0, s0.Count())
+		})
+	}
+}