Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion DESCRIPTION
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
Package: diffuseR
Title: Functional Interface to Diffusion Models in R
Version: 0.2.2.19
Version: 0.2.2.20
Authors@R: c(
person("Troy", "Hernandez", email = "troy@cornball.ai", role = c("aut", "cre"),
comment = c(ORCID = "0009-0005-4248-604X")),
Expand Down
9 changes: 9 additions & 0 deletions NEWS.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,12 @@
# diffuseR 0.2.2.20

* `serve(model = "ltx25")` serves LTX-2.5 on `/v1/videos`. It loads
the `ltx2.5-nf4-full` and `gemma4-nf4` artifacts and places them by
`recommend("ltx25")`, so 8 and 12 GB cards stream transformer blocks
and Gemma 4 layers. The encoder and render pipeline share the
process: measured on the 16 GB card, 32.9 GB host high-water and
26 s for a 512x320x49 clip with a cached prompt (35 s for a new one).

# diffuseR 0.2.2.19

* `recommend("ltx25")` covers LTX-2.5. `recommend("ltx")` (LTX-2.3)
Expand Down
146 changes: 106 additions & 40 deletions R/serve.R
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
#' \code{{prompt, size, seed, steps}} (\code{size} like
#' \code{"1024x1024"}; \code{n > 1} is not supported). Returns
#' \code{{created, data: [{b64_json}]}} with a base64 PNG.
#' \item \code{POST /v1/videos} - \code{model = "ltx"} only, a job that
#' \item \code{POST /v1/videos} - video models only, a job that
#' matches the wan2gp-api container. JSON body \code{{prompt, width,
#' height, num_frames, frame_rate, seed}} returns \code{{id, status}};
#' poll \code{GET /v1/videos/{id}} for the status and download the mp4 at
Expand Down Expand Up @@ -60,8 +60,14 @@
#' state rather than serving on with stranded components.
#'
#' @param port Integer. TCP port. Default 7812 (cornball serve range: whisper 7809, chatterbox 7810, qwen3 TTS 7811).
#' @param model One of "flux2", "zimage", "flux1" (images) or "ltx"
#' (video). SD 2.1/SDXL are not served yet.
#' @param model One of "flux2", "zimage", "flux1" (images), "ltx"
#' (LTX-2.3 video) or "ltx25" (LTX-2.5 video). SD 2.1/SDXL are not
#' served yet. "ltx25" loads the \code{ltx2.5-nf4-full} and
#' \code{gemma4-nf4} artifacts and places them by
#' \code{\link{recommend}("ltx25")}: on 8-12 GB cards the transformer
#' streams blocks and, below 12 GB, Gemma 4 streams layers. The
#' encoder and render pipeline share one process, which needs about
#' 31 GB of host RAM.
#' @param device Character. "cuda" or "cpu".
#' @param token Character or NULL. Shared secret; when set, requests
#' must send \code{Authorization: Bearer <token>}.
Expand All @@ -74,7 +80,8 @@
#' resolution clips top out at 121 frames and longer clips must
#' shrink spatially).
#' @param max_prompts Integer. Bound on the per-prompt connector-embed
#' cache for "ltx" (~9 MB per entry, LRU-evicted). Default 32.
#' cache for the video models (~9 MB per entry, LRU-evicted).
#' Default 32.
#' @param timeout Integer. Per-connection I/O timeout in seconds.
#' @param max_body Integer. Maximum request body bytes. Default 1 MB
#' (bodies are JSON).
Expand All @@ -83,11 +90,12 @@
#' Default 8.
#' @param warmup Logical. Image models: run one small generation at
#' startup so the first request doesn't pay tracing and allocator
#' growth. Ignored for "ltx".
#' growth. Ignored for the video models.
#'
#' @return Does not return normally; runs until interrupted.
#' @export
serve <- function(port = 7812L, model = c("flux2", "zimage", "flux1", "ltx"),
serve <- function(port = 7812L, model = c("flux2", "zimage", "flux1", "ltx",
"ltx25"),
device = "cuda", token = NULL, max_pixels = 1024L ^ 2,
max_frames = 161L, max_steps = 50L,
max_pixel_frames = NULL, max_prompts = 32L, timeout = 300L,
Expand Down Expand Up @@ -115,7 +123,7 @@ serve <- function(port = 7812L, model = c("flux2", "zimage", "flux1", "ltx"),
state$jobs[["__order__"]] <- character(0)
message("Model loaded.")

if (isTRUE(warmup) && model != "ltx") {
if (isTRUE(warmup) && !model %in% c("ltx", "ltx25")) {
message("Warming up...")
tryCatch({
invisible(state$generate("a red square", width = 256L,
Expand Down Expand Up @@ -190,17 +198,9 @@ serve <- function(port = 7812L, model = c("flux2", "zimage", "flux1", "ltx"),
# built, follow the machine-aware recommendation - on a card
# too small for resident nf4, recommend() prescribes streamed
# fp8 (which is also what download_ltx2() produces by default)
valid_artifact <- function(d) {
mp <- file.path(d, "manifest.json")
file.exists(mp) && isTRUE(tryCatch({
m <- jsonlite::fromJSON(mp)
length(m$shards) > 0 &&
all(file.exists(file.path(d, m$shards)))
}, error = function(e) FALSE))
}
dirs <- c(nf4 = file.path(data_dir, "ltx2.3-nf4"),
fp8 = file.path(data_dir, "ltx2.3-fp8"))
avail <- dirs[vapply(dirs, valid_artifact, logical(1))]
avail <- dirs[vapply(dirs, .dserve_valid_artifact, logical(1))]
if (!length(avail)) {
stop("No complete LTX-2.3 artifact under ", data_dir,
"; run download_ltx2() (fp8) or ltx23_quantize_nf4() first.",
Expand Down Expand Up @@ -239,39 +239,22 @@ serve <- function(port = 7812L, model = c("flux2", "zimage", "flux1", "ltx"),
} else {
"cpu"
}
# Bounded LRU of CONNECTOR outputs (~9 MB each): the raw Gemma
# stack (~0.4 GB) is encoded, projected, and dropped per prompt
cache <- new.env(parent = emptyenv())
order <- character(0)
embeds <- function(prompt) {
key <- paste0("p:", prompt)
e <- cache[[key]]
if (!is.null(e)) {
order <<- c(setdiff(order, key), key)
return(e)
}
embeds <- .dserve_embed_cache(function(prompt) {
raw <- encode_with_gemma3(prompt, model = te, tokenizer = tok,
max_sequence_length = 1024L,
device = enc_dev, verbose = FALSE)
conn <- torch::with_no_grad(pipe$connectors(
raw$prompt_embeds$to(device = "cpu",
dtype = torch::torch_bfloat16()),
raw$prompt_attention_mask$to(device = "cpu")))
e <- list(video_text_embedding = conn$video_text_embedding,
audio_text_embedding = conn$audio_text_embedding,
attention_mask = conn$attention_mask)
rm(raw, conn)
cache[[key]] <- e
order <<- c(order, key)
while (length(order) > max_prompts) {
rm(list = order[[1]], envir = cache)
order <<- order[-1]
}
gc(verbose = FALSE)
e
}
list(video_text_embedding = conn$video_text_embedding,
audio_text_embedding = conn$audio_text_embedding,
attention_mask = conn$attention_mask)
}, max_prompts)
list(model = model, video = TRUE, pipe = pipe, embeds = embeds,
device = device)
} else if (model == "ltx25") {
.dserve_load_ltx25(device, max_prompts)
} else {
loader <- switch(model, flux1 = flux_load_pipeline,
flux2 = flux2_load_pipeline,
Expand Down Expand Up @@ -299,6 +282,89 @@ serve <- function(port = 7812L, model = c("flux2", "zimage", "flux1", "ltx"),
}
}

# Bounded LRU of CONNECTOR outputs (~9 MB each): the raw Gemma stack
# (~0.4 GB) is encoded, projected, and dropped per prompt. `encode`
# maps a prompt to the connector-output list.
.dserve_embed_cache <- function(encode, max_prompts) {
cache <- new.env(parent = emptyenv())
order <- character(0)
function(prompt) {
key <- paste0("p:", prompt)
e <- cache[[key]]
if (!is.null(e)) {
order <<- c(setdiff(order, key), key)
return(e)
}
e <- encode(prompt)
cache[[key]] <- e
order <<- c(order, key)
while (length(order) > max_prompts) {
rm(list = order[[1]], envir = cache)
order <<- order[-1]
}
gc(verbose = FALSE)
e
}
}

# A complete quantized artifact: a manifest whose shards all exist (a
# partial or interrupted quantize must not be served)
.dserve_valid_artifact <- function(d) {
mp <- file.path(d, "manifest.json")
file.exists(mp) && isTRUE(tryCatch({
m <- jsonlite::fromJSON(mp)
length(m$shards) > 0 && all(file.exists(file.path(d, m$shards)))
}, error = function(e) FALSE))
}

# LTX-2.5: the nf4-full artifact with Gemma 4, placed by
# recommend("ltx25"). Below 14 GB the transformer keeps resident_gb on
# the GPU and streams the rest; below 12 GB Gemma 4 streams its layers.
# The server holds the encoder and the render pipeline in one process
# (~31 GB of host RAM with pinned staging); the split text phase of
# ltx23_text_embeds() does not apply to a persistent server.
.dserve_load_ltx25 <- function(device, max_prompts) {
data_dir <- tools::R_user_dir("diffuseR", "data")
ckpt <- file.path(data_dir, "ltx2.5-nf4-full")
gem <- file.path(data_dir, "gemma4-nf4")
if (!.dserve_valid_artifact(ckpt)) {
stop("No complete LTX-2.5 artifact at ", ckpt, "; build it with ",
"ltx23_quantize_nf4(ltx25_open_checkpoint(...), output_dir = ",
"\"", ckpt, "\") first.", call. = FALSE)
}
if (!.dserve_valid_artifact(gem)) {
stop("No complete Gemma 4 NF4 artifact at ", gem,
"; build it with gemma4_quantize_nf4() first.", call. = FALSE)
}
gpu <- identical(device, "cuda") && torch::cuda_is_available()
rec <- recommend("ltx25", vram_gb = if (gpu) NULL else 0)
if (isTRUE(rec$split_text)) {
message(sprintf(paste0(
"%.0f GB of host RAM available; serving LTX-2.5 holds Gemma 4 ",
"and the render pipeline together (~31 GB measured)."),
rec$host_ram_gb))
}
pipe <- ltx23_load_pipeline(ckpt, device = device, pin = rec$pin,
resident_gb = rec$resident_gb,
resident_from = rec$resident_from,
verbose = FALSE)
te <- load_gemma4_nf4(gem, device = "cpu", pin = rec$pin,
stream = rec$stream_text, verbose = FALSE)
tok <- gemma4_tokenizer(gem)
enc_dev <- if (gpu) "cuda" else "cpu"
embeds <- .dserve_embed_cache(function(prompt) {
raw <- torch::with_no_grad(encode_with_gemma4(prompt, te, tok,
device = enc_dev,
verbose = FALSE))
torch::with_no_grad(ltx23_text_embeds(prompt, pipe,
prompt_embeds = raw,
device = enc_dev,
verbose = FALSE))
}, max_prompts)
list(model = "ltx25", video = TRUE, pipe = pipe, embeds = embeds,
device = device)
}

.dserve_route <- function(req, state) {
if (isTRUE(req$too_large)) {
return(.dserve_err(413L, "request body too large"))
Expand Down
5 changes: 3 additions & 2 deletions inst/diffuser.service
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@
# Port 7812 sits in the cornball serve range (7809-7829): whisper 7809,
# chatterbox 7810, qwen3 TTS 7811.
#
# Video serving (model = "ltx") answers /v1/videos/generations and takes
# minutes per request - point clients at it with matching timeouts.
# Video serving (model = "ltx" for LTX-2.3, "ltx25" for LTX-2.5) answers
# /v1/videos and takes minutes per request - point clients at it with
# matching timeouts.

[Unit]
Description=diffuseR generation server (R package, base-socket HTTP)
Expand Down
38 changes: 38 additions & 0 deletions inst/tinytest/test_serve.R
Original file line number Diff line number Diff line change
Expand Up @@ -227,3 +227,41 @@ local({
expect_false(file.exists(paths[[1]]))
expect_true(!is.null(diffuseR:::.dserve_jobs_get(vs, ids[[3]])))
})

# --- embed cache: LRU, bounded, encodes each prompt once ----------------------------
calls <- character(0)
emb <- diffuseR:::.dserve_embed_cache(function(p) {
calls <<- c(calls, p)
list(prompt = p)
}, max_prompts = 2L)
expect_equal(emb("a")$prompt, "a")
emb("b")
emb("a") # hit; "a" becomes most recent
expect_equal(calls, c("a", "b"))
emb("c") # evicts "b", the least recent
emb("a")
expect_equal(calls, c("a", "b", "c"))
emb("b")
expect_equal(calls, c("a", "b", "c", "b"))

# --- artifact validity: manifest with every shard present ---------------------------
art <- tempfile("art_")
dir.create(art)
expect_false(diffuseR:::.dserve_valid_artifact(art))
writeLines('{"shards": ["s1.safetensors", "s2.safetensors"]}',
file.path(art, "manifest.json"))
file.create(file.path(art, "s1.safetensors"))
expect_false(diffuseR:::.dserve_valid_artifact(art)) # partial quantize
file.create(file.path(art, "s2.safetensors"))
expect_true(diffuseR:::.dserve_valid_artifact(art))
unlink(art, recursive = TRUE)

# ltx25 is a video model with no warmup, and a missing artifact stops
# startup with the build pointer instead of downloading
expect_true("ltx25" %in% eval(formals(serve)$model))
local({
old <- Sys.getenv("R_USER_DATA_DIR")
on.exit(Sys.setenv(R_USER_DATA_DIR = old), add = TRUE)
Sys.setenv(R_USER_DATA_DIR = tempfile("nodata_"))
expect_error(diffuseR:::.dserve_load("ltx25", "cpu"), "ltx25_open_checkpoint")
})
19 changes: 13 additions & 6 deletions man/serve.Rd
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
\usage{
serve(
port = 7812L,
model = c("flux2", "zimage", "flux1", "ltx"),
model = c("flux2", "zimage", "flux1", "ltx", "ltx25"),
device = "cuda",
token = NULL,
max_pixels = 1024L^2,
Expand All @@ -22,8 +22,14 @@ serve(
\arguments{
\item{port}{Integer. TCP port. Default 7812 (cornball serve range: whisper 7809, chatterbox 7810, qwen3 TTS 7811).}

\item{model}{One of "flux2", "zimage", "flux1" (images) or "ltx"
(video). SD 2.1/SDXL are not served yet.}
\item{model}{One of "flux2", "zimage", "flux1" (images), "ltx"
(LTX-2.3 video) or "ltx25" (LTX-2.5 video). SD 2.1/SDXL are not
served yet. "ltx25" loads the \code{ltx2.5-nf4-full} and
\code{gemma4-nf4} artifacts and places them by
\code{\link{recommend}("ltx25")}: on 8-12 GB cards the transformer
streams blocks and, below 12 GB, Gemma 4 streams layers. The
encoder and render pipeline share one process, which needs about
31 GB of host RAM.}

\item{device}{Character. "cuda" or "cpu".}

Expand All @@ -43,7 +49,8 @@ resolution clips top out at 121 frames and longer clips must
shrink spatially).}

\item{max_prompts}{Integer. Bound on the per-prompt connector-embed
cache for "ltx" (~9 MB per entry, LRU-evicted). Default 32.}
cache for the video models (~9 MB per entry, LRU-evicted).
Default 32.}

\item{timeout}{Integer. Per-connection I/O timeout in seconds.}

Expand All @@ -56,7 +63,7 @@ Default 8.}

\item{warmup}{Logical. Image models: run one small generation at
startup so the first request doesn't pay tracing and allocator
growth. Ignored for "ltx".}
growth. Ignored for the video models.}
}
\value{
Does not return normally; runs until interrupted.
Expand All @@ -77,7 +84,7 @@ single-threaded, so health only answers between requests.
\code{{prompt, size, seed, steps}} (\code{size} like
\code{"1024x1024"}; \code{n > 1} is not supported). Returns
\code{{created, data: [{b64_json}]}} with a base64 PNG.
\item \code{POST /v1/videos} - \code{model = "ltx"} only, a job that
\item \code{POST /v1/videos} - video models only, a job that
matches the wan2gp-api container. JSON body \code{{prompt, width,
height, num_frames, frame_rate, seed}} returns \code{{id, status}};
poll \code{GET /v1/videos/{id}} for the status and download the mp4 at
Expand Down
Loading