diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 019bde0..ebb27dc 100644 --- a/.github/workflows/CI.yml +++ b/.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: diff --git a/AGENTS.md b/AGENTS.md index 8a029e7..e5571f5 100644 --- a/AGENTS.md +++ b/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. diff --git a/src/runtime.jl b/src/runtime.jl index c4dbdae..42e4cdb 100644 --- a/src/runtime.jl +++ b/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,7 +235,9 @@ 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 @@ -230,13 +245,22 @@ 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) diff --git a/test/runtime.jl b/test/runtime.jl index e900116..3dfa550 100644 --- a/test/runtime.jl +++ b/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