Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Sam/stream msg app #1252

Closed
wants to merge 21 commits into from
Closed

Sam/stream msg app #1252

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
4ad62c9
Use grpc.DialContext
Jul 20, 2017
6c6d94f
Fix typo
Jul 19, 2017
388e397
Make node::send not Fatalf
Jul 18, 2017
7d2d959
Make RaftMessageStream RPC, used for MsgApp
Jul 19, 2017
5ee52b1
RaftMessageStream cleanup logic
Jul 23, 2017
9961da0
Make server graceful shutdown work
Jul 24, 2017
e08042d
Factor out sendMessages
Jul 24, 2017
d061609
Inline away n.send
Jul 24, 2017
b193f5c
Just some renamings
Jul 24, 2017
e9a2583
Tolerate stream failure in streamMsgApps
Jul 24, 2017
f513e08
Adjust comment
Jul 25, 2017
a402fb9
Add and make use of Kick() method in etcd
Jul 25, 2017
d183b28
Make kickCh not block
Jul 25, 2017
1f17412
Kick harder
Jul 25, 2017
cbee87b
kickc tweak
Jul 25, 2017
1502dc9
Make streamMsgApps WaitGroup be held per-peerPool
Jul 25, 2017
0698d6e
Remove fixed TODO
Jul 25, 2017
5554f8b
Remove spurious error return
Jul 25, 2017
91a8e21
kickValue placeholder
Jul 25, 2017
942c7f8
Make etcd use kickValue
Jul 25, 2017
240ca4c
Make kick buffering be in Dgraph, not etcd
Jul 25, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion cmd/dgraph/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -617,7 +617,10 @@ func serveHTTP(l net.Listener) {
func setupServer(che chan error) {
// By default Go GRPC traces all requests.
grpc.EnableTracing = false
go worker.RunServer(bindall) // For internal communication.
workerCancel, err := worker.SpawnServer(bindall) // For internal communication.
if err != nil {
log.Fatal(err)
}

laddr := "localhost"
if bindall {
Expand Down Expand Up @@ -662,6 +665,7 @@ func setupServer(che chan error) {
// Stops grpc/http servers; Already accepted connections are not closed.
grpcListener.Close()
httpListener.Close()
workerCancel()
}()

log.Println("gRPC server started. Listening on port", grpcPort())
Expand Down
140 changes: 106 additions & 34 deletions protos/payload.pb.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions protos/payload.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ service Worker {

// RAFT serving RPCs.
rpc RaftMessage (Payload) returns (Payload) {}
// The stream takes the same sort of Payload values that RaftMessage takes.
// Never returns.
rpc RaftMessageStream (stream Payload) returns (Payload) {}
rpc JoinCluster (RaftContext) returns (Payload) {}
rpc UpdateMembership (MembershipUpdate) returns (MembershipUpdate) {}
rpc Export (ExportPayload) returns (ExportPayload) {}
Expand Down
22 changes: 22 additions & 0 deletions vendor/github.com/coreos/etcd/raft/node.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 4 additions & 4 deletions worker/conn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ func destroyPool(pl *pool) {
}

// Returns a pool that you should call put() on.
func (p *poolsi) connect(addr string) (*pool, bool) {
func (p *poolsi) connect(ctx context.Context, addr string) (*pool, bool) {
if addr == Config.MyAddr {
return nil, false
}
Expand All @@ -118,7 +118,7 @@ func (p *poolsi) connect(addr string) (*pool, bool) {
}
p.RUnlock()

pool, err := newPool(addr)
pool, err := newPool(ctx, addr)
// TODO: Rename newPool to newConn, rename pool.
// TODO: This can get triggered with totally bogus config.
x.Checkf(err, "Unable to connect to host %s", addr)
Expand Down Expand Up @@ -172,8 +172,8 @@ func testConnection(p *pool) error {
}

// NewPool creates a new "pool" with one gRPC connection, refcount 0.
func newPool(addr string) (*pool, error) {
conn, err := grpc.Dial(addr,
func newPool(ctx context.Context, addr string) (*pool, error) {
conn, err := grpc.DialContext(ctx, addr,
grpc.WithDefaultCallOptions(
grpc.MaxCallRecvMsgSize(x.GrpcMaxSize),
grpc.MaxCallSendMsgSize(x.GrpcMaxSize)),
Expand Down
Loading