cleanup and improve ping
1. Use emit directly. 2. Abort immediately when the context is canceled (no need to wait a second for the sleep to elapse). 3. Ping once a second instead of once a second +TTL (to be more like ping) License: MIT Signed-off-by: Steven Allen <steven@stebalien.com>
Steven Allen committed
Oct 27, 2018 at 10:40 UTC
b870e8c6d111bdbf66f86669cef1e73210f61c29
1 file changed
+50
-80
core/commands/ping.go
+50
-80
@@ -8,7 +8,6 @@ import (
8
"strings"
9
"time"
10
11
- "github.com/ipfs/go-ipfs/core"
11
"github.com/ipfs/go-ipfs/core/commands/cmdenv"
12
13
cmds "gx/ipfs/QmSXUokcP4TJpFfqozT69AVAYRtzXVMUjzQVkYX41R9Svs/go-ipfs-cmds"
@@ -59,17 +58,17 @@ trip latency information.
58
return ErrNotOnline
59
}
60
62
- addr, peerID, err := ParsePeerParam(req.Arguments[0])
61
+ addr, pid, err := ParsePeerParam(req.Arguments[0])
62
if err != nil {
63
return fmt.Errorf("failed to parse peer address '%s': %s", req.Arguments[0], err)
64
}
65
67
- if peerID == n.Identity {
66
+ if pid == n.Identity {
67
return ErrPingSelf
68
}
69
70
if addr != nil {
72
- n.Peerstore.AddAddr(peerID, addr, pstore.TempAddrTTL) // temporary
71
+ n.Peerstore.AddAddr(pid, addr, pstore.TempAddrTTL) // temporary
72
}
73
74
numPings, _ := req.Options[pingCountOptionName].(int)
@@ -77,113 +76,84 @@ trip latency information.
76
return fmt.Errorf("error: ping count must be greater than 0, was %d", numPings)
77
}
78
80
- outChan := pingPeer(req.Context, n, peerID, numPings)
81
-
82
- return res.Emit(outChan)
83
- },
84
- Type: PingResult{},
85
- Encoders: cmds.EncoderMap{
86
- cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *PingResult) error {
87
- if len(out.Text) > 0 {
88
- fmt.Fprintln(w, out.Text)
89
- } else if out.Success {
90
- fmt.Fprintf(w, "Pong received: time=%.2f ms\n", out.Time.Seconds()*1000)
91
- } else {
92
- fmt.Fprintf(w, "Pong failed\n")
93
- }
94
- return nil
95
- }),
96
- },
97
-}
98
-
99
-func pingPeer(ctx context.Context, n *core.IpfsNode, pid peer.ID, numPings int) <-chan interface{} {
100
- outChan := make(chan interface{})
101
- go func() {
102
- defer close(outChan)
103
-
79
if len(n.Peerstore.Addrs(pid)) == 0 {
80
// Make sure we can find the node in question
106
- select {
107
- case outChan <- &PingResult{
81
+ if err := res.Emit(&PingResult{
82
Text: fmt.Sprintf("Looking up peer %s", pid.Pretty()),
83
Success: true,
110
- }:
111
- case <-ctx.Done():
112
- return
84
+ }); err != nil {
85
+ return err
86
}
87
115
- ctx, cancel := context.WithTimeout(ctx, kPingTimeout)
116
- defer cancel()
88
+ ctx, cancel := context.WithTimeout(req.Context, kPingTimeout)
89
p, err := n.Routing.FindPeer(ctx, pid)
90
+ cancel()
91
if err != nil {
119
- select {
120
- case outChan <- &PingResult{Text: fmt.Sprintf("Peer lookup error: %s", err)}:
121
- case <-ctx.Done():
122
- return
123
- }
124
-
125
- return
92
+ return res.Emit(&PingResult{Text: fmt.Sprintf("Peer lookup error: %s", err)})
93
}
94
n.Peerstore.AddAddrs(p.ID, p.Addrs, pstore.TempAddrTTL)
95
}
96
130
- select {
131
- case outChan <- &PingResult{
97
+ if err := res.Emit(&PingResult{
98
Text: fmt.Sprintf("PING %s.", pid.Pretty()),
99
Success: true,
134
- }:
135
- case <-ctx.Done():
136
- return
100
+ }); err != nil {
101
+ return err
102
}
103
139
- ctx, cancel := context.WithTimeout(ctx, kPingTimeout*time.Duration(numPings))
104
+ ctx, cancel := context.WithTimeout(req.Context, kPingTimeout*time.Duration(numPings))
105
defer cancel()
106
pings, err := n.Ping.Ping(ctx, pid)
107
if err != nil {
143
- select {
144
- case outChan <- &PingResult{
108
+ return res.Emit(&PingResult{
109
Success: false,
110
Text: fmt.Sprintf("Ping error: %s", err),
147
- }:
148
- case <-ctx.Done():
149
- }
150
- return
111
+ })
112
}
113
153
- var done bool
114
var total time.Duration
155
- for i := 0; i < numPings && !done; i++ {
115
+ ticker := time.NewTicker(time.Second)
116
+ defer ticker.Stop()
117
+ for i := 0; i < numPings; i++ {
118
+ t, ok := <-pings
119
+ if !ok {
120
+ break
121
+ }
122
+
123
+ if err := res.Emit(&PingResult{
124
+ Success: true,
125
+ Time: t,
126
+ }); err != nil {
127
+ return err
128
+ }
129
+
130
+ total += t
131
+
132
select {
133
+ case <-ticker.C:
134
case <-ctx.Done():
158
- done = true
159
- break
160
- case t, ok := <-pings:
161
- if !ok {
162
- done = true
163
- break
164
- }
165
- select {
166
- case outChan <- &PingResult{
167
- Success: true,
168
- Time: t,
169
- }:
170
- case <-ctx.Done():
171
- return
172
- }
173
- total += t
174
- time.Sleep(time.Second)
135
+ return ctx.Err()
136
}
137
}
138
averagems := total.Seconds() * 1000 / float64(numPings)
178
- select {
179
- case outChan <- &PingResult{
139
+ return res.Emit(&PingResult{
140
Success: true,
141
Text: fmt.Sprintf("Average latency: %.2fms", averagems),
182
- }:
183
- case <-ctx.Done():
184
- }
185
- }()
186
- return outChan
142
+ })
143
+ },
144
+ Type: PingResult{},
145
+ Encoders: cmds.EncoderMap{
146
+ cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, out *PingResult) error {
147
+ if len(out.Text) > 0 {
148
+ fmt.Fprintln(w, out.Text)
149
+ } else if out.Success {
150
+ fmt.Fprintf(w, "Pong received: time=%.2f ms\n", out.Time.Seconds()*1000)
151
+ } else {
152
+ fmt.Fprintf(w, "Pong failed\n")
153
+ }
154
+ return nil
155
+ }),
156
+ },
157
}
158
159
func ParsePeerParam(text string) (ma.Multiaddr, peer.ID, error) {