Compare commits

..
4 Commits
Author SHA1 Message Date
Ilker ManapandClaude Opus 5.5 3bb52742a4 v0.16.3: changelog and README
Release / build (push) Successful in 5m11s
Release / notes (push) Successful in 4s
CI / test (push) Successful in 10m10s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 09:43:40 +01:00
Ilker ManapandClaude Opus 5.5 1a3f385be4 Every setting of a served model from the mirror page, and its replicas per profile
CI / test (push) Successful in 10m51s
The row of a served model leads to its card, which now changes its context
and longest answer, its cards, its replicas on each profile with a profile
on CPUs to add and its speed beside, stops, starts and restarts it, and
removes it with its files or without. Stopped on the row stops the model
rather than setting its scaling to nothing. The console gains Replicas per
profile, and ctl models shows the counts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 09:40:06 +01:00
Ilker ManapandClaude Opus 5.5 e23732fbbb Replicas counted per profile: seven on cards and four on CPUs under one name
A profile's replicas, set on any of a model's profiles, is how many of its
replicas run there; the scaling becomes the sum. The reconciler starts the
ones short of their profile before stopping the ones over, so a change of
split never runs fewer. ctl model replicas sets them, adding a profile on
CPUs to a GGUF model; the router already fills the higher-priority pool
first, which the CPU profile is put below.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 09:21:50 +01:00
Ilker ManapandClaude Opus 5.5 6403e61f4b How fast a model answers on the CPU, before it is served
CI / test (push) Successful in 9m31s
Each agent measures one core's memory rate beside the whole machine's; the
estimate takes a replica's threads at about one core's rate each, up to the
machine's, calibrated by llama-bench on the PoC. The serve form shows it
beside a card's, warns when slow, and asks for the CPU on purpose when a
reply would take minutes; ctl model plan, the console and serve jobs say
it too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 08:52:15 +01:00
50 changed files with 1974 additions and 89 deletions

No files matched your search

+49 -1
View File
@@ -20,12 +20,60 @@ is only released once its **exit criterion** is met, not when its code lands.
| v0.13.0 | — | The whole system stopped and started from one command; Qwen3Guard judges, on a GPU and CPUs together | released |
| v0.14.0 | — | Knowledge bases: documents from people's own machines, kept apart per tenant, answered from | released (v0.14.1: adapters, the tree over the API, the firewall on every machine; v0.14.2: pictures from ComfyUI; v0.14.3: 3D models as jobs; v0.14.4: any ComfyUI workflow as a job, presets, licences; v0.14.5: picture and 3D models that add themselves; v0.14.6: setup that explains every part and lets it be chosen, the gateway on all three HA machines; v0.14.7: the judge sized for its machine; v0.14.8: sign-in with Active Directory, several keys for one outside provider; v0.14.9: sign-in set up from the console, a node's console that shows the machine) |
| v0.15.0 | — | The company's MCP tools and VS Code through the gateway; guardrails that ask outside services and put redacted values back; answers to similar questions | released (v0.15.1: Strata, a 125B model of experts on one card; a second mirror kept in step; pickled weights refused; a served model's context changed; v0.15.2: a served model moved to more cards; v0.15.3: a model stopped and kept, or removed with its files) |
| v0.16.0 | — | Weights pinned to a commit and checked against what Hugging Face published; documents embedded by the gateway; Strata offered where models are chosen; cluster upgrades that say how far they are | released (v0.16.1: the mirror's Strata tab, Strata's ready-made packs; v0.16.2: standbys that answer clients, FreeToken, a model fetched once per node) |
| v0.16.0 | — | Weights pinned to a commit and checked against what Hugging Face published; documents embedded by the gateway; Strata offered where models are chosen; cluster upgrades that say how far they are | released (v0.16.1: the mirror's Strata tab, Strata's ready-made packs; v0.16.2: standbys that answer clients, FreeToken, a model fetched once per node; v0.16.3: replicas per profile, a served model's settings on the mirror page, the CPU's speed before serving) |
| v1.0.0 | 5 | What phase 5 has left: precise KV routing proven on vLLM, the block cost picker, defragmentation across NVLink domains, prefill/decode disaggregation | planned |
Until v1.0.0 the minor version is the breaking one: configuration formats and Go
APIs change between minors without a deprecation period.
## v0.16.3 — 2026-11-03
Replicas counted per profile — seven on cards and four on CPUs under one name, the cards filled first — and every
setting of a served model from the mirror page; how fast a model answers on the CPU, said before it is served, by
the threads it is given rather than the whole machine's memory.
**Upgrading.**
- No migrations. `llmgw ctl upgrade v0.16.3`: the control planes carry the replicas per profile, the model card and
the speed in the serve form and in `ctl model plan`; the agents carry the measurement of one core's memory rate,
without which the speed is reckoned the old way, from the whole machine's.
### Added
- **Replicas counted per profile: seven on cards and four on CPUs, under one
name.** `llmgw ctl model replicas <model> gpu=7 cpu=4`, the console's
Replicas per profile… and the mirror page's model card set how many of a
model's replicas run on each of its profiles, rather than every replica
on the best profile that fits. For a GGUF model with no profile on CPUs
one is added (`--cpu-memory`), preferred below the cards, so the router
fills the cards first and sends to the CPUs once they are full. A change
starts the replicas short before stopping those over: the model never
runs fewer. `ctl models` shows the counts; a scaling of another total
ends them.
- **Every setting of a served model from the mirror page.** A served
model's row has a Settings link, and its card now changes the context
and longest answer, the cards a replica takes, the replicas on each
profile (with the CPU's speed beside), stops and starts it, restarts its
replicas, and takes it out of the catalog with its files or without —
beside the replicas, class, owner, memory and adapters it held before.
Stopped on the row is now the model stopped and kept, started again as
it was, rather than its scaling set to nothing.
- **How fast a model answers on the CPU, before it is served.** The serve
form says, as the cores are changed, about how many tokens a second one
reply gets on the CPU and on one card of the fleet's fastest class, and
warns when it is slow (under 15) or very slow (under 5, a reply of 500
tokens taking minutes). Very slow on the CPU is refused until "run on the
CPU anyway" is ticked, GPUs free or not. `llmgw ctl model plan` and the
console's Hugging Face screen show the same figure for every way a model
can run, and `ctl model serve --cpu` says it in its log.
- **The speed follows the threads a replica is given.** Each agent now
measures what one core reads beside what all of them read: a replica of
four threads on a large server reads at about four cores' rate, not the
server's, which the estimate used to assume and overstated several times.
Calibrated on the PoC by llama-bench at 1, 2, 4 and 8 threads (within 15%),
and a model of experts reads its experts' share, from the GGUF header.
Agents need upgrading to report one core's rate; until then the old
estimate is used.
## v0.16.2 — 2026-11-02
Standby control planes that answer clients themselves, with rate limits and budgets kept across them; FreeToken, an
+1 -1
View File
@@ -42,7 +42,7 @@ Releases track the roadmap phases; see [CHANGELOG.md](CHANGELOG.md).
| v0.13.0 | — | The whole system stopped and started from one command; Qwen3Guard judges, on a GPU and CPUs together | released |
| v0.14.0 | — | Knowledge bases: documents from people's own machines, kept apart per tenant, answered from | released (v0.14.1: adapters, the tree over the API, the firewall on every machine; v0.14.2: pictures from ComfyUI; v0.14.3: 3D models as jobs; v0.14.4: any ComfyUI workflow as a job, presets, licences; v0.14.5: picture and 3D models that add themselves; v0.14.6: setup that explains every part and lets it be chosen, the gateway on all three HA machines; v0.14.7: the judge sized for its machine; v0.14.8: sign-in with Active Directory, several keys for one outside provider; v0.14.9: sign-in set up from the console, a node's console that shows the machine) |
| v0.15.0 | — | The company's MCP tools and VS Code through the gateway; guardrails that ask outside services and put redacted values back; answers to similar questions | released (v0.15.1: Strata, a 125B model of experts on one card; a second mirror kept in step; pickled weights refused; a served model's context changed; v0.15.2: a served model moved to more cards; v0.15.3: a model stopped and kept, or removed with its files) |
| v0.16.0 | — | Weights pinned to a commit and checked against what Hugging Face published; documents embedded by the gateway; Strata offered where models are chosen; cluster upgrades that say how far they are | released (v0.16.1: the mirror's Strata tab, Strata's ready-made packs; v0.16.2: standbys that answer clients, FreeToken, a model fetched once per node) |
| v0.16.0 | — | Weights pinned to a commit and checked against what Hugging Face published; documents embedded by the gateway; Strata offered where models are chosen; cluster upgrades that say how far they are | released (v0.16.1: the mirror's Strata tab, Strata's ready-made packs; v0.16.2: standbys that answer clients, FreeToken, a model fetched once per node; v0.16.3: replicas per profile, a served model's settings on the mirror page, the CPU's speed before serving) |
| v1.0.0 | 5 | What phase 5 has left: precise KV routing proven on vLLM, the block cost picker, defragmentation across NVLink domains, prefill/decode disaggregation | planned |
As of v0.6.0 a cluster runs unattended: agents report their hardware, the
+30 -8
View File
@@ -1210,8 +1210,12 @@ type Host struct {
// mem_bandwidth_gbs is how fast the memory reads, in GB/s, measured when
// the agent started: what decides a model's speed on the CPU. 0 unknown.
MemBandwidthGbs float64 `protobuf:"fixed64,7,opt,name=mem_bandwidth_gbs,json=memBandwidthGbs,proto3" json:"mem_bandwidth_gbs,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// mem_bandwidth_core_gbs is what one core reads alone, in GB/s: a replica
// given a few threads of a large machine reads at a few times this, not at
// the whole machine's rate. 0 unknown.
MemBandwidthCoreGbs float64 `protobuf:"fixed64,8,opt,name=mem_bandwidth_core_gbs,json=memBandwidthCoreGbs,proto3" json:"mem_bandwidth_core_gbs,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Host) Reset() {
@@ -1293,6 +1297,13 @@ func (x *Host) GetMemBandwidthGbs() float64 {
return 0
}
func (x *Host) GetMemBandwidthCoreGbs() float64 {
if x != nil {
return x.MemBandwidthCoreGbs
}
return 0
}
// Storage is one filesystem a node keeps weights on.
//
// Per filesystem rather than per configured directory: free space belongs to
@@ -3049,8 +3060,10 @@ type Heartbeat struct {
// the agent has measured it — in the background after it starts, so a
// registration usually has none yet. 0 means not measured.
MemBandwidthGbs float64 `protobuf:"fixed64,20,opt,name=mem_bandwidth_gbs,json=memBandwidthGbs,proto3" json:"mem_bandwidth_gbs,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// mem_bandwidth_core_gbs is one core's read rate, as in Host.
MemBandwidthCoreGbs float64 `protobuf:"fixed64,21,opt,name=mem_bandwidth_core_gbs,json=memBandwidthCoreGbs,proto3" json:"mem_bandwidth_core_gbs,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Heartbeat) Reset() {
@@ -3223,6 +3236,13 @@ func (x *Heartbeat) GetMemBandwidthGbs() float64 {
return 0
}
func (x *Heartbeat) GetMemBandwidthCoreGbs() float64 {
if x != nil {
return x.MemBandwidthCoreGbs
}
return 0
}
// Provision is the software a node should have: which engines, at which
// versions. Nothing more.
//
@@ -5324,7 +5344,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\x1aA\n" +
"\x13EngineVersionsEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xdd\x01\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x92\x02\n" +
"\x04Host\x12\x13\n" +
"\x05os_id\x18\x01 \x01(\tR\x04osId\x12\x1d\n" +
"\n" +
@@ -5334,7 +5354,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1b\n" +
"\tcpu_flags\x18\x05 \x03(\tR\bcpuFlags\x12#\n" +
"\rnvidia_driver\x18\x06 \x01(\tR\fnvidiaDriver\x12*\n" +
"\x11mem_bandwidth_gbs\x18\a \x01(\x01R\x0fmemBandwidthGbs\"u\n" +
"\x11mem_bandwidth_gbs\x18\a \x01(\x01R\x0fmemBandwidthGbs\x123\n" +
"\x16mem_bandwidth_core_gbs\x18\b \x01(\x01R\x13memBandwidthCoreGbs\"u\n" +
"\aStorage\x12\x12\n" +
"\x04path\x18\x01 \x01(\tR\x04path\x12\x1d\n" +
"\n" +
@@ -5520,7 +5541,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\x06status\x18\x03 \x01(\tR\x06status\x12!\n" +
"\fcluster_uuid\x18\x04 \x01(\tR\vclusterUuid\x12\x1b\n" +
"\tclique_id\x18\x05 \x01(\tR\bcliqueId\x12\x1c\n" +
"\tbandwidth\x18\x06 \x01(\tR\tbandwidth\"\xd5\a\n" +
"\tbandwidth\x18\x06 \x01(\tR\tbandwidth\"\x8a\b\n" +
"\tHeartbeat\x12\x17\n" +
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12+\n" +
"\x12gpu_memory_used_mb\x18\x02 \x03(\x05R\x0fgpuMemoryUsedMb\x12!\n" +
@@ -5543,7 +5564,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\x04imex\x18\x11 \x01(\v2\x0e.llmgw.v1.IMEXR\x04imex\x12\x1b\n" +
"\tcpu_count\x18\x12 \x01(\x05R\bcpuCount\x121\n" +
"\x15gpu_foreign_memory_mb\x18\x13 \x03(\x05R\x12gpuForeignMemoryMb\x12*\n" +
"\x11mem_bandwidth_gbs\x18\x14 \x01(\x01R\x0fmemBandwidthGbs\x1aA\n" +
"\x11mem_bandwidth_gbs\x18\x14 \x01(\x01R\x0fmemBandwidthGbs\x123\n" +
"\x16mem_bandwidth_core_gbs\x18\x15 \x01(\x01R\x13memBandwidthCoreGbs\x1aA\n" +
"\x13EngineVersionsEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\";\n" +
+6
View File
@@ -241,6 +241,10 @@ message Host {
// mem_bandwidth_gbs is how fast the memory reads, in GB/s, measured when
// the agent started: what decides a model's speed on the CPU. 0 unknown.
double mem_bandwidth_gbs = 7;
// mem_bandwidth_core_gbs is what one core reads alone, in GB/s: a replica
// given a few threads of a large machine reads at a few times this, not at
// the whole machine's rate. 0 unknown.
double mem_bandwidth_core_gbs = 8;
}
// Storage is one filesystem a node keeps weights on.
@@ -618,6 +622,8 @@ message Heartbeat {
// the agent has measured it — in the background after it starts, so a
// registration usually has none yet. 0 means not measured.
double mem_bandwidth_gbs = 20;
// mem_bandwidth_core_gbs is one core's read rate, as in Host.
double mem_bandwidth_core_gbs = 21;
}
// Provision is the software a node should have: which engines, at which
+101
View File
@@ -3227,6 +3227,58 @@
}
}
},
"/api/v1/models/{name}/replicas": {
"post": {
"operationId": "postModelsNameReplicas",
"summary": "How many replicas run on each profile",
"description": "Seven on cards and four on CPUs, under the model's name. A profile not named keeps its count; cpu adds a profile on CPUs to a llama.cpp model that has none. Its scaling becomes the sum.",
"tags": [
"models"
],
"parameters": [
{
"name": "name",
"in": "path",
"required": true,
"description": "The model's name in the catalog.",
"schema": {
"type": "string"
}
},
{
"$ref": "#/components/parameters/Actor"
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProfileReplicasRequest"
}
}
}
},
"responses": {
"200": {
"description": "OK",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/catalog.Model"
}
}
}
},
"401": {
"$ref": "#/components/responses/Unauthorized"
},
"default": {
"$ref": "#/components/responses/Error"
}
}
}
},
"/api/v1/models/{name}/resources": {
"post": {
"operationId": "postModelsNameResources",
@@ -8103,6 +8155,13 @@
"type": "string"
}
},
"profile_replicas": {
"type": "object",
"additionalProperties": {
"type": "integer"
},
"description": "ProfileReplicas is how many replicas run on each profile, for a model whose replicas are counted per profile; absent otherwise."
},
"memory_gb": {
"type": "integer",
"description": "MemoryGB and CPUCores are what one replica reserves, from its first profile; MeasuredMemoryGB is the most one of its replicas uses now, 0 when none is measured."
@@ -8305,6 +8364,10 @@
"type": "number",
"description": "MemBandwidthGBs is how fast its memory reads, in GB/s, measured when its agent started; 0 when not reported. A model's speed on the CPU."
},
"mem_bandwidth_core_gbs": {
"type": "number",
"description": "MemBandwidthCoreGBs is what one core reads alone, in GB/s; 0 when not reported. A CPU replica's threads read at about this each."
},
"engines": {
"type": "array",
"items": {
@@ -8661,6 +8724,21 @@
}
}
},
"ProfileReplicasRequest": {
"type": "object",
"description": "ProfileReplicasRequest counts a model's replicas per profile: profile name to replicas, and CPU, for a llama.cpp model with no profile on CPUs, the one to add.",
"properties": {
"profiles": {
"type": "object",
"additionalProperties": {
"type": "integer"
}
},
"cpu": {
"$ref": "#/components/schemas/modelsvc.CPUProfile"
}
}
},
"ProjectView": {
"type": "object",
"description": "ProjectView is a project, with how many people and keys are in it.",
@@ -10506,6 +10584,10 @@
"type": "integer",
"description": "Priority orders profiles during placement; highest that fits wins."
},
"replicas": {
"type": "integer",
"description": "Replicas, set on any of a model's profiles, is how many of its replicas run on this one: seven on cards and four on CPUs, all answering under the model's name, rather than each replica on the best profile that fits. The model's scaling is then the sum, min and max alike; a profile left at zero runs none."
},
"startup_timeout": {
"type": "integer",
"format": "int64",
@@ -11406,6 +11488,10 @@
},
"strata_size": {
"type": "string"
},
"tokens_per_second": {
"type": "number",
"description": "TokensPerSecond is about how many tokens a second one reply gets this way, on the fastest host it fits; 0 when not known. On the CPU with the threads a replica is given by default: half the node's."
}
}
},
@@ -11898,6 +11984,21 @@
}
}
},
"modelsvc.CPUProfile": {
"type": "object",
"description": "CPUProfile is a profile on CPUs to add to a llama.cpp model, for replicas counted per profile: what a replica of it reserves.",
"properties": {
"name": {
"type": "string"
},
"memory_gb": {
"type": "integer"
},
"cpu_cores": {
"type": "integer"
}
}
},
"notify.Result": {
"type": "object",
"description": "Result is how one webhook answered a test.",
+11
View File
@@ -295,6 +295,7 @@ rather than repeated.
| `keep_within` | `string` | [Profile](#profile) |
| `weight` | `float64` | [Profile](#profile) |
| `priority` | `int` | [Profile](#profile) |
| `replicas` | `int` | [Profile](#profile) |
| `startup_timeout` | `time.Duration` | [Profile](#profile) |
| `kv_events` | `bool` | [Profile](#profile) |
| `gpu` | `*GPUNeed` | [Profile](#profile) |
@@ -2611,6 +2612,16 @@ rather than a guess (plan §6.2.1).
Priority orders profiles during placement; highest that fits wins.
### `replicas`
`int`
Replicas, set on any of a model's profiles, is how many of its
replicas run on this one: seven on cards and four on CPUs, all
answering under the model's name, rather than each replica on the
best profile that fits. The model's scaling is then the sum, min and
max alike; a profile left at zero runs none.
### `startup_timeout`
`time.Duration`
+26 -10
View File
@@ -519,7 +519,8 @@ func (a *Agent) scan(ctx context.Context) error {
inv, err := discovery.Discover(ctx, opts, a.log)
host := discovery.DiscoverHost()
host.MinComputeCapability = discovery.MinComputeCapability(inv.GPUs)
host.MemBandwidthGBs = a.memBandwidth()
bw := a.memBandwidth()
host.MemBandwidthGBs, host.MemBandwidthCoreGBs = bw.All, bw.Core
a.mu.Lock()
a.inv = inv
a.prov.Host = host
@@ -578,7 +579,8 @@ func (a *Agent) session(ctx context.Context, server string) error {
Arch: h.Arch, CpuFlags: h.CPUFlags, NvidiaDriver: h.NVIDIADriver,
// What is measured by now, not what the startup scan found before
// the measurement ended: a reconnect would otherwise say 0.
MemBandwidthGbs: a.memBandwidth(),
MemBandwidthGbs: a.memBandwidth().All,
MemBandwidthCoreGbs: a.memBandwidth().Core,
}
if err := stream.Send(&llmgwv1.AgentMessage{
@@ -676,7 +678,8 @@ func (a *Agent) heartbeat() *llmgwv1.AgentMessage {
// without anybody restarting anything.
hb.MemoryTotalMb = int64(discovery.MemoryTotalMB())
hb.CpuCount = int32(runtime.NumCPU())
hb.MemBandwidthGbs = a.memBandwidth()
bw := a.memBandwidth()
hb.MemBandwidthGbs, hb.MemBandwidthCoreGbs = bw.All, bw.Core
a.reprobeEngines()
a.mu.Lock()
@@ -1573,20 +1576,32 @@ func (a *Agent) reportCompute(ctx context.Context, out chan<- *llmgwv1.AgentMess
// VMs on one host that all restart together measure each other, and read a
// third less than any of them gets alone. Once a process rather than an
// agent, so tests running many agents in one read it once.
func (a *Agent) memBandwidth() float64 {
func (a *Agent) memBandwidth() discovery.Bandwidth {
bwOnce.Do(func() {
go func() {
time.Sleep(bandwidthDelay())
for {
if bw := measureBandwidth(); bw > math.Float64frombits(bwBits.Load()) {
bwBits.Store(math.Float64bits(bw))
a.log.Info("this node's memory", "read_gbs", math.Round(bw*10)/10)
// The best of each seen so far: an hourly pass on a node
// busy serving reads less, and that is not the memory.
bw := measureBandwidth()
cur := a.memBandwidthNow()
if bw.All > cur.All || bw.Core > cur.Core {
bwBits.Store(math.Float64bits(max(bw.All, cur.All)))
coreBits.Store(math.Float64bits(max(bw.Core, cur.Core)))
a.log.Info("this node's memory", "read_gbs", math.Round(bw.All*10)/10,
"one_core_gbs", math.Round(bw.Core*10)/10)
}
time.Sleep(time.Hour)
}
}()
})
return math.Float64frombits(bwBits.Load())
return a.memBandwidthNow()
}
// memBandwidthNow is what has been measured so far, without starting a
// measurement.
func (a *Agent) memBandwidthNow() discovery.Bandwidth {
return discovery.Bandwidth{All: math.Float64frombits(bwBits.Load()), Core: math.Float64frombits(coreBits.Load())}
}
// bandwidthDelay is how long an agent waits before measuring, so agents
@@ -1594,8 +1609,9 @@ func (a *Agent) memBandwidth() float64 {
var bandwidthDelay = func() time.Duration { return 10*time.Second + rand.N(110*time.Second) }
var (
bwOnce sync.Once
bwBits atomic.Uint64
bwOnce sync.Once
bwBits atomic.Uint64
coreBits atomic.Uint64
// measureBandwidth is replaced in tests, which have no use for it.
measureBandwidth = discovery.MeasureMemoryBandwidth
)
+4
View File
@@ -35,6 +35,10 @@ type Host struct {
// measured: what decides a model's speed on the CPU. Not read here but
// measured by the agent when it starts (MeasureMemoryBandwidth).
MemBandwidthGBs float64 `json:"mem_bandwidth_gbs,omitempty"`
// MemBandwidthCoreGBs is what one core reads alone, in GB/s, 0 when not
// measured: a replica given n threads reads at about n times this, up to
// MemBandwidthGBs.
MemBandwidthCoreGBs float64 `json:"mem_bandwidth_core_gbs,omitempty"`
}
// Family is the packaging family: "deb", "rpm", or "" when neither is known.
+28 -5
View File
@@ -28,9 +28,22 @@ const (
minBuffer = 256 << 20
)
// MeasureMemoryBandwidth is the memory's read rate in GB/s, the best of a
// few passes; 0 when the machine has too little free memory to measure.
func MeasureMemoryBandwidth() float64 {
// Bandwidth is how fast this machine's memory reads, in GB/s: with every
// CPU reading, and with one alone. A replica given a few threads of a large
// machine reads at a few times the second, not at the first: on a server of
// many memory channels one core cannot keep more than a fraction of them
// busy.
type Bandwidth struct {
All, Core float64
}
// coreBuffer is what the one-core passes read: past any cache, and short
// enough that one core reads it in a fraction of a second.
const coreBuffer = 512 << 20
// MeasureMemoryBandwidth is the memory's read rate, the best of a few passes
// each way; zero when the machine has too little free memory to measure.
func MeasureMemoryBandwidth() Bandwidth {
size := int64(bandwidthBuffer)
// Never more than a quarter of what is free: a node already holding
// models must not be pushed into swap by a measurement. Without room
@@ -40,7 +53,7 @@ func MeasureMemoryBandwidth() float64 {
size = min(size, avail/4)
}
if size < max(minBuffer, 8*lastCacheBytes("/sys/devices/system/cpu")) {
return 0
return Bandwidth{}
}
buf := make([]uint64, size/8)
for i := range buf {
@@ -68,7 +81,17 @@ func MeasureMemoryBandwidth() float64 {
best = max(best, float64(size)/s/1e9)
}
}
return best
// And one core alone, over a part of the same buffer.
part := buf[:min(len(buf), coreBuffer/8)]
core := 0.0
for pass := 0; pass < 3; pass++ {
start := time.Now()
sink(sum(part))
if s := time.Since(start).Seconds(); s > 0 {
core = max(core, float64(len(part)*8)/s/1e9)
}
}
return Bandwidth{All: best, Core: min(core, best)}
}
// sum reads every word, eight at a time so the loop is not what limits it.
+4 -4
View File
@@ -12,11 +12,11 @@ func TestMemoryBandwidthIsMeasured(t *testing.T) {
}
bw := MeasureMemoryBandwidth()
// Any machine running this reads more than a gigabyte a second and less
// than a few terabytes.
if bw < 1 || bw > 5000 {
t.Errorf("measured %.1f GB/s", bw)
// than a few terabytes, and one core no more than all of them.
if bw.All < 1 || bw.All > 5000 || bw.Core < 0.5 || bw.Core > bw.All {
t.Errorf("measured %.1f GB/s, %.1f on one core", bw.All, bw.Core)
}
t.Logf("%.1f GB/s", bw)
t.Logf("%.1f GB/s, %.1f on one core", bw.All, bw.Core)
}
func TestTheLastCacheIsReadFromSys(t *testing.T) {
+3 -1
View File
@@ -4,12 +4,14 @@ import (
"os"
"testing"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
)
func TestMain(m *testing.M) {
// Half a gigabyte read per test binary, and contending with the tests
// for the CPUs: these never look at the figure.
measureBandwidth = func() float64 { return 0 }
measureBandwidth = func() discovery.Bandwidth { return discovery.Bandwidth{} }
bandwidthDelay = func() time.Duration { return 0 }
os.Exit(m.Run())
}
+14 -1
View File
@@ -418,7 +418,7 @@ func newModelsCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
}
fmt.Fprintf(w, "%s\t%s\t%d/%d\t%s\t%s\t%s\t%s\t%s\t%s\n",
name, serving, m.Ready, m.Min, modelStatus(m), m.Class, reserves(m),
join(m.Aliases), join(m.Profiles), join(m.Fallbacks))
join(m.Aliases), join(profilesWithCounts(m)), join(m.Fallbacks))
for _, x := range m.Warnings {
warnings = append(warnings, m.Name+": "+x)
}
@@ -696,3 +696,16 @@ func reserves(m adminapi.ModelView) string {
}
return out
}
// profilesWithCounts are a model's profiles, each with its replicas when
// they are counted per profile: gpu=7, cpu=4.
func profilesWithCounts(m adminapi.ModelView) []string {
if len(m.ProfileReplicas) == 0 {
return m.Profiles
}
out := make([]string, len(m.Profiles))
for i, p := range m.Profiles {
out[i] = fmt.Sprintf("%s=%d", p, m.ProfileReplicas[p])
}
return out
}
+49 -1
View File
@@ -19,6 +19,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/sourcecheck"
)
@@ -34,7 +35,7 @@ func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
Use: "model",
Short: "Add or remove models while the cluster is running",
}
c.AddCommand(newModelStopCmd(client), newModelStartCmd(client), newModelContextCmd(client), newModelGPUsCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client),
c.AddCommand(newModelStopCmd(client), newModelStartCmd(client), newModelReplicasCmd(client), newModelContextCmd(client), newModelGPUsCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client),
newResourcesCmd(client), newClassCmd(client), newOwnerCmd(client),
newTargetCmd(client, asJSON), newAdaptersCmd(client),
newModelServeCmd(client), newModelUpdatesCmd(client, asJSON), newModelUpdateCmd(client),
@@ -598,3 +599,50 @@ func newModelGPUsCmd(client func() *ctlClient) *cobra.Command {
c.Flags().BoolVar(&inPlace, "in-place", false, "stop each replica before its replacement starts, when there is no room beside it")
return c
}
func newModelReplicasCmd(client func() *ctlClient) *cobra.Command {
var cpuMemory, cpuCores int
c := &cobra.Command{
Use: "replicas <model> <profile>=<n>...",
Short: "How many of a model's replicas run on each of its profiles: gpu=7 cpu=4",
Long: "Each replica runs on the best profile that fits unless counted here: seven on\n" +
"cards and four on CPUs, all answering under the model's name, the router filling\n" +
"the cards before the CPUs. A profile not named keeps its count. For a GGUF model\n" +
"with no profile on CPUs, --cpu-memory adds one named cpu. Its scaling becomes the\n" +
"sum; llmgw ctl model scale with another total ends the count per profile.",
Example: " llmgw ctl model replicas qwen-coder gpu-1xl40s-48g=7 cpu=4 --cpu-memory 24 --cpu-cores 8",
Args: cobra.MinimumNArgs(2),
RunE: func(cmd *cobra.Command, args []string) error {
req := adminapi.ProfileReplicasRequest{Profiles: map[string]int{}}
for _, a := range args[1:] {
name, v, ok := strings.Cut(a, "=")
n, err := strconv.Atoi(v)
if !ok || err != nil || name == "" {
return fmt.Errorf("%q: a profile and a count, as gpu=7", a)
}
req.Profiles[name] = n
}
if cpuMemory > 0 {
req.CPU = &modelsvc.CPUProfile{MemoryGB: cpuMemory, CPUCores: cpuCores}
}
body, err := json.Marshal(req)
if err != nil {
return err
}
var m catalog.Model
if err := client().post(cmd.Context(), "/api/v1/models/"+args[0]+"/replicas", body, &m); err != nil {
return err
}
var parts []string
for _, p := range m.Profiles {
parts = append(parts, fmt.Sprintf("%d on %s", p.Replicas, p.Name))
}
fmt.Fprintf(cmd.OutOrStdout(), "%s runs %s: %d in all. Those short start first, then those over stop.\n",
m.Name, strings.Join(parts, ", "), m.SplitTotal())
return nil
},
}
c.Flags().IntVar(&cpuMemory, "cpu-memory", 0, "add a profile on CPUs named cpu, reserving this many GB a replica")
c.Flags().IntVar(&cpuCores, "cpu-cores", 0, "the cores a replica on CPUs computes on (default: half the node's)")
return c
}
+40 -2
View File
@@ -321,7 +321,10 @@ func planFiles(cmd *cobra.Command, client func() *ctlClient, repo, format string
fmt.Fprintln(out, "\nNo file fits this cluster as it is: a smaller context (--context), or more memory.")
return errDoesNotFit
}
fmt.Fprintf(out, "\nBest: %s on %s: %s.\n", rec.Artifact, rec.Label(), rec.MemoryText())
fmt.Fprintf(out, "\nBest: %s on %s: %s%s.\n", rec.Artifact, rec.Label(), rec.MemoryText(), speedNote(rec))
if w := slowWarning(rec); w != "" {
fmt.Fprintln(out, w)
}
if p.KVBytesPerToken > 0 {
fmt.Fprintf(out, "The context costs %.2f GB per 1000 tokens, the whole model (--context to plan another).\n",
modelplan.GB(p.KVBytesPerToken*1000))
@@ -334,7 +337,8 @@ func planFiles(cmd *cobra.Command, client func() *ctlClient, repo, format string
if o.FreeNow {
now = "free"
}
fmt.Fprintf(w, " %s\t%s\t%s\t%d host(s)\t%s\n", o.Artifact, o.Label(), o.MemoryText(), o.Hosts, now)
fmt.Fprintf(w, " %s\t%s\t%s\t%d host(s)\t%s\t%s\n", o.Artifact, o.Label(), o.MemoryText(), o.Hosts, now,
speedCell(o.TokensPerSecond))
}
_ = w.Flush()
}
@@ -349,3 +353,37 @@ func planFiles(cmd *cobra.Command, client func() *ctlClient, repo, format string
fmt.Fprintln(out, " --fallbacks")
return nil
}
// speedCell is a way's speed in a table: about how many tokens a second one
// reply gets, and whether that is slow.
func speedCell(tps float64) string {
switch {
case tps <= 0:
return ""
case tps < modelplan.CPUTooSlowTPS:
return fmt.Sprintf("~%.1f tok/s, very slow", tps)
case tps < modelplan.CPUSlowTPS:
return fmt.Sprintf("~%.0f tok/s, slow", tps)
}
return fmt.Sprintf("~%.0f tok/s", tps)
}
// speedNote is the same for a sentence, empty when not known.
func speedNote(o modelplan.Option) string {
if o.TokensPerSecond <= 0 {
return ""
}
return fmt.Sprintf(", about %s tokens a second for one reply", strconv.FormatFloat(o.TokensPerSecond, 'f', 1, 64))
}
// slowWarning says what a slow way means for somebody waiting on a reply.
func slowWarning(o modelplan.Option) string {
tps := o.TokensPerSecond
switch {
case tps <= 0 || tps >= modelplan.CPUSlowTPS:
return ""
case tps < modelplan.CPUTooSlowTPS:
return fmt.Sprintf("Very slow: a reply of 500 tokens takes about %.0f minutes.", max(1, 500/tps/60))
}
return fmt.Sprintf("Slow: a reply of 500 tokens takes about %.0f seconds.", 500/tps)
}
+43
View File
@@ -20,6 +20,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/facility"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/release"
"gitea.mynodes.xyz/ilker/llmgw/internal/version"
)
@@ -406,6 +407,48 @@ func (m consoleModel) modelMenu(md adminapi.ModelView) *menuDialog {
}})
return nil
}},
{"Replicas per profile…", func(m *consoleModel) tea.Cmd {
var now []string
for _, p := range md.Profiles {
now = append(now, fmt.Sprintf("%s=%d", p, md.ProfileReplicas[p]))
}
m.openForm(&formDialog{title: "Replicas per profile of " + md.Name,
note: "How many replicas run on each profile, all answering under the model's name: seven on cards and " +
"four on CPUs, the router filling the cards first. A profile not named keeps its count; cpu=N with " +
"the memory below adds a profile on CPUs to a GGUF model that has none. Those short start first, " +
"then those over stop.",
fields: []formField{
textF("replicas", strings.Join(now, " "), "profile=count, separated by spaces"),
textF("CPU memory GB", "", "only to add a profile on CPUs: what a replica reserves"),
textF("CPU cores", "", "of a replica on CPUs; empty is half the node's"),
},
submit: func(m *consoleModel, v []string) (tea.Cmd, error) {
req := adminapi.ProfileReplicasRequest{Profiles: map[string]int{}}
for _, f := range strings.Fields(v[0]) {
name, c, ok := strings.Cut(f, "=")
n, err := strconv.Atoi(c)
if !ok || err != nil || n < 0 || name == "" {
return nil, fmt.Errorf("%q: a profile and a count, as gpu=7", f)
}
req.Profiles[name] = n
}
if v[1] != "" {
mem, err := strconv.Atoi(v[1])
if err != nil || mem < 1 {
return nil, fmt.Errorf("the CPU memory is a whole number of GB")
}
cores := 0
if v[2] != "" {
if cores, err = strconv.Atoi(v[2]); err != nil || cores < 1 {
return nil, fmt.Errorf("the CPU cores are a whole number from 1")
}
}
req.CPU = &modelsvc.CPUProfile{MemoryGB: mem, CPUCores: cores}
}
return m.call("Replicas of "+md.Name, http.MethodPost, path+"/replicas", req, nil), nil
}})
return nil
}},
{"Resources…", func(m *consoleModel) tea.Cmd {
profiles := md.Profiles
if len(profiles) == 0 {
+12 -1
View File
@@ -107,7 +107,15 @@ func optionLabel(o modelplan.Option) string {
if o.FreeNow {
now = sOK.Render("free")
}
return fmt.Sprintf("%s · %s · %s", o.Label(), o.MemoryText(), now)
label := fmt.Sprintf("%s · %s · %s", o.Label(), o.MemoryText(), now)
if c := speedCell(o.TokensPerSecond); c != "" {
style := sDim
if o.TokensPerSecond < modelplan.CPUTooSlowTPS {
style = sWarn
}
label += " · " + style.Render(c)
}
return label
}
// repoMenu offers a repository's files, each with the best way it runs here;
@@ -196,6 +204,9 @@ func (m *consoleModel) serveForm(p modelserve.Plan, a modelserve.PlannedArtifact
note += fmt.Sprintf("\nMemory for a %d-token context, each way it runs:", p.Context)
for _, o := range opts {
note += "\n " + o.Label() + ": " + o.MemoryText()
if w := slowWarning(o); w != "" && o.CPU {
note += "\n " + sWarn.Render(w)
}
}
if p.KVBytesPerToken > 0 {
note += fmt.Sprintf("\nThe context costs %.2f GB per 1000 tokens, the whole model; a longer one is checked\n"+
+52 -1
View File
@@ -22,6 +22,7 @@ import (
"unicode"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/gguf"
"gitea.mynodes.xyz/ilker/llmgw/internal/modelplan"
"gitea.mynodes.xyz/ilker/llmgw/internal/release"
@@ -130,6 +131,7 @@ func (a *API) Register(mux Mux) {
mux.HandleFunc("POST /api/v1/models/{name}/start", a.startModel)
mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel)
mux.HandleFunc("POST /api/v1/models/{name}/resources", a.setResources)
mux.HandleFunc("POST /api/v1/models/{name}/replicas", a.setProfileReplicas)
a.registerQuotas(mux)
a.registerKB(mux)
a.registerTenancy(mux)
@@ -259,6 +261,9 @@ type NodeView struct {
// MemBandwidthGBs is how fast its memory reads, in GB/s, measured when
// its agent started; 0 when not reported. A model's speed on the CPU.
MemBandwidthGBs float64 `json:"mem_bandwidth_gbs,omitempty"`
// MemBandwidthCoreGBs is what one core reads alone, in GB/s; 0 when not
// reported. A CPU replica's threads read at about this each.
MemBandwidthCoreGBs float64 `json:"mem_bandwidth_core_gbs,omitempty"`
// Engines are the inference engines installed here, as the agent found
// them. Absent means the agent is older than the probe and said nothing,
// which is not the same as having none — placement treats it that way too,
@@ -489,6 +494,9 @@ type ModelView struct {
Unplaceable bool `json:"unplaceable,omitempty"`
Fallbacks []string `json:"fallbacks,omitempty"`
Profiles []string `json:"profiles"`
// ProfileReplicas is how many replicas run on each profile, for a model
// whose replicas are counted per profile; absent otherwise.
ProfileReplicas map[string]int `json:"profile_replicas,omitempty"`
// MemoryGB and CPUCores are what one replica reserves, from its first
// profile; MeasuredMemoryGB is the most one of its replicas uses now, 0
// when none is measured.
@@ -723,7 +731,7 @@ func (a *API) Nodes() []NodeView {
v.EngineBuilds[e] = b
}
}
v.MemBandwidthGBs = n.Host.MemBandwidthGBs
v.MemBandwidthGBs, v.MemBandwidthCoreGBs = n.Host.MemBandwidthGBs, n.Host.MemBandwidthCoreGBs
for _, g := range n.GPUs {
if g.MIG != nil {
continue // said of the card it is carved from
@@ -865,6 +873,12 @@ func (a *API) Models() []ModelView {
Canary: m.Canary, CanaryOf: a.ctl.Catalog().CanaryOf(m.Name), Eval: m.Eval,
Role: m.Role,
}
if m.Split() {
v.ProfileReplicas = map[string]int{}
for _, p := range m.Profiles {
v.ProfileReplicas[p.Name] = p.Replicas
}
}
for _, p := range m.Profiles {
v.Profiles = append(v.Profiles, p.Name)
}
@@ -962,6 +976,15 @@ type Models interface {
SetClass(ctx context.Context, name, class, by string) (catalog.Model, error)
SetOwner(ctx context.Context, name, owner, by string) (catalog.Model, error)
SetAdapters(ctx context.Context, name string, adapters []catalog.Adapter, by string) (catalog.Model, error)
SetProfileReplicas(ctx context.Context, name string, counts map[string]int, cpu *modelsvc.CPUProfile, by string) (catalog.Model, error)
}
// ProfileReplicasRequest counts a model's replicas per profile: profile
// name to replicas, and CPU, for a llama.cpp model with no profile on CPUs,
// the one to add.
type ProfileReplicasRequest struct {
Profiles map[string]int `json:"profiles"`
CPU *modelsvc.CPUProfile `json:"cpu,omitempty"`
}
// ResourcesRequest sets what a model's profiles reserve of a node.
@@ -1288,6 +1311,25 @@ func (a *API) setResources(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, m)
}
// setProfileReplicas counts a model's replicas per profile.
func (a *API) setProfileReplicas(w http.ResponseWriter, r *http.Request) {
if a.models == nil {
writeError(w, http.StatusNotImplemented, "this control plane has no store")
return
}
var req ProfileReplicasRequest
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error())
return
}
m, err := a.models.SetProfileReplicas(r.Context(), r.PathValue("name"), req.Profiles, req.CPU, actorOf(r))
if err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
writeJSON(w, http.StatusOK, m)
}
func (a *API) removeModel(w http.ResponseWriter, r *http.Request) {
if a.models == nil {
writeError(w, http.StatusNotImplemented, "this control plane has no store")
@@ -1966,6 +2008,15 @@ func PoolNodesOf(nodes []NodeView) []modelplan.PoolNode {
if l := n.Load; l != nil && l.MemTotal > 0 {
p.HostFreeGB = int((l.MemTotal - l.MemUsed) >> 30)
}
p.CPUs, p.MemBWGBs, p.MemBWCoreGBs = n.CPUCount, n.MemBandwidthGBs, n.MemBandwidthCoreGBs
for _, g := range n.GPUs {
if bw := modelplan.GPUBandwidthGBs(g.Name); bw > 0 {
if p.CardBWGBs == nil {
p.CardBWGBs = map[string]float64{}
}
p.CardBWGBs[g.Class] = bw
}
}
out = append(out, p)
}
return out
+11
View File
@@ -16,6 +16,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
"gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
@@ -263,6 +264,16 @@ func (f *fakeModels) SetAdapters(_ context.Context, name string, as []catalog.Ad
return catalog.Model{Name: name, Adapters: as}, f.err
}
func (f *fakeModels) SetProfileReplicas(_ context.Context, name string, counts map[string]int, _ *modelsvc.CPUProfile,
_ string,
) (catalog.Model, error) {
m := catalog.Model{Name: name}
for p, n := range counts {
m.Profiles = append(m.Profiles, catalog.Profile{Name: p, Replicas: n})
}
return m, f.err
}
func (f *fakeModels) SetOwner(_ context.Context, name, owner, _ string) (catalog.Model, error) {
return catalog.Model{Name: name, Owner: owner}, f.err
}
+32 -13
View File
@@ -2,6 +2,8 @@ package adminapi
import (
"context"
"errors"
"fmt"
"log/slog"
"net/http"
"net/url"
@@ -48,10 +50,29 @@ type PurgeView struct {
// purgeModel removes a model from the catalog and its files from the mirror
// and the nodes.
func (a *API) purgeModel(w http.ResponseWriter, r *http.Request, name string) {
out, err := a.Purge(r.Context(), name, actorOf(r))
switch {
case errors.Is(err, errNoSuchModel):
writeError(w, http.StatusNotFound, err.Error())
case err != nil:
writeError(w, http.StatusBadRequest, err.Error())
default:
writeJSON(w, http.StatusOK, out)
}
}
// errNoSuchModel is a model the catalog does not have.
var errNoSuchModel = errors.New("no such model in the catalog")
// Purge removes a model from the catalog and its files from the mirror and
// the nodes, waiting a while for its replicas to stop before the nodes'.
func (a *API) Purge(ctx context.Context, name, by string) (PurgeView, error) {
if a.models == nil {
return PurgeView{}, errors.New("this control plane has no store")
}
m, ok := a.ctl.Catalog().Resolve(name)
if !ok {
writeError(w, http.StatusNotFound, "no model "+name+" in the catalog")
return
return PurgeView{}, fmt.Errorf("%w: %s", errNoSuchModel, name)
}
name = m.Name
others := []catalog.Model{}
@@ -60,37 +81,35 @@ func (a *API) purgeModel(w http.ResponseWriter, r *http.Request, name string) {
others = append(others, o)
}
}
if err := a.models.Remove(r.Context(), name); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
if err := a.models.Remove(ctx, name); err != nil {
return PurgeView{}, err
}
a.presetDeclined(r.Context(), name, actorOf(r))
a.presetDeclined(ctx, name, by)
out := PurgeView{Removed: name}
sources := purgeSources(m)
if a.purgeMirror == nil {
out.Note = "this control plane has no mirror; its files there, if any, are left"
} else {
out.Mirror, out.Kept, out.Note = purgeFromMirror(r.Context(), a.purgeMirror, sources, others)
out.Mirror, out.Kept, out.Note = purgeFromMirror(ctx, a.purgeMirror, sources, others)
}
// The nodes' copies, once nothing runs on them.
deadline := time.Now().Add(purgeWait)
for a.runs(name) && time.Now().Before(deadline) {
select {
case <-r.Context().Done():
return
case <-ctx.Done():
return out, ctx.Err()
case <-time.After(time.Second):
}
}
if a.runs(name) {
out.Note = strings.TrimSpace(out.Note + " Its replicas had not stopped yet: its copies on the nodes are " +
"left, and llmgw ctl weights --collect deletes them.")
writeJSON(w, http.StatusOK, out)
return
return out, nil
}
out.Nodes = a.purgeFromNodes(r.Context(), sources, others)
writeJSON(w, http.StatusOK, out)
out.Nodes = a.purgeFromNodes(ctx, sources, others)
return out, nil
}
// runs says whether any replica of a model is still on a node.
+4
View File
@@ -168,6 +168,10 @@ var Routes = []Route{
Response: ScaleResponse{}},
{Method: "POST", Path: "/api/v1/models/{name}/scale", Summary: "Set how many replicas a model runs",
Request: ScaleRequest{}, Response: ScaleResponse{}},
{Method: "POST", Path: "/api/v1/models/{name}/replicas", Summary: "How many replicas run on each profile",
Description: "Seven on cards and four on CPUs, under the model's name. A profile not named keeps its count; " +
"cpu adds a profile on CPUs to a llama.cpp model that has none. Its scaling becomes the sum.",
Request: ProfileReplicasRequest{}, Response: catalog.Model{}},
{Method: "POST", Path: "/api/v1/models/{name}/resources", Summary: "Set what a model's profiles reserve of a node",
Request: ResourcesRequest{}, Response: catalog.Model{}},
{Method: "POST", Path: "/api/v1/models/{name}/class", Summary: "Set a model's capacity class",
+59
View File
@@ -0,0 +1,59 @@
package adminapi
import (
"context"
"errors"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
)
// What the portal changes of a served model beside what the catalog holds:
// the same as the API's routes, called in the process.
// SetModelContext changes a served model's context and longest answer.
func (a *API) SetModelContext(ctx context.Context, name string, req modelserve.ContextRequest, by string) (modelserve.ContextChange, error) {
if a.serve == nil {
return modelserve.ContextChange{}, errors.New("this control plane cannot change a model's context")
}
return a.serve.SetContext(ctx, name, req, by)
}
// SetModelGPUs spreads a served model over more cards a replica, or fewer.
func (a *API) SetModelGPUs(ctx context.Context, name string, req modelserve.GPUsRequest, by string) (modelserve.GPUsChange, error) {
if a.serve == nil {
return modelserve.GPUsChange{}, errors.New("this control plane cannot change a model's cards")
}
return a.serve.SetGPUs(ctx, name, req, by)
}
// RestartModel starts each of a model's ready replicas again, one at a time,
// and says how many.
func (a *API) RestartModel(name string) (int, error) {
var ids []string
for _, rep := range a.ctl.Reconciler().ReadyReplicas() {
if rep.Model == name {
ids = append(ids, rep.ID)
}
}
if len(ids) == 0 {
return 0, errors.New("none of its replicas is ready to restart")
}
return len(ids), a.ctl.Reconciler().RestartPaced(reconciler.RestartPace{}, false, ids...)
}
// RemoveModel takes a model out of the catalog, with its files on the mirror
// and the nodes when purge.
func (a *API) RemoveModel(ctx context.Context, name string, purge bool, by string) (PurgeView, error) {
if purge {
return a.Purge(ctx, name, by)
}
if a.models == nil {
return PurgeView{}, errors.New("this control plane has no store")
}
if err := a.models.Remove(ctx, name); err != nil {
return PurgeView{}, err
}
a.presetDeclined(ctx, name, by)
return PurgeView{Removed: name}, nil
}
+1 -1
View File
@@ -36,7 +36,7 @@ func NodeFromRegister(r *llmgwv1.Register) inventory.Node {
n.Host = discovery.Host{
OSID: h.GetOsId(), OSVersion: h.GetOsVersion(), OSCodename: h.GetOsCodename(),
Arch: h.GetArch(), CPUFlags: h.GetCpuFlags(), NVIDIADriver: h.GetNvidiaDriver(),
MemBandwidthGBs: h.GetMemBandwidthGbs(),
MemBandwidthGBs: h.GetMemBandwidthGbs(), MemBandwidthCoreGBs: h.GetMemBandwidthCoreGbs(),
}
}
for _, w := range r.GetWeights() {
+4 -1
View File
@@ -547,7 +547,7 @@ func (s *Service) applyHeartbeat(nodeID string, hb *llmgwv1.Heartbeat) {
storage := hb.GetStorage()
memMB := int(hb.GetMemoryTotalMb())
cpus := int(hb.GetCpuCount())
bw := hb.GetMemBandwidthGbs()
bw, coreBW := hb.GetMemBandwidthGbs(), hb.GetMemBandwidthCoreGbs()
if len(used) == 0 && len(foreign) == 0 && len(problems) == 0 && len(storage) == 0 &&
!hb.GetWeightsChanged() && memMB == 0 && cpus == 0 && !hb.GetEnginesChanged() &&
!hb.GetProvisionChanged() && hb.GetLoad() == nil && len(hb.GetIbPorts()) == 0 &&
@@ -626,6 +626,9 @@ func (s *Service) applyHeartbeat(nodeID string, hb *llmgwv1.Heartbeat) {
if bw > 0 {
n.Host.MemBandwidthGBs = bw
}
if coreBW > 0 {
n.Host.MemBandwidthCoreGBs = coreBW
}
if cpus > 0 && cpus != n.CPUCount {
if n.CPUCount > 0 {
s.log.Info("node CPUs changed", "node", nodeID, "from", n.CPUCount, "to", cpus)
+56
View File
@@ -267,6 +267,12 @@ type Profile struct {
Weight float64 `json:"weight,omitempty" koanf:"weight"`
// Priority orders profiles during placement; highest that fits wins.
Priority int `json:"priority,omitempty" koanf:"priority"`
// Replicas, set on any of a model's profiles, is how many of its
// replicas run on this one: seven on cards and four on CPUs, all
// answering under the model's name, rather than each replica on the
// best profile that fits. The model's scaling is then the sum, min and
// max alike; a profile left at zero runs none.
Replicas int `json:"replicas,omitempty" koanf:"replicas"`
// StartupTimeout bounds model loading. A 70B model across two nodes takes
// tens of minutes, so this is per profile rather than global.
StartupTimeout time.Duration `json:"startup_timeout,omitempty" koanf:"startup_timeout"`
@@ -1024,10 +1030,57 @@ func (m Model) normalize() Model {
return m
}
// Split says whether the model's replicas are counted per profile: any of
// its profiles sets replicas.
func (m Model) Split() bool {
return slices.ContainsFunc(m.Profiles, func(p Profile) bool { return p.Replicas > 0 })
}
// SplitTotal is the replicas its profiles ask for together.
func (m Model) SplitTotal() int {
n := 0
for _, p := range m.Profiles {
n += max(p.Replicas, 0)
}
return n
}
// ProfileBase is the catalog profile a placement profile stands for: one
// named by the card it needs is placed as name@class.
func ProfileBase(name string) string {
base, _, _ := strings.Cut(name, "@")
return base
}
// validateSplit refuses replicas per profile that the model's scaling does
// not add up to.
func (m Model) validateSplit() error {
for _, p := range m.Profiles {
if p.Replicas < 0 {
return fmt.Errorf("model %q profile %q: replicas %d is negative", m.Name, p.Name, p.Replicas)
}
}
if !m.Split() || m.Stopped != nil {
return nil
}
n := m.SplitTotal()
if m.Scaling.Min != n || m.Scaling.Max != n {
return fmt.Errorf("model %q: its profiles' replicas add up to %d, so its scaling is min %d, max %d, not %d and %d",
m.Name, n, n, n, m.Scaling.Min, m.Scaling.Max)
}
if len(m.Scaling.Schedule) > 0 {
return fmt.Errorf("model %q: replicas per profile and a schedule are one or the other", m.Name)
}
return nil
}
func (m Model) validate() error {
if m.Name == "" {
return fmt.Errorf("a model has no name")
}
if err := m.validateSplit(); err != nil {
return err
}
if m.Source == "" {
return fmt.Errorf("model %q: source is required", m.Name)
}
@@ -1510,6 +1563,9 @@ func (m Model) Simulated() bool {
// happened, and the answer to "why is this one on the wrong node" was in
// somebody's memory of what they edited an hour ago.
func (p Profile) Fingerprint() string {
// How many replicas run on it is not how each runs: changing the count
// must not mark the running ones stale.
p.Replicas = 0
b, err := json.Marshal(p)
if err != nil {
// A profile that will not marshal cannot be compared, and saying so by
@@ -572,6 +572,14 @@ func (s *Server) run(j *Job) {
// On one card, the weights and the context's cache have to fit it: a
// 70B with a 65536-token pool on a 24 GB card loaded its weights and then
// failed to make its context, every time, as did an 8B with 131072.
// Asked for the CPU, and slow there: said, not refused — the request
// named it — so the job's log says why its replies take minutes.
if len(layout) > 0 && layout[0].CPU && layout[0].TokensPerSecond > 0 &&
layout[0].TokensPerSecond < modelplan.CPUSlowTPS {
tps := layout[0].TokensPerSecond
s.step(j, "check", Running, fmt.Sprintf("on the CPU about %.1f tokens a second for one reply: "+
"a reply of 500 tokens takes about %.0f seconds; a card would be many times faster", tps, 500/tps))
}
if len(layout) > 0 && layout[0].Class != "" && !layout[0].CPU {
fitted, why, err := fitPoolToCard(req, header, set.Bytes, layout[0].Class, max(layout[0].GPUsPerNode, 1))
if err != nil {
+4
View File
@@ -150,6 +150,7 @@ func (s *Server) plan(ctx context.Context, repo hf.Repo, req PlanRequest) (Plan,
}
shape.KVBytesPerToken = v.Header.KVBytesPerToken()
shape.KVHeads = int(v.Header.HeadCountKV) //nolint:gosec // a header's head count
shape.ActiveShare = v.Header.ActiveShare()
if p.Arch == "" {
p.Arch = v.Header.Architecture
}
@@ -181,6 +182,9 @@ func (s *Server) plan(ctx context.Context, repo hf.Repo, req PlanRequest) (Plan,
p.Note = "the model's shape was not readable, so the context's memory is not counted"
}
shape.Params = p.Params
if shape.ActiveShare == 0 || shape.ActiveShare == 1 {
shape.ActiveShare = modelplan.ActiveShare(repo.ID, p.Params)
}
p.KVBytesPerToken, p.Experts = shape.KVBytesPerToken, shape.Experts
p.Options = modelplan.Options(arts, shape, s.deps.Pool(), p.Context)
if !p.Servable {
+100
View File
@@ -22,6 +22,7 @@ import (
"fmt"
"log/slog"
"slices"
"strings"
"sync"
"time"
@@ -249,6 +250,13 @@ func (s *Service) Remove(ctx context.Context, name string) error {
// Scale changes how many replicas a model added at runtime runs.
func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by string) (catalog.Model, error) {
final, err := s.update(ctx, name, by, "change its scaling", func(m *catalog.Model) error {
// Replicas counted per profile stay so only for the same total, one
// way: any other scaling is the model's again, on whatever fits.
if m.Split() && (sc.Min != sc.Max || sc.Min != m.SplitTotal()) {
for i := range m.Profiles {
m.Profiles[i].Replicas = 0
}
}
m.Scaling = sc
// A number set is what it runs now: a stopped model is started by it.
m.Stopped = nil
@@ -261,6 +269,98 @@ func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by
return final, err
}
// CPUProfile is a profile on CPUs to add to a llama.cpp model, for replicas
// counted per profile: what a replica of it reserves.
type CPUProfile struct {
Name string `json:"name,omitempty"`
MemoryGB int `json:"memory_gb"`
CPUCores int `json:"cpu_cores,omitempty"`
}
// SetProfileReplicas sets how many of a model's replicas run on each of its
// profiles: seven on cards and four on CPUs, under the one name. A profile
// not named keeps its count. cpu, for a llama.cpp model with no profile on
// CPUs, adds one — its card profile without the card, preferred below every
// other, so the router fills the cards first and the CPUs after. Running
// replicas carry on; those over a count stop once the ones short have
// started.
func (s *Service) SetProfileReplicas(ctx context.Context, name string, counts map[string]int, cpu *CPUProfile,
by string,
) (catalog.Model, error) {
final, err := s.update(ctx, name, by, "count its replicas per profile", func(m *catalog.Model) error {
if cpu != nil {
if cpu.Name == "" {
cpu.Name = "cpu"
}
if !slices.ContainsFunc(m.Profiles, func(p catalog.Profile) bool { return p.Name == cpu.Name }) {
p, err := cpuProfileOf(*m, *cpu)
if err != nil {
return err
}
m.Profiles = append(m.Profiles, p)
}
}
for prof, n := range counts {
i := slices.IndexFunc(m.Profiles, func(p catalog.Profile) bool { return p.Name == prof })
if i < 0 {
return fmt.Errorf("model %q has no profile %q", name, prof)
}
if n < 0 {
return fmt.Errorf("profile %q: %d replicas", prof, n)
}
m.Profiles[i].Replicas = n
}
n := m.SplitTotal()
if n == 0 {
return fmt.Errorf("none of %q's profiles would run a replica; llmgw ctl model stop %s keeps it without", name, name)
}
sc := catalog.Scaling{Min: n, Max: n, IdleTimeout: m.Scaling.IdleTimeout}
if m.Stopped != nil {
sc.IdleTimeout = m.Stopped.IdleTimeout
m.Stopped = &sc
} else {
m.Scaling = sc
}
return nil
})
if err == nil {
s.log.Info("replicas counted per profile", "model", name, "counts", counts, "by", by)
}
return final, err
}
// cpuProfileOf is a model's llama.cpp profile on CPUs: the same engine and
// arguments, without the card or what spreads layers over cards.
func cpuProfileOf(m catalog.Model, cpu CPUProfile) (catalog.Profile, error) {
i := slices.IndexFunc(m.Profiles, func(p catalog.Profile) bool { return p.Engine == "llamacpp" })
if i < 0 {
return catalog.Profile{}, fmt.Errorf("model %q is not served by llama.cpp; only a GGUF model runs on CPUs", m.Name)
}
if cpu.MemoryGB <= 0 {
return catalog.Profile{}, errors.New("a profile on CPUs needs the memory a replica reserves")
}
p := m.Profiles[i]
lowest := p.Priority
for _, o := range m.Profiles {
lowest = min(lowest, o.Priority)
}
out := catalog.Profile{Name: cpu.Name, Engine: p.Engine, EngineVersion: p.EngineVersion, Source: p.Source,
Env: p.Env, NodeSelector: p.NodeSelector, StartupTimeout: p.StartupTimeout, Weight: p.Weight,
Priority: lowest - 10, MinMemoryGB: cpu.MemoryGB, CPUCores: cpu.CPUCores}
for j := 0; j < len(p.Args); j++ {
switch a := p.Args[j]; {
case slices.Contains([]string{"--n-gpu-layers", "-ngl", "--split-mode", "-sm", "--tensor-split", "-ts",
"--main-gpu", "-mg"}, a):
j++ // and its value
case strings.HasPrefix(a, "--n-gpu-layers=") || strings.HasPrefix(a, "--split-mode=") ||
strings.HasPrefix(a, "--tensor-split=") || strings.HasPrefix(a, "--main-gpu="):
default:
out.Args = append(out.Args, a)
}
}
return out, nil
}
// Stop takes a model out of service and keeps it: none of its replicas run,
// and what it was scaled to is kept for Start.
func (s *Service) Stop(ctx context.Context, name, by string) (catalog.Model, error) {
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"log/slog"
"slices"
"strings"
"testing"
"time"
@@ -458,3 +459,44 @@ func TestAStrataModelReservesTheMemoryItsExpertsTake(t *testing.T) {
t.Errorf("an unknown size: %v", err)
}
}
// A GGUF model on a card gets a profile on CPUs, and seven replicas on cards
// beside four on CPUs; a scaling of another total puts it back on whatever
// fits.
func TestReplicasCountedPerProfile(t *testing.T) {
s, _, _ := newService(t)
m := catalog.Model{Name: "g", Source: "/models/g.gguf", Scaling: catalog.Scaling{Min: 1, Max: 1},
Profiles: []catalog.Profile{{Name: "gpu", Engine: "llamacpp", GPUClass: "l40s-48g", GPUsPerNode: 2,
Priority: 100, Args: []string{"--ctx-size", "16384", "--n-gpu-layers", "999", "--split-mode=layer"}}}}
if err := s.Add(t.Context(), m, "ilker"); err != nil {
t.Fatal(err)
}
if _, err := s.SetProfileReplicas(t.Context(), "g", map[string]int{"gpu": 7, "cpu": 4}, nil, "ilker"); err == nil {
t.Fatal("a count for a profile it does not have was taken")
}
got, err := s.SetProfileReplicas(t.Context(), "g", map[string]int{"gpu": 7, "cpu": 4},
&CPUProfile{MemoryGB: 24, CPUCores: 8}, "ilker")
if err != nil {
t.Fatal(err)
}
if len(got.Profiles) != 2 || got.Scaling.Min != 11 || got.Scaling.Max != 11 {
t.Fatalf("profiles %+v, scaling %+v", got.Profiles, got.Scaling)
}
cpu := got.Profiles[1]
if cpu.Name != "cpu" || cpu.GPUClass != "" || cpu.GPUsPerNode != 0 || cpu.Replicas != 4 || cpu.Priority >= 100 ||
cpu.MinMemoryGB != 24 || cpu.CPUCores != 8 || !slices.Equal(cpu.Args, []string{"--ctx-size", "16384"}) {
t.Errorf("the profile on CPUs: %+v", cpu)
}
// Stopped and started, it comes back split.
_, _ = s.Stop(t.Context(), "g", "ilker")
if got, _ = s.Start(t.Context(), "g", "ilker"); got.Scaling.Min != 11 || got.Profiles[0].Replicas != 7 {
t.Errorf("started: %+v", got)
}
// The same total keeps the split; another one ends it.
if got, _ = s.Scale(t.Context(), "g", catalog.Scaling{Min: 11, Max: 11}, "ilker"); !got.Split() {
t.Error("the same total ended the split")
}
if got, _ = s.Scale(t.Context(), "g", catalog.Scaling{Min: 3, Max: 3}, "ilker"); got.Split() {
t.Error("another total kept the split")
}
}
+30 -4
View File
@@ -886,14 +886,25 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
}
desired := r.desiredFor(m)
// Replicas counted per profile: one short on its profile is started even
// with the total right — before one of a profile with too many is
// stopped, which the total going over then does — so changing seven on
// cards and four on CPUs to nine and two never runs fewer than eleven.
var deficit map[string]bool
var surplus []string
split := m.Split() && desired == m.SplitTotal()
if split {
deficit, surplus = r.splitGaps(m)
}
needUp := len(liveIDs) < desired || split && len(deficit) > 0 && len(liveIDs) == desired
outcome := notTried
switch {
case len(liveIDs) < desired && isParked:
case needUp && isParked:
// Not placed. Its failure is on every status the operator reads.
case len(liveIDs) < desired && r.isHeld(m.Name):
case needUp && r.isHeld(m.Name):
// It gave up room to a model of a higher class, which is placed
// first; taking it back now would undo the preemption.
case len(liveIDs) < desired:
case needUp:
if r.now().Before(blockedUntil) {
// Still in backoff. Rebuilding immediately would hot-loop on a
// model that cannot start — a missing weights path, say — and bury
@@ -912,7 +923,11 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
attribute.String("llmgw.model", m.Name),
attribute.Int("llmgw.replicas", len(liveIDs)),
attribute.Int("llmgw.desired", desired))
r.scaleDown(sctx, liveIDs[desired:], "more replicas than desired")
excess := liveIDs[desired:]
if split {
excess = pickSurplus(surplus, liveIDs, len(liveIDs)-desired)
}
r.scaleDown(sctx, excess, "more replicas than desired")
span.End()
}
@@ -976,6 +991,17 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin
attribute.String("llmgw.model", m.Name),
attribute.StringSlice("llmgw.avoid", avoid))
profiles := r.orderedProfiles(m)
if m.Split() {
// Only the profiles short of their replicas.
deficit, _ := r.splitGaps(m)
profiles = slices.DeleteFunc(profiles, func(p placement.Profile) bool {
return !deficit[catalog.ProfileBase(p.Name)]
})
if len(profiles) == 0 {
pspan.End()
return notTried
}
}
res := r.sched.Place(profiles, avoid)
if res.Placement == nil && len(avoid) > 0 {
// Spreading across failure domains is a preference, not a rule. If it
+64
View File
@@ -0,0 +1,64 @@
package reconciler
import (
"slices"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
)
// splitGaps is, for a model whose replicas are counted per profile, the
// profiles short of theirs and the replicas over: of a profile with more
// than it asks, or of one it no longer has. Counted as reconcileModel
// counts — a replica being retired is not one it has — so a replica
// restarted is replaced on its own profile.
func (r *Reconciler) splitGaps(m catalog.Model) (deficit map[string]bool, surplus []string) {
live := map[string][]string{}
r.mu.Lock()
for _, rep := range r.replicas {
if rep.Model == m.Name && rep.State.Counts() && rep.RetiringAt.IsZero() {
b := catalog.ProfileBase(rep.Profile)
live[b] = append(live[b], rep.ID)
}
}
r.mu.Unlock()
deficit = map[string]bool{}
for _, p := range m.Profiles {
ids := live[p.Name]
delete(live, p.Name)
slices.Sort(ids)
switch {
case len(ids) < p.Replicas:
deficit[p.Name] = true
case len(ids) > p.Replicas:
surplus = append(surplus, ids[p.Replicas:]...)
}
}
for _, ids := range live {
surplus = append(surplus, ids...)
}
slices.Sort(surplus)
return deficit, surplus
}
// pickSurplus is n of the live replicas to stop: those over their
// profile's count first, then any.
func pickSurplus(surplus, live []string, n int) []string {
out := make([]string, 0, n)
for _, id := range surplus {
if len(out) == n {
return out
}
if slices.Contains(live, id) {
out = append(out, id)
}
}
for _, id := range live {
if len(out) == n {
break
}
if !slices.Contains(out, id) {
out = append(out, id)
}
}
return out
}
+85
View File
@@ -0,0 +1,85 @@
package reconciler
import (
"testing"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
)
// splitModel runs gpu replicas on cards and cpu on CPUs, under one name.
func splitModel(gpu, cpu int) catalog.Model {
return catalog.Model{Name: "m", Source: "/models/m.gguf",
Scaling: catalog.Scaling{Min: gpu + cpu, Max: gpu + cpu},
Profiles: []catalog.Profile{
{Name: "gpu", Engine: "llamacpp", GPUClass: "h100-80g", GPUsPerNode: 1, Priority: 100, Replicas: gpu},
{Name: "cpu", Engine: "llamacpp", Priority: 10, Replicas: cpu},
}}
}
// byProfile counts the model's live replicas on each profile.
func byProfile(h *harness) map[string]int {
out := map[string]int{}
for _, rep := range h.r.Replicas() {
if rep.State.Counts() && rep.RetiringAt.IsZero() {
out[catalog.ProfileBase(rep.Profile)]++
}
}
return out
}
// Seven on cards and four on CPUs is what runs, not eleven on whatever fits
// best: with cards free for all eleven, four still go to the CPUs.
func TestReplicasRunAsManyOnEachProfileAsItAsks(t *testing.T) {
h := newHarness(t, splitModel(7, 4), gpuNode("1", 8), gpuNode("2", 8), cpuNode("3"), cpuNode("4"))
for range 14 {
h.r.Reconcile(t.Context())
h.markReady()
h.advance(time.Second)
}
if got := byProfile(h); got["gpu"] != 7 || got["cpu"] != 4 {
t.Fatalf("running %v, want 7 on cards and 4 on CPUs", got)
}
}
// Moving replicas from one profile to the other starts the new ones before
// stopping the old: the model never runs fewer than it asks.
func TestChangingTheSplitStartsBeforeItStops(t *testing.T) {
h := newHarness(t, splitModel(2, 1), gpuNode("1", 8), cpuNode("3"), cpuNode("4"))
for range 6 {
h.r.Reconcile(t.Context())
h.markReady()
h.advance(time.Second)
}
if got := byProfile(h); got["gpu"] != 2 || got["cpu"] != 1 {
t.Fatalf("running %v before the change", got)
}
h.setModels(t, splitModel(1, 2))
for range 10 {
h.r.Reconcile(t.Context())
h.markReady()
h.advance(h.r.cfg.DrainGrace + time.Second)
if n := replicasOf(h, "m"); n < 3 {
t.Fatalf("only %d replicas while the split changed", n)
}
}
if got := byProfile(h); got["gpu"] != 1 || got["cpu"] != 2 {
t.Errorf("running %v, want 1 on cards and 2 on CPUs", got)
}
}
// The scaling has to be the profiles' sum, so a count in one place cannot
// disagree with the other.
func TestReplicasPerProfileAndScalingAgree(t *testing.T) {
m := splitModel(2, 1)
m.Scaling = catalog.Scaling{Min: 1, Max: 5}
if _, err := catalog.New([]catalog.Model{m}); err == nil {
t.Error("profiles asking for 3 with scaling 1..5 were accepted")
}
stopped := splitModel(2, 1)
was := stopped.Scaling
stopped.Stopped, stopped.Scaling = &was, catalog.Scaling{}
if _, err := catalog.New([]catalog.Model{stopped}); err != nil {
t.Errorf("a stopped model with replicas per profile was refused: %v", err)
}
}
+1
View File
@@ -377,6 +377,7 @@ var Reference = []Block{
{Key: "keep_within", Type: "string", Doc: "KeepWithin keeps a multi-node replica's nodes inside one rack, row,\nroom or site, read from their llmgw.io/rack (and so on) labels: for a\nfabric whose leaves do not reach across racks, or cables that do not.\nNodes without the label are passed over for such a profile. Empty is\nno constraint.", Nested: ""},
{Key: "weight", Type: "float64", Doc: "Weight is the router's capacity weight. It should come from ctl bench\nrather than a guess (plan §6.2.1).", Nested: ""},
{Key: "priority", Type: "int", Doc: "Priority orders profiles during placement; highest that fits wins.", Nested: ""},
{Key: "replicas", Type: "int", Doc: "Replicas, set on any of a model's profiles, is how many of its\nreplicas run on this one: seven on cards and four on CPUs, all\nanswering under the model's name, rather than each replica on the\nbest profile that fits. The model's scaling is then the sum, min and\nmax alike; a profile left at zero runs none.", Nested: ""},
{Key: "startup_timeout", Type: "time.Duration", Doc: "StartupTimeout bounds model loading. A 70B model across two nodes takes\ntens of minutes, so this is per profile rather than global.", Nested: ""},
{Key: "kv_events", Type: "bool", Doc: "KVEvents has vLLM publish what its prefix cache holds, and the router\nroute this model's requests by it: to the replica holding the most of\na prompt's leading blocks (plan §7.5 tier 2). vLLM only.\n\nOff by default, because it costs something on every request: the\nrouter asks a replica to tokenize the prompt before placing it, a\nround trip and the chat template run twice. It pays where prompts\nshare long prefixes that the approximate tree guesses wrong — many\nreplicas, conversations that move between them, evictions. The engine\nis started with sha256_cbor block hashing and a fixed PYTHONHASHSEED,\nwhich the router's hashing assumes; arguments overriding either leave\nthe index empty, and the agent says so in its log.", Nested: ""},
{Key: "gpu", Type: "*GPUNeed", Doc: "GPU says what card the profile needs, in place of naming its class:\nit stands for every class in the cluster that meets it. A catalog\nentry written once then places sensibly on any site's cards.", Nested: "catalog.GPUNeed"},
+20
View File
@@ -48,6 +48,22 @@ type Header struct {
// (<arch>.pooling_type set and not none): an embedding model, which
// llama.cpp serves with --embedding and does not chat.
Embedding bool
// ExpertCount and ExpertUsedCount are a model of experts' experts and
// how many each token uses (<arch>.expert_count, expert_used_count);
// zero for a dense model.
ExpertCount, ExpertUsedCount uint64
}
// ActiveShare is the part of the weights a token reads: for a model of
// experts its experts' share plus a tenth for what every token reads
// whatever the experts — attention, embeddings, routing, shared experts.
// Qwen3-Coder-30B-A3B, eight of 128, read 0.16 of its weights a token on
// cuda01's card, which is 0.0625 and the tenth. 1 for a dense model.
func (h Header) ActiveShare() float64 {
if h.ExpertCount == 0 || h.ExpertUsedCount == 0 || h.ExpertUsedCount >= h.ExpertCount {
return 1
}
return min(1, float64(h.ExpertUsedCount)/float64(h.ExpertCount)+0.1)
}
// KVBytesPerToken is what one token of context costs in llama.cpp's default
@@ -276,6 +292,10 @@ func Read(r io.Reader) (Header, error) {
h.Embedding = num > 0
case "context_length":
h.ContextLength = num
case "expert_count":
h.ExpertCount = num
case "expert_used_count":
h.ExpertUsedCount = num
}
}
// The tokenizer's arrays — megabytes of vocabulary — come after the
+11
View File
@@ -210,3 +210,14 @@ func TestAPoolIsSizedToTheCard(t *testing.T) {
t.Error("class memory misread")
}
}
// A model of experts reads its experts' share a token and what every token
// reads besides; a dense one reads it all.
func TestAModelOfExpertsReadsAShareOfItsWeights(t *testing.T) {
if got := (Header{ExpertCount: 128, ExpertUsedCount: 8}).ActiveShare(); got < 0.16 || got > 0.17 {
t.Errorf("Qwen3-30B-A3B's shape reads %.3f of its weights a token, measured 0.16", got)
}
if (Header{}).ActiveShare() != 1 || (Header{ExpertCount: 8, ExpertUsedCount: 8}).ActiveShare() != 1 {
t.Error("a dense model reads all its weights")
}
}
+72 -12
View File
@@ -1,6 +1,7 @@
package modelplan
import (
"math"
"regexp"
"strconv"
)
@@ -9,28 +10,45 @@ import (
//
// Writing a token reads every weight it uses once, so on a CPU the memory
// decides the speed: a token takes the bytes read over the rate llama.cpp
// reads them at, plus a little that does not shrink with the model. Measured
// on the PoC on 2026-10-26, each node's bandwidth by MeasureMemoryBandwidth
// against llama.cpp's own predicted_per_second for 128 tokens:
// reads them at, plus a little that does not shrink with the model.
//
// node01 4 vCPU 24.9 GB/s qwen2.5-1.5b Q4_K_M 1.1 GB 14.5 tok/s 0.65 of the bandwidth
// node05 8 vCPU 41.9 GB/s qwen2.5-14b Q4_K_M 9.0 GB 3.3 tok/s 0.71
// cuda01 12 vCPU 42.4 GB/s qwen2.5-0.5b Q4_K_M 0.5 GB 36.0 tok/s (the fixed cost shows)
// The rate is the threads': each reads at about what one core reads alone,
// a little less with every one added, up to what the whole machine reads.
// On a server of many memory channels a replica given four threads reads at
// a fraction of the machine's rate, and the machine's rate alone overstated
// it several times. Measured on the PoC's node05 on 2026-11-03 (8 vCPU,
// 43.9 GB/s with every CPU, 9.7 with one; llama-bench, 64 tokens):
//
// So 0.68 of the measured rate, and 5 ms a token besides. One reply at a
// time; several at once share the same reads and each is slower. The PoC's
// host has far fewer cores than its VMs, so these are what a busy host
// gives, and a machine of its own does better.
// threads 1 2 4 8
// qwen2.5-coder-7b 1.9 3.6 6.3 9.3 tok/s 4.7 GB: 8.9 17.0 30.7 45.6 GB/s read
// qwen2.5-1.5b 8.3 15.2 25.4 37.0 tok/s 1.1 GB: 9.7 18.3 32.6 —
//
// so 0.95 of one core's rate times threads^0.9, up to the machine's, and
// 5 ms a token besides. Against what was measured before (2026-10-26):
// node01's 1.5b on two threads, 14.5 tok/s, is taken for 15.0; node05's 14b
// on four, 3.3, for 3.5. One reply at a time; several at once share the same
// reads and each is slower. The PoC's host has far fewer cores than its VMs,
// so these are what a busy host gives, and a machine of its own does better.
// cpuReadShare is the part of the measured bandwidth llama.cpp reads at.
// cpuReadShare is the part of the whole machine's rate llama.cpp reads at,
// for a node whose one-core rate is not known: an agent older than its
// measurement.
const cpuReadShare = 0.68
// coreShare and threadExp are how threads add up: each about one core's
// rate, a little less with every one added.
const (
coreShare = 0.95
threadExp = 0.9
)
// cpuTokenFixed is a token's cost that does not grow with the model.
const cpuTokenFixed = 0.005
// CPUTokensPerSecond is about how many tokens a second one reply gets on a
// CPU whose memory reads bwGBs, from a model whose tokens each read bytes.
// 0 when either is unknown.
// 0 when either is unknown. Use CPUTokensPerSecondOn where the threads and
// one core's rate are known.
func CPUTokensPerSecond(bytes int64, bwGBs float64) float64 {
if bytes <= 0 || bwGBs <= 0 {
return 0
@@ -38,6 +56,48 @@ func CPUTokensPerSecond(bytes int64, bwGBs float64) float64 {
return 1 / (float64(bytes)/(bwGBs*1e9*cpuReadShare) + cpuTokenFixed)
}
// CPUReadGBs is the rate a replica given threads reads its weights at, on a
// node whose memory reads allGBs with every CPU and coreGBs with one. Without
// one core's rate or the threads, the share of the whole that small machines
// gave.
func CPUReadGBs(allGBs, coreGBs float64, threads int) float64 {
if allGBs <= 0 {
return 0
}
if coreGBs <= 0 || threads <= 0 {
return allGBs * cpuReadShare
}
return min(coreShare*coreGBs*math.Pow(float64(threads), threadExp), allGBs)
}
// CPUTokensPerSecondOn is about how many tokens a second one reply gets
// from a model whose tokens each read bytes, on threads of a node whose
// memory reads allGBs with every CPU and coreGBs with one. 0 when unknown.
func CPUTokensPerSecondOn(bytes int64, allGBs, coreGBs float64, threads int) float64 {
read := CPUReadGBs(allGBs, coreGBs, threads)
if bytes <= 0 || read <= 0 {
return 0
}
return 1 / (float64(bytes)/(read*1e9) + cpuTokenFixed)
}
// CPUThreads is how many threads a CPU replica runs on: the cores its
// profile reserves, else half the node's, which is what the agent gives
// llama.cpp then.
func CPUThreads(cpuCores, nodeCPUs int) int {
if cpuCores > 0 {
return cpuCores
}
return max(nodeCPUs/2, 1)
}
// Slow speeds: under the first a reply is a wait somebody notices, under the
// second one of a few hundred tokens takes minutes.
const (
CPUSlowTPS = 15
CPUTooSlowTPS = 5
)
// moeRead is what a token of a model of experts reads over its active
// share.
const moeRead = 1.6
+61
View File
@@ -23,6 +23,44 @@ func TestCPUSpeedMatchesWhatThePoCMeasured(t *testing.T) {
}
}
// Each thread reads at about one core's rate, up to the machine's: what
// llama-bench measured on node05 by thread count, and what was measured on
// node01 and node05 before, all within 15%.
func TestCPUSpeedFollowsTheThreads(t *testing.T) {
for _, c := range []struct {
bytes int64
all, core float64
threads int
want float64
}{
{4683073536, 43.9, 9.7, 1, 1.88}, // node05, qwen2.5-coder-7b
{4683073536, 43.9, 9.7, 2, 3.56},
{4683073536, 43.9, 9.7, 4, 6.35},
{4683073536, 43.9, 9.7, 8, 9.29},
{1117320736, 43.9, 9.7, 1, 8.31}, // node05, qwen2.5-1.5b
{1117320736, 43.9, 9.7, 4, 25.4},
{1117320736, 34.2, 10.1, 2, 14.5}, // node01 on its two cores
{9_000_000_000, 43.9, 9.7, 4, 3.3}, // node05, qwen2.5-14b on four
} {
got := CPUTokensPerSecondOn(c.bytes, c.all, c.core, c.threads)
if math.Abs(got-c.want)/c.want > 0.15 {
t.Errorf("%d bytes, %d threads: %.1f tok/s, measured %.1f", c.bytes, c.threads, got, c.want)
}
}
// A large machine, a few threads: the threads decide, not the machine.
few := CPUTokensPerSecondOn(18e9, 300, 10, 4)
if few > 2.5 {
t.Errorf("four threads of a 300 GB/s machine: %.1f tok/s, as though they read at the whole machine's rate", few)
}
// An agent that did not measure one core: the old reckoning.
if got, want := CPUTokensPerSecondOn(1e9, 40, 0, 4), CPUTokensPerSecond(1e9, 40); got != want {
t.Errorf("without one core's rate %.2f, want %.2f", got, want)
}
if CPUThreads(0, 64) != 32 || CPUThreads(6, 64) != 6 || CPUThreads(0, 1) != 1 {
t.Error("threads: the profile's cores, else half the node's")
}
}
func TestAModelOfExpertsReadsItsActiveShare(t *testing.T) {
for name, want := range map[string]float64{
"Qwen/Qwen3-30B-A3B-Instruct-2507-GGUF": 0.16,
@@ -35,3 +73,26 @@ func TestAModelOfExpertsReadsItsActiveShare(t *testing.T) {
}
}
}
// A plan's ways say how fast they answer: on the CPU by the node's memory
// and half its threads, on a card by its bandwidth.
func TestAPlansWaysSayHowFastTheyAnswer(t *testing.T) {
nodes := []PoolNode{{ID: "big", Class: "l40s-48g", GPUMemGB: 48, GPUs: 2, FreeGPUs: 2, HostMemGB: 512, HostFreeGB: 400,
CPUs: 64, MemBWGBs: 300, MemBWCoreGBs: 10, CardBWGBs: map[string]float64{"l40s-48g": 864}}}
opts := Options([]Artifact{{Name: "m-Q4_K_M.gguf", Format: "gguf", Bytes: 18e9, Quant: "Q4_K_M"}},
Shape{Params: 32e9}, nodes, 4096)
var cpu, gpu float64
for _, o := range opts {
if o.CPU {
cpu = o.TokensPerSecond
} else {
gpu = o.TokensPerSecond
}
}
if want := CPUTokensPerSecondOn(18e9, 300, 10, 32); math.Abs(cpu-want) > 0.01 {
t.Errorf("on the CPU %.2f tok/s, want %.2f", cpu, want)
}
if gpu < 3*cpu {
t.Errorf("on a card %.1f tok/s against %.1f on the CPU", gpu, cpu)
}
}
+59
View File
@@ -35,6 +35,17 @@ type Shape struct {
Params int64
// Experts is a mixture-of-experts model's routed experts; 0 for dense.
Experts int
// ActiveShare is the part of the weights a token reads; 0 is all of
// them.
ActiveShare float64
}
// share is ActiveShare, all of the weights when not known.
func (sh Shape) share() float64 {
if sh.ActiveShare <= 0 || sh.ActiveShare > 1 {
return 1
}
return sh.ActiveShare
}
// PoolNode is one ready node, as a plan sees it.
@@ -57,6 +68,13 @@ type PoolNode struct {
Engines []string
// CC is its newest card's compute capability, 0 when not known.
CC float64
// CPUs, MemBWGBs and MemBWCoreGBs are its CPUs and how fast its memory
// reads, with every CPU and with one: a model's speed on its CPU. Zero
// when not measured.
CPUs int
MemBWGBs, MemBWCoreGBs float64
// CardBWGBs is a card's memory bandwidth by class, for its speed there.
CardBWGBs map[string]float64
}
// PoolKind is one class of a node's devices.
@@ -119,6 +137,10 @@ type Option struct {
// NeedGB and HaveGB are then about.
StrataFamily string `json:"strata_family,omitempty"`
StrataSize string `json:"strata_size,omitempty"`
// TokensPerSecond is about how many tokens a second one reply gets this
// way, on the fastest host it fits; 0 when not known. On the CPU with
// the threads a replica is given by default: half the node's.
TokensPerSecond float64 `json:"tokens_per_second,omitempty"`
}
// GPUs is every card one replica takes.
@@ -222,6 +244,9 @@ func Options(arts []Artifact, sh Shape, nodes []PoolNode, context int) []Option
out = append(out, vllmOptions(a, kv, sh, nodes, bpw)...)
}
}
for i := range out {
out[i].TokensPerSecond = speedOf(out[i], bytesOf(arts, out[i].Artifact), sh, nodes)
}
Rank(out)
return out
}
@@ -539,3 +564,37 @@ func CardBudget(class string) float64 { return float64(MemFromClass(class)) * gp
func SplitsHeads(kvHeads, width int) bool {
return kvHeads == 0 || width <= 1 || kvHeads%width == 0 || width%kvHeads == 0
}
func bytesOf(arts []Artifact, name string) int64 {
for _, a := range arts {
if a.Name == name {
return a.Bytes
}
}
return 0
}
// speedOf is about how many tokens a second one reply gets in o, on the
// fastest node it fits: on the CPU by the node's memory and the threads a
// replica is given, on cards by their bandwidth.
func speedOf(o Option, bytes int64, sh Shape, nodes []PoolNode) float64 {
best := 0.0
for _, n := range nodes {
var tps float64
switch {
case o.CPU:
if float64(n.HostMemGB)*hostShare < o.NeedGB {
continue
}
tps = CPUTokensPerSecondOn(int64(float64(bytes)*sh.share()), n.MemBWGBs, n.MemBWCoreGBs, CPUThreads(0, n.CPUs))
case o.Engine != "strata":
bw := n.CardBWGBs[o.Class]
if bw <= 0 {
continue
}
tps = GPUTokensPerSecond(o, bytes, sh.share(), sh.Params, bw, Links{})
}
best = max(best, tps)
}
return best
}
+42
View File
@@ -108,6 +108,23 @@ type modelUse struct {
// Declared models are read from server.yaml on every start, so a mode
// changed here would revert; the page says where to change it instead.
Declared bool
// What its card changes beside: the context and longest answer, whether
// it is stopped, its profile on cards and the cards a replica takes.
ContextLength, MaxOutput int
Stopped bool
GPUProfile string
GPUs int
// Profiles are where its replicas run and, Split, how many on each.
Profiles []profileUse
Split bool
// CanCPU says a profile on CPUs can be added — a llama.cpp model with
// none — reserving CPUMemoryGB and NewCPUCores, answering about
// CPUSpeed tokens a second on CPUSpeedNode.
CanCPU bool
CPUMemoryGB int
NewCPUCores int
CPUSpeed float64
CPUSpeedNode string
}
// mirrorRow is one model version as the page shows it.
@@ -324,6 +341,18 @@ func (p *AccountPages) RegisterAdminMirror(mux *http.ServeMux, d MirrorDeps) {
if name == "" || (mode != modeAlways && mode != modeOnDemand && mode != modeStopped) {
return "", errors.New(l.T("mirror.err.choose_mode"))
}
// Stopped is the model kept as it was, and started again so:
// not its scaling set to nothing, which forgot what it was —
// and its replicas per profile with it.
if mode == modeStopped {
if _, err := d.Catalog.Stop(ctx, name, admin.Email); err != nil {
return "", err
}
return l.T("mirror.msg.stopped", name), nil
}
if _, err := d.Catalog.Start(ctx, name, admin.Email); err == nil && mode == modeAlways {
return l.T("mirror.msg.model_started", name), nil
}
replicas, _ := strconv.Atoi(r.FormValue("replicas"))
sc := scalingFor(mode, replicas, 0)
if _, err := d.Catalog.Scale(ctx, name, sc, admin.Email); err != nil {
@@ -339,6 +368,7 @@ func (p *AccountPages) RegisterAdminMirror(mux *http.ServeMux, d MirrorDeps) {
}))
mux.HandleFunc("POST /admin/mirror/replicas", p.mirrorAction(m, p.setModel))
p.registerModelSettings(mux, m)
mux.HandleFunc("POST /admin/mirror/adapters", p.mirrorAction(m, p.setAdapters))
mux.HandleFunc("POST /admin/mirror/serve", p.mirrorAction(m,
@@ -370,6 +400,17 @@ func (p *AccountPages) RegisterAdminMirror(mux *http.ServeMux, d MirrorDeps) {
if err != nil {
return "", err
}
// Slow enough that a reply is minutes is a choice to make on
// purpose too, GPUs free or not: on a fleet whose every node
// has cards, the CPU is theirs, and some are slow.
if prof := model.Profiles[0]; !judge && prof.CPUOnly() && r.FormValue("cpu_ok") == "" {
share := p.checkSource(ctx, model.Source).Header.ActiveShare()
hosts := cpuHosts(p.fleet, prof.MinMemoryGB)
if tps, node := bestCPUSpeed(hosts, int64(float64(size)*share), prof.CPUCores); tps > 0 && tps < modelplan.CPUTooSlowTPS {
return "", errors.New(l.T("serve.err.cpu_too_slow", model.Name, l.Num(tps, 1), node,
l.Int(int64(500/tps/60+0.5))))
}
}
// The weights and the context's cache have to fit the card:
// an 8B asked for a 131072-token pool on 24 GB loaded its
// weights and failed to make its context, again and again.
@@ -554,6 +595,7 @@ func (p *AccountPages) renderMirror(w http.ResponseWriter, r *http.Request,
if len(m.Profiles) > 0 {
u.MemoryGB, u.CPUCores = m.Profiles[0].MinMemoryGB, m.Profiles[0].CPUCores
}
p.describeProfiles(&u, m, d.Rows[i].Status)
if v, ok := views[m.Name]; ok {
u.Fleet, u.Ready, u.Wanted = true, v.Ready, v.Desired
u.Parked, u.Unplaceable = v.Parked, v.Unplaceable
+89
View File
@@ -12,9 +12,11 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/gguf"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/modelplan"
"gitea.mynodes.xyz/ilker/llmgw/internal/strata"
)
@@ -33,6 +35,12 @@ type CatalogAdmin interface {
SetOwner(ctx context.Context, name, owner, by string) (catalog.Model, error)
Resources(ctx context.Context, name, profile string, memoryGB, cpuCores int, by string) (catalog.Model, error)
SetAdapters(ctx context.Context, name string, adapters []catalog.Adapter, by string) (catalog.Model, error)
// Stop takes a model out of service and keeps it; Start runs it again
// as it was.
Stop(ctx context.Context, name, by string) (catalog.Model, error)
Start(ctx context.Context, name, by string) (catalog.Model, error)
// SetProfileReplicas counts its replicas per profile, adding cpu.
SetProfileReplicas(ctx context.Context, name string, counts map[string]int, cpu *modelsvc.CPUProfile, by string) (catalog.Model, error)
}
// serveForm is the offer to put one mirrored version into service.
@@ -86,6 +94,20 @@ type serveForm struct {
// version of a family Strata runs, served on it.
StrataFamily, StrataSize string
// CPUHosts are the nodes a CPU replica of it could run on, with their
// memory's rates, and ReadBytes what a token of it reads: what the page
// reckons its speed on the CPU from, live, as the cores change.
CPUHosts []speedHost
ReadBytes int64
// CPUSpeed is about how many tokens a second one reply gets on the CPU,
// on CPUSpeedNode with the cores the form starts with; 0 when unknown.
CPUSpeed float64
CPUSpeedNode string
// GPUSpeed is the same on one card of the fleet's fastest class,
// GPUSpeedClass; 0 when the fleet has none it knows.
GPUSpeed float64
GPUSpeedClass string
// Refusal is set when this version cannot be served as it stands, and says
// why. Offered and then failing at placement is worse than not offered.
Refusal string
@@ -226,6 +248,13 @@ func planServe(l *Lang, st mirror.Status, files []mirror.Status, base string,
}
f.CPUCores = catalog.SuggestCores(cpus)
}
// How fast it would answer: on the CPU, and on a card for comparison.
if f.Engine == "llamacpp" {
f.ReadBytes = int64(float64(weights) * header.ActiveShare())
f.CPUHosts = cpuHosts(fleet, f.MinMemoryGB)
f.CPUSpeed, f.CPUSpeedNode = bestCPUSpeed(f.CPUHosts, f.ReadBytes, f.CPUCores)
f.GPUSpeed, f.GPUSpeedClass = bestGPUSpeed(fleet, weights, header.ActiveShare())
}
if biggest := biggestNode(fleet); biggest > 0 && f.MinMemoryGB > biggest {
big := l.T("serve.warn.too_big", biggest, f.MinMemoryGB)
if f.Warning != "" {
@@ -675,3 +704,63 @@ func (p *AccountPages) asJudge(m *catalog.Model) (judge bool, renamed string) {
}
return true, renamed
}
// speedHost is a node a CPU replica could run on, as its speed is reckoned:
// its memory's rate with every CPU and with one, and its CPUs.
type speedHost struct {
ID string `json:"id"`
All float64 `json:"all"`
Core float64 `json:"core"`
CPUs int `json:"cpus"`
}
// cpuHosts are the ready nodes whose RAM holds memGB and whose agents
// measured their memory.
func cpuHosts(fleet Source, memGB int) []speedHost {
if fleet == nil {
return nil
}
var out []speedHost
for _, n := range fleet.Nodes() {
if n.State != "ready" || n.MemBandwidthGBs <= 0 || (n.MemoryGB > 0 && n.MemoryGB < memGB) {
continue
}
out = append(out, speedHost{ID: n.ID, All: n.MemBandwidthGBs, Core: n.MemBandwidthCoreGBs, CPUs: n.CPUCount})
}
return out
}
// bestCPUSpeed is about how many tokens a second one reply gets on the
// fastest of hosts, reading read bytes a token on cores threads (zero takes
// what the agent gives: half the node's), and that node.
func bestCPUSpeed(hosts []speedHost, read int64, cores int) (float64, string) {
best, node := 0.0, ""
for _, h := range hosts {
if tps := modelplan.CPUTokensPerSecondOn(read, h.All, h.Core, modelplan.CPUThreads(cores, h.CPUs)); tps > best {
best, node = tps, h.ID
}
}
return best, node
}
// bestGPUSpeed is the same for one card of the fastest class the fleet has
// whose memory holds the weights, by its bandwidth, and that class.
func bestGPUSpeed(fleet Source, weights int64, share float64) (float64, string) {
if fleet == nil || weights <= 0 {
return 0, ""
}
best, class := 0.0, ""
for _, n := range fleet.Nodes() {
for _, g := range n.GPUs {
bw := modelplan.GPUBandwidthGBs(g.Name)
if bw <= 0 || g.MemoryTotalMB > 0 && int64(g.MemoryTotalMB)<<20 < weights+weights/10 {
continue
}
o := modelplan.Option{Engine: "llamacpp", Class: g.Class, GPUsPerNode: 1}
if tps := modelplan.GPUTokensPerSecond(o, weights, share, 0, bw, modelplan.Links{}); tps > best {
best, class = tps, g.Class
}
}
}
return best, class
}
+80
View File
@@ -12,6 +12,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/gguf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
)
@@ -27,6 +28,22 @@ type fakeCatalog struct {
owners []string
resources []string
adapterSets [][]catalog.Adapter
// stops, starts and splits are the models stopped, started and counted
// per profile, splitCPU the profile on CPUs last asked for.
stops, starts []string
splits []map[string]int
splitCPU *modelsvc.CPUProfile
}
func (f *fakeCatalog) SetProfileReplicas(_ context.Context, name string, counts map[string]int, cpu *modelsvc.CPUProfile,
_ string,
) (catalog.Model, error) {
f.splits, f.splitCPU = append(f.splits, counts), cpu
m := catalog.Model{Name: name}
for p, n := range counts {
m.Profiles = append(m.Profiles, catalog.Profile{Name: p, Replicas: n})
}
return m, f.err
}
func (f *fakeCatalog) Add(_ context.Context, m catalog.Model, _ string) error {
@@ -549,10 +566,12 @@ func TestTheCPUWithGPUsFreeIsAChoiceMadeOnPurpose(t *testing.T) {
}
func (f *fakeCatalog) Stop(_ context.Context, name, _ string) (catalog.Model, error) {
f.stops = append(f.stops, name)
return catalog.Model{Name: name, Stopped: &catalog.Scaling{Min: 1, Max: 1}}, f.err
}
func (f *fakeCatalog) Start(_ context.Context, name, _ string) (catalog.Model, error) {
f.starts = append(f.starts, name)
return catalog.Model{Name: name, Scaling: catalog.Scaling{Min: 1, Max: 1}}, f.err
}
@@ -590,3 +609,64 @@ func TestAStrataFamilysVersionIsServedOnStrata(t *testing.T) {
t.Error("an unknown size was taken")
}
}
// speedFleet is two nodes with GPUs whose memory rates were measured: a
// large server and a small one, as on a fleet where every node has cards.
type speedFleet struct{ fakeFleet }
func (speedFleet) Nodes() []adminapi.NodeView {
return []adminapi.NodeView{
{ID: "big", State: "ready", MemoryGB: 512, CPUCount: 64, MemBandwidthGBs: 300, MemBandwidthCoreGBs: 10,
Engines: []string{"llamacpp"}},
{ID: "small", State: "ready", MemoryGB: 64, CPUCount: 8, MemBandwidthGBs: 40, MemBandwidthCoreGBs: 10,
Engines: []string{"llamacpp"}},
}
}
// The speed on the CPU follows the threads a replica is given: four of a
// large server read at four cores' rate, not the server's.
func TestTheCPUsSpeedFollowsTheCoresGiven(t *testing.T) {
hosts := cpuHosts(speedFleet{}, 32)
if len(hosts) != 2 {
t.Fatalf("hosts %+v", hosts)
}
few, _ := bestCPUSpeed(hosts, 18e9, 4)
half, node := bestCPUSpeed(hosts, 18e9, 0)
if few > 2.5 || half < 8 || node != "big" {
t.Errorf("four cores %.1f tok/s, half the big node's %.1f on %s", few, half, node)
}
if hosts := cpuHosts(speedFleet{}, 100); len(hosts) != 1 || hosts[0].ID != "big" {
t.Errorf("a model of 100 GB fits the big node only: %+v", hosts)
}
}
// Slow enough that a reply is minutes, the CPU is a choice made on purpose,
// GPUs free or not.
func TestAVerySlowModelOnTheCPUIsAChoiceMadeOnPurpose(t *testing.T) {
big := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Length", "18000000000")
w.WriteHeader(http.StatusOK)
}))
t.Cleanup(big.Close)
values := func(extra ...string) url.Values {
v := url.Values{
"csrf": {"tok"}, "name": {"big-dense"}, "source": {big.URL + "/m/v1/m.gguf"}, "engine": {"llamacpp"},
"replicas": {"1"}, "context_length": {"4096"}, "placement": {"cpu"}, "min_memory_gb": {"24"},
"cpu_cores": {"4"},
}
for i := 0; i+1 < len(extra); i += 2 {
v.Set(extra[i], extra[i+1])
}
return v
}
cat := &fakeCatalog{}
page := mirrorPageDeps(t, MirrorDeps{Mirror: &fakeMirror{}, Catalog: cat, Fleet: speedFleet{}}, true)
body := mirrorPost(t, page, "/admin/mirror/serve", values()).Body.String()
if len(cat.added) != 0 || !strings.Contains(body, "tokens a second on the CPU") {
t.Fatalf("an 18 GB model on four cores went through unasked: %s", body)
}
mirrorPost(t, page, "/admin/mirror/serve", values("cpu_ok", "yes"))
if len(cat.added) != 1 {
t.Fatal("ticking the box should add it")
}
}
+18 -10
View File
@@ -43,9 +43,17 @@ type clusterFit struct {
// cpuHost is a node as a model on its CPU sees it.
type cpuHost struct {
id string
ramGB int
bwGBs float64
id string
ramGB int
cpus int
bwGBs float64
coreBW float64 // one core's rate, 0 from an older agent
}
// tps is about how many tokens a second one reply of a model whose tokens
// each read bytes gets here, with the threads a replica is given by default.
func (h cpuHost) tps(bytes int64) float64 {
return modelplan.CPUTokensPerSecondOn(bytes, h.bwGBs, h.coreBW, modelplan.CPUThreads(0, h.cpus))
}
// fp8Capability is the oldest card vLLM runs an FP8 checkpoint on: Ampere,
@@ -100,7 +108,8 @@ func (p *AccountPages) fitOf() *clusterFit {
}
}
if n.State == "ready" && n.MemBandwidthGBs > 0 {
c.cpus = append(c.cpus, cpuHost{id: n.ID, ramGB: n.MemoryGB, bwGBs: n.MemBandwidthGBs})
c.cpus = append(c.cpus, cpuHost{id: n.ID, ramGB: n.MemoryGB, cpus: n.CPUCount,
bwGBs: n.MemBandwidthGBs, coreBW: n.MemBandwidthCoreGBs})
}
for _, s := range n.ComputeCapabilities {
if v, err := strconv.ParseFloat(s, 64); err == nil && v > newest[n.ID] {
@@ -268,17 +277,16 @@ func (c *clusterFit) holds(l *Lang, name string, bytes, params int64) (gguf.Verd
// bytes on the fastest node whose RAM holds it, and that node; share is the
// part a token reads (ActiveShare). 0 when no measured node holds it.
func (c *clusterFit) cpuSpeed(bytes int64, share float64) (float64, string) {
read := int64(float64(bytes) * share)
var best cpuHost
bestTPS := 0.0
for _, n := range c.cpus {
need := float64(bytes) + float64(1<<30) + float64(bytes)/20
if need <= float64(n.ramGB)*0.85*(1<<30) && n.bwGBs > best.bwGBs {
best = n
if tps := n.tps(read); need <= float64(n.ramGB)*0.85*(1<<30) && tps > bestTPS {
best, bestTPS = n, tps
}
}
if best.id == "" {
return 0, ""
}
return modelplan.CPUTokensPerSecond(int64(float64(bytes)*share), best.bwGBs), best.id
return bestTPS, best.id
}
// speedLabel is a rate as the page shows it: whole tokens from ten up, one
+40 -1
View File
@@ -279,5 +279,44 @@
"strata.msg.served": "%s added: its first start fetches about %s GB onto the node and prepares it, which can take hours on a slow link; Replicas shows how far it is.",
"strata.pack_files": "part of Strata's ready-made %s %s pack: served whole from the mirror's Strata tab, Strata fetching it onto the node",
"strata.unknown_pack": "a pack made for Strata that Strata 0.2.7 does not know: its setup refuses it, and llama.cpp cannot read it",
"strata.old_card": "its engine needs an NVIDIA card of compute capability %s or newer (RTX 30 on); the nodes with Strata have older ones"
"strata.old_card": "its engine needs an NVIDIA card of compute capability %s or newer (RTX 30 on); the nodes with Strata have older ones",
"mirror.settings.open": "Settings",
"mirror.settings.heading": "Settings",
"mirror.settings.context": "Change the context",
"mirror.settings.context_note": "Leave one empty to keep it. The replicas start again on the new context, one at a time; a longest answer alone changes at once.",
"mirror.settings.gpus": "Cards a replica",
"mirror.settings.gpus_apply": "Spread it",
"mirror.settings.gpus_note": "Each replica moves to this many cards of its class, started again one at a time.",
"mirror.settings.start": "Start",
"mirror.settings.stop": "Stop",
"mirror.settings.stop_confirm": "Stop %s? Its replicas stop; it is kept, with its files, and Start runs it again as it was.",
"mirror.settings.restart": "Restart its replicas",
"mirror.settings.restart_confirm": "Start each of %s's replicas again, one at a time?",
"mirror.settings.unserve": "Remove from the catalog",
"mirror.settings.unserve_confirm": "Take %s out of the catalog? Its replicas stop; its files stay on the mirror and the nodes.",
"mirror.settings.purge": "Remove with its files",
"mirror.settings.purge_confirm": "Take %s out of the catalog and delete its files from the mirror and the nodes? What another model uses stays.",
"mirror.split.heading": "Replicas on each profile — all answer under the model's name; the cards are used first, the CPUs when they are full.",
"mirror.split.replicas": "replicas",
"mirror.split.on": "%d on %s",
"mirror.split.on_cpu": "On CPUs (%s)",
"mirror.split.on_gpu": "On cards (%s, %s)",
"mirror.split.add_cpu": "Also on CPUs",
"mirror.split.cpu_speed": "On the CPU about %s tokens a second for one reply, at best on %s.",
"mirror.split.note": "Those short start first, then those over stop: a change never runs fewer.",
"mirror.split.scaling_note": "Its replicas are counted per profile below; min and max are their sum.",
"mirror.msg.model_started": "%s is started again, as it was",
"mirror.msg.context": "%s: a context of %s tokens, answers of at most %s; %d replicas starting again",
"mirror.msg.gpus": "%s: %d cards a replica; %d replicas starting again",
"mirror.msg.restarting.one": "%[2]s: its replica is starting again",
"mirror.msg.restarting.other": "%[2]s: its %[1]d replicas are starting again, one at a time",
"mirror.msg.split": "%s runs %s",
"mirror.msg.unserved": "%s is out of the catalog; its files stay",
"mirror.msg.purged.one": "%[2]s is out of the catalog, its version removed from the mirror and its copies from the nodes.",
"mirror.msg.purged.other": "%[2]s is out of the catalog, %[1]d versions removed from the mirror and its copies from the nodes.",
"mirror.err.no_settings": "this control plane cannot change that",
"mirror.err.gpus": "the cards a replica takes is a whole number, 1 or more",
"mirror.err.action": "stop, start or restart",
"mirror.err.count": "%s is a whole number, 0 or more",
"mirror.err.choose_model": "choose a model"
}
+6
View File
@@ -46,6 +46,12 @@
"serve.pool": "context pool %s",
"serve.will_run_as": "Will run as: %s.",
"serve.free_gpus_warning": "%s GPUs are free; on the CPU this runs many times slower.",
"serve.speed_cpu": "about %s tokens a second on %s",
"serve.speed_gpu": "on one %s card about %s",
"serve.speed_slow": "Slow: a reply of 500 tokens takes about %s seconds.",
"serve.speed_too_slow": "Very slow: a reply of 500 tokens takes about %s minutes.",
"serve.cpu_ok_slow": "Run on the CPU anyway, knowing it will be slow",
"serve.err.cpu_too_slow": "%s would answer at about %s tokens a second on the CPU, at best on %s: a reply of 500 tokens in about %s minutes. Choose a GPU placement, or tick \"run on the CPU anyway\"",
"serve.pool_note": "The context pool is what the engine is given; llama.cpp divides it across its parallel slots, so the number one request may use is the smaller one above.",
"serve.memory_note": "Memory: %s. Change the context pool and this changes with it, by the KV cache's share — an estimate, and editable. It is filled in rather than left blank because a CPU profile has no card to be short of: with no figure, every ready node is a candidate and the first one picked is whichever the scheduler reached, not whichever can hold the model. A node label does the same job by name if you would rather pin it.",
+38 -1
View File
@@ -272,5 +272,42 @@
"strata.msg.served": "%s eklendi: ilk açılışında node'a yaklaşık %s GB indirip hazırlar; yavaş bağlantıda saatler sürebilir. İlerleyişi Replikalar gösterir.",
"strata.pack_files": "Strata'nın hazır %s %s paketinin parçası: aynanın Strata sekmesinden bütün olarak sunulur, Strata onu node'a kendisi indirir",
"strata.unknown_pack": "Strata için hazırlanmış ama Strata 0.2.7'nin tanımadığı bir paket: kurulumu onu reddeder, llama.cpp de okuyamaz",
"strata.old_card": "motoru compute capability %s ya da daha yeni bir NVIDIA kartı ister (RTX 30 ve sonrası); Strata kurulu node'ların kartları daha eski"
"strata.old_card": "motoru compute capability %s ya da daha yeni bir NVIDIA kartı ister (RTX 30 ve sonrası); Strata kurulu node'ların kartları daha eski",
"mirror.settings.open": "Ayarlar",
"mirror.settings.heading": "Ayarlar",
"mirror.settings.context": "Bağlamı değiştir",
"mirror.settings.context_note": "Değişmeyecek olanı boş bırakın. Replikalar yeni bağlamla teker teker yeniden başlar; yalnızca en uzun cevap değişirse hemen geçerli olur.",
"mirror.settings.gpus": "Replika başına kart",
"mirror.settings.gpus_apply": "Dağıt",
"mirror.settings.gpus_note": "Her replika kendi sınıfından bu kadar karta taşınır, teker teker yeniden başlatılarak.",
"mirror.settings.start": "Başlat",
"mirror.settings.stop": "Durdur",
"mirror.settings.stop_confirm": "%s durdurulsun mu? Replikaları kapanır; model dosyalarıyla birlikte saklanır, Başlat onu olduğu gibi yeniden çalıştırır.",
"mirror.settings.restart": "Replikaları yeniden başlat",
"mirror.settings.restart_confirm": "%s replikaları teker teker yeniden başlatılsın mı?",
"mirror.settings.unserve": "Katalogdan çıkar",
"mirror.settings.unserve_confirm": "%s katalogdan çıkarılsın mı? Replikaları kapanır; dosyaları aynada ve node'larda kalır.",
"mirror.settings.purge": "Dosyalarıyla birlikte sil",
"mirror.settings.purge_confirm": "%s katalogdan çıkarılıp dosyaları aynadan ve node'lardan silinsin mi? Başka bir modelin kullandıkları kalır.",
"mirror.split.heading": "Her profilde çalışan replika sayısı — hepsi modelin adıyla cevap verir; önce kartlar kullanılır, dolunca CPU'lar.",
"mirror.split.replicas": "replika sayısı",
"mirror.split.on": "%[2]s üzerinde %[1]d",
"mirror.split.on_cpu": "CPU'larda (%s)",
"mirror.split.on_gpu": "Kartlarda (%s, %s)",
"mirror.split.add_cpu": "CPU'larda da",
"mirror.split.cpu_speed": "CPU üzerinde tek bir cevap için yaklaşık %s token/sn, en iyi %s üzerinde.",
"mirror.split.note": "Önce eksik olanlar başlar, sonra fazlası kapanır: bir değişiklik sırasında daha az replika çalışmaz.",
"mirror.split.scaling_note": "Replikaları aşağıda profil başına sayılıyor; en az ve en çok bunların toplamıdır.",
"mirror.msg.model_started": "%s olduğu gibi yeniden başlatıldı",
"mirror.msg.context": "%s: %s token bağlam, en fazla %s token cevap; %d replika yeniden başlıyor",
"mirror.msg.gpus": "%s: replika başına %d kart; %d replika yeniden başlıyor",
"mirror.msg.restarting.other": "%[2]s: %[1]d replikası teker teker yeniden başlıyor",
"mirror.msg.split": "%s şöyle çalışıyor: %s",
"mirror.msg.unserved": "%s katalogdan çıkarıldı; dosyaları duruyor",
"mirror.msg.purged.other": "%[2]s katalogdan çıkarıldı, aynadan %[1]d sürümü ve node'lardaki kopyaları silindi.",
"mirror.err.no_settings": "bu control plane bunu değiştiremez",
"mirror.err.gpus": "replika başına kart sayısı 1 ya da daha büyük bir tam sayıdır",
"mirror.err.action": "durdur, başlat ya da yeniden başlat",
"mirror.err.count": "%s 0 ya da daha büyük bir tam sayıdır",
"mirror.err.choose_model": "bir model seçin"
}
+6
View File
@@ -45,6 +45,12 @@
"serve.pool": "%s token bağlam havuzu",
"serve.will_run_as": "Şöyle çalışacak: %s.",
"serve.free_gpus_warning": "%s GPU boşta; CPU üzerinde bu model kat kat yavaş çalışır.",
"serve.speed_cpu": "%[2]s üzerinde yaklaşık %[1]s token/sn",
"serve.speed_gpu": "tek bir %s kartında yaklaşık %s token/sn",
"serve.speed_slow": "Yavaş: 500 token'lık bir cevap yaklaşık %s saniye sürer.",
"serve.speed_too_slow": "Çok yavaş: 500 token'lık bir cevap yaklaşık %s dakika sürer.",
"serve.cpu_ok_slow": "Yine de CPU üzerinde çalıştır: yavaş olacağını biliyorum",
"serve.err.cpu_too_slow": "%s CPU üzerinde yaklaşık %s token/sn ile cevap verir (en iyi %s üzerinde): 500 token'lık bir cevap yaklaşık %s dakika. Bir GPU yerleşimi seçin ya da \"yine de CPU üzerinde çalıştır\" kutusunu işaretleyin",
"serve.pool_note": "Bağlam havuzu motora verilen miktardır; llama.cpp onu paralel yuvalarına böler, bu yüzden tek bir isteğin kullanabileceği sayı yukarıdaki küçük olandır.",
"serve.memory_note": "Bellek: %s. Bağlam havuzunu değiştirirseniz bu da KV önbelleğinin payı kadar değişir — bir tahmindir ve düzenlenebilir. Boş bırakılmak yerine doldurulur, çünkü bir CPU profilinin yetersiz kalacak bir kartı yoktur: bir değer olmadan hazır her düğüm adaydır ve ilk seçilen, modeli taşıyabilen değil, zamanlayıcının ilk ulaştığı düğüm olur. Sabitlemeyi tercih ederseniz bir düğüm etiketi aynı işi adla görür.",
+89
View File
@@ -2,14 +2,17 @@ package web
import (
"context"
"fmt"
"log/slog"
"net/http"
"net/url"
"slices"
"strings"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/gguf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
@@ -333,3 +336,89 @@ func TestAnAdapterHeldInTheMirrorIsAttachedFromItsModelsCard(t *testing.T) {
}
}
}
// settingsFleet is roomyFleet that can change what the catalog does not
// hold of a model, and says what it was asked.
type settingsFleet struct {
roomyFleet
asked []string
}
func (f *settingsFleet) SetModelContext(_ context.Context, name string, req modelserve.ContextRequest, _ string) (modelserve.ContextChange, error) {
f.asked = append(f.asked, fmt.Sprintf("context %s %d %d", name, req.Tokens, req.MaxOutput))
return modelserve.ContextChange{Model: name, After: req.Tokens, MaxOutput: req.MaxOutput}, nil
}
func (f *settingsFleet) SetModelGPUs(_ context.Context, name string, req modelserve.GPUsRequest, _ string) (modelserve.GPUsChange, error) {
f.asked = append(f.asked, fmt.Sprintf("gpus %s %d", name, req.GPUs))
return modelserve.GPUsChange{Model: name, After: req.GPUs}, nil
}
func (f *settingsFleet) RestartModel(name string) (int, error) {
f.asked = append(f.asked, "restart "+name)
return 1, nil
}
func (f *settingsFleet) RemoveModel(_ context.Context, name string, purge bool, _ string) (adminapi.PurgeView, error) {
f.asked = append(f.asked, fmt.Sprintf("remove %s %v", name, purge))
return adminapi.PurgeView{Removed: name}, nil
}
// A served model's row leads to its settings, and its card holds them all:
// context, cards, replicas per profile with CPUs to add, stop, restart,
// removal with its files or without.
func TestAServedModelsSettingsAreWhereItsFileIs(t *testing.T) {
cat := &fakeCatalog{sources: map[string][]catalog.Model{"http://mirror:8081/coder/v1/c.gguf": {
{Name: "coder", ContextLength: 8192, MaxOutput: 1024, Scaling: catalog.Scaling{Min: 1, Max: 1},
Profiles: []catalog.Profile{{Name: "gpu-1x6000-24g", Engine: "llamacpp", GPUClass: "6000-24g", GPUsPerNode: 1}}}}}}
fleet := &settingsFleet{roomyFleet: roomyFleet{more: 1}}
m := &fakeMirror{models: []mirror.Status{
{Model: "coder", Version: "v1", File: "c.gguf", State: mirror.Present, BytesDone: 5 << 30},
}}
mux := mirrorPageDeps(t, MirrorDeps{Mirror: m, Catalog: cat, Fleet: fleet}, true)
if row := mirrorGet(t, mux, "/admin/mirror").Body.String(); !strings.Contains(row, `open=coder%3av1">Settings</a>`) {
t.Error("the row of a served model does not lead to its settings")
}
card := mirrorGet(t, mux, "/admin/mirror?tab=held&open=coder:v1").Body.String()
for _, want := range []string{`action="/admin/mirror/context"`, `name="tokens" value="8192"`, `name="max_output" value="1024"`,
`action="/admin/mirror/gpus"`, `name="gpus" value="1"`, `action="/admin/mirror/split"`, `name="cpu_replicas"`,
`value="stop"`, `value="restart"`, `action="/admin/mirror/unserve"`, `name="purge" value="yes"`} {
if !strings.Contains(card, want) {
t.Errorf("the card has no %s", want)
}
}
post := func(path string, kv ...string) string {
v := url.Values{"csrf": {"tok"}, "model": {"coder"}}
for i := 0; i+1 < len(kv); i += 2 {
v.Add(kv[i], kv[i+1])
}
return mirrorPost(t, mux, path, v).Body.String()
}
post("/admin/mirror/context", "tokens", "16384", "max_output", "")
post("/admin/mirror/gpus", "gpus", "2")
post("/admin/mirror/state", "action", "restart")
post("/admin/mirror/unserve", "purge", "yes")
if want := []string{"context coder 16384 0", "gpus coder 2", "restart coder", "remove coder true"}; !slices.Equal(fleet.asked, want) {
t.Errorf("asked %v, want %v", fleet.asked, want)
}
post("/admin/mirror/state", "action", "stop")
post("/admin/mirror/state", "action", "start")
if !slices.Equal(cat.stops, []string{"coder"}) || !slices.Equal(cat.starts, []string{"coder"}) {
t.Errorf("stopped %v, started %v", cat.stops, cat.starts)
}
post("/admin/mirror/split", "profile", "gpu-1x6000-24g", "replicas_gpu-1x6000-24g", "7",
"cpu_replicas", "4", "cpu_memory", "12", "cpu_cores", "8")
if len(cat.splits) != 1 || cat.splits[0]["gpu-1x6000-24g"] != 7 || cat.splits[0]["cpu"] != 4 ||
cat.splitCPU == nil || cat.splitCPU.MemoryGB != 12 || cat.splitCPU.CPUCores != 8 {
t.Errorf("split %v, cpu %+v", cat.splits, cat.splitCPU)
}
}
// Stopped on the row is the model kept, not its scaling set to nothing.
func TestStoppingFromTheRowKeepsTheModel(t *testing.T) {
cat := coderCatalog()
mux := mirrorPageDeps(t, MirrorDeps{Mirror: &fakeMirror{}, Catalog: cat}, true)
mirrorPost(t, mux, "/admin/mirror/mode", url.Values{"csrf": {"tok"}, "model": {"coder"}, "mode": {"stopped"}, "replicas": {"1"}})
if !slices.Equal(cat.stops, []string{"coder"}) || len(cat.scaled) != 0 {
t.Errorf("stopped %v, scaled %v", cat.stops, cat.scaled)
}
}
+237
View File
@@ -0,0 +1,237 @@
package web
import (
"context"
"errors"
"net/http"
"strconv"
"strings"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
// ModelSettings changes a served model beyond what the catalogue holds: its
// context, its cards, its replicas started again, its removal. The cluster
// view answers it where it can.
type ModelSettings interface {
SetModelContext(ctx context.Context, name string, req modelserve.ContextRequest, by string) (modelserve.ContextChange, error)
SetModelGPUs(ctx context.Context, name string, req modelserve.GPUsRequest, by string) (modelserve.GPUsChange, error)
RestartModel(name string) (int, error)
RemoveModel(ctx context.Context, name string, purge bool, by string) (adminapi.PurgeView, error)
}
// settings is the cluster view's ModelSettings, nil where it has none.
func (p *AccountPages) settings() ModelSettings {
s, _ := p.fleet.(ModelSettings)
return s
}
// registerModelSettings mounts what the card of a served model changes.
func (p *AccountPages) registerModelSettings(mux *http.ServeMux, m MirrorAdmin) {
need := func(r *http.Request) (string, ModelSettings, error) {
name := strings.TrimSpace(r.FormValue("model"))
if name == "" {
return "", nil, errors.New(langOf(r).T("mirror.err.choose_model"))
}
s := p.settings()
if s == nil {
return "", nil, errors.New(langOf(r).T("mirror.err.no_settings"))
}
return name, s, nil
}
mux.HandleFunc("POST /admin/mirror/context", p.mirrorAction(m,
func(ctx context.Context, admin store.User, r *http.Request) (string, error) {
l := langOf(r)
name, s, err := need(r)
if err != nil {
return "", err
}
tokens, err := optionalCount(l, r.FormValue("tokens"), "serve.context_length")
if err != nil {
return "", err
}
maxOut, err := optionalCount(l, r.FormValue("max_output"), "serve.max_output")
if err != nil {
return "", err
}
ch, err := s.SetModelContext(ctx, name, modelserve.ContextRequest{Tokens: tokens, MaxOutput: maxOut}, admin.Email)
if err != nil {
return "", err
}
return l.T("mirror.msg.context", name, l.Int(int64(ch.After)), l.Int(int64(ch.MaxOutput)), ch.Restarted), nil
}))
mux.HandleFunc("POST /admin/mirror/gpus", p.mirrorAction(m,
func(ctx context.Context, admin store.User, r *http.Request) (string, error) {
l := langOf(r)
name, s, err := need(r)
if err != nil {
return "", err
}
n, err := strconv.Atoi(strings.TrimSpace(r.FormValue("gpus")))
if err != nil || n < 1 {
return "", errors.New(l.T("mirror.err.gpus"))
}
ch, err := s.SetModelGPUs(ctx, name, modelserve.GPUsRequest{GPUs: n, Profile: r.FormValue("profile")}, admin.Email)
if err != nil {
return "", err
}
return l.T("mirror.msg.gpus", name, ch.After, ch.Restarted), nil
}))
mux.HandleFunc("POST /admin/mirror/state", p.mirrorAction(m,
func(ctx context.Context, admin store.User, r *http.Request) (string, error) {
l := langOf(r)
name := strings.TrimSpace(r.FormValue("model"))
switch r.FormValue("action") {
case "stop":
if _, err := p.catalog.Stop(ctx, name, admin.Email); err != nil {
return "", err
}
return l.T("mirror.msg.stopped", name), nil
case "start":
if _, err := p.catalog.Start(ctx, name, admin.Email); err != nil {
return "", err
}
return l.T("mirror.msg.model_started", name), nil
case "restart":
s := p.settings()
if s == nil {
return "", errors.New(l.T("mirror.err.no_settings"))
}
n, err := s.RestartModel(name)
if err != nil {
return "", err
}
return l.N("mirror.msg.restarting", int64(n), name), nil
}
return "", errors.New(l.T("mirror.err.action"))
}))
mux.HandleFunc("POST /admin/mirror/split", p.mirrorAction(m,
func(ctx context.Context, admin store.User, r *http.Request) (string, error) {
l := langOf(r)
name := strings.TrimSpace(r.FormValue("model"))
counts := map[string]int{}
for _, prof := range r.Form["profile"] {
n, err := count(l, r.FormValue("replicas_"+prof), "mirror.split.replicas")
if err != nil {
return "", err
}
counts[prof] = n
}
var cpu *modelsvc.CPUProfile
if v := strings.TrimSpace(r.FormValue("cpu_replicas")); v != "" && v != "0" {
n, err := count(l, v, "mirror.split.replicas")
if err != nil {
return "", err
}
mem, err := count(l, r.FormValue("cpu_memory"), "serve.memory")
if err != nil {
return "", err
}
cores, err := optionalCount(l, r.FormValue("cpu_cores"), "serve.cpu_cores")
if err != nil {
return "", err
}
cpu = &modelsvc.CPUProfile{Name: "cpu", MemoryGB: mem, CPUCores: cores}
counts["cpu"] = n
}
got, err := p.catalog.SetProfileReplicas(ctx, name, counts, cpu, admin.Email)
if err != nil {
return "", err
}
var parts []string
for _, pr := range got.Profiles {
parts = append(parts, l.T("mirror.split.on", pr.Replicas, pr.Name))
}
return l.T("mirror.msg.split", name, strings.Join(parts, ", ")), nil
}))
mux.HandleFunc("POST /admin/mirror/unserve", p.mirrorAction(m,
func(ctx context.Context, admin store.User, r *http.Request) (string, error) {
l := langOf(r)
name, s, err := need(r)
if err != nil {
return "", err
}
purge := r.FormValue("purge") == "yes"
out, err := s.RemoveModel(ctx, name, purge, admin.Email)
if err != nil {
return "", err
}
if purge {
msg := l.N("mirror.msg.purged", int64(len(out.Mirror)), name)
if out.Note != "" {
msg += " " + out.Note
}
return msg, nil
}
return l.T("mirror.msg.unserved", name), nil
}))
}
// count is a whole number of at least zero from a form field, or why not.
func count(l *Lang, v, field string) (int, error) {
n, err := strconv.Atoi(strings.TrimSpace(v))
if err != nil || n < 0 {
return 0, errors.New(l.T("mirror.err.count", l.T(field)))
}
return n, nil
}
// optionalCount is count, with an empty field zero: left as it is.
func optionalCount(l *Lang, v, field string) (int, error) {
if strings.TrimSpace(v) == "" {
return 0, nil
}
return count(l, v, field)
}
// profileUse is one of a model's profiles as its card shows it.
type profileUse struct {
Name string
CPU bool
Class string
GPUs int
Replicas int
}
// describeProfiles fills in what the card of a served model shows of its
// profiles, and of a profile on CPUs it could have.
func (p *AccountPages) describeProfiles(u *modelUse, m catalog.Model, st mirror.Status) {
u.ContextLength, u.MaxOutput, u.Stopped, u.Split = m.ContextLength, m.MaxOutput, m.Stopped != nil, m.Split()
hasCPU, llama := false, false
for _, pr := range m.Profiles {
pu := profileUse{Name: pr.Name, CPU: pr.CPUOnly(), Class: pr.GPUClass, Replicas: pr.Replicas,
GPUs: max(pr.GPUsPerNode, pr.TP*max(pr.PP, 1))}
u.Profiles = append(u.Profiles, pu)
hasCPU = hasCPU || pu.CPU
llama = llama || pr.Engine == "llamacpp"
if !pu.CPU && u.GPUProfile == "" && pr.GPU == nil {
u.GPUProfile, u.GPUs = pr.Name, max(pu.GPUs, 1)
}
}
if !llama || hasCPU || m.IsExternal() {
return
}
u.CanCPU = true
u.CPUMemoryGB = memoryFor(st.BytesTotal, st.BytesDone)
hosts := cpuHosts(p.fleet, u.CPUMemoryGB)
var cpus []int
for _, h := range hosts {
cpus = append(cpus, h.CPUs)
}
u.NewCPUCores = catalog.SuggestCores(cpus)
weights := st.BytesTotal
if weights <= 0 {
weights = st.BytesDone
}
u.CPUSpeed, u.CPUSpeedNode = bestCPUSpeed(hosts, weights, u.NewCPUCores)
}
+8
View File
@@ -828,6 +828,14 @@ tr.held-detail > td { background: var(--bg); padding: .75rem 1rem 1rem; }
.held-card .replicas-form .field-row { align-items: flex-end; }
.held-card .replicas-form .field-row label { flex: 0 0 auto; }
.held-card .replicas-form input[type=number] { width: 8rem; margin: .25rem 0 0; }
.held-card .model-settings { margin-top: .75rem; padding-top: .5rem; border-top: 1px dashed var(--line); }
.held-card .model-settings > summary { cursor: pointer; list-style: none; }
.held-card .model-settings > summary h4 { display: inline; font-size: .95rem; margin: 0; }
.held-card .settings-form { margin: .6rem 0; }
.held-card .settings-form .field-row { align-items: flex-end; }
.held-card .settings-form .field-row label { flex: 0 0 auto; }
.held-card .settings-form input[type=number] { width: 8rem; margin: .25rem 0 0; }
.held-card .settings-actions { display: flex; gap: .75rem; flex-wrap: wrap; align-items: center; margin-top: .6rem; }
.held-card .replicas-form button { margin: 0; }
a.open-held { text-decoration: none; color: inherit; }
a.open-held:hover strong { text-decoration: underline; }
+4 -1
View File
@@ -15,7 +15,10 @@
}
});
document.addEventListener('submit', function (e) {
var msg = e.target.getAttribute && e.target.getAttribute('data-confirm');
// The button's own, where a form has several that mean different things.
var b = e.submitter;
var msg = (b && b.getAttribute && b.getAttribute('data-confirm')) ||
(e.target.getAttribute && e.target.getAttribute('data-confirm'));
if (msg && !window.confirm(msg)) { e.preventDefault(); }
});
// A form being filled in is not refreshed away. The live parts of the
+120 -9
View File
@@ -109,6 +109,7 @@
</form>
{{else if eq (printf "%s" .State) "present"}}
<button type="button" class="link copy-url" data-url="{{.URL}}" title="{{.URL}}">{{t "mirror.copy_address"}}</button>
{{if and .UsedBy (not .Detail)}}<a href="/admin/mirror?tab=held&amp;open={{.Open}}">{{t "mirror.settings.open"}}</a>{{end}}
{{if and $.HasCatalog (not .UsedBy)}}{{if .Serve}}<a href="/admin/mirror">{{t "mirror.close"}}</a>
{{else}}<a href="/admin/mirror?serve={{.Model}}:{{.Version}}">{{t "mirror.serve"}}</a>{{end}}{{end}}
<form method="post" action="/admin/mirror/{{.Model}}/{{.Version}}/remove"
@@ -177,8 +178,8 @@
{{if eq .Engine "llamacpp"}}<label>{{t "serve.parallel"}} <span class="opt">{{t "serve.parallel_hint"}}</span>
<input type="text" name="parallel" value="{{.Parallel}}" size="3"></label>{{end}}
</div>
{{if .FreeGPUs}}<label class="cpu-ok"><input type="checkbox" name="cpu_ok" value="yes">
{{t "serve.cpu_ok" .FreeGPUs}}</label>{{end}}
{{if and (eq .Engine "llamacpp") (or .FreeGPUs .CPUHosts)}}<label class="cpu-ok"><input type="checkbox" name="cpu_ok" value="yes">
{{if .FreeGPUs}}{{t "serve.cpu_ok" .FreeGPUs}}{{else}}{{t "serve.cpu_ok_slow"}}{{end}}</label>{{end}}
<div class="field-row">
<label>{{t "serve.memory"}} <span class="opt">{{t "serve.memory_hint"}}</span>
<input type="text" name="min_memory_gb" value="{{.MinMemoryGB}}" required></label>
@@ -204,7 +205,29 @@
mem: {{t "serve.host_memory"}}, cores: {{t "serve.cores.other"}},
poolPar: {{t "serve.pool_parallel"}}, pool: {{t "serve.pool"}},
runAs: {{t "serve.will_run_as"}}, free: {{t "serve.free_gpus_warning"}},
speedCPU: {{t "serve.speed_cpu"}}, speedGPU: {{t "serve.speed_gpu"}},
slow: {{t "serve.speed_slow"}}, tooSlow: {{t "serve.speed_too_slow"}},
};
// How fast one reply comes on the CPU: what a token reads over the rate a
// replica's threads read at — each about one core's, up to the machine's —
// and 5 ms a token besides. modelplan.CPUTokensPerSecondOn, in the page so
// it follows the cores as they are typed.
const hosts = {{.CPUHosts}} || [], readBytes = {{.ReadBytes}};
const gpuTPS = {{.GPUSpeed}}, gpuClass = {{.GPUSpeedClass}};
const slowTPS = 15, tooSlowTPS = 5;
function cpuTPS(cores) {
let best = 0, node = "";
for (const h of hosts) {
const threads = cores > 0 ? cores : Math.max(Math.floor(h.cpus / 2), 1);
let read = h.core > 0 && threads > 0 ? Math.min(0.95 * h.core * Math.pow(threads, 0.9), h.all) : h.all * 0.68;
if (read <= 0 || readBytes <= 0) continue;
const tps = 1 / (readBytes / (read * 1e9) + 0.005);
if (tps > best) { best = tps; node = h.id; }
}
return [best, node];
}
const nf = new Intl.NumberFormat(document.documentElement.lang || undefined, { maximumFractionDigits: 1 });
const rate = (v) => nf.format(v < 10 ? Math.round(v * 10) / 10 : Math.round(v));
const fmt = (s, ...a) => {
let i = 0;
return s.replace(/%(?:\[(\d+)\])?[sd]/g, (_, n) => { const k = n ? +n - 1 : i; i = k + 1; return String(a[k]); });
@@ -226,11 +249,21 @@
const pool = parseInt(v("ctx_size"), 10), par = parseInt(v("parallel"), 10);
if (pool && par) parts.push(fmt(T.poolPar, pool, par, Math.floor(pool / par)));
else if (pool) parts.push(fmt(T.pool, pool));
let tps = 0;
if (onCPU && f.dataset.engine === "llamacpp") {
const r = cpuTPS(parseInt(v("cpu_cores"), 10) || 0);
tps = r[0];
if (tps > 0) parts.push(fmt(T.speedCPU, rate(tps), r[1]));
if (gpuTPS > 0) parts.push(fmt(T.speedGPU, gpuClass, rate(gpuTPS)));
}
out.textContent = fmt(T.runAs, parts.join(", "));
const warn = onCPU && +f.dataset.freeGpus > 0;
out.classList.toggle("error", warn);
if (warn) out.textContent += " " + fmt(T.free, f.dataset.freeGpus);
if (ok) ok.hidden = !onCPU;
const tooSlow = tps > 0 && tps < tooSlowTPS, slow = tps > 0 && tps < slowTPS;
const free = onCPU && +f.dataset.freeGpus > 0;
out.classList.toggle("error", free || tooSlow);
if (tooSlow) out.textContent += " " + fmt(T.tooSlow, Math.max(1, Math.round(500 / tps / 60)));
else if (slow) out.textContent += " " + fmt(T.slow, Math.round(500 / tps));
if (free) out.textContent += " " + fmt(T.free, f.dataset.freeGpus);
if (ok) ok.hidden = !(onCPU && (free || tooSlow));
out.hidden = false;
}
f.addEventListener("input", draw);
@@ -631,9 +664,9 @@ document.querySelector("table.explore").addEventListener("submit", async functio
<input type="hidden" name="open" value="{{$row.Open}}">
<div class="field-row">
<label>{{t "mirror.form.min"}}
<input type="number" name="min" value="{{.Min}}" min="0"{{if .MostNow}} max="{{.MostNow}}"{{end}} required></label>
<input type="number" name="min" value="{{.Min}}" min="0"{{if .MostNow}} max="{{.MostNow}}"{{end}} required{{if .Split}} readonly{{end}}></label>
<label>{{t "mirror.form.max"}}
<input type="number" name="max" value="{{.Max}}" min="0" max="64" required></label>
<input type="number" name="max" value="{{.Max}}" min="0" max="64" required{{if .Split}} readonly{{end}}></label>
<label>{{t "mirror.form.class"}}
<select name="class">
<option value=""{{if eq .Class ""}} selected{{end}}>{{if eq .Mode "ondemand"}}{{t "mirror.class.default" (t "mirror.class.opportunistic")}}{{else}}{{t "mirror.class.default" (t "mirror.class.standard")}}{{end}}</option>
@@ -654,10 +687,88 @@ document.querySelector("table.explore").addEventListener("submit", async functio
<input type="number" name="cores" value="{{.CPUCores}}" min="0"></label>
<button type="submit">{{t "common.save"}}</button>
</div>
<p class="note">{{if .MostNow}}{{if and .Headroom .Headroom.ByPreempting}}{{t "mirror.form.most_now_preempting" .MostNow}}{{else}}{{t "mirror.form.most_now" .MostNow}}{{end}} {{end}}{{t "mirror.form.note"}}</p>
<p class="note">{{if .Split}}{{t "mirror.split.scaling_note"}} {{end}}{{if .MostNow}}{{if and .Headroom .Headroom.ByPreempting}}{{t "mirror.form.most_now_preempting" .MostNow}}{{else}}{{t "mirror.form.most_now" .MostNow}}{{end}} {{end}}{{t "mirror.form.note"}}</p>
</form>
{{template "mirror_model_settings" dict "M" . "Row" $row "CSRF" $csrf}}
{{end}}
</section>
{{end}}
</div>
{{end}}
{{/* The rest of what can be changed of a served model, where its file is:
its context, its cards, its replicas on each profile, stopped or
started, started again, and taken out of the catalog. */}}
{{define "mirror_model_settings"}}{{$m := .M}}{{$row := .Row}}{{$csrf := .CSRF}}
<details class="model-settings" open>
<summary><h4>{{t "mirror.settings.heading"}}</h4></summary>
<form method="post" action="/admin/mirror/context" class="settings-form">
<input type="hidden" name="csrf" value="{{$csrf}}">
<input type="hidden" name="model" value="{{$m.Name}}">
<input type="hidden" name="open" value="{{$row.Open}}">
<div class="field-row">
<label>{{t "serve.context_length"}} <input type="number" name="tokens" value="{{if $m.ContextLength}}{{$m.ContextLength}}{{end}}" min="1"></label>
<label>{{t "serve.max_output"}} <input type="number" name="max_output" value="{{if $m.MaxOutput}}{{$m.MaxOutput}}{{end}}" min="1"></label>
<button type="submit">{{t "mirror.settings.context"}}</button>
</div>
<p class="note">{{t "mirror.settings.context_note"}}</p>
</form>
{{if $m.GPUProfile}}
<form method="post" action="/admin/mirror/gpus" class="settings-form">
<input type="hidden" name="csrf" value="{{$csrf}}">
<input type="hidden" name="model" value="{{$m.Name}}">
<input type="hidden" name="profile" value="{{$m.GPUProfile}}">
<input type="hidden" name="open" value="{{$row.Open}}">
<div class="field-row">
<label>{{t "mirror.settings.gpus"}} <input type="number" name="gpus" value="{{$m.GPUs}}" min="1" max="16"></label>
<button type="submit">{{t "mirror.settings.gpus_apply"}}</button>
</div>
<p class="note">{{t "mirror.settings.gpus_note"}}</p>
</form>
{{end}}
{{if or (gt (len $m.Profiles) 1) $m.CanCPU}}
<form method="post" action="/admin/mirror/split" class="settings-form">
<input type="hidden" name="csrf" value="{{$csrf}}">
<input type="hidden" name="model" value="{{$m.Name}}">
<input type="hidden" name="open" value="{{$row.Open}}">
<p class="sub">{{t "mirror.split.heading"}}</p>
<div class="field-row">
{{range $m.Profiles}}<input type="hidden" name="profile" value="{{.Name}}">
<label>{{if .CPU}}{{t "mirror.split.on_cpu" .Name}}{{else}}{{t "mirror.split.on_gpu" .Name .Class}}{{end}}
<input type="number" name="replicas_{{.Name}}" value="{{.Replicas}}" min="0" max="64"></label>{{end}}
{{if $m.CanCPU}}<label>{{t "mirror.split.add_cpu"}} <input type="number" name="cpu_replicas" value="0" min="0" max="64"></label>
<label>{{t "serve.memory"}} <input type="number" name="cpu_memory" value="{{$m.CPUMemoryGB}}" min="1"></label>
<label>{{t "serve.cpu_cores"}} <input type="number" name="cpu_cores" value="{{if $m.NewCPUCores}}{{$m.NewCPUCores}}{{end}}" min="1"></label>{{end}}
<button type="submit">{{t "common.save"}}</button>
</div>
{{if and $m.CanCPU $m.CPUSpeed}}<p class="note{{if lt $m.CPUSpeed 5.0}} error{{end}}">{{t "mirror.split.cpu_speed" (num $m.CPUSpeed 1) $m.CPUSpeedNode}}</p>{{end}}
<p class="note">{{t "mirror.split.note"}}</p>
</form>
{{end}}
<div class="settings-actions">
<form method="post" action="/admin/mirror/state" class="inline">
<input type="hidden" name="csrf" value="{{$csrf}}">
<input type="hidden" name="model" value="{{$m.Name}}">
<input type="hidden" name="open" value="{{$row.Open}}">
{{if $m.Stopped}}<button type="submit" name="action" value="start">{{t "mirror.settings.start"}}</button>
{{else}}<button type="submit" name="action" value="stop" data-confirm="{{t "mirror.settings.stop_confirm" $m.Name}}">{{t "mirror.settings.stop"}}</button>
<button type="submit" name="action" value="restart" data-confirm="{{t "mirror.settings.restart_confirm" $m.Name}}">{{t "mirror.settings.restart"}}</button>{{end}}
</form>
<form method="post" action="/admin/mirror/unserve" class="inline" data-confirm="{{t "mirror.settings.unserve_confirm" $m.Name}}">
<input type="hidden" name="csrf" value="{{$csrf}}">
<input type="hidden" name="model" value="{{$m.Name}}">
<button type="submit" class="link danger">{{t "mirror.settings.unserve"}}</button>
</form>
<form method="post" action="/admin/mirror/unserve" class="inline" data-confirm="{{t "mirror.settings.purge_confirm" $m.Name}}">
<input type="hidden" name="csrf" value="{{$csrf}}">
<input type="hidden" name="model" value="{{$m.Name}}">
<input type="hidden" name="purge" value="yes">
<button type="submit" class="link danger">{{t "mirror.settings.purge"}}</button>
</form>
</div>
</details>
{{end}}