ipfs_impl.go 7.93 KB
Newer Older
1 2 3
package network

import (
4
	"context"
5
	"fmt"
Jeromy's avatar
Jeromy committed
6
	"io"
7
	"sync/atomic"
8
	"time"
Jeromy's avatar
Jeromy committed
9

Jeromy's avatar
Jeromy committed
10 11 12 13
	bsmsg "github.com/ipfs/go-bitswap/message"

	cid "github.com/ipfs/go-cid"
	logging "github.com/ipfs/go-log"
Raúl Kripalani's avatar
Raúl Kripalani committed
14
	"github.com/libp2p/go-libp2p-core/connmgr"
15
	"github.com/libp2p/go-libp2p-core/helpers"
Raúl Kripalani's avatar
Raúl Kripalani committed
16 17 18 19
	"github.com/libp2p/go-libp2p-core/host"
	"github.com/libp2p/go-libp2p-core/network"
	"github.com/libp2p/go-libp2p-core/peer"
	peerstore "github.com/libp2p/go-libp2p-core/peerstore"
20
	"github.com/libp2p/go-libp2p-core/protocol"
Raúl Kripalani's avatar
Raúl Kripalani committed
21
	"github.com/libp2p/go-libp2p-core/routing"
22
	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
Steven Allen's avatar
Steven Allen committed
23
	msgio "github.com/libp2p/go-msgio"
Jeromy's avatar
Jeromy committed
24
	ma "github.com/multiformats/go-multiaddr"
25 26
)

Jeromy's avatar
Jeromy committed
27
var log = logging.Logger("bitswap_network")
Jeromy's avatar
Jeromy committed
28

29 30
var sendMessageTimeout = time.Minute * 10

31
// NewFromIpfsHost returns a BitSwapNetwork supported by underlying IPFS host.
32
func NewFromIpfsHost(host host.Host, r routing.ContentRouting, opts ...NetOpt) BitSwapNetwork {
dirkmc's avatar
dirkmc committed
33
	s := processSettings(opts...)
34

35
	bitswapNetwork := impl{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
36
		host:    host,
37
		routing: r,
38

dirkmc's avatar
dirkmc committed
39 40 41 42 43 44
		protocolBitswapNoVers:  s.ProtocolPrefix + ProtocolBitswapNoVers,
		protocolBitswapOneZero: s.ProtocolPrefix + ProtocolBitswapOneZero,
		protocolBitswapOneOne:  s.ProtocolPrefix + ProtocolBitswapOneOne,
		protocolBitswap:        s.ProtocolPrefix + ProtocolBitswap,

		supportedProtocols: s.SupportedProtocols,
45
	}
46
	return &bitswapNetwork
47 48
}

dirkmc's avatar
dirkmc committed
49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66
func processSettings(opts ...NetOpt) Settings {
	s := Settings{
		SupportedProtocols: []protocol.ID{
			ProtocolBitswap,
			ProtocolBitswapOneOne,
			ProtocolBitswapOneZero,
			ProtocolBitswapNoVers,
		},
	}
	for _, opt := range opts {
		opt(&s)
	}
	for i, proto := range s.SupportedProtocols {
		s.SupportedProtocols[i] = s.ProtocolPrefix + proto
	}
	return s
}

67 68
// impl transforms the ipfs network interface, which sends and receives
// NetMessage objects, into the bitswap network interface.
69
type impl struct {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
70
	host    host.Host
71
	routing routing.ContentRouting
72

dirkmc's avatar
dirkmc committed
73 74 75 76 77 78
	protocolBitswapNoVers  protocol.ID
	protocolBitswapOneZero protocol.ID
	protocolBitswapOneOne  protocol.ID
	protocolBitswap        protocol.ID

	supportedProtocols []protocol.ID
79

80 81
	// inbound messages from the network are forwarded to the receiver
	receiver Receiver
82

83
	stats Stats
84 85
}

Jeromy's avatar
Jeromy committed
86
type streamMessageSender struct {
87 88
	s     network.Stream
	bsnet *impl
Jeromy's avatar
Jeromy committed
89 90 91
}

func (s *streamMessageSender) Close() error {
Raúl Kripalani's avatar
Raúl Kripalani committed
92
	return helpers.FullClose(s.s)
Jeromy's avatar
Jeromy committed
93 94
}

95 96 97 98
func (s *streamMessageSender) Reset() error {
	return s.s.Reset()
}

99
func (s *streamMessageSender) SendMsg(ctx context.Context, msg bsmsg.BitSwapMessage) error {
100
	return s.bsnet.msgToStream(ctx, s.s, msg)
101 102
}

dirkmc's avatar
dirkmc committed
103 104 105 106 107 108 109 110
func (s *streamMessageSender) SupportsHave() bool {
	return s.bsnet.SupportsHave(s.s.Protocol())
}

func (bsnet *impl) Self() peer.ID {
	return bsnet.host.ID()
}

111 112 113 114 115 116 117 118 119 120 121
func (bsnet *impl) Ping(ctx context.Context, p peer.ID) ping.Result {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	res := <-ping.Ping(ctx, bsnet.host, p)
	return res
}

func (bsnet *impl) Latency(p peer.ID) time.Duration {
	return bsnet.host.Peerstore().LatencyEWMA(p)
}

dirkmc's avatar
dirkmc committed
122 123 124 125 126 127 128 129 130
// Indicates whether the given protocol supports HAVE / DONT_HAVE messages
func (bsnet *impl) SupportsHave(proto protocol.ID) bool {
	switch proto {
	case bsnet.protocolBitswapOneOne, bsnet.protocolBitswapOneZero, bsnet.protocolBitswapNoVers:
		return false
	}
	return true
}

131
func (bsnet *impl) msgToStream(ctx context.Context, s network.Stream, msg bsmsg.BitSwapMessage) error {
132 133 134 135
	deadline := time.Now().Add(sendMessageTimeout)
	if dl, ok := ctx.Deadline(); ok {
		deadline = dl
	}
Bob Potter's avatar
Bob Potter committed
136

137
	if err := s.SetWriteDeadline(deadline); err != nil {
138
		log.Warnf("error setting deadline: %s", err)
139 140
	}

dirkmc's avatar
dirkmc committed
141 142 143
	// Older Bitswap versions use a slightly different wire format so we need
	// to convert the message to the appropriate format depending on the remote
	// peer's Bitswap version.
144
	switch s.Protocol() {
dirkmc's avatar
dirkmc committed
145
	case bsnet.protocolBitswapOneOne, bsnet.protocolBitswap:
Bob Potter's avatar
Bob Potter committed
146
		if err := msg.ToNetV1(s); err != nil {
147 148 149
			log.Debugf("error: %s", err)
			return err
		}
dirkmc's avatar
dirkmc committed
150
	case bsnet.protocolBitswapOneZero, bsnet.protocolBitswapNoVers:
Bob Potter's avatar
Bob Potter committed
151
		if err := msg.ToNetV0(s); err != nil {
152 153 154 155 156 157
			log.Debugf("error: %s", err)
			return err
		}
	default:
		return fmt.Errorf("unrecognized protocol on remote: %s", s.Protocol())
	}
158 159

	if err := s.SetWriteDeadline(time.Time{}); err != nil {
160
		log.Warnf("error resetting deadline: %s", err)
161
	}
162
	return nil
Jeromy's avatar
Jeromy committed
163 164 165 166 167 168 169 170
}

func (bsnet *impl) NewMessageSender(ctx context.Context, p peer.ID) (MessageSender, error) {
	s, err := bsnet.newStreamToPeer(ctx, p)
	if err != nil {
		return nil, err
	}

171
	return &streamMessageSender{s: s, bsnet: bsnet}, nil
Jeromy's avatar
Jeromy committed
172 173
}

Raúl Kripalani's avatar
Raúl Kripalani committed
174
func (bsnet *impl) newStreamToPeer(ctx context.Context, p peer.ID) (network.Stream, error) {
dirkmc's avatar
dirkmc committed
175
	return bsnet.host.NewStream(ctx, p, bsnet.supportedProtocols...)
176 177 178 179 180 181 182 183
}

func (bsnet *impl) SendMessage(
	ctx context.Context,
	p peer.ID,
	outgoing bsmsg.BitSwapMessage) error {

	s, err := bsnet.newStreamToPeer(ctx, p)
184 185 186
	if err != nil {
		return err
	}
187

188
	if err = bsnet.msgToStream(ctx, s, outgoing); err != nil {
Steven Allen's avatar
Steven Allen committed
189
		_ = s.Reset()
Steven Allen's avatar
Steven Allen committed
190
		return err
191
	}
192 193
	atomic.AddUint64(&bsnet.stats.MessagesSent, 1)

194
	// TODO(https://github.com/libp2p/go-libp2p-net/issues/28): Avoid this goroutine.
Steven Allen's avatar
Steven Allen committed
195
	//nolint
Raúl Kripalani's avatar
Raúl Kripalani committed
196
	go helpers.AwaitEOF(s)
197 198
	return s.Close()

199 200
}

201 202
func (bsnet *impl) SetDelegate(r Receiver) {
	bsnet.receiver = r
dirkmc's avatar
dirkmc committed
203 204 205
	for _, proto := range bsnet.supportedProtocols {
		bsnet.host.SetStreamHandler(proto, bsnet.handleNewStream)
	}
hannahhoward's avatar
hannahhoward committed
206 207 208
	bsnet.host.Network().Notify((*netNotifiee)(bsnet))
	// TODO: StopNotify.

209
}
210

211
func (bsnet *impl) ConnectTo(ctx context.Context, p peer.ID) error {
Raúl Kripalani's avatar
Raúl Kripalani committed
212
	return bsnet.host.Connect(ctx, peer.AddrInfo{ID: p})
213 214
}

dirkmc's avatar
dirkmc committed
215 216 217 218
func (bsnet *impl) DisconnectFrom(ctx context.Context, p peer.ID) error {
	panic("Not implemented: DisconnectFrom() is only used by tests")
}

219
// FindProvidersAsync returns a channel of providers for the given key.
220
func (bsnet *impl) FindProvidersAsync(ctx context.Context, k cid.Cid, max int) <-chan peer.ID {
221
	out := make(chan peer.ID, max)
222 223
	go func() {
		defer close(out)
224
		providers := bsnet.routing.FindProvidersAsync(ctx, k, max)
225
		for info := range providers {
226 227
			if info.ID == bsnet.host.ID() {
				continue // ignore self as provider
228
			}
Raúl Kripalani's avatar
Raúl Kripalani committed
229
			bsnet.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.TempAddrTTL)
230 231
			select {
			case <-ctx.Done():
232
				return
233 234 235 236 237
			case out <- info.ID:
			}
		}
	}()
	return out
238 239 240
}

// Provide provides the key to the network
241
func (bsnet *impl) Provide(ctx context.Context, k cid.Cid) error {
242
	return bsnet.routing.Provide(ctx, k, true)
243 244
}

245
// handleNewStream receives a new stream from the network.
Raúl Kripalani's avatar
Raúl Kripalani committed
246
func (bsnet *impl) handleNewStream(s network.Stream) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
247
	defer s.Close()
248 249

	if bsnet.receiver == nil {
Steven Allen's avatar
Steven Allen committed
250
		_ = s.Reset()
251 252 253
		return
	}

Steven Allen's avatar
Steven Allen committed
254
	reader := msgio.NewVarintReaderSize(s, network.MessageSizeMax)
255
	for {
Steven Allen's avatar
Steven Allen committed
256
		received, err := bsmsg.FromMsgReader(reader)
257
		if err != nil {
Jeromy's avatar
Jeromy committed
258
			if err != io.EOF {
Steven Allen's avatar
Steven Allen committed
259
				_ = s.Reset()
Jeromy's avatar
Jeromy committed
260 261 262
				go bsnet.receiver.ReceiveError(err)
				log.Debugf("bitswap net handleNewStream from %s error: %s", s.Conn().RemotePeer(), err)
			}
263 264
			return
		}
265

266 267 268 269
		p := s.Conn().RemotePeer()
		ctx := context.Background()
		log.Debugf("bitswap net handleNewStream from %s", s.Conn().RemotePeer())
		bsnet.receiver.ReceiveMessage(ctx, p, received)
270
		atomic.AddUint64(&bsnet.stats.MessagesRecvd, 1)
271
	}
272
}
273

Raúl Kripalani's avatar
Raúl Kripalani committed
274
func (bsnet *impl) ConnectionManager() connmgr.ConnManager {
275 276 277
	return bsnet.host.ConnManager()
}

278 279
func (bsnet *impl) Stats() Stats {
	return Stats{
280 281 282 283 284
		MessagesRecvd: atomic.LoadUint64(&bsnet.stats.MessagesRecvd),
		MessagesSent:  atomic.LoadUint64(&bsnet.stats.MessagesSent),
	}
}

285 286 287 288 289 290
type netNotifiee impl

func (nn *netNotifiee) impl() *impl {
	return (*impl)(nn)
}

Raúl Kripalani's avatar
Raúl Kripalani committed
291
func (nn *netNotifiee) Connected(n network.Network, v network.Conn) {
292 293
	nn.impl().receiver.PeerConnected(v.RemotePeer())
}
Raúl Kripalani's avatar
Raúl Kripalani committed
294
func (nn *netNotifiee) Disconnected(n network.Network, v network.Conn) {
295 296
	nn.impl().receiver.PeerDisconnected(v.RemotePeer())
}
dirkmc's avatar
dirkmc committed
297
func (nn *netNotifiee) OpenedStream(n network.Network, s network.Stream) {}
Raúl Kripalani's avatar
Raúl Kripalani committed
298 299 300
func (nn *netNotifiee) ClosedStream(n network.Network, v network.Stream) {}
func (nn *netNotifiee) Listen(n network.Network, a ma.Multiaddr)         {}
func (nn *netNotifiee) ListenClose(n network.Network, a ma.Multiaddr)    {}