generate random data, start deal

lotus-soup/baseline.go

@@ -1,11 +1,20 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"math/rand"
+	"time"
 
+	"github.com/davecgh/go-spew/spew"
+	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/node/impl"
+	"github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p-core/peer"
-	"github.com/testground/sdk-go/sync"
 )
 
 // This is the basline test; Filecoin 101.
@@ -59,7 +68,22 @@ func runBaselineMiner(t *TestEnvironment) error {
 
 	t.RecordMessage("got %v client addrs", len(addrs))
 
-	// subscribe to clients
+	// mine / stop mining
+	//mine := true
+	//done := make(chan struct{})
+	//go func() {
+	//defer close(done)
+	//for mine {
+	//time.Sleep(blocktime)
+	//if err := sn[0].MineOne(ctx, func(bool) {}); err != nil {
+	//t.Error(err)
+	//}
+	//}
+	//}()
+	//makeDeal(t, ctx, 6, client, miner, carExport)
+	//mine = false
+	//fmt.Println("shutting down mining")
+	//<-done
 
 	// TODO wait a bit for network to bootstrap
 	// TODO just wait until completion of test, serving requests -- the client does all the job
@@ -70,7 +94,7 @@ func runBaselineMiner(t *TestEnvironment) error {
 
 func runBaselineClient(t *TestEnvironment) error {
 	t.RecordMessage("running client")
-	client, err := prepareClient(t)
+	cl, err := prepareClient(t)
 	if err != nil {
 		return err
 	}
@@ -81,43 +105,92 @@ func runBaselineClient(t *TestEnvironment) error {
 		return err
 	}
 
+	client := cl.fullApi.(*impl.FullNodeAPI)
+
 	t.RecordMessage("got %v miner addrs", len(addrs))
 
-	if err := client.fullApi.NetConnect(ctx, addrs[0]); err != nil {
+	if err := client.NetConnect(ctx, addrs[0].PeerAddr); err != nil {
 		return err
 	}
 
 	t.RecordMessage("client connected to miner")
 
-	// TODO generate a number of random "files" and publish them to one or more miners
+	// generate a number of random "files" and publish them to one or more miners
+	data := make([]byte, 1600)
+	rand.New(rand.NewSource(time.Now().UnixNano())).Read(data)
+
+	r := bytes.NewReader(data)
+	fcid, err := client.ClientImportLocal(ctx, r)
+	if err != nil {
+		return err
+	}
+
+	t.RecordMessage("file cid: %s", fcid)
+
+	// start deal
+	deal := startDeal(ctx, addrs[0].ActorAddr, client, fcid)
+	spew.Dump(deal)
+
+	// TODO: this sleep is only necessary because deals don't immediately get logged in the dealstore, we should fix this
+	time.Sleep(time.Second)
+
+	//waitDealSealed(t, ctx, client, deal)
+	//testRetrieval(t, ctx, err, client, fcid, carExport, data)
+
 	// TODO broadcast published content CIDs to other clients
-	// TODO select a random piece of content published by some other client and retreieve it
+	// TODO select a random piece of content published by some other client and retrieve it
 
 	t.SyncClient.MustSignalAndWait(ctx, stateDone, t.TestInstanceCount)
 	return nil
 }
 
-func collectMinersAddrs(t *TestEnvironment, ctx context.Context, miners int) ([]peer.AddrInfo, error) {
+func collectMinersAddrs(t *TestEnvironment, ctx context.Context, miners int) ([]MinerAddresses, error) {
-	return collectAddrs(t, ctx, minersAddrsTopic, miners)
+	ch := make(chan MinerAddresses)
-}
+	sub := t.SyncClient.MustSubscribe(ctx, minersAddrsTopic, ch)
 
-func collectClientsAddrs(t *TestEnvironment, ctx context.Context, clients int) ([]peer.AddrInfo, error) {
+	addrs := make([]MinerAddresses, 0, miners)
-	return collectAddrs(t, ctx, clientsAddrsTopic, clients)
+	for i := 0; i < miners; i++ {
-}
-
-func collectAddrs(t *TestEnvironment, ctx context.Context, topic *sync.Topic, expectedValues int) ([]peer.AddrInfo, error) {
-	ch := make(chan peer.AddrInfo)
-	sub := t.SyncClient.MustSubscribe(ctx, topic, ch)
-
-	addrs := make([]peer.AddrInfo, 0, expectedValues)
-	for i := 0; i < expectedValues; i++ {
 		select {
 		case a := <-ch:
 			addrs = append(addrs, a)
 		case err := <-sub.Done():
-			return nil, fmt.Errorf("got error while waiting for %v addrs: %w", topic, err)
+			return nil, fmt.Errorf("got error while waiting for miners addrs: %w", err)
 		}
 	}
 
 	return addrs, nil
 }
+
+func collectClientsAddrs(t *TestEnvironment, ctx context.Context, clients int) ([]peer.AddrInfo, error) {
+	ch := make(chan peer.AddrInfo)
+	sub := t.SyncClient.MustSubscribe(ctx, clientsAddrsTopic, ch)
+
+	addrs := make([]peer.AddrInfo, 0, clients)
+	for i := 0; i < clients; i++ {
+		select {
+		case a := <-ch:
+			addrs = append(addrs, a)
+		case err := <-sub.Done():
+			return nil, fmt.Errorf("got error while waiting for clients addrs: %w", err)
+		}
+	}
+
+	return addrs, nil
+}
+
+func startDeal(ctx context.Context, minerActorAddr address.Address, client *impl.FullNodeAPI, fcid cid.Cid) *cid.Cid {
+	//func startDeal(ctx context.Context, minerActorAddr address.Address, client api.FullNodeAPI, fcid cid.Cid) *cid.Cid {
+	addr, _ := client.WalletDefaultAddress(ctx)
+
+	deal, err := client.ClientStartDeal(ctx, &api.StartDealParams{
+		Data:              &storagemarket.DataRef{Root: fcid},
+		Wallet:            addr,
+		Miner:             minerActorAddr,
+		EpochPrice:        types.NewInt(1000000),
+		MinBlocksDuration: 100,
+	})
+	if err != nil {
+		panic(err)
+	}
+	return deal
+}

lotus-soup/go.mod

@@ -3,10 +3,13 @@ module github.com/filecoin-project/oni/lotus-soup
 go 1.14
 
 require (
+	github.com/davecgh/go-spew v1.1.1
 	github.com/filecoin-project/go-address v0.0.2-0.20200504173055-8b6f2fb2b3ef
+	github.com/filecoin-project/go-fil-markets v0.3.0
 	github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b
 	github.com/filecoin-project/lotus v0.4.1-0.20200623104442-68d38eff33e4
 	github.com/filecoin-project/specs-actors v0.6.2-0.20200617175406-de392ca14121
+	github.com/ipfs/go-cid v0.0.6
 	github.com/ipfs/go-datastore v0.4.4
 	github.com/ipfs/go-log/v2 v2.1.2-0.20200609205458-f8d20c392cb7
 	github.com/libp2p/go-libp2p-core v0.6.0

lotus-soup/node.go

@@ -57,7 +57,7 @@ var (
 	presealTopic = sync.NewTopic("preseal", &PresealMsg{})
 
 	clientsAddrsTopic = sync.NewTopic("clientsAddrsTopic", &peer.AddrInfo{})
-	minersAddrsTopic  = sync.NewTopic("minersAddrsTopic", &peer.AddrInfo{})
+	minersAddrsTopic  = sync.NewTopic("minersAddrsTopic", &MinerAddresses{})
 
 	stateReady = sync.State("ready")
 	stateDone  = sync.State("done")
@@ -88,6 +88,11 @@ type GenesisMsg struct {
 	Bootstrapper []byte
 }
 
+type MinerAddresses struct {
+	PeerAddr  peer.AddrInfo
+	ActorAddr address.Address
+}
+
 func init() {
 	logging.SetLogLevel("vm", "WARN")
 
@@ -400,11 +405,15 @@ func prepareMiner(t *TestEnvironment) (*Node, error) {
 	}
 
 	t.RecordMessage("publish our address to the miners addr topic")
+	actoraddress, err := n.minerApi.ActorAddress(ctx)
+	if err != nil {
+		return nil, err
+	}
 	addrinfo, err := n.minerApi.NetAddrsListen(ctx)
 	if err != nil {
 		return nil, err
 	}
-	t.SyncClient.MustPublish(ctx, minersAddrsTopic, addrinfo)
+	t.SyncClient.MustPublish(ctx, minersAddrsTopic, MinerAddresses{addrinfo, actoraddress})
 
 	t.RecordMessage("waiting for all nodes to be ready")
 	t.SyncClient.MustSignalAndWait(ctx, stateReady, t.TestInstanceCount)