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

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

Jeromy's avatar
Jeromy committed
11 12 13 14 15 16 17 18 19 20 21 22
	bsmsg "github.com/ipfs/go-bitswap/message"

	ggio "github.com/gogo/protobuf/io"
	cid "github.com/ipfs/go-cid"
	logging "github.com/ipfs/go-log"
	host "github.com/libp2p/go-libp2p-host"
	ifconnmgr "github.com/libp2p/go-libp2p-interface-connmgr"
	inet "github.com/libp2p/go-libp2p-net"
	peer "github.com/libp2p/go-libp2p-peer"
	pstore "github.com/libp2p/go-libp2p-peerstore"
	routing "github.com/libp2p/go-libp2p-routing"
	ma "github.com/multiformats/go-multiaddr"
23 24
)

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

27 28
var sendMessageTimeout = time.Minute * 10

29
// NewFromIpfsHost returns a BitSwapNetwork supported by underlying IPFS host.
30
func NewFromIpfsHost(host host.Host, r routing.ContentRouting) BitSwapNetwork {
31
	bitswapNetwork := impl{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
32
		host:    host,
33
		routing: r,
34
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
35
	host.SetStreamHandler(ProtocolBitswap, bitswapNetwork.handleNewStream)
36 37
	host.SetStreamHandler(ProtocolBitswapOne, bitswapNetwork.handleNewStream)
	host.SetStreamHandler(ProtocolBitswapNoVers, bitswapNetwork.handleNewStream)
38 39 40
	host.Network().Notify((*netNotifiee)(&bitswapNetwork))
	// TODO: StopNotify.

41
	return &bitswapNetwork
42 43
}

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

	// inbound messages from the network are forwarded to the receiver
	receiver Receiver
52 53

	stats NetworkStats
54 55
}

Jeromy's avatar
Jeromy committed
56 57 58 59 60
type streamMessageSender struct {
	s inet.Stream
}

func (s *streamMessageSender) Close() error {
Steven Allen's avatar
Steven Allen committed
61
	return inet.FullClose(s.s)
Jeromy's avatar
Jeromy committed
62 63
}

64 65 66 67
func (s *streamMessageSender) Reset() error {
	return s.s.Reset()
}

68 69
func (s *streamMessageSender) SendMsg(ctx context.Context, msg bsmsg.BitSwapMessage) error {
	return msgToStream(ctx, s.s, msg)
70 71
}

72 73 74 75 76 77 78 79 80
func msgToStream(ctx context.Context, s inet.Stream, msg bsmsg.BitSwapMessage) error {
	deadline := time.Now().Add(sendMessageTimeout)
	if dl, ok := ctx.Deadline(); ok {
		deadline = dl
	}
	if err := s.SetWriteDeadline(deadline); err != nil {
		log.Warningf("error setting deadline: %s", err)
	}

Steven Allen's avatar
Steven Allen committed
81 82
	w := bufio.NewWriter(s)

83 84
	switch s.Protocol() {
	case ProtocolBitswap:
Steven Allen's avatar
Steven Allen committed
85
		if err := msg.ToNetV1(w); err != nil {
86 87 88 89
			log.Debugf("error: %s", err)
			return err
		}
	case ProtocolBitswapOne, ProtocolBitswapNoVers:
Steven Allen's avatar
Steven Allen committed
90
		if err := msg.ToNetV0(w); err != nil {
91 92 93 94 95 96
			log.Debugf("error: %s", err)
			return err
		}
	default:
		return fmt.Errorf("unrecognized protocol on remote: %s", s.Protocol())
	}
97

Steven Allen's avatar
Steven Allen committed
98 99 100 101 102
	if err := w.Flush(); err != nil {
		log.Debugf("error: %s", err)
		return err
	}

103 104 105
	if err := s.SetWriteDeadline(time.Time{}); err != nil {
		log.Warningf("error resetting deadline: %s", err)
	}
106
	return nil
Jeromy's avatar
Jeromy committed
107 108 109 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
	}

	return &streamMessageSender{s: s}, nil
}

118
func (bsnet *impl) newStreamToPeer(ctx context.Context, p peer.ID) (inet.Stream, error) {
119
	return bsnet.host.NewStream(ctx, p, ProtocolBitswap, ProtocolBitswapOne, ProtocolBitswapNoVers)
120 121 122 123 124 125 126 127
}

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

	s, err := bsnet.newStreamToPeer(ctx, p)
128 129 130
	if err != nil {
		return err
	}
131

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

138 139 140 141
	// TODO(https://github.com/libp2p/go-libp2p-net/issues/28): Avoid this goroutine.
	go inet.AwaitEOF(s)
	return s.Close()

142 143
}

144 145
func (bsnet *impl) SetDelegate(r Receiver) {
	bsnet.receiver = r
146
}
147

148
func (bsnet *impl) ConnectTo(ctx context.Context, p peer.ID) error {
Jeromy's avatar
Jeromy committed
149
	return bsnet.host.Connect(ctx, pstore.PeerInfo{ID: p})
150 151
}

152
// FindProvidersAsync returns a channel of providers for the given key.
153
func (bsnet *impl) FindProvidersAsync(ctx context.Context, k cid.Cid, max int) <-chan peer.ID {
154 155 156 157 158 159 160 161 162

	// Since routing queries are expensive, give bitswap the peers to which we
	// have open connections. Note that this may cause issues if bitswap starts
	// precisely tracking which peers provide certain keys. This optimization
	// would be misleading. In the long run, this may not be the most
	// appropriate place for this optimization, but it won't cause any harm in
	// the short term.
	connectedPeers := bsnet.host.Network().Peers()
	out := make(chan peer.ID, len(connectedPeers)) // just enough buffer for these connectedPeers
163
	for _, id := range connectedPeers {
164 165 166
		if id == bsnet.host.ID() {
			continue // ignore self as provider
		}
167 168 169
		out <- id
	}

170 171
	go func() {
		defer close(out)
172
		providers := bsnet.routing.FindProvidersAsync(ctx, k, max)
173
		for info := range providers {
174 175
			if info.ID == bsnet.host.ID() {
				continue // ignore self as provider
176
			}
Jeromy's avatar
Jeromy committed
177
			bsnet.host.Peerstore().AddAddrs(info.ID, info.Addrs, pstore.TempAddrTTL)
178 179
			select {
			case <-ctx.Done():
180
				return
181 182 183 184 185
			case out <- info.ID:
			}
		}
	}()
	return out
186 187 188
}

// Provide provides the key to the network
189
func (bsnet *impl) Provide(ctx context.Context, k cid.Cid) error {
190
	return bsnet.routing.Provide(ctx, k, true)
191 192
}

193 194
// handleNewStream receives a new stream from the network.
func (bsnet *impl) handleNewStream(s inet.Stream) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
195
	defer s.Close()
196 197

	if bsnet.receiver == nil {
198
		s.Reset()
199 200 201
		return
	}

Jeromy's avatar
Jeromy committed
202
	reader := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
203
	for {
Jeromy's avatar
Jeromy committed
204
		received, err := bsmsg.FromPBReader(reader)
205
		if err != nil {
Jeromy's avatar
Jeromy committed
206
			if err != io.EOF {
207
				s.Reset()
Jeromy's avatar
Jeromy committed
208 209 210
				go bsnet.receiver.ReceiveError(err)
				log.Debugf("bitswap net handleNewStream from %s error: %s", s.Conn().RemotePeer(), err)
			}
211 212
			return
		}
213

214 215 216 217
		p := s.Conn().RemotePeer()
		ctx := context.Background()
		log.Debugf("bitswap net handleNewStream from %s", s.Conn().RemotePeer())
		bsnet.receiver.ReceiveMessage(ctx, p, received)
218
		atomic.AddUint64(&bsnet.stats.MessagesRecvd, 1)
219
	}
220
}
221

222 223 224 225
func (bsnet *impl) ConnectionManager() ifconnmgr.ConnManager {
	return bsnet.host.ConnManager()
}

226 227 228 229 230 231 232
func (bsnet *impl) Stats() NetworkStats {
	return NetworkStats{
		MessagesRecvd: atomic.LoadUint64(&bsnet.stats.MessagesRecvd),
		MessagesSent:  atomic.LoadUint64(&bsnet.stats.MessagesSent),
	}
}

233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248
type netNotifiee impl

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

func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
	nn.impl().receiver.PeerConnected(v.RemotePeer())
}

func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
	nn.impl().receiver.PeerDisconnected(v.RemotePeer())
}

func (nn *netNotifiee) OpenedStream(n inet.Network, v inet.Stream) {}
func (nn *netNotifiee) ClosedStream(n inet.Network, v inet.Stream) {}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
249 250
func (nn *netNotifiee) Listen(n inet.Network, a ma.Multiaddr)      {}
func (nn *netNotifiee) ListenClose(n inet.Network, a ma.Multiaddr) {}