Compare commits

..

3 Commits

Author SHA1 Message Date
12c5f9ccca web/backend: Show elapsed time 2025-11-12 15:33:22 -06:00
f53d704e74 web: Show position in line 2025-11-12 15:09:18 -06:00
22f1f9dae5 web/backend: Allow multiple workers 2025-11-12 13:20:45 -06:00
10 changed files with 134 additions and 64 deletions

View File

@@ -6,7 +6,7 @@ WORKDIR /app
COPY Project.toml /app/Backend/
COPY src /app/Backend/src
RUN julia --project=. -e 'using Pkg; Pkg.develop(path="Backend"); Pkg.add("HiGHS"); Pkg.add("JuMP"); Pkg.precompile()'
COPY docker/startup.jl ./
COPY startup.jl ./
# Set timezone to Chicago
ENV TZ=America/Chicago
@@ -17,4 +17,4 @@ ENV UCJL_HOST="0.0.0.0"
ENV UCJL_PORT="9000"
# Run the server
CMD ["julia", "--threads", "1", "--procs", "1", "--project=.", "startup.jl"]
CMD ["julia", "--threads", "1", "--procs", "4", "--project=.", "startup.jl"]

View File

@@ -13,3 +13,11 @@ docker-run:
--memory 16g \
--cpus 4 \
ucjl-backend
test:
clear; julia --threads 1 --procs 1 --project=test -e "using BackendT; runtests()"
run:
julia --procs 1 --project=. startup.jl
.PHONY: test

View File

@@ -8,7 +8,9 @@ CodecZlib = "944b1d66-785c-5afd-91f1-9de20f533193"
Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"
HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3"
HiGHS = "87dc4568-4c63-4d18-b0c0-bb2238e4078b"
JSON = "682c06a0-de6a-54ab-a142-c8b1cf79cde6"
JuMP = "4076af6c-e467-56ae-b986-b466b2749572"
Logging = "56ddb016-857b-54e1-b83d-db4d58db5568"
Printf = "de0858da-6303-5e67-8744-51eddeeeb8d7"
Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
@@ -19,7 +21,9 @@ CodecZlib = "0.7.8"
Dates = "1.11.0"
Distributed = "1.11.0"
HTTP = "1.10.19"
HiGHS = "1.20.1"
JSON = "0.21.4"
JuMP = "1.29.2"
Logging = "1.11.0"
Printf = "1.11.0"
Random = "1.11.0"

View File

@@ -8,73 +8,115 @@ import Base: put!
Base.@kwdef mutable struct JobProcessor
pending = RemoteChannel(() -> Channel{String}(Inf))
processing = RemoteChannel(() -> Channel{String}(Inf))
completed = RemoteChannel(() -> Channel{String}(Inf))
shutdown = RemoteChannel(() -> Channel{Bool}(1))
worker_pid = nothing
monitor_task = nothing
worker_pids = []
worker_tasks = []
work_fn = nothing
master_task = nothing
job_status = Dict()
job_position = Dict()
pending_queue = []
end
function update_positions!(processor::JobProcessor)
for (i, job_id) in enumerate(processor.pending_queue)
processor.job_position[job_id] = i
end
end
function Base.put!(processor::JobProcessor, job_id::String)
return put!(processor.pending, job_id)
put!(processor.pending, job_id)
processor.job_status[job_id] = "pending"
push!(processor.pending_queue, job_id)
update_positions!(processor)
end
function isbusy(processor::JobProcessor)
return isready(processor.pending) || isready(processor.processing)
function master_loop(processor)
@info "Starting master loop"
while true
# Check for shutdown signal
if isready(processor.shutdown)
break
end
# Check for processing jobs
while isready(processor.processing)
job_id = take!(processor.processing)
processor.job_status[job_id] = "processing"
filter!(x -> x != job_id, processor.pending_queue)
delete!(processor.job_position, job_id)
update_positions!(processor)
end
# Check for completed jobs
while isready(processor.completed)
job_id = take!(processor.completed)
delete!(processor.job_status, job_id)
delete!(processor.job_position, job_id)
end
sleep(0.1)
end
end
function worker_loop(pending, processing, shutdown, work_fn)
function worker_loop(pending, processing, completed, shutdown, work_fn)
@info "Starting worker loop"
while true
# Check for shutdown signal
if isready(shutdown)
@info "Shutdown signal received"
break
end
# Wait for a job with timeout
if !isready(pending)
sleep(0.1)
continue
end
# Move job from pending to processing queue
# Check for pending tasks
if isready(pending)
job_id = take!(pending)
put!(processing, job_id)
@info "Job started: $job_id"
# Run work function
try
work_fn(job_id)
work_time = @elapsed work_fn(job_id)
@info "Job finished: $job_id ($work_time s)"
put!(completed, job_id)
catch e
@error "Job failed: job $job_id"
end
end
# Remove job from processing queue
take!(processing)
@info "Job finished: $job_id"
sleep(0.1)
end
end
function start(processor::JobProcessor)
processor.monitor_task = @spawn begin
# Get list of available worker processes
worker_pids = workers()
@info "Starting job processor with $(length(worker_pids)) worker(s)"
# Start a worker loop on each worker process
for pid in worker_pids
task = @spawnat pid begin
worker_loop(
processor.pending,
processor.processing,
processor.completed,
processor.shutdown,
processor.work_fn,
)
end
push!(processor.worker_pids, pid)
push!(processor.worker_tasks, task)
end
# Start master loop (after spawning workers to avoid serialization issues)
processor.master_task = @async master_loop(processor)
return
end
function stop(processor::JobProcessor)
put!(processor.shutdown, true)
if processor.monitor_task !== nothing
try
wait(processor.monitor_task)
catch e
@warn "Error waiting for worker task" exception=e
end
wait(processor.master_task)
for (i, task) in enumerate(processor.worker_tasks)
wait(task)
end
return
end

View File

@@ -31,6 +31,8 @@ end
function setup_logger()
global_logger(TimeLogger())
@spawn global_logger(TimeLogger())
for pid in workers()
@spawnat pid global_logger(TimeLogger())
end
return
end

View File

@@ -53,7 +53,7 @@ function submit(req, processor::JobProcessor)
return HTTP.Response(200, RESPONSE_HEADERS, response_body)
end
function jobs_view(req)
function jobs_view(req, processor)
# Extract job_id from URL path /api/jobs/{job_id}/view
path_parts = split(req.target, '/')
job_id = path_parts[4]
@@ -74,9 +74,19 @@ function jobs_view(req)
output_path = joinpath(job_dir, "output.json")
output_content = isfile(output_path) ? read(output_path, String) : nothing
# Create response JSON
response_data = Dict("log" => log_content, "solution" => output_content)
# Read job status
job_status = "unknown"
if output_content !== nothing
job_status = "completed"
elseif haskey(processor.job_status, job_id)
job_status = processor.job_status[job_id]
end
# Read job position (0 if already processed or not found)
job_position = get(processor.job_position, job_id, 0)
# Create response JSON
response_data = Dict("log" => log_content, "solution" => output_content, "status" => job_status, "position" => job_position)
response_body = JSON.json(response_data)
return HTTP.Response(200, RESPONSE_HEADERS, response_body)
end
@@ -134,7 +144,7 @@ function start_server(host, port; optimizer)
HTTP.register!(router, "POST", "/api/submit", req -> submit(req, processor))
# Register job/*/view endpoint
HTTP.register!(router, "GET", "/api/jobs/*/view", jobs_view)
HTTP.register!(router, "GET", "/api/jobs/*/view", req -> jobs_view(req, processor))
server = HTTP.serve!(router, host, port; verbose = false)
return ServerHandle(server, processor)

View File

@@ -1,10 +1,7 @@
#!/usr/bin/env julia
# UnitCommitment.jl: Optimization Package for Security-Constrained Unit Commitment
# Copyright (C) 2025, UChicago Argonne, LLC. All rights reserved.
# Released under the modified BSD license. See COPYING.md for more details.
# Load required packages
using HiGHS
using JuMP
using Backend
@@ -15,10 +12,16 @@ const UCJL_PORT = parse(Int, get(ENV, "PORT", "9000"))
println("Starting UnitCommitment Backend Server...")
println("Host: $UCJL_HOST")
println("Port: $UCJL_PORT")
println("Press Ctrl+C to stop the server")
Backend.setup_logger()
server = Backend.start_server(UCJL_HOST, UCJL_PORT; optimizer = optimizer_with_attributes(HiGHS.Optimizer, "mip_rel_gap" => 0.001))
server = Backend.start_server(
UCJL_HOST,
UCJL_PORT;
optimizer = optimizer_with_attributes(
HiGHS.Optimizer, "mip_rel_gap" => 0.001,
"threads" => 1,
)
)
try
wait()
catch e

View File

@@ -26,10 +26,7 @@ function server_test_usage()
@test length(job_id) == 16
# Wait for jobs to finish
sleep(5)
while isbusy(server.processor)
sleep(0.1)
end
sleep(10)
# Verify the compressed file was saved correctly
job_dir = joinpath(Backend.basedir, "jobs", job_id)
@@ -52,6 +49,7 @@ function server_test_usage()
@test haskey(view_data, "solution")
@test view_data["log"] !== nothing
@test view_data["solution"] !== nothing
@test view_data["status"] == "completed"
# Clean up
rm(job_dir, recursive = true)

View File

@@ -8,11 +8,11 @@ ARG REACT_APP_BACKEND_URL
ENV REACT_APP_BACKEND_URL=$REACT_APP_BACKEND_URL
RUN npm run build
# Production Stage
FROM node:18-alpine AS production
WORKDIR /app
COPY --from=build /app/build ./build
COPY server.js ./
RUN npm install --production express
EXPOSE 3000
CMD ["node", "server.js"]
# Production Stage
FROM node:18-alpine AS production
WORKDIR /app
COPY --from=build /app/build ./build
COPY server.js ./
RUN npm install --production express
EXPOSE 3000
CMD ["node", "server.js"]

View File

@@ -15,6 +15,7 @@ import formStyles from "../Common/Forms/Form.module.css";
interface JobData {
log: string;
solution: any;
position: number;
}
const Jobs = () => {
@@ -81,7 +82,9 @@ const Jobs = () => {
<SectionHeader title="Optimization log"></SectionHeader>
<div className={formStyles.FormWrapper}>
<div className={styles.SolverLog} ref={logRef}>
{jobData ? jobData.log : "Loading..."}
{jobData
? jobData.log || `Waiting for ${jobData.position} other optimization job(s) to finish...`
: "Loading..."}
</div>
</div>
</div>