switch to async reading

Author: ischasny

Diff for: go.mod

@@ -377,3 +377,5 @@ require (
 	gonum.org/v1/gonum v0.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 )
+
+replace github.com/filecoin-project/go-data-segment v0.0.1 => github.com/ischasny/go-data-segment v0.0.0-20231107120541-53b3ec9a7c69

Diff for: itests/dummydeal_podsi_test.go

@@ -38,7 +38,7 @@ func TestDummyPodsiDealOnline(t *testing.T) {
 	os.Setenv("PODSI_USE_BUFFERED_READER", "true")
 	os.Setenv("PODSI_BUFFER_SIZE", "10000")
 
-	randomFileSize := int(4e6)
+	randomFileSize := int(1e6)
 
 	ctx := context.Background()
 	log := framework.Log

Diff for: piecedirectory/piecedirectory.go

@@ -40,6 +40,7 @@ import (
 	mh "github.com/multiformats/go-multihash"
 	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/sync/errgroup"
+	"golang.org/x/xerrors"
 )
 
 var log = logging.Logger("piecedirectory")
@@ -431,13 +432,34 @@ func parsePieceWithDataSegmentIndex(pieceCid cid.Cid, unpaddedSize int64, r type
 	}
 
 	start := time.Now()
-	dataSegments, err := datasegment.ParseDataSegmentIndex(rr)
-	if err != nil {
-		return nil, fmt.Errorf("could not parse data segment index: %w", err)
+
+	results := make(chan *datasegment.SegmentDesc)
+	var parseIndexErr error
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		parseIndexErr = datasegment.ParseDataSegmentIndexAsync(ctx, rr, results)
+		close(results)
+	}()
+
+	var segments []datasegment.SegmentDesc
+	cnt := -1
+	for res := range results {
+		cnt++
+		if err := res.Validate(); err != nil {
+			if errors.Is(err, datasegment.ErrValidation) {
+				continue
+			} else {
+				cancel()
+				return nil, xerrors.Errorf("could not calculate valid entries: got unknown error for entry %d: %w", cnt, err)
+			}
+		}
+		segments = append(segments, *res)
 	}
-	segments, err := dataSegments.ValidEntries()
-	if err != nil {
-		return nil, fmt.Errorf("could not calculate valid entries: %w", err)
+
+	cancel()
+
+	if parseIndexErr != nil {
+		return nil, fmt.Errorf("could not parse data segment index: %w", parseIndexErr)
 	}
 	if len(segments) == 0 {
 		return nil, fmt.Errorf("no data segments found")