-
Notifications
You must be signed in to change notification settings - Fork 290
/
store.go
246 lines (202 loc) · 5.51 KB
/
store.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
package store
import (
"context"
"sync"
"time"
"github.com/davecgh/go-spew/spew"
"gopkg.in/d4l3k/messagediff.v1"
"github.com/windmilleng/tilt/internal/logger"
"github.com/windmilleng/tilt/internal/model"
)
// Allow actions to batch together a bit.
const actionBatchWindow = time.Millisecond
// Read-only store
type RStore interface {
Dispatch(action Action)
RLockState() EngineState
RUnlockState()
}
// A central state store, modeled after the Reactive programming UX pattern.
// Terminology is borrowed liberally from Redux. These docs in particular are helpful:
// https://redux.js.org/introduction/threeprinciples
// https://redux.js.org/basics
type Store struct {
state *EngineState
subscribers *subscriberList
actionQueue *actionQueue
actionCh chan []Action
mu sync.Mutex
stateMu sync.RWMutex
reduce Reducer
logActions bool
// TODO(nick): Define Subscribers and Reducers.
// The actionChan is an intermediate representation to make the transition easier.
}
func NewStore(reducer Reducer, logActions LogActionsFlag) *Store {
return &Store{
state: NewState(),
reduce: reducer,
actionQueue: &actionQueue{},
actionCh: make(chan []Action),
subscribers: &subscriberList{},
logActions: bool(logActions),
}
}
// Returns a Store for testing that saves observed actions and makes them available
// via the return value `getActions`
func NewStoreForTesting() (st *Store, getActions func() []Action) {
var mu sync.Mutex
actions := []Action{}
reducer := Reducer(func(ctx context.Context, s *EngineState, action Action) {
mu.Lock()
defer mu.Unlock()
actions = append(actions, action)
errorAction, isErrorAction := action.(ErrorAction)
if isErrorAction {
s.PermanentError = errorAction.Error
}
})
getActions = func() []Action {
mu.Lock()
defer mu.Unlock()
return append([]Action{}, actions...)
}
return NewStore(reducer, false), getActions
}
func (s *Store) AddSubscriber(ctx context.Context, sub Subscriber) {
s.subscribers.Add(ctx, sub)
}
func (s *Store) RemoveSubscriber(ctx context.Context, sub Subscriber) error {
return s.subscribers.Remove(ctx, sub)
}
// Sends messages to all the subscribers asynchronously.
func (s *Store) NotifySubscribers(ctx context.Context) {
s.subscribers.NotifyAll(ctx, s)
}
// TODO(nick): Clone the state to ensure it's not mutated.
// For now, we use RW locks to simulate the same behavior, but the
// onus is on the caller to RUnlockState.
func (s *Store) RLockState() EngineState {
s.stateMu.RLock()
return *(s.state)
}
func (s *Store) RUnlockState() {
s.stateMu.RUnlock()
}
func (s *Store) LockMutableStateForTesting() *EngineState {
s.stateMu.Lock()
return s.state
}
func (s *Store) UnlockMutableState() {
s.stateMu.Unlock()
}
func (s *Store) Dispatch(action Action) {
s.actionQueue.add(action)
go s.drainActions()
}
func (s *Store) Close() {
close(s.actionCh)
}
func (s *Store) SetUpSubscribersForTesting(ctx context.Context) {
s.subscribers.SetUp(ctx)
}
func (s *Store) Loop(ctx context.Context) error {
s.subscribers.SetUp(ctx)
defer s.subscribers.TeardownAll(context.Background())
for {
select {
case <-ctx.Done():
return ctx.Err()
case actions := <-s.actionCh:
s.stateMu.Lock()
for _, action := range actions {
var oldState EngineState
if s.logActions {
oldState = s.cheapCopyState()
}
s.reduce(ctx, s.state, action)
if s.logActions {
newState := s.cheapCopyState()
go func() {
diff, equal := messagediff.PrettyDiff(oldState, newState)
if !equal {
logger.Get(ctx).Infof("action %T:\n%s\ncaused state change:\n%s\n", action, spew.Sdump(action), diff)
}
}()
}
}
s.stateMu.Unlock()
}
// Subscribers
done, err := s.maybeFinished()
if done {
return err
}
s.NotifySubscribers(ctx)
}
}
func (s *Store) maybeFinished() (bool, error) {
state := s.RLockState()
defer s.RUnlockState()
if state.PermanentError != nil {
return true, state.PermanentError
}
if state.UserExited {
return true, nil
}
if len(state.ManifestTargets) == 0 {
return false, nil
}
finished := !state.WatchFiles &&
state.CompletedBuildCount == state.InitialBuildCount
return finished, nil
}
func (s *Store) drainActions() {
time.Sleep(actionBatchWindow)
// The mutex here ensures that the actions appear on the channel in-order.
// Otherwise, two drains can interleave badly.
s.mu.Lock()
defer s.mu.Unlock()
actions := s.actionQueue.drain()
if len(actions) > 0 {
s.actionCh <- actions
}
}
type Action interface {
Action()
}
type actionQueue struct {
actions []Action
mu sync.Mutex
}
func (q *actionQueue) add(action Action) {
q.mu.Lock()
defer q.mu.Unlock()
q.actions = append(q.actions, action)
}
func (q *actionQueue) drain() []Action {
q.mu.Lock()
defer q.mu.Unlock()
result := append([]Action{}, q.actions...)
q.actions = nil
return result
}
type LogActionsFlag bool
// This does a partial deep copy for the purposes of comparison
// i.e., it ensures fields that will be useful in action logging get copied
// some fields might not be copied and might still point to the same instance as s.state
// and thus might reflect changes that happened as part of the current action or any future action
func (s *Store) cheapCopyState() EngineState {
ret := *s.state
targets := ret.ManifestTargets
ret.ManifestTargets = make(map[model.ManifestName]*ManifestTarget)
for k, v := range targets {
ms := *(v.State)
target := &ManifestTarget{
Manifest: v.Manifest,
State: &ms,
}
ret.ManifestTargets[k] = target
}
return ret
}