From fefac028ec25d18e229c02aa3b65758b03cbd9b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A7=D0=B5=D0=BF=D1=83=D1=80=D0=BD=D0=BE=D0=B2=20=D0=A1?= =?UTF-8?q?=D0=B5=D0=BC=D1=91=D0=BD?= Date: Sun, 25 Oct 2026 19:37:26 +0600 Subject: [PATCH] feat: send EC shards in parallel per destination disk (#12050) * feat: send EC shards in parallel per destination disk One stream per destination disk. max_shard_transfers unset means no global cap, 0 sends shards sequentially, and N is a global cap that still keeps one stream per disk. The value is persisted through TaskPolicy. The maintenance form is a checkbox plus Max transfers because unset and 0 differ. The key is intentionally omitted from GetConfigSpec. * ec shard distribution: treat negative max_shard_transfers as sequential The config layer (normalizedMaxShardTransfers) clamps negative values to 0, which selects sequential distribution. shardTransferCap diverged: a negative cap that reached ShardDistributionOptions directly would have selected uncapped parallel mode, the most aggressive setting. Align the options layer with the same normalization so invalid input degrades to the conservative mode. * ec: serialize shard transfers to a server when a target leaves disk choice to the receiver disk_id 0 is auto-placement on the receiving volume server, so an auto-placed stream can land on a disk another queued stream is pinned to. Grouping by (destNode, disk_id) alone let both run at once; collapse all of that server's transfers under the destNode key instead. * worker: clamp max_shard_transfers to the int32 wire range ReadIntConfig returned the 0 fallback for values above MaxInt32, turning a large positive cap into sequential mode. Read int64 and clamp into range instead. * worker: drop unused shard-transfer form helper The maintenance form renders and collects max_shard_transfers in plugin.templ JavaScript; the Go copy had no callers. * worker: bound max_shard_transfers conversion under an in-range guard --------- Co-authored-by: Chepurnov Simon <195125934+CHEPURNOV-S-S@users.noreply.github.com> Co-authored-by: Chris Lu --- weed/admin/dash/config_toml.go | 11 + weed/admin/dash/config_toml_test.go | 94 ++++ weed/admin/view/app/plugin.templ | 112 +++++ weed/admin/view/app/plugin_templ.go | 2 +- weed/command/scaffold/admin.toml | 3 + weed/pb/worker.proto | 1 + weed/pb/worker_pb/worker.pb.go | 31 +- .../erasure_coding/shard_distribution.go | 228 +++++++++- .../erasure_coding/shard_distribution_test.go | 428 ++++++++++++++++++ weed/worker/tasks/erasure_coding/config.go | 46 ++ .../erasure_coding/config_policy_test.go | 56 +++ weed/worker/tasks/erasure_coding/ec_task.go | 8 +- .../tasks/erasure_coding/plugin_handler.go | 37 ++ .../erasure_coding/plugin_handler_test.go | 50 ++ weed/worker/tasks/erasure_coding/register.go | 8 +- 15 files changed, 1089 insertions(+), 26 deletions(-) create mode 100644 weed/storage/erasure_coding/shard_distribution_test.go create mode 100644 weed/worker/tasks/erasure_coding/config_policy_test.go diff --git a/weed/admin/dash/config_toml.go b/weed/admin/dash/config_toml.go index bdc00a7328..a47c777617 100644 --- a/weed/admin/dash/config_toml.go +++ b/weed/admin/dash/config_toml.go @@ -111,6 +111,16 @@ func (cp *ConfigPersistence) ApplyMaintenanceConfigFromToml(v TomlConfig) error } ecChanged = true } + if k := "maintenance.erasure_coding.max_shard_transfers"; v.IsSet(k) { + n := v.GetInt(k) + if n < 0 { + n = 0 + } else if n > math.MaxInt32 { + n = math.MaxInt32 + } + ecConf.MaxShardTransfers = &n + ecChanged = true + } if !maintenanceChanged && !vacuumChanged && !balanceChanged && !ecChanged { return nil @@ -239,6 +249,7 @@ var pluginConfigSections = []pluginConfigSection{ "replica_placement": stringValue, "strict_placement": boolValue, "free_slots_deadband": int64Value, + "max_shard_transfers": int64Value, }, // workers read collection_filter from the admin values, not the worker values adminKeys: map[string]func(v TomlConfig, key string) *plugin_pb.ConfigValue{ diff --git a/weed/admin/dash/config_toml_test.go b/weed/admin/dash/config_toml_test.go index 111c3ccff7..e3ff08f9eb 100644 --- a/weed/admin/dash/config_toml_test.go +++ b/weed/admin/dash/config_toml_test.go @@ -307,3 +307,97 @@ preferred_tags = ["fast,SSD"] t.Errorf("preferred_tags = %v, want [fast ssd]", got) } } + +func TestApplyMaintenanceConfigFromTomlMaxShardTransfers(t *testing.T) { + dir := t.TempDir() + cp := NewConfigPersistence(dir) + if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, ` +[maintenance.erasure_coding] +fullness_ratio = 0.9 +`)); err != nil { + t.Fatalf("apply fullness: %v", err) + } + ec := erasure_coding.LoadConfigFromPersistence(cp) + if ec.MaxShardTransfers != nil { + t.Fatalf("absent key persisted as %d, want unset", *ec.MaxShardTransfers) + } + if ec.FullnessRatio != 0.9 { + t.Fatalf("fullness_ratio = %v, want 0.9", ec.FullnessRatio) + } + + if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, ` +[maintenance.erasure_coding] +max_shard_transfers = 0 +`)); err != nil { + t.Fatalf("apply 0: %v", err) + } + ec = erasure_coding.LoadConfigFromPersistence(cp) + if ec.MaxShardTransfers == nil || *ec.MaxShardTransfers != 0 { + t.Fatalf("0: got %v, want sequential 0", ec.MaxShardTransfers) + } + + if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, ` +[maintenance.erasure_coding] +max_shard_transfers = 4 +`)); err != nil { + t.Fatalf("apply 4: %v", err) + } + ec = erasure_coding.LoadConfigFromPersistence(cp) + if ec.MaxShardTransfers == nil || *ec.MaxShardTransfers != 4 { + t.Fatalf("4: got %v, want 4", ec.MaxShardTransfers) + } + + if err := cp.ApplyMaintenanceConfigFromToml(tomlConfig(t, ` +[maintenance.erasure_coding] +max_shard_transfers = -3 +`)); err != nil { + t.Fatalf("apply negative: %v", err) + } + ec = erasure_coding.LoadConfigFromPersistence(cp) + if ec.MaxShardTransfers == nil || *ec.MaxShardTransfers != 0 { + t.Fatalf("negative: got %v, want sequential 0", ec.MaxShardTransfers) + } +} + +func TestApplyPluginConfigFromTomlMaxShardTransfers(t *testing.T) { + s := newPluginServer(t) + seed := &plugin_pb.PersistedJobTypeConfig{ + JobType: "erasure_coding", + AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true}, + } + if err := s.GetPlugin().SaveJobTypeConfig(seed); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := s.ApplyPluginConfigFromToml(tomlConfig(t, ` +[maintenance.erasure_coding] +replica_placement = "020" +`)); err != nil { + t.Fatalf("apply placement: %v", err) + } + cfg, err := s.GetPlugin().LoadJobTypeConfig("erasure_coding") + if err != nil || cfg == nil { + t.Fatalf("load: %v %v", cfg, err) + } + if _, ok := cfg.WorkerConfigValues["max_shard_transfers"]; ok { + t.Fatal("absent max_shard_transfers was written into worker config") + } + + if err := s.ApplyPluginConfigFromToml(tomlConfig(t, ` +[maintenance.erasure_coding] +max_shard_transfers = 0 +`)); err != nil { + t.Fatalf("apply 0: %v", err) + } + cfg, err = s.GetPlugin().LoadJobTypeConfig("erasure_coding") + if err != nil || cfg == nil { + t.Fatalf("load 0: %v %v", cfg, err) + } + got, ok := cfg.WorkerConfigValues["max_shard_transfers"] + if !ok { + t.Fatal("max_shard_transfers = 0 missing from worker config") + } + if got.GetInt64Value() != 0 { + t.Fatalf("max_shard_transfers = %d, want 0", got.GetInt64Value()) + } +} diff --git a/weed/admin/view/app/plugin.templ b/weed/admin/view/app/plugin.templ index e1a3263a98..5aab757a9c 100644 --- a/weed/admin/view/app/plugin.templ +++ b/weed/admin/view/app/plugin.templ @@ -2272,6 +2272,105 @@ templ Plugin(page string, initialJob string, lane string) { return input; } + function shardTransferParallelOn(nativeValue) { + if (nativeValue === null || nativeValue === undefined || nativeValue === '') { + return true; + } + var n = parseInt(nativeValue, 10); + if (Number.isNaN(n)) { + return true; + } + return n !== 0; + } + + function shardTransferMaxText(nativeValue) { + if (nativeValue === null || nativeValue === undefined || nativeValue === '') { + return ''; + } + var n = parseInt(nativeValue, 10); + if (Number.isNaN(n) || n <= 0) { + return ''; + } + return String(n); + } + + // Presentation of the single persisted max_shard_transfers value. + // Unchecked stores 0. Checked with an empty max omits the key. + function renderShardTransferControls(scope, field, nativeValue) { + var parallelOn = shardTransferParallelOn(nativeValue); + var wrapper = document.createElement('div'); + wrapper.className = 'mb-3 plugin-field'; + wrapper.dataset.scope = scope; + wrapper.dataset.fieldName = field.name; + wrapper.dataset.fieldType = normalizeEnum(field.field_type); + wrapper.dataset.widget = 'SHARD_TRANSFERS'; + + var checkWrap = document.createElement('div'); + checkWrap.className = 'form-check form-switch mb-2'; + var box = document.createElement('input'); + box.type = 'checkbox'; + box.className = 'form-check-input'; + box.id = toInputID(scope, field.name) + '-parallel'; + box.checked = parallelOn; + var boxLabel = document.createElement('label'); + boxLabel.className = 'form-check-label'; + boxLabel.setAttribute('for', box.id); + boxLabel.textContent = 'Parallel shard distribution'; + checkWrap.appendChild(box); + checkWrap.appendChild(boxLabel); + wrapper.appendChild(checkWrap); + + var numberLabel = document.createElement('label'); + numberLabel.className = 'form-label'; + numberLabel.setAttribute('for', toInputID(scope, field.name)); + numberLabel.textContent = 'Max transfers'; + wrapper.appendChild(numberLabel); + + var input = document.createElement('input'); + input.type = 'number'; + input.min = '1'; + input.step = '1'; + input.className = 'form-control'; + input.id = toInputID(scope, field.name); + input.placeholder = 'no global cap'; + input.value = shardTransferMaxText(nativeValue); + input.disabled = !parallelOn; + wrapper.appendChild(input); + + box.addEventListener('change', function() { + input.disabled = !box.checked; + }); + + var help = document.createElement('div'); + help.className = 'form-text'; + help.textContent = field.description || 'Unchecked sends shards sequentially. Checked with an empty max uses one in-flight copy per destination disk. A number is a global cap, still one copy per disk.'; + wrapper.appendChild(help); + return wrapper; + } + + function collectShardTransferValue(scope) { + var box = document.getElementById(toInputID(scope, 'max_shard_transfers') + '-parallel'); + var input = document.getElementById(toInputID(scope, 'max_shard_transfers')); + if (!box || !input) { + return null; + } + if (!box.checked) { + return { int64_value: '0' }; + } + var raw = String(input.value || '').trim(); + if (raw === '') { + return null; + } + if (!/^\d+$/.test(raw)) { + throw new Error('Invalid int64 value for field Max transfers'); + } + var n = parseInt(raw, 10); + if (n <= 0) { + return { int64_value: '0' }; + } + return { int64_value: String(n) }; + } + function renderConfigForm(scope, rootID, form) { var root = document.getElementById(rootID); if (!root) { @@ -2321,6 +2420,11 @@ templ Plugin(page string, initialJob string, lane string) { var configValue = resolveFieldConfigValue(scope, field); var nativeValue = configValueToNative(configValue); + if (field.name === 'max_shard_transfers') { + sectionBody.appendChild(renderShardTransferControls(scope, field, nativeValue)); + continue; + } + var wrapper = document.createElement('div'); wrapper.className = 'mb-3 plugin-field'; wrapper.dataset.scope = scope; @@ -2440,6 +2544,14 @@ templ Plugin(page string, initialJob string, lane string) { continue; } + if (field.name === 'max_shard_transfers') { + var shardCap = collectShardTransferValue(scope); + if (shardCap !== null) { + out[field.name] = shardCap; + } + continue; + } + var wrapper = document.querySelector('.plugin-field[data-scope="' + scope + '"][data-field-name="' + field.name + '"]'); if (wrapper && wrapper.classList.contains('plugin-field-hidden')) { continue; diff --git a/weed/admin/view/app/plugin_templ.go b/weed/admin/view/app/plugin_templ.go index e6110ad062..7af5ba12b6 100644 --- a/weed/admin/view/app/plugin_templ.go +++ b/weed/admin/view/app/plugin_templ.go @@ -104,7 +104,7 @@ func Plugin(page string, initialJob string, lane string) templ.Component { if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 6, "

Workers

0

Active Jobs

0

Activities (recent)

0

Next Run

-

Per Job Type Summary
Job TypeActive JobsRecent Activities
Loading...
Scheduler State
Sequential scheduler with per-job runtime limits
Job TypeEnabledDetectorIn FlightMax RuntimeExec GlobalExec/WorkerExecutor WorkersEffective ExecLast Run
Loading...
Workers
WorkerAddressCapabilitiesLoad
Loading...
Job Type Configuration
Not loaded
Selected Job Type
-
Descriptor
Select a job type to load schema and config.
Admin Config Form
No admin form loaded.
Worker Config Form
No worker form loaded.
Job Scheduling Settings
How often to check for new work.
Per-attempt deadline for one job. Size-aware tasks may extend this automatically.
Next Run
Scheduler
-
Not scheduled
Run History
Keep last 10 success + last 10 errors
Successful Runs
TimeJob IDWorkerDuration
No data
Error Runs
TimeJob IDWorkerError
No data
Detection Results
Run detection to see proposals.
Job Queue
States: pending/assigned/running
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Detection Jobs
Detection activities for selected job type
TimeJob TypeRequest IDWorkerStageSourceMessage
Loading...
Execution Jobs
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Execution Activities
Non-detection events only
TimeJob TypeJob IDSourceStageMessage
Loading...
Job Detail
Select a job to view details.
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 6, "

Workers

0

Active Jobs

0

Activities (recent)

0

Next Run

-

Per Job Type Summary
Job TypeActive JobsRecent Activities
Loading...
Scheduler State
Sequential scheduler with per-job runtime limits
Job TypeEnabledDetectorIn FlightMax RuntimeExec GlobalExec/WorkerExecutor WorkersEffective ExecLast Run
Loading...
Workers
WorkerAddressCapabilitiesLoad
Loading...
Job Type Configuration
Not loaded
Selected Job Type
-
Descriptor
Select a job type to load schema and config.
Admin Config Form
No admin form loaded.
Worker Config Form
No worker form loaded.
Job Scheduling Settings
How often to check for new work.
Per-attempt deadline for one job. Size-aware tasks may extend this automatically.
Next Run
Scheduler
-
Not scheduled
Run History
Keep last 10 success + last 10 errors
Successful Runs
TimeJob IDWorkerDuration
No data
Error Runs
TimeJob IDWorkerError
No data
Detection Results
Run detection to see proposals.
Job Queue
States: pending/assigned/running
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Detection Jobs
Detection activities for selected job type
TimeJob TypeRequest IDWorkerStageSourceMessage
Loading...
Execution Jobs
Job IDTypeStateProgressWorkerUpdatedMessage
Loading...
Execution Activities
Non-detection events only
TimeJob TypeJob IDSourceStageMessage
Loading...
Job Detail
Select a job to view details.
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } diff --git a/weed/command/scaffold/admin.toml b/weed/command/scaffold/admin.toml index 99709abc03..3f43b4b439 100644 --- a/weed/command/scaffold/admin.toml +++ b/weed/command/scaffold/admin.toml @@ -69,6 +69,9 @@ # volume-slot deadband for shard placement: nodes whose free volume slots differ by at # most this much tie and fewer live shards wins; applied times data shards (0 = strict) # free_slots_deadband = 0 +# shard copy concurrency. Leave unset for one in-flight copy per destination disk. +# 0 sends shards sequentially. A positive N is a global cap, still one per disk. +# max_shard_transfers = 4 # max retry attempts for a failed erasure coding job # retry_limit = 1 # seconds to wait between retry attempts diff --git a/weed/pb/worker.proto b/weed/pb/worker.proto index 822f89269c..399c195514 100644 --- a/weed/pb/worker.proto +++ b/weed/pb/worker.proto @@ -382,6 +382,7 @@ message ErasureCodingTaskConfig { string replica_placement = 6; // EC shard replica placement (e.g. "020"); empty falls back to master default replication bool strict_placement = 7; // fail planning instead of relaxing placement constraints int32 free_slots_deadband = 8; // Place free-volume-slot deadband; |ΔfreeSlots| ≤ M×dataShards ranks by liveShards; 0 = strict + optional int32 max_shard_transfers = 9; // unset = no global cap (one in-flight per dest disk); 0 = sequential; N>0 = global cap } // BalanceTaskConfig contains balance-specific configuration diff --git a/weed/pb/worker_pb/worker.pb.go b/weed/pb/worker_pb/worker.pb.go index e2a81e3f5f..bfb05bfe0e 100644 --- a/weed/pb/worker_pb/worker.pb.go +++ b/weed/pb/worker_pb/worker.pb.go @@ -2974,14 +2974,15 @@ func (x *VacuumTaskConfig) GetMinIntervalSeconds() int32 { // ErasureCodingTaskConfig contains EC-specific configuration type ErasureCodingTaskConfig struct { state protoimpl.MessageState `protogen:"open.v1"` - FullnessRatio float64 `protobuf:"fixed64,1,opt,name=fullness_ratio,json=fullnessRatio,proto3" json:"fullness_ratio,omitempty"` // Minimum fullness ratio to trigger EC (0.0-1.0) - QuietForSeconds int32 `protobuf:"varint,2,opt,name=quiet_for_seconds,json=quietForSeconds,proto3" json:"quiet_for_seconds,omitempty"` // Minimum quiet time before EC - MinVolumeSizeMb int32 `protobuf:"varint,3,opt,name=min_volume_size_mb,json=minVolumeSizeMb,proto3" json:"min_volume_size_mb,omitempty"` // Minimum volume size for EC - CollectionFilter string `protobuf:"bytes,4,opt,name=collection_filter,json=collectionFilter,proto3" json:"collection_filter,omitempty"` // Only process volumes from specific collections - PreferredTags []string `protobuf:"bytes,5,rep,name=preferred_tags,json=preferredTags,proto3" json:"preferred_tags,omitempty"` // Disk tags to prioritize for EC shard placement - ReplicaPlacement string `protobuf:"bytes,6,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"` // EC shard replica placement (e.g. "020"); empty falls back to master default replication - StrictPlacement bool `protobuf:"varint,7,opt,name=strict_placement,json=strictPlacement,proto3" json:"strict_placement,omitempty"` // fail planning instead of relaxing placement constraints - FreeSlotsDeadband int32 `protobuf:"varint,8,opt,name=free_slots_deadband,json=freeSlotsDeadband,proto3" json:"free_slots_deadband,omitempty"` // Place free-volume-slot deadband; |ΔfreeSlots| ≤ M×dataShards ranks by liveShards; 0 = strict + FullnessRatio float64 `protobuf:"fixed64,1,opt,name=fullness_ratio,json=fullnessRatio,proto3" json:"fullness_ratio,omitempty"` // Minimum fullness ratio to trigger EC (0.0-1.0) + QuietForSeconds int32 `protobuf:"varint,2,opt,name=quiet_for_seconds,json=quietForSeconds,proto3" json:"quiet_for_seconds,omitempty"` // Minimum quiet time before EC + MinVolumeSizeMb int32 `protobuf:"varint,3,opt,name=min_volume_size_mb,json=minVolumeSizeMb,proto3" json:"min_volume_size_mb,omitempty"` // Minimum volume size for EC + CollectionFilter string `protobuf:"bytes,4,opt,name=collection_filter,json=collectionFilter,proto3" json:"collection_filter,omitempty"` // Only process volumes from specific collections + PreferredTags []string `protobuf:"bytes,5,rep,name=preferred_tags,json=preferredTags,proto3" json:"preferred_tags,omitempty"` // Disk tags to prioritize for EC shard placement + ReplicaPlacement string `protobuf:"bytes,6,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"` // EC shard replica placement (e.g. "020"); empty falls back to master default replication + StrictPlacement bool `protobuf:"varint,7,opt,name=strict_placement,json=strictPlacement,proto3" json:"strict_placement,omitempty"` // fail planning instead of relaxing placement constraints + FreeSlotsDeadband int32 `protobuf:"varint,8,opt,name=free_slots_deadband,json=freeSlotsDeadband,proto3" json:"free_slots_deadband,omitempty"` // Place free-volume-slot deadband; |ΔfreeSlots| ≤ M×dataShards ranks by liveShards; 0 = strict + MaxShardTransfers *int32 `protobuf:"varint,9,opt,name=max_shard_transfers,json=maxShardTransfers,proto3,oneof" json:"max_shard_transfers,omitempty"` // unset = no global cap (one in-flight per dest disk); 0 = sequential; N>0 = global cap unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3072,6 +3073,13 @@ func (x *ErasureCodingTaskConfig) GetFreeSlotsDeadband() int32 { return 0 } +func (x *ErasureCodingTaskConfig) GetMaxShardTransfers() int32 { + if x != nil && x.MaxShardTransfers != nil { + return *x.MaxShardTransfers + } + return 0 +} + // BalanceTaskConfig contains balance-specific configuration type BalanceTaskConfig struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4301,7 +4309,7 @@ const file_worker_proto_rawDesc = "" + "\x10VacuumTaskConfig\x12+\n" + "\x11garbage_threshold\x18\x01 \x01(\x01R\x10garbageThreshold\x12/\n" + "\x14min_volume_age_hours\x18\x02 \x01(\x05R\x11minVolumeAgeHours\x120\n" + - "\x14min_interval_seconds\x18\x03 \x01(\x05R\x12minIntervalSeconds\"\xf5\x02\n" + + "\x14min_interval_seconds\x18\x03 \x01(\x05R\x12minIntervalSeconds\"\xc2\x03\n" + "\x17ErasureCodingTaskConfig\x12%\n" + "\x0efullness_ratio\x18\x01 \x01(\x01R\rfullnessRatio\x12*\n" + "\x11quiet_for_seconds\x18\x02 \x01(\x05R\x0fquietForSeconds\x12+\n" + @@ -4310,7 +4318,9 @@ const file_worker_proto_rawDesc = "" + "\x0epreferred_tags\x18\x05 \x03(\tR\rpreferredTags\x12+\n" + "\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\x12)\n" + "\x10strict_placement\x18\a \x01(\bR\x0fstrictPlacement\x12.\n" + - "\x13free_slots_deadband\x18\b \x01(\x05R\x11freeSlotsDeadband\"\x9b\x01\n" + + "\x13free_slots_deadband\x18\b \x01(\x05R\x11freeSlotsDeadband\x123\n" + + "\x13max_shard_transfers\x18\t \x01(\x05H\x00R\x11maxShardTransfers\x88\x01\x01B\x16\n" + + "\x14_max_shard_transfers\"\x9b\x01\n" + "\x11BalanceTaskConfig\x12/\n" + "\x13imbalance_threshold\x18\x01 \x01(\x01R\x12imbalanceThreshold\x12(\n" + "\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\x12+\n" + @@ -4585,6 +4595,7 @@ func file_worker_proto_init() { (*TaskPolicy_ReplicationConfig)(nil), (*TaskPolicy_EcBalanceConfig)(nil), } + file_worker_proto_msgTypes[31].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ diff --git a/weed/storage/erasure_coding/shard_distribution.go b/weed/storage/erasure_coding/shard_distribution.go index 34ea64fbf9..8d311a6893 100644 --- a/weed/storage/erasure_coding/shard_distribution.go +++ b/weed/storage/erasure_coding/shard_distribution.go @@ -14,9 +14,66 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "golang.org/x/sync/errgroup" "google.golang.org/grpc" ) +// ShardDistributionOptions controls DistributeEcShards concurrency. +// MaxShardTransfers: +// - nil / unset: no global cap; one in-flight per (destNode, disk_id), or +// per destNode (host:port, not rack) when any target leaves disk choice +// to the receiver (disk_id 0) +// - N<=0: historical sequential nested loops +// - N>0: global cap of N, still at most one in-flight per key +type ShardDistributionOptions struct { + MaxShardTransfers *int + send shardSender +} + +type shardSender func(ctx context.Context, volumeID uint32, collection string, dialOption grpc.DialOption, destServer string, diskID uint32, filePath, shardType string) error + +type shardJob struct { + destNode string + shardType string + filePath string + diskID uint32 + hasDisk bool +} + +func shardSenderOrDefault(opts *ShardDistributionOptions) shardSender { + if opts != nil && opts.send != nil { + return opts.send + } + return sendShardFileToDestination +} + +func shardTransferCap(opts *ShardDistributionOptions) (sequential bool, capN int) { + if opts == nil || opts.MaxShardTransfers == nil { + return false, 0 + } + n := *opts.MaxShardTransfers + if n <= 0 { + return true, 0 + } + return false, n +} + +func lookupShardDisk(shardDisks map[string]map[string]uint32, destNode, shardType string) (uint32, bool) { + if byShard, ok := shardDisks[destNode]; ok { + if id, exists := byShard[shardType]; exists { + return id, true + } + } + return 0, false +} + +func shardTransferKey(j shardJob) string { + if j.hasDisk { + return fmt.Sprintf("%s#%d", j.destNode, j.diskID) + } + return j.destNode +} + func ensureLogger(logger logger) logger { if logger == nil { return &glogFallbackLogger{} @@ -98,7 +155,8 @@ func (g *glogFallbackLogger) Error(msg string, args ...interface{}) { // DistributeEcShards distributes locally generated EC shards to destination servers. // Returns the shard assignment map used for mounting. -func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb.TaskTarget, shardFiles map[string]string, dialOption grpc.DialOption, logger logger) (map[string][]string, error) { +// opts nil is the default: one in-flight per dest disk/node, no global cap. +func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb.TaskTarget, shardFiles map[string]string, dialOption grpc.DialOption, logger logger, opts *ShardDistributionOptions) (map[string][]string, error) { if len(targets) == 0 { return nil, fmt.Errorf("no targets specified for EC shard distribution") } @@ -173,6 +231,22 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb return nil, fmt.Errorf("no shard assignments found from planning phase") } + send := shardSenderOrDefault(opts) + sequential, capN := shardTransferCap(opts) + if sequential { + if err := distributeEcShardsSequential(volumeID, collection, shardAssignment, shardFiles, shardDisks, dialOption, log, send); err != nil { + return nil, err + } + } else if err := distributeEcShardsParallel(volumeID, collection, shardAssignment, shardFiles, shardDisks, dialOption, log, send, capN); err != nil { + return nil, err + } + + glog.V(1).Infof("Successfully distributed EC shards to %d destinations", len(shardAssignment)) + return shardAssignment, nil +} + +func distributeEcShardsSequential(volumeID uint32, collection string, shardAssignment map[string][]string, shardFiles map[string]string, shardDisks map[string]map[string]uint32, dialOption grpc.DialOption, log logger, send shardSender) error { + ctx := context.Background() for destNode, assignedShards := range shardAssignment { withFields(log, map[string]interface{}{ "destination": destNode, @@ -184,7 +258,7 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb for _, shardType := range assignedShards { filePath, exists := shardFiles[shardType] if !exists { - return nil, fmt.Errorf("shard file %s not found for destination %s", shardType, destNode) + return fmt.Errorf("shard file %s not found for destination %s", shardType, destNode) } if info, err := os.Stat(filePath); err == nil { @@ -198,12 +272,9 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb }).Info("Starting shard file transfer") } - diskID := uint32(0) - if byShard, ok := shardDisks[destNode]; ok { - diskID = byShard[shardType] - } - if err := sendShardFileToDestination(volumeID, collection, dialOption, destNode, diskID, filePath, shardType); err != nil { - return nil, fmt.Errorf("failed to send %s to %s: %w", shardType, destNode, err) + diskID, _ := lookupShardDisk(shardDisks, destNode, shardType) + if err := send(ctx, volumeID, collection, dialOption, destNode, diskID, filePath, shardType); err != nil { + return fmt.Errorf("failed to send %s to %s: %w", shardType, destNode, err) } withFields(log, map[string]interface{}{ @@ -219,9 +290,139 @@ func DistributeEcShards(volumeID uint32, collection string, targets []*worker_pb "total_mb": float64(transferredBytes) / (1024 * 1024), }).Info("All shards distributed to destination server") } + return nil +} - glog.V(1).Infof("Successfully distributed EC shards to %d destinations", len(shardAssignment)) - return shardAssignment, nil +func collectShardJobs(shardAssignment map[string][]string, shardFiles map[string]string, shardDisks map[string]map[string]uint32, log logger) (map[string][]shardJob, error) { + // disk_id 0 leaves placement to the receiver's auto-select, which can land + // on a disk another queued stream is pinned to — serialize every transfer + // to that server under one key. + autoPlacement := make(map[string]bool) + for destNode, assignedShards := range shardAssignment { + for _, shardType := range assignedShards { + if diskID, ok := lookupShardDisk(shardDisks, destNode, shardType); ok && diskID == 0 { + autoPlacement[destNode] = true + break + } + } + } + + byKey := make(map[string][]shardJob) + for destNode, assignedShards := range shardAssignment { + withFields(log, map[string]interface{}{ + "destination": destNode, + "assigned_shards": len(assignedShards), + "shard_types": assignedShards, + }).Info("Starting shard distribution to destination server") + + for _, shardType := range assignedShards { + filePath, exists := shardFiles[shardType] + if !exists { + return nil, fmt.Errorf("shard file %s not found for destination %s", shardType, destNode) + } + diskID, hasDisk := lookupShardDisk(shardDisks, destNode, shardType) + job := shardJob{ + destNode: destNode, + shardType: shardType, + filePath: filePath, + diskID: diskID, + hasDisk: hasDisk, + } + key := shardTransferKey(job) + if autoPlacement[destNode] { + key = destNode + } + byKey[key] = append(byKey[key], job) + } + } + return byKey, nil +} + +func distributeEcShardsParallel(volumeID uint32, collection string, shardAssignment map[string][]string, shardFiles map[string]string, shardDisks map[string]map[string]uint32, dialOption grpc.DialOption, log logger, send shardSender, capN int) error { + byKey, err := collectShardJobs(shardAssignment, shardFiles, shardDisks, log) + if err != nil { + return err + } + + g, ctx := errgroup.WithContext(context.Background()) + var sem chan struct{} + if capN > 0 { + sem = make(chan struct{}, capN) + } + + for _, jobs := range byKey { + jobs := jobs + g.Go(func() error { + for _, job := range jobs { + if err := ctx.Err(); err != nil { + return err + } + if sem != nil { + select { + case sem <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + if err := ctx.Err(); err != nil { + <-sem + return err + } + } + err := transferOneShard(ctx, volumeID, collection, dialOption, log, send, job) + if err != nil { + // Hold the slot so another key cannot start a send in the + // window before errgroup cancels ctx. + return err + } + if sem != nil { + <-sem + } + } + return nil + }) + } + + if err := g.Wait(); err != nil { + return err + } + + for destNode, assignedShards := range shardAssignment { + var transferredBytes int64 + for _, shardType := range assignedShards { + if filePath, ok := shardFiles[shardType]; ok { + if info, statErr := os.Stat(filePath); statErr == nil { + transferredBytes += info.Size() + } + } + } + withFields(log, map[string]interface{}{ + "destination": destNode, + "shards_transferred": len(assignedShards), + "total_bytes": transferredBytes, + "total_mb": float64(transferredBytes) / (1024 * 1024), + }).Info("All shards distributed to destination server") + } + return nil +} + +func transferOneShard(ctx context.Context, volumeID uint32, collection string, dialOption grpc.DialOption, log logger, send shardSender, job shardJob) error { + if info, err := os.Stat(job.filePath); err == nil { + withFields(log, map[string]interface{}{ + "destination": job.destNode, + "shard_type": job.shardType, + "file_path": job.filePath, + "size_bytes": info.Size(), + "size_kb": float64(info.Size()) / 1024, + }).Info("Starting shard file transfer") + } + if err := send(ctx, volumeID, collection, dialOption, job.destNode, job.diskID, job.filePath, job.shardType); err != nil { + return fmt.Errorf("failed to send %s to %s: %w", job.shardType, job.destNode, err) + } + withFields(log, map[string]interface{}{ + "destination": job.destNode, + "shard_type": job.shardType, + }).Info("Shard file transfer completed") + return nil } // MountEcShards mounts EC shards on destination servers using an assignment map. @@ -300,7 +501,10 @@ func MountEcShards(volumeID uint32, collection string, shardAssignment map[strin } // diskID=0 leaves disk placement to the server's auto-select. -func sendShardFileToDestination(volumeID uint32, collection string, dialOption grpc.DialOption, destServer string, diskID uint32, filePath, shardType string) error { +func sendShardFileToDestination(ctx context.Context, volumeID uint32, collection string, dialOption grpc.DialOption, destServer string, diskID uint32, filePath, shardType string) error { + if ctx == nil { + ctx = context.Background() + } return operation.WithVolumeServerClient(false, pb.ServerAddress(destServer), dialOption, func(client volume_server_pb.VolumeServerClient) error { file, err := os.Open(filePath) @@ -335,7 +539,7 @@ func sendShardFileToDestination(volumeID uint32, collection string, dialOption g return fmt.Errorf("unknown shard type: %s", shardType) } - stream, err := client.ReceiveFile(context.Background()) + stream, err := client.ReceiveFile(ctx) if err != nil { return fmt.Errorf("failed to create receive stream: %v", err) } diff --git a/weed/storage/erasure_coding/shard_distribution_test.go b/weed/storage/erasure_coding/shard_distribution_test.go new file mode 100644 index 0000000000..e99ad8bc4f --- /dev/null +++ b/weed/storage/erasure_coding/shard_distribution_test.go @@ -0,0 +1,428 @@ +package erasure_coding + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "google.golang.org/grpc" +) + +func intPtr(n int) *int { return &n } + +func writeShardFiles(t *testing.T, names ...string) map[string]string { + t.Helper() + dir := t.TempDir() + files := make(map[string]string, len(names)) + for _, name := range names { + p := filepath.Join(dir, name) + if err := os.WriteFile(p, []byte(name), 0644); err != nil { + t.Fatal(err) + } + files[name] = p + } + return files +} + +func shardTarget(node string, disk uint32, shards ...uint32) *worker_pb.TaskTarget { + return &worker_pb.TaskTarget{Node: node, DiskId: disk, ShardIds: shards} +} + +type flightRecorder struct { + mu sync.Mutex + inFlight int + maxFlight int + keyFlight map[string]int + keyMax int + attempts []string + sent []string + failOn string + failFirst bool + enter chan struct{} + goAhead chan struct{} + noGate atomic.Bool +} + +func newFlightRecorder() *flightRecorder { + return &flightRecorder{ + keyFlight: make(map[string]int), + enter: make(chan struct{}, 32), + goAhead: make(chan struct{}), + } +} + +func (r *flightRecorder) release() { + r.noGate.Store(true) + close(r.goAhead) +} + +func (r *flightRecorder) send(ctx context.Context, _ uint32, _ string, _ grpc.DialOption, destServer string, diskID uint32, _ string, shardType string) error { + id := destServer + "/" + shardType + key := shardTransferKey(shardJob{destNode: destServer, diskID: diskID, hasDisk: true}) + + r.mu.Lock() + r.inFlight++ + if r.inFlight > r.maxFlight { + r.maxFlight = r.inFlight + } + r.keyFlight[key]++ + if r.keyFlight[key] > r.keyMax { + r.keyMax = r.keyFlight[key] + } + r.attempts = append(r.attempts, id) + r.mu.Unlock() + + defer func() { + r.mu.Lock() + r.keyFlight[key]-- + r.inFlight-- + r.mu.Unlock() + }() + + if r.failOn == id || r.failFirst { + r.failFirst = false + return errors.New("mock send failed") + } + if !r.noGate.Load() { + select { + case r.enter <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + select { + case <-r.goAhead: + case <-ctx.Done(): + return ctx.Err() + } + } + r.mu.Lock() + r.sent = append(r.sent, id) + r.mu.Unlock() + return nil +} + +func waitEnters(t *testing.T, rec *flightRecorder, n int) { + t.Helper() + deadline := time.After(2 * time.Second) + for i := 0; i < n; i++ { + select { + case <-rec.enter: + case <-deadline: + t.Fatalf("timed out waiting for %d in-flight sends (got %d)", n, i) + } + } +} + +func assertNoExtraEnter(t *testing.T, rec *flightRecorder, wait time.Duration) { + t.Helper() + select { + case <-rec.enter: + t.Fatal("unexpected extra in-flight send") + case <-time.After(wait): + } +} + +func runDistribute(t *testing.T, targets []*worker_pb.TaskTarget, files map[string]string, rec *flightRecorder, max *int) (map[string][]string, error) { + t.Helper() + opts := &ShardDistributionOptions{ + MaxShardTransfers: max, + send: rec.send, + } + return DistributeEcShards(1, "col", targets, files, nil, nil, opts) +} + +func TestDistributeEcShardsZeroIsSequential(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01", "ec02", "ec03") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 0, 0), + shardTarget("full-01:8080", 1, 1), + shardTarget("full-02:8080", 0, 2), + shardTarget("full-02:8080", 1, 3), + } + rec := newFlightRecorder() + errCh := make(chan error, 1) + go func() { + _, err := runDistribute(t, targets, files, rec, intPtr(0)) + errCh <- err + }() + + waitEnters(t, rec, 1) + assertNoExtraEnter(t, rec, 40*time.Millisecond) + rec.mu.Lock() + if rec.maxFlight != 1 { + rec.mu.Unlock() + t.Fatalf("max in-flight=%d, want 1 for max_shard_transfers=0", rec.maxFlight) + } + rec.mu.Unlock() + rec.release() + if err := <-errCh; err != nil { + t.Fatal(err) + } + rec.mu.Lock() + defer rec.mu.Unlock() + if rec.maxFlight != 1 { + t.Fatalf("max in-flight=%d after release, want 1", rec.maxFlight) + } + if rec.keyMax != 1 { + t.Fatalf("per-key in-flight=%d, want 1", rec.keyMax) + } + if len(rec.sent) != 4 { + t.Fatalf("sent %d shards, want 4: %v", len(rec.sent), rec.sent) + } +} + +func TestDistributeEcShardsNoCapParallelByKey(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01", "ec02", "ec03") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 1, 0), + shardTarget("full-01:8080", 2, 1), + shardTarget("full-02:8080", 0, 2), + shardTarget("full-02:8081", 0, 3), // same rack host, different destNode + } + rec := newFlightRecorder() + errCh := make(chan error, 1) + go func() { + _, err := runDistribute(t, targets, files, rec, nil) + errCh <- err + }() + + waitEnters(t, rec, 4) + rec.mu.Lock() + if rec.maxFlight != 4 { + rec.mu.Unlock() + t.Fatalf("no cap: max in-flight=%d, want 4 (one per dest disk/node)", rec.maxFlight) + } + if rec.keyMax != 1 { + rec.mu.Unlock() + t.Fatalf("per-key in-flight=%d, want 1", rec.keyMax) + } + rec.mu.Unlock() + rec.release() + if err := <-errCh; err != nil { + t.Fatal(err) + } + rec.mu.Lock() + defer rec.mu.Unlock() + if len(rec.sent) != 4 { + t.Fatalf("sent %d, want 4: %v", len(rec.sent), rec.sent) + } +} + +func TestDistributeEcShardsSameDiskQueued(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 0, 0, 1), + } + rec := newFlightRecorder() + errCh := make(chan error, 1) + go func() { + _, err := runDistribute(t, targets, files, rec, nil) + errCh <- err + }() + + waitEnters(t, rec, 1) + assertNoExtraEnter(t, rec, 40*time.Millisecond) + rec.release() + if err := <-errCh; err != nil { + t.Fatal(err) + } + rec.mu.Lock() + defer rec.mu.Unlock() + if rec.maxFlight != 1 { + t.Fatalf("same disk: max in-flight=%d, want 1", rec.maxFlight) + } + if rec.sent[0] != "full-01:8080/ec00" || rec.sent[1] != "full-01:8080/ec01" { + t.Fatalf("same-disk order %v, want ec00 then ec01", rec.sent) + } +} + +func TestDistributeEcShardsAutoDiskSerializesServer(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01", "ec02") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 0, 0), // disk picked by the receiver + shardTarget("full-01:8080", 1, 1), + shardTarget("full-02:8080", 1, 2), + } + rec := newFlightRecorder() + errCh := make(chan error, 1) + go func() { + _, err := runDistribute(t, targets, files, rec, nil) + errCh <- err + }() + + waitEnters(t, rec, 2) + rec.mu.Lock() + if rec.maxFlight != 2 { + rec.mu.Unlock() + t.Fatalf("auto-placement server must serialize: in-flight=%d, want 2 (one per server)", rec.maxFlight) + } + rec.mu.Unlock() + rec.release() + if err := <-errCh; err != nil { + t.Fatal(err) + } + rec.mu.Lock() + defer rec.mu.Unlock() + if len(rec.sent) != 3 { + t.Fatalf("sent %d, want 3", len(rec.sent)) + } +} + +func TestDistributeEcShardsTwoDisksTwoStreams(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 1, 0), + shardTarget("full-01:8080", 2, 1), + } + rec := newFlightRecorder() + errCh := make(chan error, 1) + go func() { + _, err := runDistribute(t, targets, files, rec, nil) + errCh <- err + }() + + waitEnters(t, rec, 2) + rec.mu.Lock() + if rec.maxFlight != 2 { + rec.mu.Unlock() + t.Fatalf("two disks on one VS: in-flight=%d, want 2", rec.maxFlight) + } + rec.mu.Unlock() + rec.release() + if err := <-errCh; err != nil { + t.Fatal(err) + } +} + +func TestDistributeEcShardsCapNeverExceedsN(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01", "ec02", "ec03") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 1, 0), + shardTarget("full-01:8080", 2, 1), + shardTarget("full-02:8080", 1, 2), + shardTarget("full-02:8080", 2, 3), + } + rec := newFlightRecorder() + errCh := make(chan error, 1) + go func() { + _, err := runDistribute(t, targets, files, rec, intPtr(2)) + errCh <- err + }() + + waitEnters(t, rec, 2) + assertNoExtraEnter(t, rec, 40*time.Millisecond) + rec.mu.Lock() + if rec.maxFlight != 2 { + rec.mu.Unlock() + t.Fatalf("cap 2: in-flight=%d, want 2", rec.maxFlight) + } + rec.mu.Unlock() + rec.release() + if err := <-errCh; err != nil { + t.Fatal(err) + } + rec.mu.Lock() + defer rec.mu.Unlock() + if rec.maxFlight > 2 { + t.Fatalf("cap 2: max in-flight=%d", rec.maxFlight) + } + if rec.keyMax != 1 { + t.Fatalf("per-key in-flight=%d, want 1 under cap", rec.keyMax) + } + if len(rec.sent) != 4 { + t.Fatalf("sent %d, want 4", len(rec.sent)) + } +} + +func TestDistributeEcShardsErrorAbortSameKey(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01", "ec02") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 0, 0, 1, 2), + } + rec := newFlightRecorder() + rec.noGate.Store(true) + rec.failOn = "full-01:8080/ec01" + + _, err := runDistribute(t, targets, files, rec, nil) + if err == nil { + t.Fatal("expected send error") + } + if !strings.Contains(err.Error(), "failed to send ec01 to full-01:8080") { + t.Fatalf("error wrap: %v", err) + } + rec.mu.Lock() + defer rec.mu.Unlock() + if got := strings.Join(rec.attempts, ","); got != "full-01:8080/ec00,full-01:8080/ec01" { + t.Fatalf("attempts %v, want ec00 then ec01 (ec02 aborted)", rec.attempts) + } + if len(rec.sent) != 1 || rec.sent[0] != "full-01:8080/ec00" { + t.Fatalf("successful sends %v, want only ec00", rec.sent) + } +} + +func TestDistributeEcShardsErrorAbortCancelsOtherKeys(t *testing.T) { + files := writeShardFiles(t, "ec00", "ec01", "ec02") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 0, 0), + shardTarget("full-02:8080", 0, 1), + shardTarget("full-03:8080", 0, 2), + } + rec := newFlightRecorder() + rec.noGate.Store(true) + rec.failFirst = true + + _, err := runDistribute(t, targets, files, rec, intPtr(1)) + if err == nil { + t.Fatal("expected send error") + } + rec.mu.Lock() + defer rec.mu.Unlock() + if len(rec.attempts) != 1 { + t.Fatalf("cap=1 abort: attempts %v, want only the failing send", rec.attempts) + } + if len(rec.sent) != 0 { + t.Fatalf("successful sends %v, want none", rec.sent) + } +} + +func TestDistributeEcShardsMissingFile(t *testing.T) { + files := writeShardFiles(t, "ec00") + targets := []*worker_pb.TaskTarget{ + shardTarget("full-01:8080", 0, 1), + } + for _, max := range []*int{intPtr(0), nil} { + _, err := DistributeEcShards(1, "col", targets, files, nil, nil, &ShardDistributionOptions{ + MaxShardTransfers: max, + send: func(context.Context, uint32, string, grpc.DialOption, string, uint32, string, string) error { + t.Fatal("send should not run for missing file") + return nil + }, + }) + if err == nil || !strings.Contains(err.Error(), "shard file ec01 not found") { + t.Fatalf("missing file error (max=%v): %v", max, err) + } + } +} + +func TestShardTransferKeyUsesDestNodeNotRack(t *testing.T) { + a := shardTransferKey(shardJob{destNode: "full-01:8080", diskID: 0, hasDisk: true}) + b := shardTransferKey(shardJob{destNode: "full-01:8081", diskID: 0, hasDisk: true}) + if a == b { + t.Fatalf("host:port keys collided: %s", a) + } + noDisk := shardTransferKey(shardJob{destNode: "full-01:8080", hasDisk: false}) + if noDisk != "full-01:8080" { + t.Fatalf("no disk id key=%q, want destNode", noDisk) + } + if fmt.Sprint(a) == noDisk { + t.Fatalf("disk 0 key should differ from destNode-only when hasDisk") + } +} diff --git a/weed/worker/tasks/erasure_coding/config.go b/weed/worker/tasks/erasure_coding/config.go index e1eba8fb03..7a56b217da 100644 --- a/weed/worker/tasks/erasure_coding/config.go +++ b/weed/worker/tasks/erasure_coding/config.go @@ -2,6 +2,7 @@ package erasure_coding import ( "fmt" + "math" "github.com/seaweedfs/seaweedfs/weed/admin/config" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -20,6 +21,9 @@ type Config struct { ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication StrictPlacement bool `json:"strict_placement"` // fail planning instead of relaxing placement constraints FreeSlotsDeadband int `json:"free_slots_deadband"` // Place volume-slot M; |ΔfreeSlots| ≤ M×dataShards is a tie, then liveShards; 0 = strict + // MaxShardTransfers is the global in-flight cap for EC shard copies. + // nil = no cap (one in-flight per dest disk); 0 = sequential; N>0 = cap. + MaxShardTransfers *int `json:"max_shard_transfers,omitempty"` } // NewDefaultConfig creates a new default erasure coding configuration @@ -224,11 +228,51 @@ func (c *Config) ToTaskPolicy() *worker_pb.TaskPolicy { ReplicaPlacement: c.ReplicaPlacement, StrictPlacement: c.StrictPlacement, FreeSlotsDeadband: int32(c.FreeSlotsDeadband), + MaxShardTransfers: optionalInt32(c.MaxShardTransfers), }, }, } } +// optionalInt32 keeps unset (nil) distinct from sequential (0) on the proto field. +func optionalInt32(n *int) *int32 { + n = normalizedMaxShardTransfers(n) + if n == nil { + return nil + } + v := int32(*n) + return &v +} + +// normalizedMaxShardTransfers clamps a set cap into the proto int32 range. +// nil stays nil so an absent knob is not stored as sequential. +func normalizedMaxShardTransfers(n *int) *int { + if n == nil { + return nil + } + v := *n + if v < 0 { + v = 0 + } + if v > math.MaxInt32 { + v = math.MaxInt32 + } + return &v +} + +func optionalIntFromProto(n *int32) *int { + if n == nil { + return nil + } + v := int(*n) + return &v +} + +// cloneMaxShardTransfers copies the cap so a task does not alias the live config. +func cloneMaxShardTransfers(n *int) *int { + return normalizedMaxShardTransfers(n) +} + // FromTaskPolicy loads configuration from a TaskPolicy protobuf message func (c *Config) FromTaskPolicy(policy *worker_pb.TaskPolicy) error { if policy == nil { @@ -250,6 +294,8 @@ func (c *Config) FromTaskPolicy(policy *worker_pb.TaskPolicy) error { c.ReplicaPlacement = ecConfig.ReplicaPlacement c.StrictPlacement = ecConfig.StrictPlacement c.FreeSlotsDeadband = int(ecConfig.FreeSlotsDeadband) + // Read the pointer, not GetMaxShardTransfers: the getter maps unset to 0. + c.MaxShardTransfers = normalizedMaxShardTransfers(optionalIntFromProto(ecConfig.MaxShardTransfers)) } return nil diff --git a/weed/worker/tasks/erasure_coding/config_policy_test.go b/weed/worker/tasks/erasure_coding/config_policy_test.go new file mode 100644 index 0000000000..6c465f1d71 --- /dev/null +++ b/weed/worker/tasks/erasure_coding/config_policy_test.go @@ -0,0 +1,56 @@ +package erasure_coding + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +func TestMaxShardTransfersPolicyRoundTrip(t *testing.T) { + cases := []struct { + name string + in *int + }{ + {name: "unset", in: nil}, + {name: "sequential", in: intPtr(0)}, + {name: "cap", in: intPtr(4)}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cfg := NewDefaultConfig() + cfg.MaxShardTransfers = cloneMaxShardTransfers(tc.in) + + raw, err := proto.Marshal(cfg.ToTaskPolicy()) + require.NoError(t, err) + var decoded worker_pb.TaskPolicy + require.NoError(t, proto.Unmarshal(raw, &decoded)) + + restored := NewDefaultConfig() + require.NoError(t, restored.FromTaskPolicy(&decoded)) + ec := decoded.GetErasureCodingConfig() + require.NotNil(t, ec) + if tc.in == nil { + require.Nil(t, ec.MaxShardTransfers) + require.Nil(t, restored.MaxShardTransfers) + return + } + require.NotNil(t, ec.MaxShardTransfers) + require.Equal(t, int32(*tc.in), *ec.MaxShardTransfers) + require.NotNil(t, restored.MaxShardTransfers) + require.Equal(t, *tc.in, *restored.MaxShardTransfers) + }) + } +} + +func TestMaxShardTransfersNegativeClampsToSequential(t *testing.T) { + cfg := NewDefaultConfig() + cfg.MaxShardTransfers = intPtr(-3) + restored := NewDefaultConfig() + require.NoError(t, restored.FromTaskPolicy(cfg.ToTaskPolicy())) + require.NotNil(t, restored.MaxShardTransfers) + require.Equal(t, 0, *restored.MaxShardTransfers) +} + +func intPtr(n int) *int { return &n } diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index b9df6e3ed3..77a9db780d 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -64,6 +64,10 @@ type ErasureCodingTask struct { // with, read back off the EC context. Every holder must report serving the // same one before the source volume may be deleted. encodedBlockSize int64 + + // maxShardTransfers is the optional global cap for shard fan-out. + // nil = no cap (one in-flight per dest disk/node); 0 = sequential; N>0 = cap. + maxShardTransfers *int } // NewErasureCodingTask creates a new unified EC task instance @@ -797,7 +801,9 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string // distributeEcShards distributes locally generated EC shards to destination servers // using pre-assigned shard IDs from planning phase func (t *ErasureCodingTask) distributeEcShards(shardFiles map[string]string) error { - assignment, err := erasure_coding.DistributeEcShards(t.volumeID, t.collection, t.targets, shardFiles, t.grpcDialOption, t.GetLogger()) + assignment, err := erasure_coding.DistributeEcShards(t.volumeID, t.collection, t.targets, shardFiles, t.grpcDialOption, t.GetLogger(), &erasure_coding.ShardDistributionOptions{ + MaxShardTransfers: t.maxShardTransfers, + }) if err != nil { return err } diff --git a/weed/worker/tasks/erasure_coding/plugin_handler.go b/weed/worker/tasks/erasure_coding/plugin_handler.go index 5071529b6d..51b7dbbcb0 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler.go @@ -3,6 +3,7 @@ package erasure_coding import ( "context" "fmt" + "math" "path/filepath" "strings" "time" @@ -174,6 +175,20 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor { }, }, }, + { + SectionId: "shard_distribution", + Title: "Shard distribution", + Description: "How many EC shard copies are in flight. The checkbox is presentation only; the saved value is max_shard_transfers.", + Fields: []*plugin_pb.ConfigField{ + { + Name: "max_shard_transfers", + Label: "Parallel shard distribution", + Description: "Unchecked sends shards sequentially. Checked with an empty max uses one in-flight copy per destination disk. A number is a global cap, still one copy per disk.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + }, + }, + }, }, DefaultValues: map[string]*plugin_pb.ConfigValue{ "quiet_for_seconds": { @@ -545,6 +560,7 @@ func (h *ErasureCodingHandler) Execute( h.grpcDialOption, ) task.workerAddress = h.workerAddress + task.maxShardTransfers = deriveErasureCodingWorkerConfig(request.GetWorkerConfigValues()).TaskConfig.MaxShardTransfers execCtx, execCancel := context.WithCancel(ctx) defer execCancel() task.SetProgressCallback(func(progress float64, stage string) { @@ -669,12 +685,33 @@ func deriveErasureCodingWorkerConfig(values map[string]*plugin_pb.ConfigValue) * deadband = 100000 } taskConfig.FreeSlotsDeadband = deadband + taskConfig.MaxShardTransfers = maxShardTransfersFromValues(values) return &erasureCodingWorkerConfig{ TaskConfig: taskConfig, } } +// maxShardTransfersFromValues returns nil when the key is absent so shard +// distribution keeps the uncapped default. 0 is sequential; N>0 is a global cap. +func maxShardTransfersFromValues(values map[string]*plugin_pb.ConfigValue) *int { + if values == nil { + return nil + } + value, ok := values["max_shard_transfers"] + if !ok || value == nil { + return nil + } + n := pluginworker.ReadInt64Config(values, "max_shard_transfers", 0) + v := 0 + if n > math.MaxInt32 { + v = math.MaxInt32 + } else if n > 0 { + v = int(n) + } + return &v +} + func buildErasureCodingProposal( result *workertypes.TaskDetectionResult, baseWorkingDir string, diff --git a/weed/worker/tasks/erasure_coding/plugin_handler_test.go b/weed/worker/tasks/erasure_coding/plugin_handler_test.go index 71e221ce4d..237d2819c9 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler_test.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler_test.go @@ -2,6 +2,7 @@ package erasure_coding import ( "context" + "math" "strings" "testing" "time" @@ -139,6 +140,46 @@ func TestDeriveErasureCodingWorkerConfig(t *testing.T) { if cfg.TaskConfig.FreeSlotsDeadband != 12 { t.Fatalf("expected free_slots_deadband 12, got %d", cfg.TaskConfig.FreeSlotsDeadband) } + if cfg.TaskConfig.MaxShardTransfers != nil { + t.Fatalf("missing max_shard_transfers must stay unset, got %d", *cfg.TaskConfig.MaxShardTransfers) + } +} + +func TestDeriveErasureCodingWorkerConfigMaxShardTransfers(t *testing.T) { + if got := deriveErasureCodingWorkerConfig(nil).TaskConfig.MaxShardTransfers; got != nil { + t.Fatalf("nil values: got %d, want unset", *got) + } + if got := deriveErasureCodingWorkerConfig(map[string]*plugin_pb.ConfigValue{}).TaskConfig.MaxShardTransfers; got != nil { + t.Fatalf("missing key: got %d, want unset", *got) + } + + cfg := deriveErasureCodingWorkerConfig(map[string]*plugin_pb.ConfigValue{ + "max_shard_transfers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + }) + if cfg.TaskConfig.MaxShardTransfers == nil || *cfg.TaskConfig.MaxShardTransfers != 0 { + t.Fatalf("0: got %v, want 0 (sequential)", cfg.TaskConfig.MaxShardTransfers) + } + + cfg = deriveErasureCodingWorkerConfig(map[string]*plugin_pb.ConfigValue{ + "max_shard_transfers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}}, + }) + if cfg.TaskConfig.MaxShardTransfers == nil || *cfg.TaskConfig.MaxShardTransfers != 4 { + t.Fatalf("4: got %v, want 4", cfg.TaskConfig.MaxShardTransfers) + } + + cfg = deriveErasureCodingWorkerConfig(map[string]*plugin_pb.ConfigValue{ + "max_shard_transfers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -3}}, + }) + if cfg.TaskConfig.MaxShardTransfers == nil || *cfg.TaskConfig.MaxShardTransfers != 0 { + t.Fatalf("negative: got %v, want 0", cfg.TaskConfig.MaxShardTransfers) + } + + cfg = deriveErasureCodingWorkerConfig(map[string]*plugin_pb.ConfigValue{ + "max_shard_transfers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: math.MaxInt32 + 1}}, + }) + if cfg.TaskConfig.MaxShardTransfers == nil || *cfg.TaskConfig.MaxShardTransfers != math.MaxInt32 { + t.Fatalf("above int32: got %v, want clamp to %d", cfg.TaskConfig.MaxShardTransfers, int64(math.MaxInt32)) + } } func TestBuildErasureCodingProposal(t *testing.T) { @@ -291,6 +332,15 @@ func TestErasureCodingDescriptorExposesFreeSlotsDeadband(t *testing.T) { if sectionID != "placement" { t.Fatalf("free_slots_deadband section = %q, want placement", sectionID) } + if !workerConfigFormHasField(descriptor.WorkerConfigForm, "max_shard_transfers") { + t.Fatal("worker config form omitted max_shard_transfers") + } + if _, ok := descriptor.WorkerConfigForm.DefaultValues["max_shard_transfers"]; ok { + t.Fatal("max_shard_transfers default would collapse unset into 0") + } + if _, ok := descriptor.WorkerDefaultValues["max_shard_transfers"]; ok { + t.Fatal("worker default max_shard_transfers would collapse unset into 0") + } if prev != "placement" { t.Fatalf("replica_placement section = %q, want placement next to free_slots_deadband", prev) } diff --git a/weed/worker/tasks/erasure_coding/register.go b/weed/worker/tasks/erasure_coding/register.go index ad7e48cc82..821e5eba14 100644 --- a/weed/worker/tasks/erasure_coding/register.go +++ b/weed/worker/tasks/erasure_coding/register.go @@ -51,13 +51,17 @@ func RegisterErasureCodingTask() { if len(params.Sources) == 0 { return nil, fmt.Errorf("at least one source is required for erasure coding task") } - return NewErasureCodingTask( + task := NewErasureCodingTask( fmt.Sprintf("erasure_coding-%d", params.VolumeId), params.Sources[0].Node, // Use first source node params.VolumeId, params.Collection, dialOpt, - ), nil + ) + if cfg, ok := globalTaskDef.Config.(*Config); ok { + task.maxShardTransfers = cloneMaxShardTransfers(cfg.MaxShardTransfers) + } + return task, nil }, DetectionFunc: func(metrics []*types.VolumeHealthMetrics, clusterInfo *types.ClusterInfo, config base.TaskConfig) ([]*types.TaskDetectionResult, error) { results, _, err := Detection(context.Background(), metrics, clusterInfo, config, 0)