ipfs_impl.go 6.49 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"
Steven Allen's avatar
Steven Allen committed
22
	msgio "github.com/libp2p/go-msgio"
Jeromy's avatar
Jeromy committed
23
	ma "github.com/multiformats/go-multiaddr"
24 25
)

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

28 29
var sendMessageTimeout = time.Minute * 10

30
// NewFromIpfsHost returns a BitSwapNetwork supported by underlying IPFS host.
31 32 33 34 35 36
func NewFromIpfsHost(host host.Host, r routing.ContentRouting, opts ...NetOpt) BitSwapNetwork {
	s := Settings{}
	for _, opt := range opts {
		opt(&s)
	}

37
	bitswapNetwork := impl{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
38
		host:    host,
39
		routing: r,
40 41 42 43

		protocolBitswap:       s.ProtocolPrefix + ProtocolBitswap,
		protocolBitswapOne:    s.ProtocolPrefix + ProtocolBitswapOne,
		protocolBitswapNoVers: s.ProtocolPrefix + ProtocolBitswapNoVers,
44
	}
45
	return &bitswapNetwork
46 47
}

48 49
// impl transforms the ipfs network interface, which sends and receives
// NetMessage objects, into the bitswap network interface.
50
type impl struct {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
51
	host    host.Host
52
	routing routing.ContentRouting
53

54 55 56 57
	protocolBitswap       protocol.ID
	protocolBitswapOne    protocol.ID
	protocolBitswapNoVers protocol.ID

58 59
	// inbound messages from the network are forwarded to the receiver
	receiver Receiver
60

61
	stats Stats
62 63
}

Jeromy's avatar
Jeromy committed
64
type streamMessageSender struct {
65 66
	s     network.Stream
	bsnet *impl
Jeromy's avatar
Jeromy committed
67 68 69
}

func (s *streamMessageSender) Close() error {
Raúl Kripalani's avatar
Raúl Kripalani committed
70
	return helpers.FullClose(s.s)
Jeromy's avatar
Jeromy committed
71 72
}

73 74 75 76
func (s *streamMessageSender) Reset() error {
	return s.s.Reset()
}

77
func (s *streamMessageSender) SendMsg(ctx context.Context, msg bsmsg.BitSwapMessage) error {
78
	return s.bsnet.msgToStream(ctx, s.s, msg)
79 80
}

81
func (bsnet *impl) msgToStream(ctx context.Context, s network.Stream, msg bsmsg.BitSwapMessage) error {
82 83 84 85
	deadline := time.Now().Add(sendMessageTimeout)
	if dl, ok := ctx.Deadline(); ok {
		deadline = dl
	}
Bob Potter's avatar
Bob Potter committed
86

87 88 89 90
	if err := s.SetWriteDeadline(deadline); err != nil {
		log.Warningf("error setting deadline: %s", err)
	}

91
	switch s.Protocol() {
92
	case bsnet.protocolBitswap:
Bob Potter's avatar
Bob Potter committed
93
		if err := msg.ToNetV1(s); err != nil {
94 95 96
			log.Debugf("error: %s", err)
			return err
		}
97
	case bsnet.protocolBitswapOne, bsnet.protocolBitswapNoVers:
Bob Potter's avatar
Bob Potter committed
98
		if err := msg.ToNetV0(s); err != nil {
99 100 101 102 103 104
			log.Debugf("error: %s", err)
			return err
		}
	default:
		return fmt.Errorf("unrecognized protocol on remote: %s", s.Protocol())
	}
105 106 107 108

	if err := s.SetWriteDeadline(time.Time{}); err != nil {
		log.Warningf("error resetting deadline: %s", err)
	}
109
	return nil
Jeromy's avatar
Jeromy committed
110 111 112 113 114 115 116 117
}

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

118
	return &streamMessageSender{s: s, bsnet: bsnet}, nil
Jeromy's avatar
Jeromy committed
119 120
}

Raúl Kripalani's avatar
Raúl Kripalani committed
121
func (bsnet *impl) newStreamToPeer(ctx context.Context, p peer.ID) (network.Stream, error) {
122
	return bsnet.host.NewStream(ctx, p, bsnet.protocolBitswap, bsnet.protocolBitswapOne, bsnet.protocolBitswapNoVers)
123 124 125 126 127 128 129 130
}

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

	s, err := bsnet.newStreamToPeer(ctx, p)
131 132 133
	if err != nil {
		return err
	}
134

135
	if err = bsnet.msgToStream(ctx, s, outgoing); err != nil {
Steven Allen's avatar
Steven Allen committed
136
		_ = s.Reset()
Steven Allen's avatar
Steven Allen committed
137
		return err
138
	}
139 140
	atomic.AddUint64(&bsnet.stats.MessagesSent, 1)

141
	// TODO(https://github.com/libp2p/go-libp2p-net/issues/28): Avoid this goroutine.
Steven Allen's avatar
Steven Allen committed
142
	//nolint
Raúl Kripalani's avatar
Raúl Kripalani committed
143
	go helpers.AwaitEOF(s)
144 145
	return s.Close()

146 147
}

148 149
func (bsnet *impl) SetDelegate(r Receiver) {
	bsnet.receiver = r
150 151 152
	bsnet.host.SetStreamHandler(bsnet.protocolBitswap, bsnet.handleNewStream)
	bsnet.host.SetStreamHandler(bsnet.protocolBitswapOne, bsnet.handleNewStream)
	bsnet.host.SetStreamHandler(bsnet.protocolBitswapNoVers, bsnet.handleNewStream)
hannahhoward's avatar
hannahhoward committed
153 154 155
	bsnet.host.Network().Notify((*netNotifiee)(bsnet))
	// TODO: StopNotify.

156
}
157

158
func (bsnet *impl) ConnectTo(ctx context.Context, p peer.ID) error {
Raúl Kripalani's avatar
Raúl Kripalani committed
159
	return bsnet.host.Connect(ctx, peer.AddrInfo{ID: p})
160 161
}

162
// FindProvidersAsync returns a channel of providers for the given key.
163
func (bsnet *impl) FindProvidersAsync(ctx context.Context, k cid.Cid, max int) <-chan peer.ID {
164
	out := make(chan peer.ID, max)
165 166
	go func() {
		defer close(out)
167
		providers := bsnet.routing.FindProvidersAsync(ctx, k, max)
168
		for info := range providers {
169 170
			if info.ID == bsnet.host.ID() {
				continue // ignore self as provider
171
			}
Raúl Kripalani's avatar
Raúl Kripalani committed
172
			bsnet.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.TempAddrTTL)
173 174
			select {
			case <-ctx.Done():
175
				return
176 177 178 179 180
			case out <- info.ID:
			}
		}
	}()
	return out
181 182 183
}

// Provide provides the key to the network
184
func (bsnet *impl) Provide(ctx context.Context, k cid.Cid) error {
185
	return bsnet.routing.Provide(ctx, k, true)
186 187
}

188
// handleNewStream receives a new stream from the network.
Raúl Kripalani's avatar
Raúl Kripalani committed
189
func (bsnet *impl) handleNewStream(s network.Stream) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
190
	defer s.Close()
191 192

	if bsnet.receiver == nil {
Steven Allen's avatar
Steven Allen committed
193
		_ = s.Reset()
194 195 196
		return
	}

Steven Allen's avatar
Steven Allen committed
197
	reader := msgio.NewVarintReaderSize(s, network.MessageSizeMax)
198
	for {
Steven Allen's avatar
Steven Allen committed
199
		received, err := bsmsg.FromMsgReader(reader)
200
		if err != nil {
Jeromy's avatar
Jeromy committed
201
			if err != io.EOF {
Steven Allen's avatar
Steven Allen committed
202
				_ = s.Reset()
Jeromy's avatar
Jeromy committed
203 204 205
				go bsnet.receiver.ReceiveError(err)
				log.Debugf("bitswap net handleNewStream from %s error: %s", s.Conn().RemotePeer(), err)
			}
206 207
			return
		}
208

209 210 211 212
		p := s.Conn().RemotePeer()
		ctx := context.Background()
		log.Debugf("bitswap net handleNewStream from %s", s.Conn().RemotePeer())
		bsnet.receiver.ReceiveMessage(ctx, p, received)
213
		atomic.AddUint64(&bsnet.stats.MessagesRecvd, 1)
214
	}
215
}
216

Raúl Kripalani's avatar
Raúl Kripalani committed
217
func (bsnet *impl) ConnectionManager() connmgr.ConnManager {
218 219 220
	return bsnet.host.ConnManager()
}

221 222
func (bsnet *impl) Stats() Stats {
	return Stats{
223 224 225 226 227
		MessagesRecvd: atomic.LoadUint64(&bsnet.stats.MessagesRecvd),
		MessagesSent:  atomic.LoadUint64(&bsnet.stats.MessagesSent),
	}
}

228 229 230 231 232 233
type netNotifiee impl

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

Raúl Kripalani's avatar
Raúl Kripalani committed
234
func (nn *netNotifiee) Connected(n network.Network, v network.Conn) {
235 236 237
	nn.impl().receiver.PeerConnected(v.RemotePeer())
}

Raúl Kripalani's avatar
Raúl Kripalani committed
238
func (nn *netNotifiee) Disconnected(n network.Network, v network.Conn) {
239 240 241
	nn.impl().receiver.PeerDisconnected(v.RemotePeer())
}

Raúl Kripalani's avatar
Raúl Kripalani committed
242 243 244 245
func (nn *netNotifiee) OpenedStream(n network.Network, v network.Stream) {}
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)    {}