this repo has no description
use parallel event processor in bgs (#241)
There might be other test failures as a result of this, but i ran them
20 or so times and this was the only one that popped up locally.
We might want to make the settings on the consumer pool tuneable, but
for now this should get us some much better numbers on our load testing
authored by
Whyrusleeping
and committed by
GitHub
420fa008
988c1e0d
+11
-13
+2
-2
bgs/fedmgr.go
+2
-2
bgs/fedmgr.go
···
332
332
},
333
333
}
334
334
335
-
// TODO: probably make this use the parallel handler after some thought
336
-
return events.HandleRepoStream(ctx, con, &events.SequentialScheduler{rsc.EventHandler})
335
+
pool := events.NewConsumerPool(32, 20, rsc.EventHandler)
336
+
return events.HandleRepoStream(ctx, con, pool)
337
337
}
338
338
339
339
func (s *Slurper) updateCursor(host *models.PDS, curs int64) error {
+9
-11
testing/integ_test.go
+9
-11
testing/integ_test.go
···
44
44
defer evts.Cancel()
45
45
46
46
bob := p1.MustNewUser(t, "bob.tpds")
47
-
alice := p1.MustNewUser(t, "alice.tpds")
48
-
49
-
bp1 := bob.Post(t, "cats for cats")
50
-
ap1 := alice.Post(t, "no i like dogs")
51
-
52
-
_ = bp1
53
-
_ = ap1
54
-
55
-
fmt.Println("bob:", bob.DID())
56
-
fmt.Println("alice:", alice.DID())
57
-
58
47
fmt.Println("event 1")
59
48
e1 := evts.Next()
60
49
assert.NotNil(e1.RepoCommit)
61
50
assert.Equal(e1.RepoCommit.Repo, bob.DID())
62
51
52
+
alice := p1.MustNewUser(t, "alice.tpds")
63
53
fmt.Println("event 2")
64
54
e2 := evts.Next()
65
55
assert.NotNil(e2.RepoCommit)
66
56
assert.Equal(e2.RepoCommit.Repo, alice.DID())
67
57
58
+
bp1 := bob.Post(t, "cats for cats")
59
+
ap1 := alice.Post(t, "no i like dogs")
60
+
61
+
_ = bp1
62
+
_ = ap1
63
+
64
+
fmt.Println("bob:", bob.DID())
68
65
fmt.Println("event 3")
69
66
e3 := evts.Next()
70
67
assert.Equal(e3.RepoCommit.Repo, bob.DID())
71
68
//assert.Equal(e3.RepoCommit.Ops[0].Kind, "createRecord")
72
69
70
+
fmt.Println("alice:", alice.DID())
73
71
fmt.Println("event 4")
74
72
e4 := evts.Next()
75
73
assert.Equal(e4.RepoCommit.Repo, alice.DID())