opensource/kubo
1
0
Fork 0
mirror of https://github.com/ipfs/kubo.git synced 2025-06-30 18:13:54 +08:00
Code Issues Projects Releases Wiki Activity

Merge pull request #1472 from ipfs/feat/proto-ping

Browse Source 
Feat/proto ping
This commit is contained in:
Juan Batiz-Benet
2015-07-13 20:52:26 -07:00
parent eee61b06fd a19ad97ea5
commit 498e927ac7
8 changed files with 187 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
core/commands/ping.go
View File

@ -138,30 +138,35 @@ func pingPeer(ctx context.Context, n *core.IpfsNode, pid peer.ID, numPings int)
outChan <- &PingResult{Text: fmt.Sprintf("PING %s.", pid.Pretty())}
ctx, cancel := context.WithTimeout(ctx, kPingTimeout*time.Duration(numPings))
defer cancel()
pings, err := n.Ping.Ping(ctx, pid)
if err != nil {
log.Debugf("Ping error: %s", err)
outChan <- &PingResult{Text: fmt.Sprintf("Ping error: %s", err)}
return
}
var done bool
var total time.Duration
for i := 0; i < numPings && !done; i++ {
select {
case <-ctx.Done():
done = true
continue
default:
}
ctx, cancel := context.WithTimeout(ctx, kPingTimeout)
defer cancel()
took, err := n.Routing.Ping(ctx, pid)
if err != nil {
log.Debugf("Ping error: %s", err)
outChan <- &PingResult{Text: fmt.Sprintf("Ping error: %s", err)}
break
case t, ok := <-pings:
if !ok {
done = true
break
}
outChan <- &PingResult{
Success: true,
Time: t,
}
total += t
time.Sleep(time.Second)
}
outChan <- &PingResult{
Success: true,
Time: took,
}
total += took
time.Sleep(time.Second)
}
averagems := total.Seconds() * 1000 / float64(numPings)
outChan <- &PingResult{

5
core/core.go
View File

@ -33,6 +33,7 @@ import (
swarm "github.com/ipfs/go-ipfs/p2p/net/swarm"
addrutil "github.com/ipfs/go-ipfs/p2p/net/swarm/addr"
peer "github.com/ipfs/go-ipfs/p2p/peer"
ping "github.com/ipfs/go-ipfs/p2p/protocol/ping"
eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
routing "github.com/ipfs/go-ipfs/routing"
@ -102,7 +103,8 @@ type IpfsNode struct {
Exchange exchange.Interface // the block exchange + strategy (bitswap)
Namesys namesys.NameSystem // the name system, resolves paths to hashes
Diagnostics *diag.Diagnostics // the diagnostics service
Reprovider *rp.Reprovider // the value reprovider system
Ping *ping.PingService
Reprovider *rp.Reprovider // the value reprovider system
IpnsFs *ipnsfs.Filesystem
@ -324,6 +326,7 @@ func (n *IpfsNode) HandlePeerFound(p peer.PeerInfo) {
func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost.Host, routingOption RoutingOption) error {
// setup diagnostics service
n.Diagnostics = diag.NewDiagnostics(n.Identity, host)
n.Ping = ping.NewPingService(host)
// setup routing service
r, err := routingOption(ctx, host, n.Repo.Datastore())

105
p2p/protocol/ping/ping.go Normal file
View File

@ -0,0 +1,105 @@
package ping
import (
"bytes"
"errors"
"io"
"time"
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
host "github.com/ipfs/go-ipfs/p2p/host"
inet "github.com/ipfs/go-ipfs/p2p/net"
peer "github.com/ipfs/go-ipfs/p2p/peer"
eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
u "github.com/ipfs/go-ipfs/util"
)
var log = eventlog.Logger("ping")
const PingSize = 32
const ID = "/ipfs/ping"
type PingService struct {
Host host.Host
}
func NewPingService(h host.Host) *PingService {
ps := &PingService{h}
h.SetStreamHandler(ID, ps.PingHandler)
return ps
}
func (p *PingService) PingHandler(s inet.Stream) {
buf := make([]byte, PingSize)
for {
_, err := io.ReadFull(s, buf)
if err != nil {
log.Debug(err)
return
}
_, err = s.Write(buf)
if err != nil {
log.Debug(err)
return
}
}
}
func (ps *PingService) Ping(ctx context.Context, p peer.ID) (<-chan time.Duration, error) {
s, err := ps.Host.NewStream(ID, p)
if err != nil {
return nil, err
}
out := make(chan time.Duration)
go func() {
defer close(out)
for {
select {
case <-ctx.Done():
return
default:
t, err := ping(s)
if err != nil {
log.Debugf("ping error: %s", err)
return
}
select {
case out <- t:
case <-ctx.Done():
return
}
}
}
}()
return out, nil
}
func ping(s inet.Stream) (time.Duration, error) {
buf := make([]byte, PingSize)
u.NewTimeSeededRand().Read(buf)
before := time.Now()
_, err := s.Write(buf)
if err != nil {
return 0, err
}
rbuf := make([]byte, PingSize)
_, err = io.ReadFull(s, rbuf)
if err != nil {
return 0, err
}
if !bytes.Equal(buf, rbuf) {
return 0, errors.New("ping packet was incorrect!")
}
return time.Now().Sub(before), nil
}

51
p2p/protocol/ping/ping_test.go Normal file
View File

@ -0,0 +1,51 @@
package ping
import (
"testing"
"time"
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
peer "github.com/ipfs/go-ipfs/p2p/peer"
netutil "github.com/ipfs/go-ipfs/p2p/test/util"
)
func TestPing(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
h1 := netutil.GenHostSwarm(t, ctx)
h2 := netutil.GenHostSwarm(t, ctx)
err := h1.Connect(ctx, peer.PeerInfo{
ID: h2.ID(),
Addrs: h2.Addrs(),
})
if err != nil {
t.Fatal(err)
}
ps1 := NewPingService(h1)
ps2 := NewPingService(h2)
testPing(t, ps1, h2.ID())
testPing(t, ps2, h1.ID())
}
func testPing(t *testing.T, ps *PingService, p peer.ID) {
pctx, cancel := context.WithCancel(context.Background())
defer cancel()
ts, err := ps.Ping(pctx, p)
if err != nil {
t.Fatal(err)
}
for i := 0; i < 5; i++ {
select {
case took := <-ts:
t.Log("ping took: ", took)
case <-time.After(time.Second * 4):
t.Fatal("failed to receive ping")
}
}
}

57
routing/dht/dht.go
View File

@ -4,7 +4,6 @@ package dht
import (
"bytes"
"crypto/rand"
"errors"
"fmt"
"sync"
@ -33,8 +32,6 @@ var log = eventlog.Logger("dht")
var ProtocolDHT protocol.ID = "/ipfs/dht"
const doPinging = false
// NumBootstrapQueries defines the number of random dht queries to do to
// collect members of the routing table.
const NumBootstrapQueries = 5
@ -92,11 +89,6 @@ func NewDHT(ctx context.Context, h host.Host, dstore ds.ThreadSafeDatastore) *Ip
dht.Validator = make(record.Validator)
dht.Validator["pk"] = record.PublicKeyValidator
if doPinging {
dht.proc.Go(func(p goprocess.Process) {
dht.PingRoutine(time.Second * 10)
})
}
return dht
}
@ -110,23 +102,6 @@ func (dht *IpfsDHT) log() eventlog.EventLogger {
return log // TODO rm
}
// Connect to a new peer at the given address, ping and add to the routing table
func (dht *IpfsDHT) Connect(ctx context.Context, npeer peer.ID) error {
// TODO: change interface to accept a PeerInfo as well.
if err := dht.host.Connect(ctx, peer.PeerInfo{ID: npeer}); err != nil {
return err
}
// Ping new peer to register in their routing table
// NOTE: this should be done better...
if _, err := dht.Ping(ctx, npeer); err != nil {
return fmt.Errorf("failed to ping newly connected peer: %s", err)
}
log.Event(ctx, "connect", dht.self, npeer)
dht.Update(ctx, npeer)
return nil
}
// putValueToPeer stores the given key/value pair at the peer 'p'
func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID,
key key.Key, rec *pb.Record) error {
@ -343,38 +318,6 @@ func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, p peer.ID, count int) [
return filtered
}
func (dht *IpfsDHT) ensureConnectedToPeer(ctx context.Context, p peer.ID) error {
if p == dht.self {
return errors.New("attempting to ensure connection to self")
}
// dial connection
return dht.host.Connect(ctx, peer.PeerInfo{ID: p})
}
// PingRoutine periodically pings nearest neighbors.
func (dht *IpfsDHT) PingRoutine(t time.Duration) {
tick := time.Tick(t)
for {
select {
case <-tick:
id := make([]byte, 16)
rand.Read(id)
peers := dht.routingTable.NearestPeers(kb.ConvertKey(key.Key(id)), 5)
for _, p := range peers {
ctx, cancel := context.WithTimeout(dht.Context(), time.Second*5)
_, err := dht.Ping(ctx, p)
if err != nil {
log.Debugf("Ping error: %s", err)
}
cancel()
}
case <-dht.proc.Closing():
return
}
}
}
// Context return dht's context
func (dht *IpfsDHT) Context() context.Context {
return dht.ctx

38
routing/dht/dht_test.go
View File

@ -74,7 +74,8 @@ func connect(t *testing.T, ctx context.Context, a, b *IpfsDHT) {
}
a.peerstore.AddAddrs(idB, addrB, peer.TempAddrTTL)
if err := a.Connect(ctx, idB); err != nil {
pi := peer.PeerInfo{ID: idB}
if err := a.host.Connect(ctx, pi); err != nil {
t.Fatal(err)
}
}
@ -101,35 +102,6 @@ func bootstrap(t *testing.T, ctx context.Context, dhts []*IpfsDHT) {
cancel()
}
func TestPing(t *testing.T) {
// t.Skip("skipping test to debug another")
ctx := context.Background()
dhtA := setupDHT(ctx, t)
dhtB := setupDHT(ctx, t)
peerA := dhtA.self
peerB := dhtB.self
defer dhtA.Close()
defer dhtB.Close()
defer dhtA.host.Close()
defer dhtB.host.Close()
connect(t, ctx, dhtA, dhtB)
//Test that we can ping the node
ctxT, _ := context.WithTimeout(ctx, 100*time.Millisecond)
if _, err := dhtA.Ping(ctxT, peerB); err != nil {
t.Fatal(err)
}
ctxT, _ = context.WithTimeout(ctx, 100*time.Millisecond)
if _, err := dhtB.Ping(ctxT, peerA); err != nil {
t.Fatal(err)
}
}
func TestValueGetSet(t *testing.T) {
// t.Skip("skipping test to debug another")
@ -789,12 +761,14 @@ func TestConnectCollision(t *testing.T) {
errs := make(chan error)
go func() {
dhtA.peerstore.AddAddr(peerB, addrB, peer.TempAddrTTL)
err := dhtA.Connect(ctx, peerB)
pi := peer.PeerInfo{ID: peerB}
err := dhtA.host.Connect(ctx, pi)
errs <- err
}()
go func() {
dhtB.peerstore.AddAddr(peerA, addrA, peer.TempAddrTTL)
err := dhtB.Connect(ctx, peerA)
pi := peer.PeerInfo{ID: peerA}
err := dhtB.host.Connect(ctx, pi)
errs <- err
}()

14
routing/dht/routing.go
View File

@ -2,7 +2,6 @@ package dht
import (
"sync"
"time"
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
key "github.com/ipfs/go-ipfs/blocks/key"
@ -397,16 +396,3 @@ func (dht *IpfsDHT) FindPeersConnectedToPeer(ctx context.Context, id peer.ID) (<
return peerchan, nil
}
// Ping a peer, log the time it took
func (dht *IpfsDHT) Ping(ctx context.Context, p peer.ID) (time.Duration, error) {
// Thoughts: maybe this should accept an ID and do a peer lookup?
log.Debugf("ping %s start", p)
before := time.Now()
pmes := pb.NewMessage(pb.Message_PING, "", 0)
_, err := dht.sendRequest(ctx, p, pmes)
log.Debugf("ping %s end (err = %s)", p, err)
return time.Now().Sub(before), err
}

4
routing/routing.go
View File

@ -3,7 +3,6 @@ package routing
import (
"errors"
"time"
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
key "github.com/ipfs/go-ipfs/blocks/key"
@ -38,9 +37,6 @@ type IpfsRouting interface {
// with relevant addresses.
FindPeer(context.Context, peer.ID) (peer.PeerInfo, error)
// Ping a peer, log the time it took
Ping(context.Context, peer.ID) (time.Duration, error)
// Bootstrap allows callers to hint to the routing system to get into a
// Boostrapped state
Bootstrap(context.Context) error