mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-14 18:18:35 +02:00
Чепурнов СемёнandChepurnov Simon
624a13ecc6
* feat: prefer same-host worker and local EC replica Admin executor pick ranked idle workers only, so a job whose volume lived on full-03 could still run on full-01 and pull the whole .dat across the LAN. Rank workers whose Hello host matches any replica first (fallback to others if the local worker is full). Detection named Sources[0] / the lowest server id; execute then SyncAndSelectBestReplica and could switch the copy source to another rack. After sync, if a replica on this machine has a complete file count, copy from that URL so affinity and pull stay on one host. * feat: pin scheduler jobs to replica-host workers until load K Reserve prefers the least-loaded replica-host worker and waits while that load is below 5 instead of spilling to a free remote worker. K=0 restores immediate spill. Waterfill uses reservation load so four idle preferred workers each get a job. * feat: park scheduled jobs on per-worker admin queues Assign a detection batch to the least-loaded replica-host worker in one pass, then let ExecutionConcurrency goroutines pop a home-runnable job or steal when that home queue reaches K. Dispatch no longer holds a job in reserve-backoff, so a free worker is not blocked by another home's pin. Tracked jobs stay pending/queued and record the home worker for the admin UI. * fix: rehome queued jobs when the home worker dies A live home still pins below K. A home missing from live executors is stolen immediately onto a free worker so a short orphaned queue does not sit until the dispatch window ends. * feat: wait before stealing a dead home worker queue Orphaned queues are held for 30s so a worker reconnect blink does not dump jobs onto other machines. After the grace, steal without K as before. Zero disables the wait. A home that comes back during grace keeps its queue. * fix: late-bind EC from to worker hostname After replicas are consistent (equal FileCount), rebind encode source to a volume server on the same hostname as worker -address. Any port on that host is local. Diverged FileCount still uses the max replica. * feat: assign homeless jobs to global FIFO Jobs whose replica hostname has no registered worker skip pin and orphan grace. Free workers take own home, then steal at K/orphan, then leftover FIFO. * fix: take homeless FIFO before steal for no-home workers Workers without a replica-host match take leftover first; steal (K/orphan) only when FIFO is empty. Home workers keep own-queue, steal, then leftover. --------- Co-authored-by: Chepurnov Simon <195125934+CHEPURNOV-S-S@users.noreply.github.com>
67 lines
2.0 KiB
Go
67 lines
2.0 KiB
Go
package plugin
|
|
|
|
import (
|
|
"strings"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
|
|
pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
// JobReplicaHosts are the volume-replica hosts a job will read from.
|
|
// Used so executor pick prefers a worker on one of those machines.
|
|
// Sources with ShardIds are EC-cleanup rows, not encode replicas.
|
|
func JobReplicaHosts(job *plugin_pb.JobSpec) []string {
|
|
if job == nil {
|
|
return nil
|
|
}
|
|
var addrs []string
|
|
addrs = append(addrs, pluginworker.ReadStringListConfig(job.Parameters, "replica_servers")...)
|
|
if src := strings.TrimSpace(pluginworker.ReadStringConfig(job.Parameters, "source_server", "")); src != "" {
|
|
addrs = append(addrs, src)
|
|
}
|
|
if payload := pluginworker.ReadBytesConfig(job.Parameters, "task_params_pb"); len(payload) > 0 {
|
|
params := &worker_pb.TaskParams{}
|
|
if err := proto.Unmarshal(payload, params); err == nil {
|
|
for _, s := range params.GetSources() {
|
|
if s == nil || s.Node == "" || len(s.ShardIds) > 0 {
|
|
continue
|
|
}
|
|
addrs = append(addrs, s.Node)
|
|
}
|
|
}
|
|
}
|
|
return util.HostKeysFromAddresses(addrs)
|
|
}
|
|
|
|
// partitionByPreferredHost splits candidates into (host matches, others).
|
|
// No preferred hosts, or no matches, returns (nil, all) so callers keep
|
|
// the existing slots+round-robin path.
|
|
func partitionByPreferredHost(candidates []*WorkerSession, preferredHosts []string) (pref, rest []*WorkerSession) {
|
|
want := util.HostKeysFromAddresses(preferredHosts)
|
|
if len(want) == 0 || len(candidates) == 0 {
|
|
return nil, candidates
|
|
}
|
|
set := make(map[string]struct{}, len(want))
|
|
for _, h := range want {
|
|
set[h] = struct{}{}
|
|
}
|
|
for _, c := range candidates {
|
|
if c == nil {
|
|
continue
|
|
}
|
|
if _, ok := set[util.HostKey(c.Address)]; ok {
|
|
pref = append(pref, c)
|
|
continue
|
|
}
|
|
rest = append(rest, c)
|
|
}
|
|
if len(pref) == 0 {
|
|
return nil, candidates
|
|
}
|
|
return pref, rest
|
|
}
|