protocol and muxer pkg
Juan Batiz-Benet committed
Dec 29, 2014 at 19:34 UTC
e0b9a368b8d1de8689127b2988143cff8d767364
3 files changed
+251
p2p/protocol/mux/mux.go
new
+143
@@ -0,0 +1,143 @@
1
+package mux
2
+
3
+import (
4
+ "fmt"
5
+ "io"
6
+ "sync"
7
+
8
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
+
10
+ inet "github.com/jbenet/go-ipfs/p2p/net"
11
+ protocol "github.com/jbenet/go-ipfs/p2p/protocol"
12
+ eventlog "github.com/jbenet/go-ipfs/util/eventlog"
13
+ lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
14
+)
15
+
16
+var log = eventlog.Logger("net/mux")
17
+
18
+type StreamHandlerMap map[protocol.ID]inet.StreamHandler
19
+
20
+// Mux provides simple stream multixplexing.
21
+// It helps you precisely when:
22
+// * You have many streams
23
+// * You have function handlers
24
+//
25
+// It contains the handlers for each protocol accepted.
26
+// It dispatches handlers for streams opened by remote peers.
27
+//
28
+// WARNING: this datastructure IS NOT threadsafe.
29
+// do not modify it once the network is using it.
30
+type Mux struct {
31
+ Default inet.StreamHandler // handles unknown protocols.
32
+ Handlers StreamHandlerMap
33
+
34
+ sync.RWMutex
35
+}
36
+
37
+// Protocols returns the list of protocols this muxer has handlers for
38
+func (m *Mux) Protocols() []protocol.ID {
39
+ m.RLock()
40
+ l := make([]protocol.ID, 0, len(m.Handlers))
41
+ for p := range m.Handlers {
42
+ l = append(l, p)
43
+ }
44
+ m.RUnlock()
45
+ return l
46
+}
47
+
48
+// readHeader reads the stream and returns the next Handler function
49
+// according to the muxer encoding.
50
+func (m *Mux) readHeader(s io.Reader) (protocol.ID, inet.StreamHandler, error) {
51
+ // log.Error("ReadProtocolHeader")
52
+ p, err := protocol.ReadHeader(s)
53
+ if err != nil {
54
+ return "", nil, err
55
+ }
56
+
57
+ // log.Debug("readHeader got:", p)
58
+ m.RLock()
59
+ h, found := m.Handlers[p]
60
+ m.RUnlock()
61
+
62
+ switch {
63
+ case !found && m.Default != nil:
64
+ return p, m.Default, nil
65
+ case !found && m.Default == nil:
66
+ return p, nil, fmt.Errorf("%s no handler with name: %s (%d)", m, p, len(p))
67
+ default:
68
+ return p, h, nil
69
+ }
70
+}
71
+
72
+// String returns the muxer's printing representation
73
+func (m *Mux) String() string {
74
+ m.RLock()
75
+ defer m.RUnlock()
76
+ return fmt.Sprintf("<Muxer %p %d>", m, len(m.Handlers))
77
+}
78
+
79
+// SetHandler sets the protocol handler on the Network's Muxer.
80
+// This operation is threadsafe.
81
+func (m *Mux) SetHandler(p protocol.ID, h inet.StreamHandler) {
82
+ log.Debugf("%s setting handler for protocol: %s (%d)", m, p, len(p))
83
+ m.Lock()
84
+ m.Handlers[p] = h
85
+ m.Unlock()
86
+}
87
+
88
+// Handle reads the next name off the Stream, and calls a handler function
89
+// This is done in its own goroutine, to avoid blocking the caller.
90
+func (m *Mux) Handle(s inet.Stream) {
91
+
92
+ // Flow control and backpressure of Opening streams is broken.
93
+ // I believe that spdystream has one set of workers that both send
94
+ // data AND accept new streams (as it's just more data). there
95
+ // is a problem where if the new stream handlers want to throttle,
96
+ // they also eliminate the ability to read/write data, which makes
97
+ // forward-progress impossible. Thus, throttling this function is
98
+ // -- at this moment -- not the solution. Either spdystream must
99
+ // change, or we must throttle another way.
100
+ //
101
+ // In light of this, we use a goroutine for now (otherwise the
102
+ // spdy worker totally blocks, and we can't even read the protocol
103
+ // header). The better route in the future is to use a worker pool.
104
+ go m.HandleSync(s)
105
+}
106
+
107
+// HandleSync reads the next name off the Stream, and calls a handler function
108
+// This is done synchronously. The handler function will return before
109
+// HandleSync returns.
110
+func (m *Mux) HandleSync(s inet.Stream) {
111
+ ctx := context.Background()
112
+
113
+ name, handler, err := m.readHeader(s)
114
+ if err != nil {
115
+ err = fmt.Errorf("protocol mux error: %s", err)
116
+ log.Error(err)
117
+ log.Event(ctx, "muxError", lgbl.Error(err))
118
+ return
119
+ }
120
+
121
+ log.Infof("muxer handle protocol: %s", name)
122
+ log.Event(ctx, "muxHandle", eventlog.Metadata{"protocol": name})
123
+ handler(s)
124
+}
125
+
126
+// ReadLengthPrefix reads the name from Reader with a length-byte-prefix.
127
+func ReadLengthPrefix(r io.Reader) (string, error) {
128
+ // c-string identifier
129
+ // the first byte is our length
130
+ l := make([]byte, 1)
131
+ if _, err := io.ReadFull(r, l); err != nil {
132
+ return "", err
133
+ }
134
+ length := int(l[0])
135
+
136
+ // the next are our identifier
137
+ name := make([]byte, length)
138
+ if _, err := io.ReadFull(r, name); err != nil {
139
+ return "", err
140
+ }
141
+
142
+ return string(name), nil
143
+}
p2p/protocol/mux/mux_test.go
new
+68
@@ -0,0 +1,68 @@
1
+package mux
2
+
3
+import (
4
+ "bytes"
5
+ "testing"
6
+
7
+ inet "github.com/jbenet/go-ipfs/p2p/net"
8
+ protocol "github.com/jbenet/go-ipfs/p2p/protocol"
9
+)
10
+
11
+var testCases = map[string]string{
12
+ "/bitswap": "\u0009/bitswap\n",
13
+ "/dht": "\u0005/dht\n",
14
+ "/ipfs": "\u0006/ipfs\n",
15
+ "/ipfs/dksnafkasnfkdajfkdajfdsjadosiaaodj": ")/ipfs/dksnafkasnfkdajfkdajfdsjadosiaaodj\n",
16
+}
17
+
18
+func TestWrite(t *testing.T) {
19
+ for k, v := range testCases {
20
+ var buf bytes.Buffer
21
+ if err := protocol.WriteHeader(&buf, protocol.ID(k)); err != nil {
22
+ t.Fatal(err)
23
+ }
24
+
25
+ v2 := buf.Bytes()
26
+ if !bytes.Equal(v2, []byte(v)) {
27
+ t.Errorf("failed: %s - %v != %v", k, []byte(v), v2)
28
+ }
29
+ }
30
+}
31
+
32
+func TestHandler(t *testing.T) {
33
+
34
+ outs := make(chan string, 10)
35
+
36
+ h := func(n string) func(s inet.Stream) {
37
+ return func(s inet.Stream) {
38
+ outs <- n
39
+ }
40
+ }
41
+
42
+ m := Mux{Handlers: StreamHandlerMap{}}
43
+ m.Default = h("default")
44
+ m.Handlers["/dht"] = h("bitswap")
45
+ // m.Handlers["/ipfs"] = h("bitswap") // default!
46
+ m.Handlers["/bitswap"] = h("bitswap")
47
+ m.Handlers["/ipfs/dksnafkasnfkdajfkdajfdsjadosiaaodj"] = h("bitswap")
48
+
49
+ for k, v := range testCases {
50
+ var buf bytes.Buffer
51
+ if _, err := buf.Write([]byte(v)); err != nil {
52
+ t.Error(err)
53
+ continue
54
+ }
55
+
56
+ name, err := protocol.ReadHeader(&buf)
57
+ if err != nil {
58
+ t.Error(err)
59
+ continue
60
+ }
61
+
62
+ if name != protocol.ID(k) {
63
+ t.Errorf("name mismatch: %s != %s", k, name)
64
+ continue
65
+ }
66
+ }
67
+
68
+}
p2p/protocol/protocol.go
new
+40
@@ -0,0 +1,40 @@
1
+package protocol
2
+
3
+import (
4
+ "io"
5
+
6
+ msgio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
7
+)
8
+
9
+// ID is an identifier used to write protocol headers in streams.
10
+type ID string
11
+
12
+// These are reserved protocol.IDs.
13
+const (
14
+ TestingID ID = "/p2p/_testing"
15
+)
16
+
17
+// WriteHeader writes a protocol.ID header to an io.Writer. This is so
18
+// multiple protocols can be multiplexed on top of the same transport.
19
+//
20
+// We use go-msgio varint encoding:
21
+// <varint length><string name>\n
22
+// (the varint includes the \n)
23
+func WriteHeader(w io.Writer, id ID) error {
24
+ vw := msgio.NewVarintWriter(w)
25
+ s := string(id) + "\n" // add \n
26
+ return vw.WriteMsg([]byte(s))
27
+}
28
+
29
+// ReadHeader reads a protocol.ID header from an io.Reader. This is so
30
+// multiple protocols can be multiplexed on top of the same transport.
31
+// See WriteHeader.
32
+func ReadHeader(r io.Reader) (ID, error) {
33
+ vr := msgio.NewVarintReader(r)
34
+ msg, err := vr.ReadMsg()
35
+ if err != nil {
36
+ return ID(""), err
37
+ }
38
+ msg = msg[:len(msg)-1] // remove \n
39
+ return ID(msg), nil
40
+}