perf(runtime): read schema graph and subschema caches without locking

.github/workflows/CI.yml

@@ -13,6 +13,8 @@ jobs:
     runs-on: ${{ matrix.os }}
     env:
       OPENAPI_SCHEMA_SUITE: ${{ matrix.version == '1' && 'all' || '' }}
+      # Pkg.test forwards the thread count, so the runtime cache stress test races.
+      JULIA_NUM_THREADS: 4
     strategy:
       fail-fast: false
       matrix:

AGENTS.md

@@ -115,6 +115,9 @@ julia +1.12 --project=. -e 'using Pkg; Pkg.test()'
 julia +1.10 --project=. -e 'using Pkg; Pkg.test()'
 ```
 
+Set `JULIA_NUM_THREADS=4` (as CI does) so the runtime schema-cache stress test
+in `test/runtime.jl` runs threaded; `Pkg.test` forwards the thread count.
+
 The focused files separate concerns:
 
 - `test/normalization.jl`: loading, versions, immutability, generated source.

src/runtime.jl

@@ -56,6 +56,18 @@ function require_contract(version::Integer, generator::AbstractString)
     )
 end
 
+"""
+Compiled schema graphs and subschema views, published as immutable snapshots.
+Readers load a snapshot without locking; a writer holds `Spec.graph_lock`,
+copies the current snapshot, inserts, and publishes the copy. A published
+`Dict` is never mutated.
+"""
+mutable struct SchemaCache
+    @atomic graphs::Dict{Symbol,Any}
+    @atomic subschemas::Dict{Tuple{Symbol,String,String},Any}
+    SchemaCache() = new(Dict{Symbol,Any}(), Dict{Tuple{Symbol,String,String},Any}())
+end
+
 """
 Document-specific data a generated module supplies to the shared runtime:
 schema resources for validation, security schemes, and server defaults.
@@ -71,8 +83,7 @@ struct Spec
     directional_required::Vector{Any}
     default_server::String
     server::Base.RefValue{String}
-    graphs::Dict{Symbol,Any}
-    subschemas::Dict{Tuple{Symbol,String,String},Any}
+    cache::SchemaCache
     graph_lock::ReentrantLock
     # Generated data is required and name-mapped so an omitted keyword or a
     # declaration reorder cannot silently substitute an empty or adjacent
@@ -94,8 +105,7 @@ struct Spec
             directional_required,
             default_server = normalized_server,
             server = Ref(normalized_server),
-            graphs = Dict{Symbol,Any}(),
-            subschemas = Dict{Tuple{Symbol,String,String},Any}(),
+            cache = SchemaCache(),
             graph_lock = ReentrantLock(),
         )
         ordered = map(field -> getproperty(values, field), fieldnames(Spec))
@@ -139,9 +149,12 @@ function _schema_graph(spec::Spec, direction::Symbol = :neutral)
     direction in (:neutral, :input, :output) ||
         throw(ArgumentError("schema direction must be :neutral, :input, or :output"))
     isempty(spec.roots) && return nothing
-    haskey(spec.graphs, direction) && return spec.graphs[direction]
+    cached = get((@atomic spec.cache.graphs), direction, nothing)
+    cached === nothing || return cached
     return lock(spec.graph_lock) do
-        haskey(spec.graphs, direction) && return spec.graphs[direction]
+        # Recheck under the lock so each graph is compiled once.
+        published = get((@atomic spec.cache.graphs), direction, nothing)
+        published === nothing || return published
         documents = Dict{String,Any}(
             entry.id => JSON.parse(entry.json; duplicate_keys = :error) for
             entry in spec.resources
@@ -222,21 +235,32 @@ function _schema_graph(spec::Spec, direction::Symbol = :neutral)
             root_dialects,
             dialect_aliases,
         )
-        spec.graphs[direction] = graph
+        graphs = copy(@atomic spec.cache.graphs)
+        graphs[direction] = graph
+        @atomic spec.cache.graphs = graphs
         return graph
     end
 end
 
 function _schema_at(spec::Spec, descriptor, direction::Symbol = :neutral)
     descriptor === nothing && return nothing
     # Every validation of a generated model asks for the same few views, and building one
-    # parses the resource URI and walks its pointer; keep each view once it exists.
+    # parses the resource URI and walks its pointer; keep each view once it exists. Hits
+    # read the published snapshot without locking.
     key = (direction, String(descriptor.resource), String(descriptor.pointer))
-    cached = lock(() -> get(spec.subschemas, key, nothing), spec.graph_lock)
+    cached = get((@atomic spec.cache.subschemas), key, nothing)
     cached === nothing || return cached
     schema = _build_schema_at(spec, descriptor, direction)
-    lock(() -> (spec.subschemas[key] = schema), spec.graph_lock)
-    return schema
+    return lock(spec.graph_lock) do
+        # A racing miss may have published first; return its view so callers share one.
+        current = @atomic spec.cache.subschemas
+        published = get(current, key, nothing)
+        published === nothing || return published
+        subschemas = copy(current)
+        subschemas[key] = schema
+        @atomic spec.cache.subschemas = subschemas
+        return schema
+    end
 end
 
 function _build_schema_at(spec::Spec, descriptor, direction::Symbol)

test/runtime.jl

@@ -680,4 +680,159 @@ end
     @test Runtime._schema_at(spec, descriptor) === view
     @test Runtime._schema_valid(spec, descriptor, 1)
     @test !Runtime._schema_valid(spec, descriptor, "one")
+    # Each direction keeps its own graph and view, and a new key leaves earlier entries.
+    graphs = Dict(
+        direction => Runtime._schema_graph(spec, direction) for
+        direction in (:neutral, :input, :output)
+    )
+    @test graphs[:neutral] !== graphs[:input] !== graphs[:output]
+    for direction in (:neutral, :input, :output)
+        directional = Runtime._schema_at(spec, descriptor, direction)
+        @test Runtime._schema_at(spec, descriptor, direction) === directional
+        @test Runtime._schema_graph(spec, direction) === graphs[direction]
+    end
+    @test Runtime._schema_at(spec, descriptor) === view
+    @test length(@atomic spec.cache.subschemas) == 3
+end
+
+@testset "runtime schema cache under concurrent decoding" begin
+    Runtime = OpenAPI.Runtime
+    Threads.nthreads() == 1 && @info(
+        "schema cache stress test runs on one thread; set JULIA_NUM_THREADS to exercise races",
+    )
+    tasks_count = 4 * Threads.nthreads()
+    directions = (:neutral, :input, :output)
+    names = ["d$index" for index in 1:8]
+
+    # Every round starts cold, so racing tasks contend for the first graph build and the
+    # first insertion of each view. Graphs carry mutable state, so `===` proves each one
+    # was compiled once; views are values bound to their graph.
+    resource = "https://example.test/concurrent-schema"
+    for round in 1:20
+        spec = Runtime.Spec(;
+            security_schemes = Dict{String,NamedTuple}(),
+            resources = Any[(
+                id = resource,
+                retrieval = resource,
+                media_type = "application/schema+json",
+                json = JSON.json(
+                    OpenAPI.obj(
+                        "\$defs" => OpenAPI.obj(
+                            (name => OpenAPI.obj("type" => "integer", "minimum" => index) for
+                            (index, name) in enumerate(names))...,
+                        ),
+                    ),
+                ),
+            )],
+            roots = Any[(
+                resource = resource,
+                pointer = "",
+                dialect = SchemaEngine.DRAFT202012,
+            )],
+            dialects = Any[],
+            directional_required = Any[],
+            default_server = "",
+        )
+        descriptors = [(resource = resource, pointer = "/\$defs/" * name) for name in names]
+        ready = Threads.Atomic{Int}(0)
+        tasks = map(1:tasks_count) do task_index
+            Threads.@spawn begin
+                Threads.atomic_add!(ready, 1)
+                while ready[] < tasks_count
+                    yield()
+                end
+                order = circshift(collect(eachindex(descriptors)), task_index)
+                seen = Dict{Any,Any}()
+                valid = true
+                for direction in directions, index in order
+                    descriptor = descriptors[index]
+                    seen[(direction, index)] = Runtime._schema_at(spec, descriptor, direction)
+                    valid &= Runtime._schema_valid(spec, descriptor, index; direction)
+                    valid &= !Runtime._schema_valid(spec, descriptor, index - 1; direction)
+                end
+                for direction in directions
+                    seen[direction] = Runtime._schema_graph(spec, direction)
+                end
+                (seen, valid)
+            end
+        end
+        results = fetch.(tasks)
+        @test all(last, results)
+        reference = first(first(results))
+        @test all(keys(seen) == keys(reference) && all(key -> seen[key] === reference[key], keys(seen))
+                  for (seen, _) in results)
+        @test length(@atomic spec.cache.subschemas) == length(directions) * length(names)
+        @test length(@atomic spec.cache.graphs) == length(directions)
+    end
+
+    # Generated model decoding validates through the module's shared spec.
+    document = OpenAPI.obj(
+        "openapi" => "3.2.0",
+        "info" => OpenAPI.obj("title" => "Concurrent", "version" => "1"),
+        "paths" => OpenAPI.obj(
+            "/order" => OpenAPI.obj(
+                "get" => OpenAPI.obj(
+                    "operationId" => "getOrder",
+                    "responses" => OpenAPI.obj(
+                        "200" => OpenAPI.obj(
+                            "description" => "order",
+                            "content" => OpenAPI.obj(
+                                "application/json" => OpenAPI.obj(
+                                    "schema" => OpenAPI.obj(
+                                        "\$ref" => "#/components/schemas/Order",
+                                    ),
+                                ),
+                            ),
+                        ),
+                    ),
+                ),
+            ),
+        ),
+        "components" => OpenAPI.obj(
+            "schemas" => OpenAPI.obj(
+                "Widget" => OpenAPI.obj(
+                    "type" => "object",
+                    "required" => ["id"],
+                    "properties" => OpenAPI.obj(
+                        "id" => OpenAPI.obj("type" => "integer", "minimum" => 1),
+                    ),
+                ),
+                "Order" => OpenAPI.obj(
+                    "type" => "object",
+                    "required" => ["widget", "count"],
+                    "properties" => OpenAPI.obj(
+                        "widget" => OpenAPI.obj("\$ref" => "#/components/schemas/Widget"),
+                        "count" => OpenAPI.obj("type" => "integer", "maximum" => 10),
+                    ),
+                ),
+            ),
+        ),
+    )
+    host = Module(:ConcurrentDecodeClientHost)
+    Base.include_string(
+        host,
+        OpenAPI.client(document; name = "ConcurrentDecodeClient"),
+        "ConcurrentDecodeClient.jl",
+    )
+    C = Base.invokelatest(getfield, host, :ConcurrentDecodeClient)
+    decode(T, value) = Base.invokelatest(getfield(C, :_decode), T, value)
+    Order = Base.invokelatest(getfield, C, :OrderModel)
+    decoded = fetch.(map(1:tasks_count) do task_index
+        Threads.@spawn begin
+            ok = true
+            for count in 1:200
+                value = Dict("widget" => Dict("id" => task_index), "count" => count % 10)
+                ok &= decode(Order, value).widget.id == task_index
+                rejected = try
+                    decode(Order, Dict("widget" => Dict("id" => 0), "count" => count))
+                    false
+                catch error
+                    error isa Runtime.SchemaValidationError
+                end
+                ok &= rejected
+            end
+            ok
+        end
+    end)
+    @test all(decoded)
 end