dht dynamic mode switching.

dht.go

@@ -42,6 +42,10 @@ import (
 
 var logger = logging.Logger("dht")
 
+// DynamicModeSwitchDebouncePeriod is the amount of time to wait before making a dynamic dht mode switch effective.
+// It helps mitigate flapping from routability events.
+var DynamicModeSwitchDebouncePeriod = 1 * time.Minute
+
 // NumBootstrapQueries defines the number of random dht queries to do to
 // collect members of the routing table.
 const NumBootstrapQueries = 5
@@ -86,6 +90,7 @@ type IpfsDHT struct {
 
 	subscriptions struct {
 		evtPeerIdentification event.Subscription
+		evtLocalRoutability   event.Subscription
 	}
 }
 
@@ -119,14 +124,20 @@ func New(ctx context.Context, h host.Host, options ...opts.Option) (*IpfsDHT, er
 		// remove ourselves from network notifs.
 		dht.host.Network().StopNotify((*subscriberNotifee)(dht))
 
-		if dht.subscriptions.evtPeerIdentification != nil {
-			_ = dht.subscriptions.evtPeerIdentification.Close()
+		for _, sub := range []event.Subscription{
+			dht.subscriptions.evtPeerIdentification,
+			dht.subscriptions.evtLocalRoutability,
+		} {
+			if sub != nil {
+				_ = sub.Close()
+			}
 		}
 		return nil
 	})
 
 	dht.proc.AddChild(subnot.Process(ctx))
 	dht.proc.AddChild(dht.providers.Process())
+	dht.proc.AddChild(dht.dynamicModeSwitching(ctx))
 	dht.Validator = cfg.Validator
 	dht.mode = ModeClient
 
@@ -187,16 +198,33 @@ func makeDHT(ctx context.Context, h host.Host, dstore ds.Batching, protocols []p
 		bucketSize:   bucketSize,
 	}
 
+	dht.setupEventSubscribers()
+
+	dht.ctx = dht.newContextWithLocalTags(ctx)
+
+	return dht
+}
+
+func (dht *IpfsDHT) setupEventSubscribers() {
 	var err error
-	evts := []interface{}{&event.EvtPeerIdentificationCompleted{}, &event.EvtPeerIdentificationFailed{}}
-	dht.subscriptions.evtPeerIdentification, err = h.EventBus().Subscribe(evts, eventbus.BufSize(256))
+	evts := []interface{}{
+		&event.EvtPeerIdentificationCompleted{},
+		&event.EvtPeerIdentificationFailed{},
+	}
+	dht.subscriptions.evtPeerIdentification, err = dht.host.EventBus().Subscribe(evts, eventbus.BufSize(256))
 	if err != nil {
 		logger.Errorf("dht not subscribed to peer identification events; things will fail; err: %s", err)
 	}
 
-	dht.ctx = dht.newContextWithLocalTags(ctx)
-
-	return dht
+	evts = []interface{}{
+		&event.EvtLocalRoutabilityPublic{},
+		&event.EvtLocalRoutabilityPrivate{},
+		&event.EvtLocalRoutabilityUnknown{},
+	}
+	dht.subscriptions.evtLocalRoutability, err = dht.host.EventBus().Subscribe(evts, eventbus.BufSize(256))
+	if err != nil {
+		logger.Errorf("dht not subscribed to local routability events; dynamic mode switching will not work; err: %s", err)
+	}
 }
 
 // putValueToPeer stores the given key/value pair at the peer 'p'
@@ -498,6 +526,46 @@ func (dht *IpfsDHT) SetMode(m DHTMode) error {
 	}
 }
 
+func (dht *IpfsDHT) dynamicModeSwitching(ctx context.Context) goprocess.Process {
+	proc := goprocessctx.WithContext(ctx)
+	watch := func(proc goprocess.Process) {
+		var (
+			target   DHTMode
+			debounce <-chan time.Time
+		)
+		for {
+			select {
+			case ev := <-dht.subscriptions.evtLocalRoutability.Out():
+				switch ev.(type) {
+				case event.EvtLocalRoutabilityPrivate, event.EvtLocalRoutabilityUnknown:
+					target = ModeClient
+				case event.EvtLocalRoutabilityPublic:
+					target = ModeServer
+				}
+				if debounce == nil {
+					debounce = time.After(DynamicModeSwitchDebouncePeriod)
+				}
+				logger.Infof("processed event %T; scheduled dht mode switch", ev)
+
+			case <-debounce:
+				err := dht.SetMode(target)
+				// NOTE: the mode will be printed out as a decimal.
+				if err == nil {
+					logger.Infof("switched DHT mode successfully; new mode: %d", target)
+				} else {
+					logger.Warningf("switching DHT mode failed; new mode: %d, err: %s", target, err)
+				}
+				debounce, target = nil, 0
+
+			case <-proc.Closing():
+				return
+			}
+		}
+	}
+	proc.Go(watch)
+	return proc
+}
+
 func (dht *IpfsDHT) moveToServerMode() error {
 	dht.mode = ModeServer
 	for _, p := range dht.protocols {

dht_test.go

@@ -12,11 +12,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/libp2p/go-libp2p-core/event"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/routing"
 
-	multistream "github.com/multiformats/go-multistream"
+	"github.com/multiformats/go-multistream"
 
 	"golang.org/x/xerrors"
 
@@ -26,12 +27,12 @@ import (
 	opts "github.com/libp2p/go-libp2p-kad-dht/opts"
 	pb "github.com/libp2p/go-libp2p-kad-dht/pb"
 
-	cid "github.com/ipfs/go-cid"
+	"github.com/ipfs/go-cid"
 	u "github.com/ipfs/go-ipfs-util"
 	kb "github.com/libp2p/go-libp2p-kbucket"
 	record "github.com/libp2p/go-libp2p-record"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
-	ci "github.com/libp2p/go-libp2p-testing/ci"
+	"github.com/libp2p/go-libp2p-testing/ci"
 	travisci "github.com/libp2p/go-libp2p-testing/ci/travis"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	ma "github.com/multiformats/go-multiaddr"
@@ -1405,3 +1406,79 @@ func TestModeChange(t *testing.T) {
 	err = clientOnly.Ping(ctx, clientToServer.PeerID())
 	assert.NotNil(t, err)
 }
+
+func TestDynamicModeSwitching(t *testing.T) {
+	// remove the debounce period.
+	DynamicModeSwitchDebouncePeriod = 0
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	prober := setupDHT(ctx, t, true) // our test harness
+	node := setupDHT(ctx, t, true)   // the node under test
+	prober.Host().Peerstore().AddAddrs(node.PeerID(), node.Host().Addrs(), peerstore.AddressTTL)
+	if _, err := prober.Host().Network().DialPeer(ctx, node.PeerID()); err != nil {
+		t.Fatal(err)
+	}
+
+	var err error
+	var emitters struct {
+		evtLocalRoutabilityPrivate event.Emitter
+		evtLocalRoutabilityPublic  event.Emitter
+		evtLocalRoutabilityUnknown event.Emitter
+	}
+
+	emitters.evtLocalRoutabilityPublic, err = node.host.EventBus().Emitter(new(event.EvtLocalRoutabilityPublic))
+	if err != nil {
+		t.Fatal(err)
+	}
+	emitters.evtLocalRoutabilityPrivate, err = node.host.EventBus().Emitter(new(event.EvtLocalRoutabilityPrivate))
+	if err != nil {
+		t.Fatal(err)
+	}
+	emitters.evtLocalRoutabilityUnknown, err = node.host.EventBus().Emitter(new(event.EvtLocalRoutabilityUnknown))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	emitters.evtLocalRoutabilityPrivate.Emit(event.EvtLocalRoutabilityPrivate{})
+	time.Sleep(500 * time.Millisecond)
+
+	err = prober.Ping(ctx, node.PeerID())
+	assert.True(t, xerrors.Is(err, multistream.ErrNotSupported))
+	if l := len(prober.RoutingTable().ListPeers()); l != 0 {
+		t.Errorf("expected routing table length to be 0; instead is %d", l)
+	}
+
+	emitters.evtLocalRoutabilityPublic.Emit(event.EvtLocalRoutabilityPublic{})
+	time.Sleep(500 * time.Millisecond)
+
+	err = prober.Ping(ctx, node.PeerID())
+	assert.Nil(t, err)
+	if l := len(prober.RoutingTable().ListPeers()); l != 1 {
+		t.Errorf("expected routing table length to be 1; instead is %d", l)
+	}
+
+	emitters.evtLocalRoutabilityUnknown.Emit(event.EvtLocalRoutabilityUnknown{})
+	time.Sleep(500 * time.Millisecond)
+
+	err = prober.Ping(ctx, node.PeerID())
+	assert.True(t, xerrors.Is(err, multistream.ErrNotSupported))
+	if l := len(prober.RoutingTable().ListPeers()); l != 0 {
+		t.Errorf("expected routing table length to be 0; instead is %d", l)
+	}
+
+	//////////////////////////////////////////////////////////////
+	// let's activate the debounce, to check we do not flap.
+	// receiving a "routability public" event should have no immediate effect now.
+	DynamicModeSwitchDebouncePeriod = 1 * time.Minute
+
+	emitters.evtLocalRoutabilityPublic.Emit(event.EvtLocalRoutabilityPublic{})
+	time.Sleep(500 * time.Millisecond)
+
+	err = prober.Ping(ctx, node.PeerID())
+	assert.True(t, xerrors.Is(err, multistream.ErrNotSupported))
+	if l := len(prober.RoutingTable().ListPeers()); l != 0 {
+		t.Errorf("expected routing table length to be 0; instead is %d", l)
+	}
+}