ec: read a needle's intervals in parallel (#10911)

* ec: read a needle's intervals in parallel A needle spanning more than one EC block gets one interval per block, and consecutive blocks live on different shards. We read those intervals in sequence, so a 4MB chunk landing in a volume's 1MB small-block region cost five round trips to five different servers. Read them concurrently into disjoint slices of a single buffer, at most 8 in flight. Same change in the Rust volume server's phase C. * ec test: seed the random payload instead of the deprecated rand.Read
2026-08-28 11:56:07 +00:00 · 2026-08-24 14:03:44 -07:00
parent 69cc2869ad
commit 51eb5333d3
3 changed files with 204 additions and 42 deletions
@@ -33,6 +33,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};

 use futures::future::join_all;
+use futures::stream::{self, StreamExt};
 use reed_solomon_erasure::galois_8::ReedSolomon;
 use tonic::Request;

@@ -49,6 +50,10 @@ use crate::storage::store_ec_reconcile::EcVolumeMissingIndex;
 use crate::storage::types::*;
 use crate::storage::volume::volume_file_name;

+/// Bounds the fan-out of a single needle read. Mirrors Go's
+/// `ecIntervalReadConcurrency`.
+const INTERVAL_READ_CONCURRENCY: usize = 8;
+
 /// One interval's data after Phase A.
 enum IntervalResult {
    /// Already read from a locally-mounted shard.
@@ -90,7 +95,7 @@ pub async fn read_ec_shard_needle_distributed(
    // intervals, and read any locally-mounted shard intervals. We must
    // not `.await` while holding this guard (std::sync::RwLockReadGuard
    // is !Send).
-    let snapshot = match snapshot_under_lock(state, vid, needle_id)? {
+    let mut snapshot = match snapshot_under_lock(state, vid, needle_id)? {
        Some(s) => s,
        None => return Ok(None),
    };
@@ -138,39 +143,56 @@ pub async fn read_ec_shard_needle_distributed(
        }
    }

-    // Phase C — fetch missing intervals, reconstructing when the
-    // direct peer read fails.
-    let mut assembled: Vec<Vec<u8>> = Vec::with_capacity(snapshot.intervals.len());
-    for res in snapshot.intervals {
-        match res {
-            IntervalResult::Local(buf) => assembled.push(buf),
-            IntervalResult::NeedRemote {
-                shard_id,
-                shard_offset,
-                size,
-            } => {
-                let (buf, is_deleted) = fetch_one_interval(
-                    state,
-                    vid,
-                    needle_id,
+    // Phase C — fetch missing intervals, reconstructing when the direct peer
+    // read fails. Blocks that follow each other in the .dat live on different
+    // shards, so a needle spanning several of them costs one round trip per
+    // block when fetched in sequence; `buffered` keeps the order while letting
+    // INTERVAL_READ_CONCURRENCY of them fly at once.
+    let data_shards = snapshot.data_shards as usize;
+    let parity_shards = snapshot.parity_shards as usize;
+    let encode_ts_ns = snapshot.encode_ts_ns;
+    let intervals = std::mem::take(&mut snapshot.intervals);
+    let fetched: Vec<io::Result<(Vec<u8>, bool)>> = stream::iter(intervals.into_iter().map(|res| {
+        let shard_locations = &shard_locations;
+        async move {
+            match res {
+                IntervalResult::Local(buf) => Ok((buf, false)),
+                IntervalResult::NeedRemote {
                    shard_id,
                    shard_offset,
                    size,
-                    &shard_locations,
-                    snapshot.data_shards as usize,
-                    snapshot.parity_shards as usize,
-                    snapshot.encode_ts_ns,
-                )
-                .await?;
-                // A peer reports the needle deleted (a cross-server window where the
-                // local index still shows it live): treat as not-found rather than
-                // serving zeros, mirroring Go's ErrorDeleted.
-                if is_deleted {
-                    return Ok(None);
+                } => {
+                    fetch_one_interval(
+                        state,
+                        vid,
+                        needle_id,
+                        shard_id,
+                        shard_offset,
+                        size,
+                        shard_locations,
+                        data_shards,
+                        parity_shards,
+                        encode_ts_ns,
+                    )
+                    .await
                }
-                assembled.push(buf);
            }
        }
+    }))
+    .buffered(INTERVAL_READ_CONCURRENCY)
+    .collect()
+    .await;
+
+    let mut assembled: Vec<Vec<u8>> = Vec::with_capacity(fetched.len());
+    for res in fetched {
+        let (buf, is_deleted) = res?;
+        // A peer reports the needle deleted (a cross-server window where the
+        // local index still shows it live): treat as not-found rather than
+        // serving zeros, mirroring Go's ErrorDeleted.
+        if is_deleted {
+            return Ok(None);
+        }
+        assembled.push(buf);
    }

    // Phase D — assemble and parse the Needle. Mirrors the tail of
@@ -436,31 +436,65 @@ func (s *Store) IntervalToShardIdAndOffset(iv erasure_coding.Interval) (erasure_
 	return iv.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize)
 }

+// ecIntervalReadConcurrency bounds the fan-out of a single needle read. Blocks
+// that follow each other in the .dat live on different shards, so a needle
+// spanning several of them costs one round trip per block when read in sequence.
+const ecIntervalReadConcurrency = 8
+
 func (s *Store) readEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, is_deleted bool, err error) {
 	if err = s.cachedLookupEcShardLocations(ecVolume); err != nil {
 		return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)
 	}

-	for i, interval := range intervals {
-		if d, isDeleted, e := s.readOneEcShardInterval(needleId, ecVolume, interval); e != nil {
-			return nil, isDeleted, e
-		} else {
-			if isDeleted {
-				is_deleted = true
-			}
-			if i == 0 {
-				data = d
-			} else {
-				data = append(data, d...)
+	var totalSize int
+	for _, interval := range intervals {
+		totalSize += int(interval.Size)
+	}
+	data = make([]byte, totalSize)
+
+	if len(intervals) <= 1 {
+		for _, interval := range intervals {
+			if is_deleted, err = s.readOneEcShardInterval(needleId, ecVolume, interval, data); err != nil {
+				return nil, is_deleted, err
 			}
 		}
+		return data, is_deleted, nil
 	}
-	return
+
+	errs := make([]error, len(intervals))
+	var deleted atomic.Bool
+	var wg sync.WaitGroup
+	sem := make(chan struct{}, ecIntervalReadConcurrency)
+	var pos int
+	for i, interval := range intervals {
+		buf := data[pos : pos+int(interval.Size)]
+		pos += int(interval.Size)
+		wg.Add(1)
+		sem <- struct{}{}
+		go func() {
+			defer wg.Done()
+			defer func() { <-sem }()
+			isDeleted, e := s.readOneEcShardInterval(needleId, ecVolume, interval, buf)
+			if isDeleted {
+				deleted.Store(true)
+			}
+			errs[i] = e
+		}()
+	}
+	wg.Wait()
+
+	is_deleted = deleted.Load()
+	for _, e := range errs {
+		if e != nil {
+			return nil, is_deleted, e
+		}
+	}
+	return data, is_deleted, nil
 }

-func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, interval erasure_coding.Interval) (data []byte, is_deleted bool, err error) {
+// readOneEcShardInterval fills data, which must be interval.Size long.
+func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, interval erasure_coding.Interval, data []byte) (is_deleted bool, err error) {
 	shardId, actualOffset := s.IntervalToShardIdAndOffset(interval)
-	data = make([]byte, interval.Size)

 	// try local read
 	err = s.readLocalEcShardInterval(ecVolume, shardId, data, actualOffset)
@@ -0,0 +1,106 @@
+package storage
+
+import (
+	"bytes"
+	"math/rand"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
+)
+
+// A needle larger than one EC block is split over consecutive blocks, which
+// live on different shards. The intervals are read concurrently, so check they
+// still come back in order.
+func TestReadEcShardNeedleSpanningBlocks(t *testing.T) {
+	store := newTestStore(t, 1)
+	dir := store.Locations[0].Directory
+	const vid = needle.VolumeId(7)
+
+	v, err := NewVolume(dir, dir, "", vid, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("new volume: %v", err)
+	}
+	n := new(needle.Needle)
+	n.Id = types.Uint64ToNeedleId(42)
+	n.Data = make([]byte, 3*erasure_coding.ErasureCodingSmallBlockSize+1234)
+	rand.New(rand.NewSource(42)).Read(n.Data)
+	n.Checksum = needle.NewCRC(n.Data)
+	if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	baseFileName := v.DataFileName()
+	v.Close()
+
+	datSize, err := os.Stat(baseFileName + ".dat")
+	if err != nil {
+		t.Fatalf("stat .dat: %v", err)
+	}
+	if _, err := erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()); err != nil {
+		t.Fatalf("write ec files: %v", err)
+	}
+	if err := erasure_coding.WriteSortedFileFromIdx(baseFileName, ".ecx"); err != nil {
+		t.Fatalf("write .ecx: %v", err)
+	}
+	if err := os.WriteFile(baseFileName+".ecj", nil, 0o644); err != nil {
+		t.Fatalf("write .ecj: %v", err)
+	}
+	if err := volume_info.SaveVolumeInfo(baseFileName+".vif", &volume_server_pb.VolumeInfo{
+		Version:     uint32(needle.GetCurrentVersion()),
+		DatFileSize: datSize.Size(),
+		EcShardConfig: &volume_server_pb.EcShardConfig{
+			DataShards:   erasure_coding.DataShardsCount,
+			ParityShards: erasure_coding.ParityShardsCount,
+		},
+	}); err != nil {
+		t.Fatalf("save .vif: %v", err)
+	}
+	for _, ext := range []string{".dat", ".idx"} {
+		if err := os.Remove(baseFileName + ext); err != nil {
+			t.Fatalf("remove %s: %v", ext, err)
+		}
+	}
+
+	for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ {
+		if err := store.MountEcShards("", vid, erasure_coding.ShardId(shardId), ""); err != nil {
+			t.Fatalf("mount shard %d: %v", shardId, err)
+		}
+	}
+	ecVolume, found := store.Locations[0].FindEcVolume(vid)
+	if !found {
+		t.Fatal("ec volume not mounted")
+	}
+
+	// Every shard is local, so seed the location cache to keep the read off the
+	// master this test does not have.
+	ecVolume.ShardLocationsLock.Lock()
+	for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ {
+		ecVolume.ShardLocations[erasure_coding.ShardId(shardId)] = []pb.ServerAddress{"localhost:8080"}
+	}
+	ecVolume.ShardLocationsRefreshTime = time.Now()
+	ecVolume.ShardLocationsLock.Unlock()
+
+	_, _, intervals, err := ecVolume.LocateEcShardNeedle(n.Id, ecVolume.Version)
+	if err != nil {
+		t.Fatalf("locate needle: %v", err)
+	}
+	if len(intervals) < 2 {
+		t.Fatalf("needle covers %d interval(s), want it split over several", len(intervals))
+	}
+
+	got := new(needle.Needle)
+	got.Id = n.Id
+	if _, err := store.ReadEcShardNeedle(vid, got, nil); err != nil {
+		t.Fatalf("read ec needle: %v", err)
+	}
+	if !bytes.Equal(got.Data, n.Data) {
+		t.Fatalf("read back %d bytes, want the %d written", len(got.Data), len(n.Data))
+	}
+}