1
=== modified file 'mstate/machine.go'
2
--- mstate/machine.go	2012-09-10 15:29:01 +0000
3
+++ mstate/machine.go	2012-09-11 13:29:53 +0000
4
@@ -80,35 +80,29 @@
5
80
}
80
}
6
81
81
7
82
// AgentAlive returns whether the respective remote agent is alive.
82
// AgentAlive returns whether the respective remote agent is alive.
10
83
func (m *Machine) AgentAlive() bool {
83
func (m *Machine) AgentAlive() (bool, error) {
11
84
	return m.st.presencew.Alive(m.globalKey())
84
	return m.st.pwatcher.Alive(m.globalKey())
12
85
}
85
}
13
86
86
14
87
// WaitAgentAlive blocks until the respective agent is alive.
87
// WaitAgentAlive blocks until the respective agent is alive.
16
88
func (m *Machine) WaitAgentAlive(timeout time.Duration) error {
88
func (m *Machine) WaitAgentAlive(timeout time.Duration) (err error) {
17
89
	defer trivial.ErrorContextf(&err, "waiting for agent of machine %v", m)
18
89
	ch := make(chan presence.Change)
90
	ch := make(chan presence.Change)
41
90
	m.st.presencew.Add(m.globalKey(), ch)
91
	m.st.pwatcher.Watch(m.globalKey(), ch)
42
91
	defer m.st.presencew.Remove(m.globalKey(), ch)
92
	defer m.st.pwatcher.Unwatch(m.globalKey(), ch)
43
92
	// Initial check.
93
	for i := 0; i < 2; i++ {
44
93
	select {
94
		select {
45
94
	case change := <-ch:
95
		case change := <-ch:
46
95
		if change.Alive {
96
			if change.Alive {
47
96
			return nil
97
				return nil
48
97
		}
98
			}
49
98
	case <-time.After(timeout):
99
		case <-time.After(timeout):
50
99
		return fmt.Errorf("waiting for agent of machine %v: still not alive after timeout", m)
100
			return fmt.Errorf("still not alive after timeout")
51
100
	}
101
		case <-m.st.pwatcher.Dying():
52
101
	// Hasn't been alive, so now wait for change.
102
			return m.st.pwatcher.Err()
53
102
	select {
103
		}
54
103
	case change := <-ch:
104
	}
55
104
		if change.Alive {
105
	panic(fmt.Sprintf("presence reported dead status twice in a row for machine %v", m))
34
105
			return nil
35
106
		}
36
107
		panic(fmt.Sprintf("presence reported dead status twice in a row for machine %v", m))
37
108
	case <-time.After(timeout):
38
109
		return fmt.Errorf("waiting for agent of machine %v: still not alive after timeout", m)
39
110
	}
40
111
	panic("unreachable")
56
112
}
106
}
57
113
107
58
114
// SetAgentAlive signals that the agent for machine m is alive. 
108
// SetAgentAlive signals that the agent for machine m is alive. 
59
115
109
60
=== modified file 'mstate/machine_test.go'
61
--- mstate/machine_test.go	2012-09-10 15:29:01 +0000
62
+++ mstate/machine_test.go	2012-09-11 13:29:53 +0000
63
@@ -23,7 +23,8 @@
64
23
}
23
}
65
24
24
66
25
func (s *MachineSuite) TestMachineSetAgentAlive(c *C) {
25
func (s *MachineSuite) TestMachineSetAgentAlive(c *C) {
68
26
	alive := s.machine.AgentAlive()
26
	alive, err := s.machine.AgentAlive()
69
27
	c.Assert(err, IsNil)
70
27
	c.Assert(alive, Equals, false)
28
	c.Assert(alive, Equals, false)
71
28
29
72
29
	pinger, err := s.machine.SetAgentAlive()
30
	pinger, err := s.machine.SetAgentAlive()
73
@@ -31,37 +32,39 @@
74
31
	c.Assert(pinger, Not(IsNil))
32
	c.Assert(pinger, Not(IsNil))
75
32
	defer pinger.Stop()
33
	defer pinger.Stop()
76
33
34
79
34
	s.State.ForcePresenceRefresh()
35
	s.State.Sync()
80
35
	alive = s.machine.AgentAlive()
36
	alive, err = s.machine.AgentAlive()
81
37
	c.Assert(err, IsNil)
82
36
	c.Assert(alive, Equals, true)
38
	c.Assert(alive, Equals, true)
83
37
}
39
}
84
38
40
85
39
func (s *MachineSuite) TestMachineWaitAgentAlive(c *C) {
41
func (s *MachineSuite) TestMachineWaitAgentAlive(c *C) {
86
40
	// test -gocheck.f TestMachineWaitAgentAlive
42
	// test -gocheck.f TestMachineWaitAgentAlive
87
41
	timeout := 5 * time.Second
43
	timeout := 5 * time.Second
89
42
	alive := s.machine.AgentAlive()
44
	alive, err := s.machine.AgentAlive()
90
45
	c.Assert(err, IsNil)
91
43
	c.Assert(alive, Equals, false)
46
	c.Assert(alive, Equals, false)
92
44
47
95
45
	s.State.ForcePresenceRefresh()
48
	s.State.StartSync()
96
46
	err := s.machine.WaitAgentAlive(timeout)
49
	err = s.machine.WaitAgentAlive(timeout)
97
47
	c.Assert(err, ErrorMatches, `waiting for agent of machine 0: still not alive after timeout`)
50
	c.Assert(err, ErrorMatches, `waiting for agent of machine 0: still not alive after timeout`)
98
48
51
99
49
	pinger, err := s.machine.SetAgentAlive()
52
	pinger, err := s.machine.SetAgentAlive()
100
50
	c.Assert(err, IsNil)
53
	c.Assert(err, IsNil)
101
51
54
103
52
	s.State.ForcePresenceRefresh()
55
	s.State.StartSync()
104
53
	err = s.machine.WaitAgentAlive(timeout)
56
	err = s.machine.WaitAgentAlive(timeout)
105
54
	c.Assert(err, IsNil)
57
	c.Assert(err, IsNil)
106
55
58
108
56
	alive = s.machine.AgentAlive()
59
	alive, err = s.machine.AgentAlive()
109
57
	c.Assert(err, IsNil)
60
	c.Assert(err, IsNil)
110
58
	c.Assert(alive, Equals, true)
61
	c.Assert(alive, Equals, true)
111
59
62
112
60
	err = pinger.Kill()
63
	err = pinger.Kill()
113
61
	c.Assert(err, IsNil)
64
	c.Assert(err, IsNil)
114
62
65
117
63
	s.State.ForcePresenceRefresh()
66
	s.State.Sync()
118
64
	alive = s.machine.AgentAlive()
67
	alive, err = s.machine.AgentAlive()
119
65
	c.Assert(err, IsNil)
68
	c.Assert(err, IsNil)
120
66
	c.Assert(alive, Equals, false)
69
	c.Assert(alive, Equals, false)
121
67
}
70
}
122
68
71
123
=== modified file 'mstate/open.go'
124
--- mstate/open.go	2012-09-11 08:42:28 +0000
125
+++ mstate/open.go	2012-09-11 13:29:53 +0000
126
@@ -50,7 +50,7 @@
127
50
	st.runner = txn.NewRunner(db.C("txns"))
50
	st.runner = txn.NewRunner(db.C("txns"))
128
51
	st.runner.ChangeLog(db.C("txns.log"))
51
	st.runner.ChangeLog(db.C("txns.log"))
129
52
	st.watcher = watcher.New(db.C("txns.log"))
52
	st.watcher = watcher.New(db.C("txns.log"))
131
53
	st.presencew = presence.NewWatcher(pdb.C("presence"))
53
	st.pwatcher = presence.NewWatcher(pdb.C("presence"))
132
54
	for _, index := range indexes {
54
	for _, index := range indexes {
133
55
		err = st.relations.EnsureIndex(index)
55
		err = st.relations.EnsureIndex(index)
134
56
		if err != nil {
56
		if err != nil {
135
@@ -61,8 +61,8 @@
136
61
}
61
}
137
62
62
138
63
func (st *State) Close() error {
63
func (st *State) Close() error {
141
64
	err1 := st.presencew.Stop()
64
	err1 := st.watcher.Stop()
142
65
	err2 := st.watcher.Stop()
65
	err2 := st.pwatcher.Stop()
143
66
	st.db.Session.Close()
66
	st.db.Session.Close()
144
67
	for _, err := range []error{err1, err2} {
67
	for _, err := range []error{err1, err2} {
145
68
		if err != nil {
68
		if err != nil {
146
69
69
147
=== modified file 'mstate/presence/presence.go'
148
--- mstate/presence/presence.go	2012-09-06 22:12:07 +0000
149
+++ mstate/presence/presence.go	2012-09-11 13:29:53 +0000
150
@@ -58,7 +58,7 @@
151
58
	beingKey map[int64]string
58
	beingKey map[int64]string
152
59
	beingSeq map[string]int64
59
	beingSeq map[string]int64
153
60
60
155
61
	// watches has the per-key observer channels from Add/Remove.
61
	// watches has the per-key observer channels from Watch/Unwatch.
156
62
	watches map[string][]chan<- Change
62
	watches map[string][]chan<- Change
157
63
63
158
64
	// pending contains all the events to be dispatched to the watcher
64
	// pending contains all the events to be dispatched to the watcher
159
@@ -70,13 +70,12 @@
160
70
	// the the gorotuine loop.
70
	// the the gorotuine loop.
161
71
	request chan interface{}
71
	request chan interface{}
162
72
72
166
73
	// refreshed contains pending ForceRefresh done channels
73
	// syncDone contains pending done channels from sync requests.
167
74
	// that are waiting for the completion notice.
74
	syncDone []chan bool
165
75
	refreshed []chan bool
168
76
75
170
77
	// next will dispatch when it's time to refresh the database
76
	// next will dispatch when it's time to sync the database
171
78
	// knowledge. It's maintained here so that ForceRefresh
77
	// knowledge. It's maintained here so that ForceRefresh
173
79
	// can manipulate it to force a refresh sooner.
78
	// can manipulate it to force a sync sooner.
174
80
	next <-chan time.Time
79
	next <-chan time.Time
175
81
}
80
}
176
82
81
177
@@ -116,17 +115,31 @@
178
116
	return w.tomb.Wait()
115
	return w.tomb.Wait()
179
117
}
116
}
180
118
117
192
119
type reqAdd struct {
118
// Dying returns a channel that is closed when the watcher is stopping
193
120
	key string
119
// due to an error or because Stop was called explicitly.
194
121
	ch  chan<- Change
120
func (w *Watcher) Dying() <-chan struct{} {
195
122
}
121
	return w.tomb.Dying()
196
123
122
}
197
124
type reqRemove struct {
123
198
125
	key string
124
// Err returns the error with which the watcher stopped.
199
126
	ch  chan<- Change
125
// It returns nil if the watcher stopped cleanly, tomb.ErrStillAlive
200
127
}
126
// if the watcher is still running properly, or the respective error
201
128
127
// if the watcher is terminating or has terminated with an error.
202
129
type reqRefresh struct {
128
func (w *Watcher) Err() error {
203
129
	return w.tomb.Err()
204
130
}
205
131
206
132
type reqWatch struct {
207
133
	key string
208
134
	ch  chan<- Change
209
135
}
210
136
211
137
type reqUnwatch struct {
212
138
	key string
213
139
	ch  chan<- Change
214
140
}
215
141
216
142
type reqSync struct {
217
130
	done chan bool
143
	done chan bool
218
131
}
144
}
219
132
145
220
@@ -135,55 +148,55 @@
221
135
	result chan bool
148
	result chan bool
222
136
}
149
}
223
137
150
225
138
// Add includes key into w for liveness monitoring. An event will
151
func (w *Watcher) sendReq(req interface{}) {
226
152
	select {
227
153
	case w.request <- req:
228
154
	case <-w.tomb.Dying():
229
155
	}
230
156
}
231
157
232
158
// Watch starts watching the liveness of key. An event will
233
139
// be sent onto ch to report the initial status for the key, and
159
// be sent onto ch to report the initial status for the key, and
234
140
// from then on a new event will be sent whenever a change is
160
// from then on a new event will be sent whenever a change is
235
141
// detected. Change values sent to the channel must be consumed,
161
// detected. Change values sent to the channel must be consumed,
236
142
// or the whole watcher will blocked.
162
// or the whole watcher will blocked.
257
143
func (w *Watcher) Add(key string, ch chan<- Change) {
163
func (w *Watcher) Watch(key string, ch chan<- Change) {
258
144
	select {
164
	w.sendReq(reqWatch{key, ch})
259
145
	case w.request <- reqAdd{key, ch}:
165
}
260
146
	case <-w.tomb.Dying():
166
261
147
	}
167
// Unwatch stops watching the liveness of key via ch.
262
148
}
168
func (w *Watcher) Unwatch(key string, ch chan<- Change) {
263
149
169
	w.sendReq(reqUnwatch{key, ch})
264
150
// Remove removes key and ch from liveness monitoring.
170
}
265
151
func (w *Watcher) Remove(key string, ch chan<- Change) {
171
266
152
	select {
172
// StartSync forces the watcher to load new events from the database.
267
153
	case w.request <- reqRemove{key, ch}:
173
func (w *Watcher) StartSync() {
268
154
	case <-w.tomb.Dying():
174
	w.sendReq(reqSync{nil})
269
155
	}
175
}
270
156
}
176
271
157
177
// Sync forces the watcher to load new events from the database and blocks
272
158
// ForceRefresh forces a synchronous refresh of the watcher knowledge.
178
// until all events have been dispatched.
273
159
// It blocks until the database state has been loaded and the events
179
func (w *Watcher) Sync() {
254
160
// have been prepared, but unblocks before changes are sent onto the
255
161
// registered channels.
256
162
func (w *Watcher) ForceRefresh() {
274
163
	done := make(chan bool)
180
	done := make(chan bool)
279
164
	select {
181
	w.sendReq(reqSync{done})
276
165
	case w.request <- reqRefresh{done}:
277
166
	case <-w.tomb.Dying():
278
167
	}
280
168
	select {
182
	select {
281
169
	case <-done:
183
	case <-done:
282
170
	case <-w.tomb.Dying():
184
	case <-w.tomb.Dying():
283
171
	}
185
	}
284
172
}
186
}
285
173
187
288
174
// Alive returns whether the key is currently considered alive by w.
188
// Alive returns whether the key is currently considered alive by w,
289
175
func (w *Watcher) Alive(key string) bool {
189
// or an error in case the watcher is dying.
290
190
func (w *Watcher) Alive(key string) (bool, error) {
291
176
	result := make(chan bool, 1)
191
	result := make(chan bool, 1)
296
177
	select {
192
	w.sendReq(reqAlive{key, result})
293
178
	case w.request <- reqAlive{key, result}:
294
179
	case <-w.tomb.Dying():
295
180
	}
297
181
	var alive bool
193
	var alive bool
298
182
	select {
194
	select {
299
183
	case alive = <-result:
195
	case alive = <-result:
300
184
	case <-w.tomb.Dying():
196
	case <-w.tomb.Dying():
301
197
		return false, fmt.Errorf("cannot check liveness: watcher is dying")
302
185
	}
198
	}
304
186
	return alive
199
	return alive, nil
305
187
}
200
}
306
188
201
307
189
// period is the length of each time slot in seconds.
202
// period is the length of each time slot in seconds.
308
@@ -205,18 +218,19 @@
309
205
			return tomb.ErrDying
218
			return tomb.ErrDying
310
206
		case <-w.next:
219
		case <-w.next:
311
207
			w.next = time.After(time.Duration(period) * time.Second)
220
			w.next = time.After(time.Duration(period) * time.Second)
315
208
			refreshed := w.refreshed
221
			syncDone := w.syncDone
316
209
			w.refreshed = nil
222
			w.syncDone = nil
317
210
			if err := w.refresh(); err != nil {
223
			if err := w.sync(); err != nil {
318
211
				return err
224
				return err
319
212
			}
225
			}
321
213
			for _, done := range refreshed {
226
			w.flush()
322
227
			for _, done := range syncDone {
323
214
				close(done)
228
				close(done)
324
215
			}
229
			}
325
216
		case req := <-w.request:
230
		case req := <-w.request:
326
217
			w.handle(req)
231
			w.handle(req)
327
232
			w.flush()
328
218
		}
233
		}
329
219
		w.flush()
330
220
	}
234
	}
331
221
	return nil
235
	return nil
332
222
}
236
}
333
@@ -224,20 +238,18 @@
334
224
// flush sends all pending events to their respective channels.
238
// flush sends all pending events to their respective channels.
335
225
func (w *Watcher) flush() {
239
func (w *Watcher) flush() {
336
226
	// w.pending may get new requests as we handle other requests.
240
	// w.pending may get new requests as we handle other requests.
339
227
	i := 0
241
	for i := 0; i < len(w.pending); i++ {
338
228
	for i < len(w.pending) {
340
229
		e := &w.pending[i]
242
		e := &w.pending[i]
352
230
		if e.ch == nil {
243
		for e.ch != nil {
353
231
			i++ // Removed meanwhile.
244
			select {
354
232
			continue
245
			case <-w.tomb.Dying():
355
233
		}
246
				return
356
234
		select {
247
			case req := <-w.request:
357
235
		case <-w.tomb.Dying():
248
				w.handle(req)
358
236
			return
249
				continue
359
237
		case req := <-w.request:
250
			case e.ch <- Change{e.key, e.alive}:
360
238
			w.handle(req)
251
			}
361
239
		case e.ch <- Change{e.key, e.alive}:
252
			break
351
240
			i++
362
241
		}
253
		}
363
242
	}
254
	}
364
243
	w.pending = w.pending[:0]
255
	w.pending = w.pending[:0]
365
@@ -248,10 +260,12 @@
366
248
func (w *Watcher) handle(req interface{}) {
260
func (w *Watcher) handle(req interface{}) {
367
249
	log.Debugf("presence: got request: %#v", req)
261
	log.Debugf("presence: got request: %#v", req)
368
250
	switch r := req.(type) {
262
	switch r := req.(type) {
370
251
	case reqRefresh:
263
	case reqSync:
371
252
		w.next = time.After(0)
264
		w.next = time.After(0)
374
253
		w.refreshed = append(w.refreshed, r.done)
265
		if r.done != nil {
375
254
	case reqAdd:
266
			w.syncDone = append(w.syncDone, r.done)
376
267
		}
377
268
	case reqWatch:
378
255
		for _, ch := range w.watches[r.key] {
269
		for _, ch := range w.watches[r.key] {
379
256
			if ch == r.ch {
270
			if ch == r.ch {
380
257
				panic("adding channel twice for same key")
271
				panic("adding channel twice for same key")
381
@@ -260,7 +274,7 @@
382
260
		w.watches[r.key] = append(w.watches[r.key], r.ch)
274
		w.watches[r.key] = append(w.watches[r.key], r.ch)
383
261
		_, alive := w.beingSeq[r.key]
275
		_, alive := w.beingSeq[r.key]
384
262
		w.pending = append(w.pending, event{r.ch, r.key, alive})
276
		w.pending = append(w.pending, event{r.ch, r.key, alive})
386
263
	case reqRemove:
277
	case reqUnwatch:
387
264
		watches := w.watches[r.key]
278
		watches := w.watches[r.key]
388
265
		for i, ch := range watches {
279
		for i, ch := range watches {
389
266
			if ch == r.ch {
280
			if ch == r.ch {
390
@@ -294,11 +308,11 @@
391
294
	Dead  map[string]int64 ",omitempty"
308
	Dead  map[string]int64 ",omitempty"
392
295
}
309
}
393
296
310
395
297
// refresh updates the watcher knowledge from the database, and
311
// sync updates the watcher knowledge from the database, and
396
298
// queues events to observing channels. It fetches the last two time
312
// queues events to observing channels. It fetches the last two time
397
299
// slots and compares the union of both to the in-memory state.
313
// slots and compares the union of both to the in-memory state.
400
300
func (w *Watcher) refresh() error {
314
func (w *Watcher) sync() error {
401
301
	log.Debugf("presence: refreshing watcher knowledge from database...")
315
	log.Debugf("presence: synchronizing watcher knowledge with database...")
402
302
	slot := timeSlot(time.Now(), w.delta)
316
	slot := timeSlot(time.Now(), w.delta)
403
303
	var ping []pingInfo
317
	var ping []pingInfo
404
304
	err := w.pings.Find(bson.D{{"$or", []pingInfo{{Slot: slot}, {Slot: slot - period}}}}).All(&ping)
318
	err := w.pings.Find(bson.D{{"$or", []pingInfo{{Slot: slot}, {Slot: slot - period}}}}).All(&ping)
405
305
319
406
=== modified file 'mstate/presence/presence_test.go'
407
--- mstate/presence/presence_test.go	2012-09-06 22:12:07 +0000
408
+++ mstate/presence/presence_test.go	2012-09-11 13:29:53 +0000
409
@@ -5,6 +5,7 @@
410
5
	. "launchpad.net/gocheck"
5
	. "launchpad.net/gocheck"
411
6
	"launchpad.net/juju-core/mstate/presence"
6
	"launchpad.net/juju-core/mstate/presence"
412
7
	"launchpad.net/juju-core/testing"
7
	"launchpad.net/juju-core/testing"
413
8
	"launchpad.net/tomb"
414
8
	"strconv"
9
	"strconv"
415
9
	stdtesting "testing"
10
	stdtesting "testing"
416
10
	"time"
11
	"time"
417
@@ -71,6 +72,40 @@
418
71
	}
72
	}
419
72
}
73
}
420
73
74
421
75
func assertAlive(c *C, w *presence.Watcher, key string, alive bool) {
422
76
	alive, err := w.Alive("a")
423
77
	c.Assert(err, IsNil)
424
78
	c.Assert(alive, Equals, alive)
425
79
}
426
80
427
81
func (s *PresenceSuite) TestErrAndDying(c *C) {
428
82
	w := presence.NewWatcher(s.presence)
429
83
	defer w.Stop()
430
84
431
85
	c.Assert(w.Err(), Equals, tomb.ErrStillAlive)
432
86
	select {
433
87
	case <-w.Dying():
434
88
		c.Fatalf("Dying channel fired unexpectedly")
435
89
	default:
436
90
	}
437
91
	c.Assert(w.Stop(), IsNil)
438
92
	c.Assert(w.Err(), IsNil)
439
93
	select {
440
94
	case <-w.Dying():
441
95
	default:
442
96
		c.Fatalf("Dying channel should have fired")
443
97
	}
444
98
}
445
99
446
100
func (s *PresenceSuite) TestAliveError(c *C) {
447
101
	w := presence.NewWatcher(s.presence)
448
102
	c.Assert(w.Stop(), IsNil)
449
103
450
104
	alive, err := w.Alive("a")
451
105
	c.Assert(err, ErrorMatches, ".*: watcher is dying")
452
106
	c.Assert(alive, Equals, false)
453
107
}
454
108
455
74
func (s *PresenceSuite) TestWorkflow(c *C) {
109
func (s *PresenceSuite) TestWorkflow(c *C) {
456
75
	w := presence.NewWatcher(s.presence)
110
	w := presence.NewWatcher(s.presence)
457
76
	pa := presence.NewPinger(s.presence, "a")
111
	pa := presence.NewPinger(s.presence, "a")
458
@@ -79,61 +114,61 @@
459
79
	defer pa.Stop()
114
	defer pa.Stop()
460
80
	defer pb.Stop()
115
	defer pb.Stop()
461
81
116
464
82
	c.Assert(w.Alive("a"), Equals, false)
117
	assertAlive(c, w, "a", false)
465
83
	c.Assert(w.Alive("b"), Equals, false)
118
	assertAlive(c, w, "b", false)
466
84
119
467
85
	// Buffer one entry to avoid blocking the watcher here.
120
	// Buffer one entry to avoid blocking the watcher here.
468
86
	cha := make(chan presence.Change, 1)
121
	cha := make(chan presence.Change, 1)
469
87
	chb := make(chan presence.Change, 1)
122
	chb := make(chan presence.Change, 1)
472
88
	w.Add("a", cha)
123
	w.Watch("a", cha)
473
89
	w.Add("b", chb)
124
	w.Watch("b", chb)
474
90
125
475
91
	// Initial events with current status.
126
	// Initial events with current status.
476
92
	assertChange(c, cha, presence.Change{"a", false})
127
	assertChange(c, cha, presence.Change{"a", false})
477
93
	assertChange(c, chb, presence.Change{"b", false})
128
	assertChange(c, chb, presence.Change{"b", false})
478
94
129
480
95
	w.ForceRefresh()
130
	w.StartSync()
481
96
	assertNoChange(c, cha)
131
	assertNoChange(c, cha)
482
97
	assertNoChange(c, chb)
132
	assertNoChange(c, chb)
483
98
133
484
99
	c.Assert(pa.Start(), IsNil)
134
	c.Assert(pa.Start(), IsNil)
485
100
135
487
101
	w.ForceRefresh()
136
	w.StartSync()
488
102
	assertChange(c, cha, presence.Change{"a", true})
137
	assertChange(c, cha, presence.Change{"a", true})
489
103
	assertNoChange(c, cha)
138
	assertNoChange(c, cha)
490
104
	assertNoChange(c, chb)
139
	assertNoChange(c, chb)
491
105
140
494
106
	c.Assert(w.Alive("a"), Equals, true)
141
	assertAlive(c, w, "a", true)
495
107
	c.Assert(w.Alive("b"), Equals, false)
142
	assertAlive(c, w, "b", false)
496
108
143
497
109
	// Changes while the channel is out are not observed.
144
	// Changes while the channel is out are not observed.
499
110
	w.Remove("a", cha)
145
	w.Unwatch("a", cha)
500
111
	assertNoChange(c, cha)
146
	assertNoChange(c, cha)
501
112
	pa.Kill()
147
	pa.Kill()
503
113
	w.ForceRefresh()
148
	w.Sync()
504
114
	pa = presence.NewPinger(s.presence, "a")
149
	pa = presence.NewPinger(s.presence, "a")
505
115
	pa.Start()
150
	pa.Start()
507
116
	w.ForceRefresh()
151
	w.StartSync()
508
117
	assertNoChange(c, cha)
152
	assertNoChange(c, cha)
509
118
153
510
119
	// We can still query it manually, though.
154
	// We can still query it manually, though.
513
120
	c.Assert(w.Alive("a"), Equals, true)
155
	assertAlive(c, w, "a", true)
514
121
	c.Assert(w.Alive("b"), Equals, false)
156
	assertAlive(c, w, "b", false)
515
122
157
516
123
	// Initial positive event. No refresh needed.
158
	// Initial positive event. No refresh needed.
518
124
	w.Add("a", cha)
159
	w.Watch("a", cha)
519
125
	assertChange(c, cha, presence.Change{"a", true})
160
	assertChange(c, cha, presence.Change{"a", true})
520
126
161
521
127
	c.Assert(pb.Start(), IsNil)
162
	c.Assert(pb.Start(), IsNil)
522
128
163
524
129
	w.ForceRefresh()
164
	w.StartSync()
525
130
	assertChange(c, chb, presence.Change{"b", true})
165
	assertChange(c, chb, presence.Change{"b", true})
526
131
	assertNoChange(c, cha)
166
	assertNoChange(c, cha)
527
132
	assertNoChange(c, chb)
167
	assertNoChange(c, chb)
528
133
168
529
134
	c.Assert(pa.Stop(), IsNil)
169
	c.Assert(pa.Stop(), IsNil)
530
135
170
532
136
	w.ForceRefresh()
171
	w.StartSync()
533
137
	assertNoChange(c, cha)
172
	assertNoChange(c, cha)
534
138
	assertNoChange(c, chb)
173
	assertNoChange(c, chb)
535
139
174
536
@@ -141,7 +176,7 @@
537
141
	c.Assert(pa.Kill(), IsNil)
176
	c.Assert(pa.Kill(), IsNil)
538
142
	c.Assert(pb.Kill(), IsNil)
177
	c.Assert(pb.Kill(), IsNil)
539
143
178
541
144
	w.ForceRefresh()
179
	w.StartSync()
542
145
	assertChange(c, cha, presence.Change{"a", false})
180
	assertChange(c, cha, presence.Change{"a", false})
543
146
	assertChange(c, chb, presence.Change{"b", false})
181
	assertChange(c, chb, presence.Change{"b", false})
544
147
182
545
@@ -172,11 +207,11 @@
546
172
	c.Logf("Checking who's still alive...")
207
	c.Logf("Checking who's still alive...")
547
173
	w := presence.NewWatcher(s.presence)
208
	w := presence.NewWatcher(s.presence)
548
174
	defer w.Stop()
209
	defer w.Stop()
550
175
	w.ForceRefresh()
210
	w.Sync()
551
176
	ch := make(chan presence.Change)
211
	ch := make(chan presence.Change)
552
177
	for i := 0; i < N; i++ {
212
	for i := 0; i < N; i++ {
553
178
		k := strconv.Itoa(i)
213
		k := strconv.Itoa(i)
555
179
		w.Add(k, ch)
214
		w.Watch(k, ch)
556
180
		if i%2 == 0 {
215
		if i%2 == 0 {
557
181
			assertChange(c, ch, presence.Change{k, true})
216
			assertChange(c, ch, presence.Change{k, true})
558
182
		} else {
217
		} else {
559
@@ -192,26 +227,26 @@
560
192
	defer p.Stop()
227
	defer p.Stop()
561
193
228
562
194
	ch := make(chan presence.Change)
229
	ch := make(chan presence.Change)
564
195
	w.Add("a", ch)
230
	w.Watch("a", ch)
565
196
	assertChange(c, ch, presence.Change{"a", false})
231
	assertChange(c, ch, presence.Change{"a", false})
566
197
232
567
198
	c.Assert(p.Start(), IsNil)
233
	c.Assert(p.Start(), IsNil)
569
199
	w.ForceRefresh()
234
	w.StartSync()
570
200
	assertChange(c, ch, presence.Change{"a", true})
235
	assertChange(c, ch, presence.Change{"a", true})
571
201
236
572
202
	// Still alive in previous slot.
237
	// Still alive in previous slot.
573
203
	presence.FakeTimeSlot(1)
238
	presence.FakeTimeSlot(1)
575
204
	w.ForceRefresh()
239
	w.StartSync()
576
205
	assertNoChange(c, ch)
240
	assertNoChange(c, ch)
577
206
241
578
207
	// Two last slots are empty.
242
	// Two last slots are empty.
579
208
	presence.FakeTimeSlot(2)
243
	presence.FakeTimeSlot(2)
581
209
	w.ForceRefresh()
244
	w.StartSync()
582
210
	assertChange(c, ch, presence.Change{"a", false})
245
	assertChange(c, ch, presence.Change{"a", false})
583
211
246
584
212
	// Already dead so killing isn't noticed.
247
	// Already dead so killing isn't noticed.
585
213
	p.Kill()
248
	p.Kill()
587
214
	w.ForceRefresh()
249
	w.StartSync()
588
215
	assertNoChange(c, ch)
250
	assertNoChange(c, ch)
589
216
}
251
}
590
217
252
591
@@ -225,7 +260,7 @@
592
225
	defer p.Stop()
260
	defer p.Stop()
593
226
261
594
227
	ch := make(chan presence.Change)
262
	ch := make(chan presence.Change)
596
228
	w.Add("a", ch)
263
	w.Watch("a", ch)
597
229
	assertChange(c, ch, presence.Change{"a", false})
264
	assertChange(c, ch, presence.Change{"a", false})
598
230
265
599
231
	// A single ping.
266
	// A single ping.
600
@@ -237,18 +272,18 @@
601
237
	assertChange(c, ch, presence.Change{"a", true})
272
	assertChange(c, ch, presence.Change{"a", true})
602
238
}
273
}
603
239
274
605
240
func (s *PresenceSuite) TestAddRemoveOnQueue(c *C) {
275
func (s *PresenceSuite) TestWatchUnwatchOnQueue(c *C) {
606
241
	w := presence.NewWatcher(s.presence)
276
	w := presence.NewWatcher(s.presence)
607
242
	ch := make(chan presence.Change)
277
	ch := make(chan presence.Change)
608
243
	for i := 0; i < 100; i++ {
278
	for i := 0; i < 100; i++ {
609
244
		key := strconv.Itoa(i)
279
		key := strconv.Itoa(i)
610
245
		c.Logf("Adding %q", key)
280
		c.Logf("Adding %q", key)
612
246
		w.Add(key, ch)
281
		w.Watch(key, ch)
613
247
	}
282
	}
614
248
	for i := 1; i < 100; i += 2 {
283
	for i := 1; i < 100; i += 2 {
615
249
		key := strconv.Itoa(i)
284
		key := strconv.Itoa(i)
616
250
		c.Logf("Removing %q", key)
285
		c.Logf("Removing %q", key)
618
251
		w.Remove(key, ch)
286
		w.Unwatch(key, ch)
619
252
	}
287
	}
620
253
	alive := make(map[string]bool)
288
	alive := make(map[string]bool)
621
254
	for i := 0; i < 50; i++ {
289
	for i := 0; i < 50; i++ {
622
@@ -288,10 +323,10 @@
623
288
		stop := false
323
		stop := false
624
289
		for !stop {
324
		for !stop {
625
290
			w := presence.NewWatcher(s.presence)
325
			w := presence.NewWatcher(s.presence)
628
291
			w.ForceRefresh()
326
			w.Sync()
629
292
			alive := w.Alive("a")
327
			alive, err := w.Alive("a")
630
293
			c.Check(w.Stop(), IsNil)
328
			c.Check(w.Stop(), IsNil)
632
294
			if !c.Check(alive, Equals, true) {
329
			if !c.Check(err, IsNil) || !c.Check(alive, Equals, true) {
633
295
				break
330
				break
634
296
			}
331
			}
635
297
			select {
332
			select {
636
@@ -325,22 +360,87 @@
637
325
	// Start p1 and let it go on.
360
	// Start p1 and let it go on.
638
326
	c.Assert(p1.Start(), IsNil)
361
	c.Assert(p1.Start(), IsNil)
639
327
362
642
328
	w.ForceRefresh()
363
	w.Sync()
643
329
	c.Assert(w.Alive("a"), Equals, true)
364
	assertAlive(c, w, "a", true)
644
330
365
645
331
	// Start and kill p2, which will temporarily
366
	// Start and kill p2, which will temporarily
646
332
	// invalidate p1 and set the key as dead.
367
	// invalidate p1 and set the key as dead.
647
333
	c.Assert(p2.Start(), IsNil)
368
	c.Assert(p2.Start(), IsNil)
648
334
	c.Assert(p2.Kill(), IsNil)
369
	c.Assert(p2.Kill(), IsNil)
649
335
370
652
336
	w.ForceRefresh()
371
	w.Sync()
653
337
	c.Assert(w.Alive("a"), Equals, false)
372
	assertAlive(c, w, "a", false)
654
338
373
655
339
	// Wait for two periods, and check again. Since
374
	// Wait for two periods, and check again. Since
656
340
	// p1 is still alive, p2's death will expire and
375
	// p1 is still alive, p2's death will expire and
657
341
	// the key will come back.
376
	// the key will come back.
658
342
	time.Sleep(period * 2 * time.Second)
377
	time.Sleep(period * 2 * time.Second)
659
343
378
662
344
	w.ForceRefresh()
379
	w.Sync()
663
345
	c.Assert(w.Alive("a"), Equals, true)
380
	assertAlive(c, w, "a", true)
664
381
}
665
382
666
383
func (s *PresenceSuite) TestStartSync(c *C) {
667
384
	w := presence.NewWatcher(s.presence)
668
385
	p := presence.NewPinger(s.presence, "a")
669
386
	defer w.Stop()
670
387
	defer p.Stop()
671
388
672
389
	ch := make(chan presence.Change)
673
390
	w.Watch("a", ch)
674
391
	assertChange(c, ch, presence.Change{"a", false})
675
392
676
393
	c.Assert(p.Start(), IsNil)
677
394
678
395
	done := make(chan bool)
679
396
	go func() {
680
397
		w.StartSync()
681
398
		w.StartSync()
682
399
		w.StartSync()
683
400
		done <- true
684
401
	}()
685
402
686
403
	select {
687
404
	case <-done:
688
405
	case <-time.After(100 * time.Millisecond):
689
406
		c.Fatalf("StartSync failed to return")
690
407
	}
691
408
692
409
	assertChange(c, ch, presence.Change{"a", true})
693
410
}
694
411
695
412
func (s *PresenceSuite) TestSync(c *C) {
696
413
	w := presence.NewWatcher(s.presence)
697
414
	p := presence.NewPinger(s.presence, "a")
698
415
	defer w.Stop()
699
416
	defer p.Stop()
700
417
701
418
	ch := make(chan presence.Change)
702
419
	w.Watch("a", ch)
703
420
	assertChange(c, ch, presence.Change{"a", false})
704
421
705
422
	// Nothing to do here.
706
423
	w.Sync()
707
424
708
425
	c.Assert(p.Start(), IsNil)
709
426
710
427
	done := make(chan bool)
711
428
	go func() {
712
429
		w.Sync()
713
430
		done <- true
714
431
	}()
715
432
716
433
	select {
717
434
	case <-done:
718
435
		c.Fatalf("Sync returned too early")
719
436
	case <-time.After(200 * time.Millisecond):
720
437
	}
721
438
722
439
	assertChange(c, ch, presence.Change{"a", true})
723
440
724
441
	select {
725
442
	case <-done:
726
443
	case <-time.After(100 * time.Millisecond):
727
444
		c.Fatalf("Sync failed to returned")
728
445
	}
729
346
}
446
}
730
347
447
731
=== modified file 'mstate/state.go'
732
--- mstate/state.go	2012-09-10 15:29:01 +0000
733
+++ mstate/state.go	2012-09-11 13:29:53 +0000
734
@@ -28,17 +28,17 @@
735
28
// State represents the state of an environment
28
// State represents the state of an environment
736
29
// managed by juju.
29
// managed by juju.
737
30
type State struct {
30
type State struct {
749
31
	db         *mgo.Database
31
	db        *mgo.Database
750
32
	charms     *mgo.Collection
32
	charms    *mgo.Collection
751
33
	machines   *mgo.Collection
33
	machines  *mgo.Collection
752
34
	relations  *mgo.Collection
34
	relations *mgo.Collection
753
35
	services   *mgo.Collection
35
	services  *mgo.Collection
754
36
	settings   *mgo.Collection
36
	settings  *mgo.Collection
755
37
	units      *mgo.Collection
37
	units     *mgo.Collection
756
38
	presence   *mgo.Collection
38
	presence  *mgo.Collection
757
39
	runner     *txn.Runner
39
	runner    *txn.Runner
758
40
	watcher    *watcher.Watcher
40
	watcher   *watcher.Watcher
759
41
	presencew  *presence.Watcher
41
	pwatcher  *presence.Watcher
760
42
}
42
}
761
43
43
762
44
func deadOnAbort(err error) error {
44
func deadOnAbort(err error) error {
763
@@ -365,16 +365,16 @@
764
365
	return newUnit(s, &doc), nil
365
	return newUnit(s, &doc), nil
765
366
}
366
}
766
367
367
767
368
// ForcePresenceRefresh forces a synchronous refresh of 
768
369
// the presence watcher knowledge
769
370
func (s *State) ForcePresenceRefresh() {
770
371
	s.presencew.ForceRefresh()
771
372
}
772
373
773
374
// StartSync forces watchers to resynchronize their state with the
368
// StartSync forces watchers to resynchronize their state with the
774
375
// database immediately. This will happen periodically automatically.
369
// database immediately. This will happen periodically automatically.
775
376
func (s *State) StartSync() {
370
func (s *State) StartSync() {
776
377
	// TODO Make presence more like watcher, add it here, and
777
378
	// remove ForcePresenceRefresh.
778
379
	s.watcher.StartSync()
371
	s.watcher.StartSync()
779
372
	s.pwatcher.StartSync()
780
373
}
781
374
782
375
// Sync forces watchers to resynchronize their state with the
783
376
// database immediately, and waits until all events are known.
784
377
func (s *State) Sync() {
785
378
	s.watcher.Sync()
786
379
	s.pwatcher.Sync()
787
380
}
380
}
788
381
381
789
=== modified file 'mstate/unit.go'
790
--- mstate/unit.go	2012-09-06 16:54:22 +0000
791
+++ mstate/unit.go	2012-09-11 13:29:53 +0000
792
@@ -196,7 +196,10 @@
793
196
	case UnitStopped:
196
	case UnitStopped:
794
197
		return UnitStopped, "", nil
197
		return UnitStopped, "", nil
795
198
	}
198
	}
797
199
	alive := u.AgentAlive()
199
	alive, err := u.AgentAlive()
798
200
	if err != nil {
799
201
		return "", "", err
800
202
	}
801
200
	if !alive {
203
	if !alive {
802
201
		s = UnitDown
204
		s = UnitDown
803
202
	}
205
	}
804
@@ -222,35 +225,29 @@
805
222
}
225
}
806
223
226
807
224
// AgentAlive returns whether the respective remote agent is alive.
227
// AgentAlive returns whether the respective remote agent is alive.
810
225
func (u *Unit) AgentAlive() bool {
228
func (u *Unit) AgentAlive() (bool, error) {
811
226
	return u.st.presencew.Alive(u.globalKey())
229
	return u.st.pwatcher.Alive(u.globalKey())
812
227
}
230
}
813
228
231
814
229
// WaitAgentAlive blocks until the respective agent is alive.
232
// WaitAgentAlive blocks until the respective agent is alive.
816
230
func (u *Unit) WaitAgentAlive(timeout time.Duration) error {
233
func (u *Unit) WaitAgentAlive(timeout time.Duration) (err error) {
817
234
	defer trivial.ErrorContextf(&err, "waiting for agent of unit %q", u)
818
231
	ch := make(chan presence.Change)
235
	ch := make(chan presence.Change)
841
232
	u.st.presencew.Add(u.globalKey(), ch)
236
	u.st.pwatcher.Watch(u.globalKey(), ch)
842
233
	defer u.st.presencew.Remove(u.globalKey(), ch)
237
	defer u.st.pwatcher.Unwatch(u.globalKey(), ch)
843
234
	// Initial check.
238
	for i := 0; i < 2; i++ {
844
235
	select {
239
		select {
845
236
	case change := <-ch:
240
		case change := <-ch:
846
237
		if change.Alive {
241
			if change.Alive {
847
238
			return nil
242
				return nil
848
239
		}
243
			}
849
240
	case <-time.After(timeout):
244
		case <-time.After(timeout):
850
241
		return fmt.Errorf("waiting for agent of unit %q: still not alive after timeout", u)
245
			return fmt.Errorf("still not alive after timeout")
851
242
	}
246
		case <-u.st.pwatcher.Dying():
852
243
	// Hasn't been alive, so now wait for change.
247
			return u.st.pwatcher.Err()
853
244
	select {
248
		}
854
245
	case change := <-ch:
249
	}
855
246
		if change.Alive {
250
	panic(fmt.Sprintf("presence reported dead status twice in a row for unit %q", u))
834
247
			return nil
835
248
		}
836
249
		panic(fmt.Sprintf("presence reported dead status twice in a row for unit %q", u))
837
250
	case <-time.After(timeout):
838
251
		return fmt.Errorf("waiting for agent of unit %q: still not alive after timeout", u)
839
252
	}
840
253
	panic("unreachable")
856
254
}
251
}
857
255
252
858
256
// SetAgentAlive signals that the agent for unit u is alive. 
253
// SetAgentAlive signals that the agent for unit u is alive. 
859
257
254
860
=== modified file 'mstate/unit_test.go'
861
--- mstate/unit_test.go	2012-09-06 17:04:54 +0000
862
+++ mstate/unit_test.go	2012-09-11 13:29:53 +0000
863
@@ -90,7 +90,7 @@
864
90
		c.Assert(p.Kill(), IsNil)
90
		c.Assert(p.Kill(), IsNil)
865
91
	}()
91
	}()
866
92
92
868
93
	s.State.ForcePresenceRefresh()
93
	s.State.StartSync()
869
94
	status, info, err = s.unit.Status()
94
	status, info, err = s.unit.Status()
870
95
	c.Assert(err, IsNil)
95
	c.Assert(err, IsNil)
871
96
	c.Assert(status, Equals, state.UnitStarted)
96
	c.Assert(status, Equals, state.UnitStarted)
872
@@ -105,7 +105,8 @@
873
105
}
105
}
874
106
106
875
107
func (s *UnitSuite) TestUnitSetAgentAlive(c *C) {
107
func (s *UnitSuite) TestUnitSetAgentAlive(c *C) {
877
108
	alive := s.unit.AgentAlive()
108
	alive, err := s.unit.AgentAlive()
878
109
	c.Assert(err, IsNil)
879
109
	c.Assert(alive, Equals, false)
110
	c.Assert(alive, Equals, false)
880
110
111
881
111
	pinger, err := s.unit.SetAgentAlive()
112
	pinger, err := s.unit.SetAgentAlive()
882
@@ -113,34 +114,38 @@
883
113
	c.Assert(pinger, Not(IsNil))
114
	c.Assert(pinger, Not(IsNil))
884
114
	defer pinger.Stop()
115
	defer pinger.Stop()
885
115
116
888
116
	s.State.ForcePresenceRefresh()
117
	s.State.Sync()
889
117
	alive = s.unit.AgentAlive()
118
	alive, err = s.unit.AgentAlive()
890
119
	c.Assert(err, IsNil)
891
118
	c.Assert(alive, Equals, true)
120
	c.Assert(alive, Equals, true)
892
119
}
121
}
893
120
122
894
121
func (s *UnitSuite) TestUnitWaitAgentAlive(c *C) {
123
func (s *UnitSuite) TestUnitWaitAgentAlive(c *C) {
895
122
	timeout := 5 * time.Second
124
	timeout := 5 * time.Second
897
123
	alive := s.unit.AgentAlive()
125
	alive, err := s.unit.AgentAlive()
898
126
	c.Assert(err, IsNil)
899
124
	c.Assert(alive, Equals, false)
127
	c.Assert(alive, Equals, false)
900
125
128
902
126
	err := s.unit.WaitAgentAlive(timeout)
129
	err = s.unit.WaitAgentAlive(timeout)
903
127
	c.Assert(err, ErrorMatches, `waiting for agent of unit "wordpress/0": still not alive after timeout`)
130
	c.Assert(err, ErrorMatches, `waiting for agent of unit "wordpress/0": still not alive after timeout`)
904
128
131
905
129
	pinger, err := s.unit.SetAgentAlive()
132
	pinger, err := s.unit.SetAgentAlive()
906
130
	c.Assert(err, IsNil)
133
	c.Assert(err, IsNil)
907
131
134
909
132
	s.State.ForcePresenceRefresh()
135
	s.State.StartSync()
910
133
	err = s.unit.WaitAgentAlive(timeout)
136
	err = s.unit.WaitAgentAlive(timeout)
911
134
	c.Assert(err, IsNil)
137
	c.Assert(err, IsNil)
912
135
138
914
136
	alive = s.unit.AgentAlive()
139
	alive, err = s.unit.AgentAlive()
915
140
	c.Assert(err, IsNil)
916
137
	c.Assert(alive, Equals, true)
141
	c.Assert(alive, Equals, true)
917
138
142
918
139
	err = pinger.Kill()
143
	err = pinger.Kill()
919
140
	c.Assert(err, IsNil)
144
	c.Assert(err, IsNil)
920
141
145
923
142
	s.State.ForcePresenceRefresh()
146
	s.State.Sync()
924
143
	alive = s.unit.AgentAlive()
147
	alive, err = s.unit.AgentAlive()
925
148
	c.Assert(err, IsNil)
926
144
	c.Assert(alive, Equals, false)
149
	c.Assert(alive, Equals, false)
927
145
}
150
}
928
146
151
929
147
152
930
=== modified file 'mstate/watcher/watcher_test.go'
931
--- mstate/watcher/watcher_test.go	2012-09-10 16:19:57 +0000
932
+++ mstate/watcher/watcher_test.go	2012-09-11 13:29:53 +0000
933
@@ -440,9 +440,6 @@
934
440
func (s *WatcherSuite) TestStartSync(c *C) {
440
func (s *WatcherSuite) TestStartSync(c *C) {
935
441
	s.w.Watch("test", "a", -1, s.ch)
441
	s.w.Watch("test", "a", -1, s.ch)
936
442
442
937
443
	// Nothing to do here.
938
444
	s.w.StartSync()
939
445
940
446
	revno := s.insert(c, "test", "a")
443
	revno := s.insert(c, "test", "a")
941
447
444
942
448
	done := make(chan bool)
445
	done := make(chan bool)
943
@@ -456,7 +453,7 @@
944
456
	select {
453
	select {
945
457
	case <-done:
454
	case <-done:
946
458
	case <-time.After(100 * time.Millisecond):
455
	case <-time.After(100 * time.Millisecond):
948
459
		c.Fatalf("SyncStart failed to return")
456
		c.Fatalf("StartSync failed to return")
949
460
	}
457
	}
950
461
458
951
462
	assertChange(c, s.ch, watcher.Change{"test", "a", revno})
459
	assertChange(c, s.ch, watcher.Change{"test", "a", revno})
952
@@ -479,7 +476,7 @@
953
479
	select {
476
	select {
954
480
	case <-done:
477
	case <-done:
955
481
		c.Fatalf("Sync returned too early")
478
		c.Fatalf("Sync returned too early")
957
482
	case <-time.After(500 * time.Millisecond):
479
	case <-time.After(200 * time.Millisecond):
958
483
	}
480
	}
959
484
481
960
485
	assertChange(c, s.ch, watcher.Change{"test", "a", revno})
482
	assertChange(c, s.ch, watcher.Change{"test", "a", revno})
Reviewer	Review Type	Date Requested	Status
The Go Language Gophers		2012-09-10	Pending
Review via email: mp+123627@code.launchpad.net