Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
whqwjb
go-ethereum
提交
f475a013
G
go-ethereum
项目概览
whqwjb
/
go-ethereum
与 Fork 源项目一致
从无法访问的项目Fork
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
G
go-ethereum
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
f475a013
编写于
6月 15, 2015
作者:
J
Jeffrey Wilcke
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #1261 from fjl/p2p-no-writes-at-shutdown
p2p: prevent writes at shutdown time
上级
26390337
70da79f0
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
61 addition
and
28 deletion
+61
-28
p2p/peer.go
p2p/peer.go
+60
-27
p2p/peer_test.go
p2p/peer_test.go
+1
-1
未找到文件。
p2p/peer.go
浏览文件 @
f475a013
...
...
@@ -115,41 +115,60 @@ func newPeer(conn *conn, protocols []Protocol) *Peer {
}
func
(
p
*
Peer
)
run
()
DiscReason
{
readErr
:=
make
(
chan
error
,
1
)
var
(
writeStart
=
make
(
chan
struct
{},
1
)
writeErr
=
make
(
chan
error
,
1
)
readErr
=
make
(
chan
error
,
1
)
reason
DiscReason
requested
bool
)
p
.
wg
.
Add
(
2
)
go
p
.
readLoop
(
readErr
)
go
p
.
pingLoop
()
p
.
startProtocols
()
// Start all protocol handlers.
writeStart
<-
struct
{}{}
p
.
startProtocols
(
writeStart
,
writeErr
)
// Wait for an error or disconnect.
var
(
reason
DiscReason
requested
bool
)
select
{
case
err
:=
<-
readErr
:
if
r
,
ok
:=
err
.
(
DiscReason
);
ok
{
reason
=
r
}
else
{
// Note: We rely on protocols to abort if there is a write
// error. It might be more robust to handle them here as well.
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"%v: Read error: %v
\n
"
,
p
,
err
)
reason
=
DiscNetworkError
loop
:
for
{
select
{
case
err
:=
<-
writeErr
:
// A write finished. Allow the next write to start if
// there was no error.
if
err
!=
nil
{
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"%v: write error: %v
\n
"
,
p
,
err
)
reason
=
DiscNetworkError
break
loop
}
writeStart
<-
struct
{}{}
case
err
:=
<-
readErr
:
if
r
,
ok
:=
err
.
(
DiscReason
);
ok
{
glog
.
V
(
logger
.
Debug
)
.
Infof
(
"%v: remote requested disconnect: %v
\n
"
,
p
,
r
)
requested
=
true
reason
=
r
}
else
{
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"%v: read error: %v
\n
"
,
p
,
err
)
reason
=
DiscNetworkError
}
break
loop
case
err
:=
<-
p
.
protoErr
:
reason
=
discReasonForError
(
err
)
glog
.
V
(
logger
.
Debug
)
.
Infof
(
"%v: protocol error: %v (%v)
\n
"
,
p
,
err
,
reason
)
break
loop
case
reason
=
<-
p
.
disc
:
glog
.
V
(
logger
.
Debug
)
.
Infof
(
"%v: locally requested disconnect: %v
\n
"
,
p
,
reason
)
break
loop
}
case
err
:=
<-
p
.
protoErr
:
reason
=
discReasonForError
(
err
)
case
reason
=
<-
p
.
disc
:
requested
=
true
}
close
(
p
.
closed
)
p
.
rw
.
close
(
reason
)
p
.
wg
.
Wait
()
if
requested
{
reason
=
DiscRequested
}
glog
.
V
(
logger
.
Debug
)
.
Infof
(
"%v: Disconnected: %v
\n
"
,
p
,
reason
)
return
reason
}
...
...
@@ -196,7 +215,6 @@ func (p *Peer) handle(msg Msg) error {
// This is the last message. We don't need to discard or
// check errors because, the connection will be closed after it.
rlp
.
Decode
(
msg
.
Payload
,
&
reason
)
glog
.
V
(
logger
.
Debug
)
.
Infof
(
"%v: Disconnect Requested: %v
\n
"
,
p
,
reason
[
0
])
return
reason
[
0
]
case
msg
.
Code
<
baseProtocolLength
:
// ignore other base protocol messages
...
...
@@ -247,11 +265,13 @@ outer:
return
result
}
func
(
p
*
Peer
)
startProtocols
()
{
func
(
p
*
Peer
)
startProtocols
(
writeStart
<-
chan
struct
{},
writeErr
chan
<-
error
)
{
p
.
wg
.
Add
(
len
(
p
.
running
))
for
_
,
proto
:=
range
p
.
running
{
proto
:=
proto
proto
.
closed
=
p
.
closed
proto
.
wstart
=
writeStart
proto
.
werr
=
writeErr
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"%v: Starting protocol %s/%d
\n
"
,
p
,
proto
.
Name
,
proto
.
Version
)
go
func
()
{
err
:=
proto
.
Run
(
p
,
proto
)
...
...
@@ -280,18 +300,31 @@ func (p *Peer) getProto(code uint64) (*protoRW, error) {
type
protoRW
struct
{
Protocol
in
chan
Msg
closed
<-
chan
struct
{}
in
chan
Msg
// receices read messages
closed
<-
chan
struct
{}
// receives when peer is shutting down
wstart
<-
chan
struct
{}
// receives when write may start
werr
chan
<-
error
// for write results
offset
uint64
w
MsgWriter
}
func
(
rw
*
protoRW
)
WriteMsg
(
msg
Msg
)
error
{
func
(
rw
*
protoRW
)
WriteMsg
(
msg
Msg
)
(
err
error
)
{
if
msg
.
Code
>=
rw
.
Length
{
return
newPeerError
(
errInvalidMsgCode
,
"not handled"
)
}
msg
.
Code
+=
rw
.
offset
return
rw
.
w
.
WriteMsg
(
msg
)
select
{
case
<-
rw
.
wstart
:
err
=
rw
.
w
.
WriteMsg
(
msg
)
// Report write status back to Peer.run. It will initiate
// shutdown if the error is non-nil and unblock the next write
// otherwise. The calling protocol code should exit for errors
// as well but we don't want to rely on that.
rw
.
werr
<-
err
case
<-
rw
.
closed
:
err
=
fmt
.
Errorf
(
"shutting down"
)
}
return
err
}
func
(
rw
*
protoRW
)
ReadMsg
()
(
Msg
,
error
)
{
...
...
p2p/peer_test.go
浏览文件 @
f475a013
...
...
@@ -121,7 +121,7 @@ func TestPeerDisconnect(t *testing.T) {
}
select
{
case
reason
:=
<-
disc
:
if
reason
!=
Disc
Quitting
{
if
reason
!=
Disc
Requested
{
t
.
Errorf
(
"run returned wrong reason: got %v, want %v"
,
reason
,
DiscRequested
)
}
case
<-
time
.
After
(
500
*
time
.
Millisecond
)
:
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录