-
Notifications
You must be signed in to change notification settings - Fork 9.7k
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.
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
*: fix races from clientv3/integration tests #4876
Changes from all commits
47db0a2
e129223
2c83362
91dc6b2
813afc3
5ee85be
4bdfc0a
161bc5e
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,64 @@ | ||
// Copyright 2016 CoreOS, Inc. | ||
// | ||
// Licensed under the Apache License, Version 2.0 (the "License"); | ||
// you may not use this file except in compliance with the License. | ||
// You may obtain a copy of the License at | ||
// | ||
// http://www.apache.org/licenses/LICENSE-2.0 | ||
// | ||
// Unless required by applicable law or agreed to in writing, software | ||
// distributed under the License is distributed on an "AS IS" BASIS, | ||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
// See the License for the specific language governing permissions and | ||
// limitations under the License. | ||
|
||
package clientv3 | ||
|
||
import ( | ||
"log" | ||
"os" | ||
"sync" | ||
|
||
"google.golang.org/grpc/grpclog" | ||
) | ||
|
||
type Logger grpclog.Logger | ||
|
||
var ( | ||
logger settableLogger | ||
) | ||
|
||
type settableLogger struct { | ||
l grpclog.Logger | ||
mu sync.RWMutex | ||
} | ||
|
||
func init() { | ||
// use go's standard logger by default like grpc | ||
logger.mu.Lock() | ||
logger.l = log.New(os.Stderr, "", log.LstdFlags) | ||
grpclog.SetLogger(&logger) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. lgtm. defer to @xiang90 |
||
logger.mu.Unlock() | ||
} | ||
|
||
func (s *settableLogger) Set(l Logger) { | ||
s.mu.Lock() | ||
logger.l = l | ||
s.mu.Unlock() | ||
} | ||
|
||
func (s *settableLogger) Get() Logger { | ||
s.mu.RLock() | ||
l := logger.l | ||
s.mu.RUnlock() | ||
return l | ||
} | ||
|
||
// implement the grpclog.Logger interface | ||
|
||
func (s *settableLogger) Fatal(args ...interface{}) { s.Get().Fatal(args...) } | ||
func (s *settableLogger) Fatalf(format string, args ...interface{}) { s.Get().Fatalf(format, args...) } | ||
func (s *settableLogger) Fatalln(args ...interface{}) { s.Get().Fatalln(args...) } | ||
func (s *settableLogger) Print(args ...interface{}) { s.Get().Print(args...) } | ||
func (s *settableLogger) Printf(format string, args ...interface{}) { s.Get().Printf(format, args...) } | ||
func (s *settableLogger) Println(args ...interface{}) { s.Get().Println(args...) } |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -16,6 +16,7 @@ package v3rpc | |
|
||
import ( | ||
"io" | ||
"sync" | ||
"time" | ||
|
||
"github.com/coreos/etcd/etcdserver" | ||
|
@@ -42,8 +43,9 @@ func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer { | |
|
||
var ( | ||
// expose for testing purpose. External test can change this to a | ||
// small value to finish fast. | ||
ProgressReportInterval = 10 * time.Minute | ||
// small value to finish fast. The type is int32 instead of time.Duration | ||
// in order to placate the race detector by setting the value with atomic stores. | ||
ProgressReportIntervalMilliseconds = int32(10 * 60 * 1000) // 10 minutes | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add a comment for why we use int32 instead of time duration? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. done |
||
) | ||
|
||
const ( | ||
|
@@ -71,6 +73,8 @@ type serverWatchStream struct { | |
// progress tracks the watchID that stream might need to send | ||
// progress to. | ||
progress map[storage.WatchID]bool | ||
// mu protects progress | ||
mu sync.Mutex | ||
|
||
// closec indicates the stream is closed. | ||
closec chan struct{} | ||
|
@@ -144,7 +148,9 @@ func (sws *serverWatchStream) recvLoop() error { | |
WatchId: id, | ||
Canceled: true, | ||
} | ||
sws.mu.Lock() | ||
delete(sws.progress, storage.WatchID(id)) | ||
sws.mu.Unlock() | ||
} | ||
} | ||
// TODO: do we need to return error back to client? | ||
|
@@ -160,7 +166,8 @@ func (sws *serverWatchStream) sendLoop() { | |
// watch responses pending on a watch id creation message | ||
pending := make(map[storage.WatchID][]*pb.WatchResponse) | ||
|
||
progressTicker := time.NewTicker(ProgressReportInterval) | ||
interval := time.Duration(ProgressReportIntervalMilliseconds) * time.Millisecond | ||
progressTicker := time.NewTicker(interval) | ||
defer progressTicker.Stop() | ||
|
||
for { | ||
|
@@ -198,9 +205,11 @@ func (sws *serverWatchStream) sendLoop() { | |
return | ||
} | ||
|
||
sws.mu.Lock() | ||
if _, ok := sws.progress[wresp.WatchID]; ok { | ||
sws.progress[wresp.WatchID] = false | ||
} | ||
sws.mu.Unlock() | ||
|
||
case c, ok := <-sws.ctrlStream: | ||
if !ok { | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -165,19 +165,19 @@ func (t *Transport) Send(msgs []raftpb.Message) { | |
to := types.ID(m.To) | ||
|
||
t.mu.RLock() | ||
p, ok := t.peers[to] | ||
p, pok := t.peers[to] | ||
g, rok := t.remotes[to] | ||
t.mu.RUnlock() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. good catch. Would like to test this code to see if we still get #4855. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nvm, I still see failures even with this. Will investigate more. |
||
|
||
if ok { | ||
if pok { | ||
if m.Type == raftpb.MsgApp { | ||
t.ServerStats.SendAppendReq(m.Size()) | ||
} | ||
p.send(m) | ||
continue | ||
} | ||
|
||
g, ok := t.remotes[to] | ||
if ok { | ||
if rok { | ||
g.send(m) | ||
continue | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
it is not a good idea to hold the lock while doing network io i think.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think it's OK here since switchRemote only happens on connection failure; IO would have to wait for a new connection anyway and canceling the client context can break the lock if it stalls.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok. add a comment for this?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sure