DHT API

bootstrap.go

@@ -0,0 +1,123 @@
+package p2pd
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"time"
+
+	inet "github.com/libp2p/go-libp2p-net"
+	pstore "github.com/libp2p/go-libp2p-peerstore"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+var BootstrapPeers = []string{
+	"/dnsaddr/bootstrap.libp2p.io/ipfs/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
+	"/dnsaddr/bootstrap.libp2p.io/ipfs/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
+	"/dnsaddr/bootstrap.libp2p.io/ipfs/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
+	"/dnsaddr/bootstrap.libp2p.io/ipfs/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
+	"/ip4/104.131.131.82/tcp/4001/ipfs/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ",            // mars.i.ipfs.io
+	"/ip4/104.236.179.241/tcp/4001/ipfs/QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM",           // pluto.i.ipfs.io
+	"/ip4/128.199.219.111/tcp/4001/ipfs/QmSoLSafTMBsPKadTEgaXctDQVcqN88CNLHXMkTNwMKPnu",           // saturn.i.ipfs.io
+	"/ip4/104.236.76.40/tcp/4001/ipfs/QmSoLV4Bbm51jM9C4gDYZQ9Cy3U6aXMJDAbzgu2fzaDs64",             // venus.i.ipfs.io
+	"/ip4/178.62.158.247/tcp/4001/ipfs/QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd",            // earth.i.ipfs.io
+	"/ip6/2604:a880:1:20::203:d001/tcp/4001/ipfs/QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM",  // pluto.i.ipfs.io
+	"/ip6/2400:6180:0:d0::151:6001/tcp/4001/ipfs/QmSoLSafTMBsPKadTEgaXctDQVcqN88CNLHXMkTNwMKPnu",  // saturn.i.ipfs.io
+	"/ip6/2604:a880:800:10::4a:5001/tcp/4001/ipfs/QmSoLV4Bbm51jM9C4gDYZQ9Cy3U6aXMJDAbzgu2fzaDs64", // venus.i.ipfs.io
+	"/ip6/2a03:b0c0:0:1010::23:1001/tcp/4001/ipfs/QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd", // earth.i.ipfs.io
+}
+
+const BootstrapConnections = 4
+
+func bootstrapPeerInfo() ([]*pstore.PeerInfo, error) {
+	pis := make([]*pstore.PeerInfo, len(BootstrapPeers))
+	for x, p := range BootstrapPeers {
+		a, err := ma.NewMultiaddr(p)
+		if err != nil {
+			return nil, err
+		}
+
+		pi, err := pstore.InfoFromP2pAddr(a)
+		if err != nil {
+			return nil, err
+		}
+
+		pis[x] = pi
+	}
+
+	return pis, nil
+}
+
+func shufflePeerInfos(peers []*pstore.PeerInfo) {
+	for i := range peers {
+		j := rand.Intn(i + 1)
+		peers[i], peers[j] = peers[j], peers[i]
+	}
+}
+
+func (d *Daemon) Bootstrap() error {
+	pis, err := bootstrapPeerInfo()
+	if err != nil {
+		return err
+	}
+
+	for _, pi := range pis {
+		d.host.Peerstore().AddAddrs(pi.ID, pi.Addrs, pstore.PermanentAddrTTL)
+	}
+
+	count := d.connectBootstrapPeers(pis, BootstrapConnections)
+	if count == 0 {
+		return errors.New("Failed to connect to bootstrap peers")
+	}
+
+	go d.keepBootstrapConnections(pis)
+
+	if d.dht != nil {
+		return d.dht.Bootstrap(d.ctx)
+	}
+
+	return nil
+}
+
+func (d *Daemon) connectBootstrapPeers(pis []*pstore.PeerInfo, toconnect int) int {
+	count := 0
+
+	shufflePeerInfos(pis)
+
+	ctx, cancel := context.WithTimeout(d.ctx, 60*time.Second)
+	defer cancel()
+
+	for _, pi := range pis {
+		if d.host.Network().Connectedness(pi.ID) == inet.Connected {
+			continue
+		}
+		err := d.host.Connect(ctx, *pi)
+		if err != nil {
+			log.Debugf("Error connecting to bootstrap peer %s: %s", pi.ID, err.Error())
+		} else {
+			count++
+			toconnect--
+		}
+		if toconnect == 0 {
+			break
+		}
+	}
+
+	return count
+
+}
+
+func (d *Daemon) keepBootstrapConnections(pis []*pstore.PeerInfo) {
+	ticker := time.NewTicker(15 * time.Minute)
+	for {
+		<-ticker.C
+
+		conns := d.host.Network().Conns()
+		if len(conns) >= BootstrapConnections {
+			continue
+		}
+
+		toconnect := BootstrapConnections - len(conns)
+		d.connectBootstrapPeers(pis, toconnect)
+	}
+}

conn.go

@@ -2,7 +2,6 @@ package p2pd
 
 import (
 	"context"
-	"errors"
 	"io"
 	"net"
 	"time"
@@ -79,8 +78,44 @@ func (d *Daemon) handleConn(c net.Conn) {
 				return
 			}
 
+		case pb.Request_DHT:
+			res, ch, cancel := d.doDHT(&req)
+			err := w.WriteMsg(res)
+			if err != nil {
+				log.Debugf("Error writing response: %s", err.Error())
+				if ch != nil {
+					cancel()
+				}
+				return
+			}
+
+			if ch != nil {
+				for res := range ch {
+					err = w.WriteMsg(res)
+					if err != nil {
+						log.Debugf("Error writing response: %s", err.Error())
+						cancel()
+						return
+					}
+				}
+
+				err = w.WriteMsg(dhtResponseEnd())
+				if err != nil {
+					log.Debugf("Error writing response: %s", err.Error())
+					return
+				}
+			}
+
+		case pb.Request_LIST_PEERS:
+			res := d.doListPeers(&req)
+			err := w.WriteMsg(res)
+			if err != nil {
+				log.Debugf("Error writing response: %s", err.Error())
+				return
+			}
+
 		default:
-			log.Debugf("Unexpected request type: %s", req.Type)
+			log.Debugf("Unexpected request type: %d", *req.Type)
 			return
 		}
 	}
@@ -104,7 +139,7 @@ func (d *Daemon) doConnect(req *pb.Request) *pb.Response {
 	defer cancel()
 
 	if req.Connect == nil {
-		return errorResponse(errors.New("Malformed request; missing parameters"))
+		return errorResponseString("Malformed request; missing parameters")
 	}
 
 	pid, err := peer.IDFromBytes(req.Connect.Peer)
@@ -141,7 +176,7 @@ func (d *Daemon) doStreamOpen(req *pb.Request) (*pb.Response, inet.Stream) {
 	defer cancel()
 
 	if req.StreamOpen == nil {
-		return errorResponse(errors.New("Malformed request; missing parameters")), nil
+		return errorResponseString("Malformed request; missing parameters"), nil
 	}
 
 	pid, err := peer.IDFromBytes(req.StreamOpen.Peer)
@@ -169,7 +204,7 @@ func (d *Daemon) doStreamOpen(req *pb.Request) (*pb.Response, inet.Stream) {
 
 func (d *Daemon) doStreamHandler(req *pb.Request) *pb.Response {
 	if req.StreamHandler == nil {
-		return errorResponse(errors.New("Malformed request; missing parameters"))
+		return errorResponseString("Malformed request; missing parameters")
 	}
 
 	d.mx.Lock()
@@ -189,6 +224,21 @@ func (d *Daemon) doStreamHandler(req *pb.Request) *pb.Response {
 	return okResponse()
 }
 
+func (d *Daemon) doListPeers(req *pb.Request) *pb.Response {
+	conns := d.host.Network().Conns()
+	peers := make([]*pb.PeerInfo, len(conns))
+	for x, conn := range conns {
+		peers[x] = &pb.PeerInfo{
+			Id:    []byte(conn.RemotePeer()),
+			Addrs: [][]byte{conn.RemoteMultiaddr().Bytes()},
+		}
+	}
+
+	res := okResponse()
+	res.Peers = peers
+	return res
+}
+
 func okResponse() *pb.Response {
 	return &pb.Response{
 		Type: pb.Response_OK.Enum(),
@@ -203,6 +253,13 @@ func errorResponse(err error) *pb.Response {
 	}
 }
 
+func errorResponseString(err string) *pb.Response {
+	return &pb.Response{
+		Type:  pb.Response_ERROR.Enum(),
+		Error: &pb.ErrorResponse{Msg: &err},
+	}
+}
+
 func makeStreamInfo(s inet.Stream) *pb.StreamInfo {
 	proto := string(s.Protocol())
 	return &pb.StreamInfo{

daemon.go

@@ -8,8 +8,11 @@ import (
 	logging "github.com/ipfs/go-log"
 	libp2p "github.com/libp2p/go-libp2p"
 	host "github.com/libp2p/go-libp2p-host"
+	dht "github.com/libp2p/go-libp2p-kad-dht"
+	dhtopts "github.com/libp2p/go-libp2p-kad-dht/opts"
 	peer "github.com/libp2p/go-libp2p-peer"
 	proto "github.com/libp2p/go-libp2p-protocol"
+	rhost "github.com/libp2p/go-libp2p/p2p/host/routed"
 	ma "github.com/multiformats/go-multiaddr"
 )
 
@@ -20,6 +23,8 @@ type Daemon struct {
 	host     host.Host
 	listener net.Listener
 
+	dht *dht.IpfsDHT
+
 	mx sync.Mutex
 	// stream handlers: map of protocol.ID to unix socket path
 	handlers map[proto.ID]string
@@ -49,6 +54,24 @@ func NewDaemon(ctx context.Context, path string, opts ...libp2p.Option) (*Daemon
 	return d, nil
 }
 
+func (d *Daemon) EnableDHT(client bool) error {
+	var opts []dhtopts.Option
+
+	if client {
+		opts = append(opts, dhtopts.Client(true))
+	}
+
+	dht, err := dht.New(d.ctx, d.host, opts...)
+	if err != nil {
+		return err
+	}
+
+	d.dht = dht
+	d.host = rhost.Wrap(d.host, d.dht)
+
+	return nil
+}
+
 func (d *Daemon) ID() peer.ID {
 	return d.host.ID()
 }