added swarm (fanout/in with peers)
Juan Batiz-Benet committed
Jul 12, 2014 at 01:51 UTC
deb7dd487286a80f1d4a9aa738159d660589f4f2
4 files changed
+366
-124
swarm/conn.go
+49
-47
@@ -1,72 +1,74 @@
1
package swarm
2
3
import (
4
- "fmt"
5
- "net"
6
- ma "github.com/jbenet/go-multiaddr"
7
- peer "github.com/jbenet/go-ipfs/peer"
8
- msgio "github.com/jbenet/go-msgio"
4
+ "fmt"
5
+ peer "github.com/jbenet/go-ipfs/peer"
6
+ u "github.com/jbenet/go-ipfs/util"
7
+ msgio "github.com/jbenet/go-msgio"
8
+ ma "github.com/jbenet/go-multiaddr"
9
+ "net"
10
)
11
12
const ChanBuffer = 10
13
14
type Conn struct {
14
- Peer *peer.Peer
15
- Addr *ma.Multiaddr
16
- Conn net.Conn
15
+ Peer *peer.Peer
16
+ Addr *ma.Multiaddr
17
+ Conn net.Conn
18
18
- Closed chan bool
19
- Outgoing *msgio.Chan
20
- Incoming *msgio.Chan
19
+ Closed chan bool
20
+ Outgoing *msgio.Chan
21
+ Incoming *msgio.Chan
22
}
23
24
+type ConnMap map[u.Key]*Conn
25
26
func Dial(network string, peer *peer.Peer) (*Conn, error) {
25
- addr := peer.NetAddress(network)
26
- if addr == nil {
27
- return nil, fmt.Errorf("No address for network %s", network)
28
- }
27
+ addr := peer.NetAddress(network)
28
+ if addr == nil {
29
+ return nil, fmt.Errorf("No address for network %s", network)
30
+ }
31
30
- network, host, err := addr.DialArgs()
31
- if err != nil {
32
- return nil, err
33
- }
32
+ network, host, err := addr.DialArgs()
33
+ if err != nil {
34
+ return nil, err
35
+ }
36
35
- nconn, err := net.Dial(network, host)
36
- if err != nil {
37
- return nil, err
38
- }
37
+ nconn, err := net.Dial(network, host)
38
+ if err != nil {
39
+ return nil, err
40
+ }
41
40
- out := msgio.NewChan(10)
41
- inc := msgio.NewChan(10)
42
+ out := msgio.NewChan(10)
43
+ inc := msgio.NewChan(10)
44
43
- conn := &Conn{
44
- Peer: peer,
45
- Addr: addr,
46
- Conn: nconn,
45
+ conn := &Conn{
46
+ Peer: peer,
47
+ Addr: addr,
48
+ Conn: nconn,
49
48
- Outgoing: out,
49
- Incoming: inc,
50
- Closed: make(chan bool, 1),
51
- }
50
+ Outgoing: out,
51
+ Incoming: inc,
52
+ Closed: make(chan bool, 1),
53
+ }
54
53
- go out.WriteTo(nconn)
54
- go inc.ReadFrom(nconn, 1 << 12)
55
+ go out.WriteTo(nconn)
56
+ go inc.ReadFrom(nconn, 1<<12)
57
56
- return conn, nil
58
+ return conn, nil
59
}
60
61
func (s *Conn) Close() error {
60
- if s.Conn == nil {
61
- return fmt.Errorf("Already closed.") // already closed
62
- }
62
+ if s.Conn == nil {
63
+ return fmt.Errorf("Already closed.") // already closed
64
+ }
65
64
- // closing net connection
65
- err := s.Conn.Close()
66
- s.Conn = nil
67
- // closing channels
68
- s.Incoming.Close()
69
- s.Outgoing.Close()
70
- s.Closed<- true
71
- return err
66
+ // closing net connection
67
+ err := s.Conn.Close()
68
+ s.Conn = nil
69
+ // closing channels
70
+ s.Incoming.Close()
71
+ s.Outgoing.Close()
72
+ s.Closed <- true
73
+ return err
74
}
swarm/conn_test.go
+73
-76
@@ -1,93 +1,90 @@
1
package swarm
2
3
import (
4
- "fmt"
5
- "net"
6
- "testing"
7
- ma "github.com/jbenet/go-multiaddr"
8
- mh "github.com/jbenet/go-multihash"
9
- peer "github.com/jbenet/go-ipfs/peer"
4
+ "fmt"
5
+ peer "github.com/jbenet/go-ipfs/peer"
6
+ ma "github.com/jbenet/go-multiaddr"
7
+ mh "github.com/jbenet/go-multihash"
8
+ "net"
9
+ "testing"
10
)
11
12
func setupPeer(id string, addr string) (*peer.Peer, error) {
13
- tcp, err := ma.NewMultiaddr(addr)
14
- if err != nil {
15
- return nil, err
16
- }
17
-
18
- mh, err := mh.FromHexString(id)
19
- if err != nil {
20
- return nil, err
21
- }
22
-
23
- p := &peer.Peer{Id: peer.PeerId(mh)}
24
- p.AddAddress(tcp)
25
- return p, nil
13
+ tcp, err := ma.NewMultiaddr(addr)
14
+ if err != nil {
15
+ return nil, err
16
+ }
17
+
18
+ mh, err := mh.FromHexString(id)
19
+ if err != nil {
20
+ return nil, err
21
+ }
22
+
23
+ p := &peer.Peer{Id: peer.PeerId(mh)}
24
+ p.AddAddress(tcp)
25
+ return p, nil
26
}
27
28
func echoListen(listener *net.TCPListener) {
29
- for {
30
- c, err := listener.Accept()
31
- if err == nil {
32
- fmt.Println("accepeted")
33
- go echo(c)
34
- }
35
- }
29
+ for {
30
+ c, err := listener.Accept()
31
+ if err == nil {
32
+ fmt.Println("accepeted")
33
+ go echo(c)
34
+ }
35
+ }
36
}
37
38
func echo(c net.Conn) {
39
- for {
40
- data := make([]byte, 1024)
41
- i, err := c.Read(data)
42
- if err != nil {
43
- fmt.Printf("error %v\n", err)
44
- return
45
- }
46
- _, err = c.Write(data[:i])
47
- if err != nil {
48
- fmt.Printf("error %v\n", err)
49
- return
50
- }
51
- fmt.Println("echoing", data[:i])
52
- }
39
+ for {
40
+ data := make([]byte, 1024)
41
+ i, err := c.Read(data)
42
+ if err != nil {
43
+ fmt.Printf("error %v\n", err)
44
+ return
45
+ }
46
+ _, err = c.Write(data[:i])
47
+ if err != nil {
48
+ fmt.Printf("error %v\n", err)
49
+ return
50
+ }
51
+ fmt.Println("echoing", data[:i])
52
+ }
53
}
54
55
func TestDial(t *testing.T) {
56
57
- listener, err := net.Listen("tcp", "127.0.0.1:1234")
58
- if err != nil {
59
- t.Fatal("error setting up listener", err)
60
- }
61
- go echoListen(listener.(*net.TCPListener))
62
-
63
- p, err := setupPeer("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33", "/ip4/127.0.0.1/tcp/1234")
64
- if err != nil {
65
- t.Fatal("error setting up peer", err)
66
- }
67
-
68
- c, err := Dial("tcp", p)
69
- if err != nil {
70
- t.Fatal("error dialing peer", err)
71
- }
72
-
73
-
74
- fmt.Println("sending")
75
- c.Outgoing.MsgChan<- []byte("beep")
76
- c.Outgoing.MsgChan<- []byte("boop")
77
- out := <-c.Incoming.MsgChan
78
- fmt.Println("recving", string(out))
79
- if string(out) != "beep" {
80
- t.Error("unexpected conn output")
81
- }
82
-
83
- out = <-c.Incoming.MsgChan
84
- if string(out) != "boop" {
85
- t.Error("unexpected conn output")
86
- }
87
-
88
- fmt.Println("closing")
89
- c.Close()
90
- listener.Close()
57
+ listener, err := net.Listen("tcp", "127.0.0.1:1234")
58
+ if err != nil {
59
+ t.Fatal("error setting up listener", err)
60
+ }
61
+ go echoListen(listener.(*net.TCPListener))
62
+
63
+ p, err := setupPeer("11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33", "/ip4/127.0.0.1/tcp/1234")
64
+ if err != nil {
65
+ t.Fatal("error setting up peer", err)
66
+ }
67
+
68
+ c, err := Dial("tcp", p)
69
+ if err != nil {
70
+ t.Fatal("error dialing peer", err)
71
+ }
72
+
73
+ fmt.Println("sending")
74
+ c.Outgoing.MsgChan <- []byte("beep")
75
+ c.Outgoing.MsgChan <- []byte("boop")
76
+ out := <-c.Incoming.MsgChan
77
+ fmt.Println("recving", string(out))
78
+ if string(out) != "beep" {
79
+ t.Error("unexpected conn output")
80
+ }
81
+
82
+ out = <-c.Incoming.MsgChan
83
+ if string(out) != "boop" {
84
+ t.Error("unexpected conn output")
85
+ }
86
+
87
+ fmt.Println("closing")
88
+ c.Close()
89
+ listener.Close()
90
}
92
-
93
-
swarm/swarm.go
+135
-1
@@ -1,9 +1,143 @@
1
package swarm
2
3
import (
4
+ "fmt"
5
+ peer "github.com/jbenet/go-ipfs/peer"
6
+ "sync"
7
)
8
9
+type Message struct {
10
+ // To or from, depending on direction.
11
+ Peer *peer.Peer
12
+
13
+ // Opaque data
14
+ Data []byte
15
+}
16
+
17
+type Chan struct {
18
+ Outgoing chan Message
19
+ Incoming chan Message
20
+ Errors chan error
21
+ Close chan bool
22
+}
23
+
24
+func NewChan(bufsize int) *Chan {
25
+ return &Chan{
26
+ Outgoing: make(chan Message, bufsize),
27
+ Incoming: make(chan Message, bufsize),
28
+ Errors: make(chan error),
29
+ Close: make(chan bool, bufsize),
30
+ }
31
+}
32
+
33
type Swarm struct {
7
- Conns map[string]*Conn
34
+ Chan *Chan
35
+ conns ConnMap
36
+ connsLock sync.RWMutex
37
}
38
39
+func NewSwarm() *Swarm {
40
+ s := &Swarm{
41
+ Chan: NewChan(10),
42
+ conns: ConnMap{},
43
+ }
44
+ go s.fanOut()
45
+ return s
46
+}
47
+
48
+func (s *Swarm) Close() {
49
+ s.connsLock.RLock()
50
+ l := len(s.conns)
51
+ s.connsLock.RUnlock()
52
+
53
+ for i := 0; i < l; i++ {
54
+ s.Chan.Close <- true // fan ins
55
+ }
56
+ s.Chan.Close <- true // fan out
57
+ s.Chan.Close <- true // listener
58
+}
59
+
60
+func (s *Swarm) Dial(peer *peer.Peer) (*Conn, error) {
61
+ k := peer.Key()
62
+
63
+ // check if we already have an open connection first
64
+ s.connsLock.RLock()
65
+ conn, found := s.conns[k]
66
+ s.connsLock.RUnlock()
67
+ if found {
68
+ return conn, nil
69
+ }
70
+
71
+ // open connection to peer
72
+ conn, err := Dial("tcp", peer)
73
+ if err != nil {
74
+ return nil, err
75
+ }
76
+
77
+ // add to conns
78
+ s.connsLock.Lock()
79
+ s.conns[k] = conn
80
+ s.connsLock.Unlock()
81
+
82
+ // kick off reader goroutine
83
+ go s.fanIn(conn)
84
+ return conn, nil
85
+}
86
+
87
+// Handles the unwrapping + sending of messages to the right connection.
88
+func (s *Swarm) fanOut() {
89
+ for {
90
+ select {
91
+ case <-s.Chan.Close:
92
+ return // told to close.
93
+ case msg, ok := <-s.Chan.Outgoing:
94
+ if !ok {
95
+ return
96
+ }
97
+
98
+ s.connsLock.RLock()
99
+ conn, found := s.conns[msg.Peer.Key()]
100
+ s.connsLock.RUnlock()
101
+ if !found {
102
+ e := fmt.Errorf("Sent msg to peer without open conn: %v", msg.Peer)
103
+ s.Chan.Errors <- e
104
+ }
105
+
106
+ // queue it in the connection's buffer
107
+ conn.Outgoing.MsgChan <- msg.Data
108
+ }
109
+ }
110
+}
111
+
112
+// Handles the receiving + wrapping of messages, per conn.
113
+// Consider using reflect.Select with one goroutine instead of n.
114
+func (s *Swarm) fanIn(conn *Conn) {
115
+Loop:
116
+ for {
117
+ select {
118
+ case <-s.Chan.Close:
119
+ // close Conn.
120
+ conn.Close()
121
+ break Loop
122
+
123
+ case <-conn.Closed:
124
+ break Loop
125
+
126
+ case data, ok := <-conn.Incoming.MsgChan:
127
+ fmt.Println("got back data", data)
128
+ if !ok {
129
+ e := fmt.Errorf("Error retrieving from conn: %v", conn)
130
+ s.Chan.Errors <- e
131
+ break Loop
132
+ }
133
+
134
+ // wrap it for consumers.
135
+ msg := Message{Peer: conn.Peer, Data: data}
136
+ s.Chan.Incoming <- msg
137
+ }
138
+ }
139
+
140
+ s.connsLock.Lock()
141
+ delete(s.conns, conn.Peer.Key())
142
+ s.connsLock.Unlock()
143
+}
swarm/swarm_test.go
new
+109
@@ -0,0 +1,109 @@
1
+package swarm
2
+
3
+import (
4
+ "fmt"
5
+ peer "github.com/jbenet/go-ipfs/peer"
6
+ u "github.com/jbenet/go-ipfs/util"
7
+ msgio "github.com/jbenet/go-msgio"
8
+ "net"
9
+ "testing"
10
+)
11
+
12
+func pingListen(listener *net.TCPListener, peer *peer.Peer) {
13
+ for {
14
+ c, err := listener.Accept()
15
+ if err == nil {
16
+ fmt.Println("accepeted")
17
+ go pong(c, peer)
18
+ }
19
+ }
20
+}
21
+
22
+func pong(c net.Conn, peer *peer.Peer) {
23
+ mrw := msgio.NewReadWriter(c)
24
+ for {
25
+ data := make([]byte, 1024)
26
+ n, err := mrw.ReadMsg(data)
27
+ if err != nil {
28
+ fmt.Printf("error %v\n", err)
29
+ return
30
+ }
31
+ if string(data[:n]) != "ping" {
32
+ fmt.Printf("error: didn't receive ping: '%v'\n", data[:n])
33
+ return
34
+ }
35
+ err = mrw.WriteMsg([]byte("pong"))
36
+ if err != nil {
37
+ fmt.Printf("error %v\n", err)
38
+ return
39
+ }
40
+ fmt.Println("pong")
41
+ }
42
+}
43
+
44
+func TestSwarm(t *testing.T) {
45
+
46
+ swarm := NewSwarm()
47
+ peers := []*peer.Peer{}
48
+ listeners := []*net.Listener{}
49
+ peerNames := map[string]string{
50
+ "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a30": "/ip4/127.0.0.1/tcp/1234",
51
+ "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a31": "/ip4/127.0.0.1/tcp/2345",
52
+ "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a32": "/ip4/127.0.0.1/tcp/3456",
53
+ "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33": "/ip4/127.0.0.1/tcp/4567",
54
+ }
55
+
56
+ for k, n := range peerNames {
57
+ peer, err := setupPeer(k, n)
58
+ if err != nil {
59
+ t.Fatal("error setting up peer", err)
60
+ }
61
+ a := peer.NetAddress("tcp")
62
+ if a == nil {
63
+ t.Fatal("error setting up peer (addr is nil)", peer)
64
+ }
65
+ n, h, err := a.DialArgs()
66
+ if err != nil {
67
+ t.Fatal("error getting dial args from addr")
68
+ }
69
+ listener, err := net.Listen(n, h)
70
+ if err != nil {
71
+ t.Fatal("error setting up listener", err)
72
+ }
73
+ go pingListen(listener.(*net.TCPListener), peer)
74
+
75
+ _, err = swarm.Dial(peer)
76
+ if err != nil {
77
+ t.Fatal("error swarm dialing to peer", err)
78
+ }
79
+
80
+ // ok done, add it.
81
+ peers = append(peers, peer)
82
+ listeners = append(listeners, &listener)
83
+ }
84
+
85
+ for i, p := range peers {
86
+ swarm.Chan.Outgoing <- Message{Peer: p, Data: []byte("ping")}
87
+ fmt.Println("ping", i)
88
+ }
89
+
90
+ got := map[u.Key]bool{}
91
+ for _, _ = range peers {
92
+ msg := <-swarm.Chan.Incoming
93
+ fmt.Println("recving", string(msg.Data))
94
+ if string(msg.Data) != "pong" {
95
+ t.Error("unexpected conn output", msg.Data)
96
+ }
97
+ got[msg.Peer.Key()] = true
98
+ }
99
+
100
+ if len(peers) != len(got) {
101
+ t.Error("got less messages than sent")
102
+ }
103
+
104
+ fmt.Println("closing")
105
+ swarm.Close()
106
+ for _, listener := range listeners {
107
+ (*listener).(*net.TCPListener).Close()
108
+ }
109
+}