p2p: Optimize registry, move stream stuff around
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Apr 6, 2018 at 15:59 UTC
a0ad8cfd5c49d3a2f3a3de1004a8cad65369d315
7 files changed
+134
-150
core/commands/p2p.go
+6
-6
@@ -91,7 +91,7 @@ var p2pListenerLsCmd = &cmds.Command{
91
Tagline: "List active p2p listeners.",
92
},
93
Options: []cmdkit.Option{
94
- cmdkit.BoolOption("headers", "v", "Print table headers (HandlerID, Protocol, Local, Remote)."),
94
+ cmdkit.BoolOption("headers", "v", "Print table headers (Id, Protocol, Local, Remote)."),
95
},
96
Run: func(req cmds.Request, res cmds.Response) {
97
@@ -156,7 +156,7 @@ var p2pStreamLsCmd = &cmds.Command{
156
157
for _, s := range n.P2P.Streams.Streams {
158
output.Streams = append(output.Streams, P2PStreamInfoOutput{
159
- HandlerID: strconv.FormatUint(s.HandlerID, 10),
159
+ HandlerID: strconv.FormatUint(s.Id, 10),
160
161
Protocol: s.Protocol,
162
@@ -184,7 +184,7 @@ var p2pStreamLsCmd = &cmds.Command{
184
w := tabwriter.NewWriter(buf, 1, 2, 1, ' ', 0)
185
for _, stream := range list.Streams {
186
if headers {
187
- fmt.Fprintln(w, "HandlerID\tProtocol\tLocal\tRemote")
187
+ fmt.Fprintln(w, "Id\tProtocol\tLocal\tRemote")
188
}
189
190
fmt.Fprintf(w, "%s\t%s\t%s\t%s\n", stream.HandlerID, stream.Protocol, stream.LocalAddress, stream.RemotePeer)
@@ -347,7 +347,7 @@ var p2pStreamCloseCmd = &cmds.Command{
347
Tagline: "Close active p2p stream.",
348
},
349
Arguments: []cmdkit.Argument{
350
- cmdkit.StringArg("HandlerID", false, false, "Stream HandlerID"),
350
+ cmdkit.StringArg("Id", false, false, "Stream Id"),
351
},
352
Options: []cmdkit.Option{
353
cmdkit.BoolOption("all", "a", "Close all streams."),
@@ -366,7 +366,7 @@ var p2pStreamCloseCmd = &cmds.Command{
366
367
if !closeAll {
368
if len(req.Arguments()) == 0 {
369
- res.SetError(errors.New("no HandlerID specified"), cmdkit.ErrNormal)
369
+ res.SetError(errors.New("no Id specified"), cmdkit.ErrNormal)
370
return
371
}
372
@@ -378,7 +378,7 @@ var p2pStreamCloseCmd = &cmds.Command{
378
}
379
380
for _, stream := range n.P2P.Streams.Streams {
381
- if !closeAll && handlerID != stream.HandlerID {
381
+ if !closeAll && handlerID != stream.Id {
382
continue
383
}
384
stream.Close()
p2p/inbound.go
+3
-2
@@ -32,7 +32,7 @@ func (p2p *P2P) NewListener(ctx context.Context, proto string, addr ma.Multiaddr
32
return
33
}
34
35
- stream := StreamInfo{
35
+ stream := Stream{
36
Protocol: proto,
37
38
LocalPeer: p2p.identity,
@@ -66,5 +66,6 @@ func (l *inboundListener) Address() string {
66
67
func (l *inboundListener) Close() error {
68
l.p2p.peerHost.RemoveStreamHandler(protocol.ID(l.proto))
69
- return l.p2p.Listeners.Deregister(l.proto)
69
+ l.p2p.Listeners.Deregister(l.proto)
70
+ return nil
71
}
p2p/listener.go
new
+39
@@ -0,0 +1,39 @@
1
+package p2p
2
+
3
+import (
4
+ pstore "gx/ipfs/QmZb7hAgQEhW9dBbzBudU39gCeD4zbe6xafD52LUuF4cUN/go-libp2p-peerstore"
5
+ peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
6
+ p2phost "gx/ipfs/QmdHyfNVTZ5VtUx4Xz23z8wtnioSrFQ28XSfpVkdhQBkGA/go-libp2p-host"
7
+)
8
+
9
+type Listener interface {
10
+ Protocol() string
11
+ Address() string
12
+
13
+ // Close closes the listener. Does not affect child streams
14
+ Close() error
15
+}
16
+
17
+// NewP2P creates new P2P struct
18
+func NewP2P(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore) *P2P {
19
+ return &P2P{
20
+ identity: identity,
21
+ peerHost: peerHost,
22
+ peerstore: peerstore,
23
+ }
24
+}
25
+
26
+// ListenerRegistry is a collection of local application proto listeners.
27
+type ListenerRegistry struct {
28
+ Listeners map[string]Listener
29
+}
30
+
31
+// Register registers listenerInfo2 in this registry
32
+func (c *ListenerRegistry) Register(listenerInfo Listener) {
33
+ c.Listeners[listenerInfo.Protocol()] = listenerInfo
34
+}
35
+
36
+// Deregister removes p2p listener from this registry
37
+func (c *ListenerRegistry) Deregister(proto string) {
38
+ delete(c.Listeners, proto)
39
+}
p2p/outbound.go
+3
-3
@@ -84,7 +84,7 @@ func (l *outboundListener) acceptConns() {
84
return
85
}
86
87
- stream := StreamInfo{
87
+ stream := Stream{
88
Protocol: l.proto,
89
90
LocalPeer: l.id,
@@ -106,8 +106,8 @@ func (l *outboundListener) acceptConns() {
106
107
func (l *outboundListener) Close() error {
108
l.listener.Close()
109
- err := l.p2p.Listeners.Deregister(l.proto)
110
- return err
109
+ l.p2p.Listeners.Deregister(l.proto)
110
+ return nil
111
}
112
113
func (l *outboundListener) Protocol() string {
p2p/p2p.go
-17
@@ -16,23 +16,6 @@ type P2P struct {
16
peerstore pstore.Peerstore
17
}
18
19
-type Listener interface {
20
- Protocol() string
21
- Address() string
22
-
23
- // Close closes the listener. Does not affect child streams
24
- Close() error
25
-}
26
-
27
-// NewP2P creates new P2P struct
28
-func NewP2P(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore) *P2P {
29
- return &P2P{
30
- identity: identity,
31
- peerHost: peerHost,
32
- peerstore: peerstore,
33
- }
34
-}
35
-
19
// CheckProtoExists checks whether a proto handler is registered to
20
// mux handler
21
func (p2p *P2P) CheckProtoExists(proto string) bool {
p2p/registry.go
deleted
-122
@@ -1,122 +0,0 @@
1
-package p2p
2
-
3
-import (
4
- "fmt"
5
- "io"
6
-
7
- peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
8
- manet "gx/ipfs/QmV6FjemM1K8oXjrvuq3wuVWWoU2TLDPmNnKrxHzY3v6Ai/go-multiaddr-net"
9
- ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"
10
- net "gx/ipfs/QmZNJyx9GGCX4GeuHnLB8fxaxMLs4MjTjHokxfQcCd6Nve/go-libp2p-net"
11
-)
12
-
13
-// ListenerRegistry is a collection of local application proto listeners.
14
-type ListenerRegistry struct {
15
- Listeners []Listener
16
-}
17
-
18
-// Register registers listenerInfo2 in this registry
19
-func (c *ListenerRegistry) Register(listenerInfo Listener) {
20
- c.Listeners = append(c.Listeners, listenerInfo)
21
-}
22
-
23
-// Deregister removes p2p listener from this registry
24
-func (c *ListenerRegistry) Deregister(proto string) error {
25
- foundAt := -1
26
- for i, a := range c.Listeners {
27
- if a.Protocol() == proto {
28
- foundAt = i
29
- break
30
- }
31
- }
32
-
33
- if foundAt != -1 {
34
- c.Listeners = append(c.Listeners[:foundAt], c.Listeners[foundAt+1:]...)
35
- return nil
36
- }
37
-
38
- return fmt.Errorf("failed to deregister proto %s", proto)
39
-}
40
-
41
-// StreamInfo holds information on active incoming and outgoing p2p streams.
42
-type StreamInfo struct {
43
- HandlerID uint64
44
-
45
- Protocol string
46
-
47
- LocalPeer peer.ID
48
- LocalAddr ma.Multiaddr
49
-
50
- RemotePeer peer.ID
51
- RemoteAddr ma.Multiaddr
52
-
53
- Local manet.Conn
54
- Remote net.Stream
55
-
56
- Registry *StreamRegistry
57
-}
58
-
59
-// Close closes stream endpoints and deregisters it
60
-func (s *StreamInfo) Close() error {
61
- s.Local.Close()
62
- s.Remote.Close()
63
- s.Registry.Deregister(s.HandlerID)
64
- return nil
65
-}
66
-
67
-// Reset closes stream endpoints and deregisters it
68
-func (s *StreamInfo) Reset() error {
69
- s.Local.Close()
70
- s.Remote.Reset()
71
- s.Registry.Deregister(s.HandlerID)
72
- return nil
73
-}
74
-
75
-func (s *StreamInfo) startStreaming() {
76
- go func() {
77
- _, err := io.Copy(s.Local, s.Remote)
78
- if err != nil {
79
- s.Reset()
80
- } else {
81
- s.Close()
82
- }
83
- }()
84
-
85
- go func() {
86
- _, err := io.Copy(s.Remote, s.Local)
87
- if err != nil {
88
- s.Reset()
89
- } else {
90
- s.Close()
91
- }
92
- }()
93
-}
94
-
95
-// StreamRegistry is a collection of active incoming and outgoing proto app streams.
96
-type StreamRegistry struct {
97
- Streams []*StreamInfo
98
-
99
- nextID uint64
100
-}
101
-
102
-// Register registers a stream to the registry
103
-func (c *StreamRegistry) Register(streamInfo *StreamInfo) {
104
- streamInfo.HandlerID = c.nextID
105
- c.Streams = append(c.Streams, streamInfo)
106
- c.nextID++
107
-}
108
-
109
-// Deregister deregisters stream from the registry
110
-func (c *StreamRegistry) Deregister(handlerID uint64) {
111
- foundAt := -1
112
- for i, s := range c.Streams {
113
- if s.HandlerID == handlerID {
114
- foundAt = i
115
- break
116
- }
117
- }
118
-
119
- if foundAt != -1 {
120
- c.Streams = append(c.Streams[:foundAt], c.Streams[foundAt+1:]...)
121
- }
122
-}
p2p/stream.go
new
+83
@@ -0,0 +1,83 @@
1
+package p2p
2
+
3
+import (
4
+ "io"
5
+
6
+ ma "gx/ipfs/QmWWQ2Txc2c6tqjsBpzg5Ar652cHPGNsQQp2SejkNmkUMb/go-multiaddr"
7
+ net "gx/ipfs/QmYj8wdn5sZEHX2XMDWGBvcXJNdzVbaVpHmXvhHBVZepen/go-libp2p-net"
8
+ manet "gx/ipfs/QmcGXGdw9BWDysPJQHxJinjGHha3eEg4vzFETre4woNwcX/go-multiaddr-net"
9
+ peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
10
+)
11
+
12
+// Stream holds information on active incoming and outgoing p2p streams.
13
+type Stream struct {
14
+ Id uint64
15
+
16
+ Protocol string
17
+
18
+ LocalPeer peer.ID
19
+ LocalAddr ma.Multiaddr
20
+
21
+ RemotePeer peer.ID
22
+ RemoteAddr ma.Multiaddr
23
+
24
+ Local manet.Conn
25
+ Remote net.Stream
26
+
27
+ Registry *StreamRegistry
28
+}
29
+
30
+// Close closes stream endpoints and deregisters it
31
+func (s *Stream) Close() error {
32
+ s.Local.Close()
33
+ s.Remote.Close()
34
+ s.Registry.Deregister(s.Id)
35
+ return nil
36
+}
37
+
38
+// Reset closes stream endpoints and deregisters it
39
+func (s *Stream) Reset() error {
40
+ s.Local.Close()
41
+ s.Remote.Reset()
42
+ s.Registry.Deregister(s.Id)
43
+ return nil
44
+}
45
+
46
+func (s *Stream) startStreaming() {
47
+ go func() {
48
+ _, err := io.Copy(s.Local, s.Remote)
49
+ if err != nil {
50
+ s.Reset()
51
+ } else {
52
+ s.Close()
53
+ }
54
+ }()
55
+
56
+ go func() {
57
+ _, err := io.Copy(s.Remote, s.Local)
58
+ if err != nil {
59
+ s.Reset()
60
+ } else {
61
+ s.Close()
62
+ }
63
+ }()
64
+}
65
+
66
+// StreamRegistry is a collection of active incoming and outgoing proto app streams.
67
+type StreamRegistry struct {
68
+ Streams map[uint64]*Stream
69
+
70
+ nextId uint64
71
+}
72
+
73
+// Register registers a stream to the registry
74
+func (c *StreamRegistry) Register(streamInfo *Stream) {
75
+ streamInfo.Id = c.nextId
76
+ c.Streams[c.nextId] = streamInfo
77
+ c.nextId++
78
+}
79
+
80
+// Deregister deregisters stream from the registry
81
+func (c *StreamRegistry) Deregister(streamId uint64) {
82
+ delete(c.Streams, streamId)
83
+}