diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 4c6d6438f..6c8485e08 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -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::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, 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::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 diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 32acd01e1..85165330a 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -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) diff --git a/weed/storage/store_ec_interval_read_test.go b/weed/storage/store_ec_interval_read_test.go new file mode 100644 index 000000000..dbab475ee --- /dev/null +++ b/weed/storage/store_ec_interval_read_test.go @@ -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)) + } +}