Introduce pub-sub on mutable state (uber#2233)

service/history/MockWorkflowExecutionContext.go

@@ -317,3 +317,18 @@ func (_m *mockWorkflowExecutionContext) persistNonFirstWorkflowEvents(_a0 *persi
 
 	return r0, r1
 }
+
+func (_m *mockWorkflowExecutionContext) getWorkflowWatcher() WorkflowWatcher {
+	ret := _m.Called()
+
+	var r0 WorkflowWatcher
+	if rf, ok := ret.Get(0).(func() WorkflowWatcher); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(WorkflowWatcher)
+		}
+	}
+
+	return r0
+}

service/history/workflowExecutionContext.go

@@ -113,6 +113,7 @@ type (
 			baseRunID string,
 			baseRunNextEventID int64,
 		) (retError error)
+		getWorkflowWatcher() WorkflowWatcher
 	}
 )
 
@@ -131,6 +132,7 @@ type (
 		msBuilder       mutableState
 		stats           *persistence.ExecutionStats
 		updateCondition int64
+		watcher         WorkflowWatcher
 	}
 )
 
@@ -167,6 +169,7 @@ func newWorkflowExecutionContext(
 		stats: &persistence.ExecutionStats{
 			HistorySize: 0,
 		},
+		watcher: NewWorkflowWatcher(),
 	}
 }
 
@@ -499,7 +502,12 @@ func (c *workflowExecutionContextImpl) updateWorkflowExecutionWithNew(
 	// TODO remove updateCondition in favor of condition in mutable state
 	c.updateCondition = currentWorkflow.ExecutionInfo.NextEventID
 
+	c.watcher.Publish(&WatcherSnapshot{
+		CloseStatus: c.msBuilder.GetExecutionInfo().CloseStatus,
+	})
+
 	// for any change in the workflow, send a event
+	// TODO: @andrewjdawson2016 remove historyEventNotifier once plumbing for MutableStatePubSub is finished
 	c.engine.NotifyNewHistoryEvent(newHistoryEventNotification(
 		c.domainID,
 		&c.workflowExecution,
@@ -1001,3 +1009,7 @@ func (c *workflowExecutionContextImpl) resetWorkflowExecution(
 	}
 	return nil
 }
+
+func (c *workflowExecutionContextImpl) getWorkflowWatcher() WorkflowWatcher {
+	return c.watcher
+}

service/history/workflowWatcher.go

@@ -0,0 +1,119 @@
+// 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 history
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+type (
+	// WatcherSnapshot contains the union of all workflow state subscribers care about.
+	WatcherSnapshot struct {
+		CloseStatus int
+	}
+
+	// WorkflowWatcher is used to get WatcherSnapshots when workflow updates occur.
+	WorkflowWatcher interface {
+		Publish(*WatcherSnapshot)
+		Subscribe() (int64, <-chan struct{})
+		Unsubscribe(int64)
+		GetLatestSnapshot() *WatcherSnapshot
+	}
+
+	workflowWatcher struct {
+		// An atomic is used to keep track of latest snapshot rather than pushing snapshots on a channel.
+		// If a channel (buffered or unbuffered) is used, then at some point either blocking will occur or a snapshot will be dropped.
+		// Since clients only care about getting the latest, simplify notifying clients
+		// of updates and providing the ability to get the latest snapshot satisfies all current use cases.
+		latestSnapshot atomic.Value
+		lockableSubscribers
+	}
+
+	lockableSubscribers struct {
+		sync.Mutex
+		nextSubscriberID int64
+		subscribers      map[int64]chan struct{}
+	}
+)
+
+func (ls *lockableSubscribers) add() (int64, chan struct{}) {
+	ls.Lock()
+	defer ls.Unlock()
+	id := ls.nextSubscriberID
+	ch := make(chan struct{}, 1)
+	ls.nextSubscriberID = ls.nextSubscriberID + 1
+	ls.subscribers[id] = ch
+	return id, ch
+}
+
+func (ls *lockableSubscribers) remove(id int64) {
+	ls.Lock()
+	defer ls.Unlock()
+	delete(ls.subscribers, id)
+}
+
+func (ls *lockableSubscribers) notifyAll() {
+	ls.Lock()
+	defer ls.Unlock()
+	for _, ch := range ls.subscribers {
+		select {
+		case ch <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// NewWorkflowWatcher returns new WorkflowWatcher.
+// This should only be called from workflowExecutionContext.go.
+func NewWorkflowWatcher() WorkflowWatcher {
+	return &workflowWatcher{
+		lockableSubscribers: lockableSubscribers{
+			subscribers: make(map[int64]chan struct{}),
+		},
+	}
+}
+
+// Publish is used to indicate that the workflow has been updated and subscribers should be notified that there is a new snapshot.
+func (w *workflowWatcher) Publish(update *WatcherSnapshot) {
+	w.latestSnapshot.Store(update)
+	w.notifyAll()
+}
+
+// Subscribe to updates.
+// Every time returned channel is received on it is guaranteed that at least one new snapshot is available.
+func (w *workflowWatcher) Subscribe() (int64, <-chan struct{}) {
+	return w.add()
+}
+
+// Unsubscribe from updates. This must be called when client is finished watching in order to avoid resource leak.
+func (w *workflowWatcher) Unsubscribe(id int64) {
+	w.remove(id)
+}
+
+// GetLatestSnapshot returns the latest WatcherSnapshot.
+func (w *workflowWatcher) GetLatestSnapshot() *WatcherSnapshot {
+	v := w.latestSnapshot.Load()
+	if v == nil {
+		return nil
+	}
+	return v.(*WatcherSnapshot)
+}

service/history/workflowWatcher_test.go

@@ -0,0 +1,104 @@
+// 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 history
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+	"go.uber.org/cadence/testsuite"
+)
+
+type workflowWatcherSuite struct {
+	suite.Suite
+	testsuite.WorkflowTestSuite
+}
+
+func TestWorkflowWatcherSuite(t *testing.T) {
+	suite.Run(t, new(workflowWatcherSuite))
+}
+
+func (s *workflowWatcherSuite) TestWorkflowWatcher_NoSubscribers() {
+	ps := NewWorkflowWatcher()
+	state := s.newRandomWatcherUpdate()
+	ps.Publish(state)
+	s.assertWatcherUpdateEqual(state, ps.GetLatestSnapshot())
+}
+
+func (s *workflowWatcherSuite) TestWorkflowWatcher_NilPublishAndAccess() {
+	ps := NewWorkflowWatcher()
+	s.Nil(ps.GetLatestSnapshot())
+	ps.Publish(nil)
+	s.Nil(ps.GetLatestSnapshot())
+}
+
+func (s *workflowWatcherSuite) TestWorkflowWatcher_ZombieUnsubscribe() {
+	ps := NewWorkflowWatcher()
+	ps.Unsubscribe(-1)
+}
+
+func (s *workflowWatcherSuite) TestWorkflowWatcher_ManySubscribers() {
+	ps := NewWorkflowWatcher()
+	var ids []int64
+	var chans []<-chan struct{}
+	for i := 0; i < 10; i++ {
+		id, ch := ps.Subscribe()
+		ids = append(ids, id)
+		chans = append(chans, ch)
+	}
+	for i := 0; i < 3; i++ {
+		index := rand.Intn(len(ids))
+		ids = append(ids[:index], ids[index+1:]...)
+		chans = append(chans[:index], chans[index+1:]...)
+	}
+	s.assertChanLengthsEqual(0, chans)
+	state1 := s.newRandomWatcherUpdate()
+	ps.Publish(state1)
+	s.assertChanLengthsEqual(1, chans)
+	s.assertWatcherUpdateEqual(state1, ps.GetLatestSnapshot())
+	state2 := s.newRandomWatcherUpdate()
+	ps.Publish(state2)
+	s.assertChanLengthsEqual(1, chans)
+	s.assertWatcherUpdateEqual(state2, ps.GetLatestSnapshot())
+	for _, ch := range chans {
+		<-ch
+	}
+	s.assertChanLengthsEqual(0, chans)
+	s.assertWatcherUpdateEqual(state2, ps.GetLatestSnapshot())
+
+}
+
+func (s *workflowWatcherSuite) newRandomWatcherUpdate() *WatcherSnapshot {
+	return &WatcherSnapshot{
+		CloseStatus: rand.Intn(100),
+	}
+}
+
+func (s *workflowWatcherSuite) assertWatcherUpdateEqual(expected, actual *WatcherSnapshot) {
+	s.Equal(expected.CloseStatus, actual.CloseStatus)
+}
+
+func (s *workflowWatcherSuite) assertChanLengthsEqual(expectedLengths int, chans []<-chan struct{}) {
+	for _, ch := range chans {
+		s.Equal(expectedLengths, len(ch))
+	}
+}