Merge remote-tracking branch 'origin/devnet/7' into feat/interactive-porep

lib/jsonrpc/client.go

@@ -137,6 +137,7 @@ func (c *client) makeOutChan(ctx context.Context, ftyp reflect.Type, valOut int)
 		// unpack chan type to make sure it's reflect.BothDir
 		ctyp := reflect.ChanOf(reflect.BothDir, ftyp.Out(valOut).Elem())
 		ch := reflect.MakeChan(ctyp, 0) // todo: buffer?
+		chCtx, chCancel := context.WithCancel(ctx)
 		retVal = ch.Convert(ftyp.Out(valOut))
 
 		buf := (&list.List{}).Init()
@@ -144,6 +145,7 @@ func (c *client) makeOutChan(ctx context.Context, ftyp reflect.Type, valOut int)
 
 		return ctx, func(result []byte, ok bool) {
 			if !ok {
+				chCancel()
 				// remote channel closed, close ours too
 				ch.Close()
 				return
@@ -173,13 +175,29 @@ func (c *client) makeOutChan(ctx context.Context, ftyp reflect.Type, valOut int)
 			go func() {
 				for buf.Len() > 0 {
 					front := buf.Front()
-
 					bufLk.Unlock()
 
-					ch.Send(front.Value.(reflect.Value).Elem()) // todo: select on ctx is probably a good idea
+					cases :=  []reflect.SelectCase{
+						{
+							Dir: reflect.SelectRecv,
+							Chan: reflect.ValueOf(chCtx.Done()),
+						},
+						{
+							Dir:  reflect.SelectSend,
+							Chan: ch,
+							Send: front.Value.(reflect.Value).Elem(),
+						},
+					}
 
+					chosen, _, _ := reflect.Select(cases)
 					bufLk.Lock()
-					buf.Remove(front)
+
+					switch chosen {
+					case 0:
+						buf.Init()
+					case 1:
+						buf.Remove(front)
+					}
 				}
 
 				bufLk.Unlock()

lib/sectorbuilder/mock.go

@@ -0,0 +1,46 @@
+package sectorbuilder
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/filecoin-project/lotus/chain/address"
+)
+
+func TempSectorbuilder(sectorSize uint64) (*SectorBuilder, func(), error) {
+	dir, err := ioutil.TempDir("", "sbtest")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	addr, err := address.NewFromString("t3vfxagwiegrywptkbmyohqqbfzd7xzbryjydmxso4hfhgsnv6apddyihltsbiikjf3lm7x2myiaxhuc77capq")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	metadata := filepath.Join(dir, "meta")
+	sealed := filepath.Join(dir, "sealed")
+	staging := filepath.Join(dir, "staging")
+	cache := filepath.Join(dir, "cache")
+
+	sb, err := New(&Config{
+		SectorSize:  sectorSize,
+		SealedDir:   sealed,
+		StagedDir:   staging,
+		MetadataDir: metadata,
+		CacheDir:    cache,
+
+		WorkerThreads: 2,
+		Miner:       addr,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return sb, func() {
+		if err := os.RemoveAll(dir); err != nil {
+			log.Warn("failed to clean up temp sectorbuilder: ", err)
+		}
+	}, nil
+}

lib/sectorbuilder/sectorbuilder.go

@@ -41,6 +41,8 @@ const CommLen = sectorbuilder.CommitmentBytesLen
 type SectorBuilder struct {
 	handle unsafe.Pointer
 
+	Miner address.Address
+
 	rateLimit chan struct{}
 }
 
@@ -69,7 +71,8 @@ func New(cfg *Config) (*SectorBuilder, error) {
 	}
 
 	return &SectorBuilder{
-		handle:    sbp,
+		handle: sbp,
+		Miner:  cfg.Miner,
 		rateLimit: make(chan struct{}, cfg.WorkerThreads-PoStReservedWorkers),
 	}, nil
 }

lib/sectorbuilder/sectorbuilder_test.go

@@ -2,13 +2,10 @@ package sectorbuilder_test
 
 import (
 	"io"
-	"io/ioutil"
 	"math/rand"
-	"path/filepath"
 	"testing"
 
 	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain/address"
 	"github.com/filecoin-project/lotus/lib/sectorbuilder"
 )
 
@@ -24,35 +21,11 @@ func TestSealAndVerify(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	dir, err := ioutil.TempDir("", "sbtest")
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	addr, err := address.NewFromString("t3vfxagwiegrywptkbmyohqqbfzd7xzbryjydmxso4hfhgsnv6apddyihltsbiikjf3lm7x2myiaxhuc77capq")
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	cache := filepath.Join(dir, "cache")
-	metadata := filepath.Join(dir, "meta")
-	sealed := filepath.Join(dir, "sealed")
-	staging := filepath.Join(dir, "staging")
-
-	sb, err := sectorbuilder.New(&sectorbuilder.Config{
-		SectorSize: sectorSize,
-		Miner:      addr,
-
-		WorkerThreads: 2,
-
-		CacheDir:    cache,
-		SealedDir:   sealed,
-		StagedDir:   staging,
-		MetadataDir: metadata,
-	})
+	sb, cleanup, err := sectorbuilder.TempSectorbuilder(sectorSize)
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer cleanup()
 
 	dlen := sectorbuilder.UserBytesForSectorSize(sectorSize)
 
@@ -82,7 +55,7 @@ func TestSealAndVerify(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	ok, err := sectorbuilder.VerifySeal(sectorSize, pco.CommR[:], pco.CommD[:], addr, ticket.TicketBytes[:], seed.TicketBytes[:], sid, sco.Proof)
+	ok, err := sectorbuilder.VerifySeal(sectorSize, pco.CommR[:], pco.CommD[:], sb.Miner, ticket.TicketBytes[:], seed.TicketBytes[:], sid, sco.Proof)
 	if err != nil {
 		t.Fatal(err)
 	}