Merge pull request #2328 from ipfs/feat/parallel-pub

put pubkey and ipns entry to dht in parallel

namesys/publisher.go

@@ -146,12 +146,22 @@ func PutRecordToRouting(ctx context.Context, k ci.PrivKey, value path.Path, seqn
 		entry.Ttl = proto.Uint64(uint64(ttl.Nanoseconds()))
 	}
 
-	err = PublishEntry(ctx, r, ipnskey, entry)
+	errs := make(chan error)
+
+	go func() {
+		errs <- PublishEntry(ctx, r, ipnskey, entry)
+	}()
+
+	go func() {
+		errs <- PublishPublicKey(ctx, r, namekey, k.GetPublic())
+	}()
+
+	err = waitOnErrChan(ctx, errs)
 	if err != nil {
 		return err
 	}
 
-	err = PublishPublicKey(ctx, r, namekey, k.GetPublic())
+	err = waitOnErrChan(ctx, errs)
 	if err != nil {
 		return err
 	}
@@ -159,6 +169,15 @@ func PutRecordToRouting(ctx context.Context, k ci.PrivKey, value path.Path, seqn
 	return nil
 }
 
+func waitOnErrChan(ctx context.Context, errs chan error) error {
+	select {
+	case err := <-errs:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
 func PublishPublicKey(ctx context.Context, r routing.IpfsRouting, k key.Key, pubk ci.PubKey) error {
 	log.Debugf("Storing pubkey at: %s", k)
 	pkbytes, err := pubk.Bytes()

namesys/resolve_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/ipfs/go-datastore"
+	dssync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/ipfs/go-datastore/sync"
 	key "github.com/ipfs/go-ipfs/blocks/key"
 	path "github.com/ipfs/go-ipfs/path"
 	mockrouting "github.com/ipfs/go-ipfs/routing/mock"
@@ -16,8 +17,10 @@ import (
 )
 
 func TestRoutingResolve(t *testing.T) {
-	d := mockrouting.NewServer().Client(testutil.RandIdentityOrFatal(t))
-	dstore := ds.NewMapDatastore()
+	dstore := dssync.MutexWrap(ds.NewMapDatastore())
+	serv := mockrouting.NewServer()
+	id := testutil.RandIdentityOrFatal(t)
+	d := serv.ClientWithDatastore(context.Background(), id, dstore)
 
 	resolver := NewRoutingResolver(d, 0)
 	publisher := NewRoutingPublisher(d, dstore)
@@ -50,7 +53,7 @@ func TestRoutingResolve(t *testing.T) {
 }
 
 func TestPrexistingExpiredRecord(t *testing.T) {
-	dstore := ds.NewMapDatastore()
+	dstore := dssync.MutexWrap(ds.NewMapDatastore())
 	d := mockrouting.NewServer().ClientWithDatastore(context.Background(), testutil.RandIdentityOrFatal(t), dstore)
 
 	resolver := NewRoutingResolver(d, 0)
@@ -87,7 +90,7 @@ func TestPrexistingExpiredRecord(t *testing.T) {
 }
 
 func TestPrexistingRecord(t *testing.T) {
-	dstore := ds.NewMapDatastore()
+	dstore := dssync.MutexWrap(ds.NewMapDatastore())
 	d := mockrouting.NewServer().ClientWithDatastore(context.Background(), testutil.RandIdentityOrFatal(t), dstore)
 
 	resolver := NewRoutingResolver(d, 0)

routing/mock/centralized_server.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/ipfs/go-datastore"
+	dssync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/ipfs/go-datastore/sync"
 	key "github.com/ipfs/go-ipfs/blocks/key"
 	"github.com/ipfs/go-ipfs/util/testutil"
 	peer "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/peer"
@@ -74,7 +75,7 @@ func (rs *s) Providers(k key.Key) []peer.PeerInfo {
 }
 
 func (rs *s) Client(p testutil.Identity) Client {
-	return rs.ClientWithDatastore(context.Background(), p, ds.NewMapDatastore())
+	return rs.ClientWithDatastore(context.Background(), p, dssync.MutexWrap(ds.NewMapDatastore()))
 }
 
 func (rs *s) ClientWithDatastore(_ context.Context, p testutil.Identity, datastore ds.Datastore) Client {