From 89b561cfb6f5f80f9a04c829f84b98429a2e03f5 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Sun, 16 Aug 2026 23:34:39 +0200 Subject: [PATCH 01/10] fix to allow timing collection to be off --- src/io.jl | 3 +++ src/prob_cp.jl | 13 ++++++++----- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/src/io.jl b/src/io.jl index 526c47a..f18084d 100644 --- a/src/io.jl +++ b/src/io.jl @@ -926,6 +926,9 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) db.output.timing_sp[(scenix, subix)][stepnr, :] .= fetch(f) @spawnat core reset_maintiming_sp(scenix, subix) end + + db.output.timing_cp[stepnr, :] .= db.cp.div[MainTiming] + fill!(db.cp.div[MainTiming], 0.0) end if has_result_scenarios(settings) diff --git a/src/prob_cp.jl b/src/prob_cp.jl index 94ee185..319a8bb 100644 --- a/src/prob_cp.jl +++ b/src/prob_cp.jl @@ -55,7 +55,10 @@ function create_cp() probmethod = parse_methods(settings["problems"]["clearing"]["solver"]) prob = TuLiPa.buildprob(probmethod, modelobjects) - db.cp = ClearingProblem(prob, Dict{String, Float64}(), Dict()) + div = Dict() + div[MainTiming] = zeros(3) + + db.cp = ClearingProblem(prob, Dict{String, Float64}(), div) return end @@ -64,15 +67,15 @@ function solve_cp(t, stepnr, skipmed) db = get_local_db() if db.core_main == db.core - timing = db.output.timing_cp - timing[stepnr, 3] = @elapsed begin + maintiming = db.cp.div[MainTiming] + maintiming[3] = @elapsed begin update_startstates_cp(db.cp.prob, db.startstates, stepnr, t) update_cuts(db.dist_mp, db.cp.prob, skipmed) update_nonstoragestates_cp(db.dist_ppp, db.cp.prob) update_statedependent_cp(stepnr, t) - timing[stepnr, 1] = @elapsed TuLiPa.update!(db.cp.prob, t) + maintiming[1] = @elapsed TuLiPa.update!(db.cp.prob, t) set_minstoragevalue!(db.cp.prob, minstoragevaluerule) - timing[stepnr, 2] = @elapsed TuLiPa.solve!(db.cp.prob) + maintiming[2] = @elapsed TuLiPa.solve!(db.cp.prob) get_startstates!(db.cp.prob, db.input.dataset["detailedrescopl"], db.input.dataset["enekvglobaldict"], db.cp.endstates) end end From 10580d6bf7e73aa6d443cd7b3e34e2b13bcb0824 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 00:16:41 +0200 Subject: [PATCH 02/10] timing collection one collection per core --- src/io.jl | 61 +++++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 46 insertions(+), 15 deletions(-) diff --git a/src/io.jl b/src/io.jl index f18084d..2d61212 100644 --- a/src/io.jl +++ b/src/io.jl @@ -903,28 +903,29 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) end if has_result_times(settings) - for (scenix, core) in db.dist_ppp - f = @spawnat core get_maintiming_ppp(scenix) - db.output.timing_ppp[scenix][stepnr, :, :] .= fetch(f) - @spawnat core reset_maintiming_ppp(scenix) + futures = Pair{CoreId, Any}[] + @sync for core in get_cores(db) + if core != db.core + f = @spawnat core collect_and_reset_timings_local() + push!(futures, core => f) + end + end + all_timings = Dict{CoreId, Any}(db.core => collect_and_reset_timings_local()) + for (core, f) in futures + all_timings[core] = fetch(f) end + for (scenix, core) in db.dist_ppp + db.output.timing_ppp[scenix][stepnr, :, :] .= all_timings[core][1][scenix] + end for (scenix, subix, core) in db.dist_evp - f = @spawnat core get_maintiming_evp(scenix, subix) - db.output.timing_evp[(scenix, subix)][stepnr, :] .= fetch(f) - @spawnat core reset_maintiming_evp(scenix, subix) + db.output.timing_evp[(scenix, subix)][stepnr, :] .= all_timings[core][2][(scenix, subix)] end - for (subix, core) in db.dist_mp - f = @spawnat core get_maintiming_mp(subix) - db.output.timing_mp[subix][stepnr, :] .= fetch(f) - @spawnat core reset_maintiming_mp(subix) + db.output.timing_mp[subix][stepnr, :] .= all_timings[core][3][subix] end - for (scenix, subix, core) in db.dist_sp - f = @spawnat core get_maintiming_sp(scenix, subix) - db.output.timing_sp[(scenix, subix)][stepnr, :] .= fetch(f) - @spawnat core reset_maintiming_sp(scenix, subix) + db.output.timing_sp[(scenix, subix)][stepnr, :] .= all_timings[core][4][(scenix, subix)] end db.output.timing_cp[stepnr, :] .= db.cp.div[MainTiming] @@ -1230,6 +1231,36 @@ reset_maintiming_evp(scenix, subix) = fill!(get_local_db().evp[(scenix, subix)]. reset_maintiming_mp(subix) = fill!(get_local_db().mp[subix].div[MainTiming], 0.0) reset_maintiming_sp(scenix, subix) = fill!(get_local_db().sp[(scenix, subix)].div[MainTiming], 0.0) +function collect_and_reset_timings_local() + db = get_local_db() + + ppp_timings = Dict{Int, Matrix{Float64}}() + for (scenix, ppp) in db.ppp + ppp_timings[scenix] = copy(ppp.div[MainTiming]) + fill!(ppp.div[MainTiming], 0.0) + end + + evp_timings = Dict{Tuple{Int,Int}, Vector{Float64}}() + for ((scenix, subix), evp) in db.evp + evp_timings[(scenix, subix)] = copy(evp.div[MainTiming]) + fill!(evp.div[MainTiming], 0.0) + end + + mp_timings = Dict{Int, Vector{Float64}}() + for (subix, mp) in db.mp + mp_timings[subix] = copy(mp.div[MainTiming]) + fill!(mp.div[MainTiming], 0.0) + end + + sp_timings = Dict{Tuple{Int,Int}, Vector{Float64}}() + for ((scenix, subix), sp) in db.sp + sp_timings[(scenix, subix)] = copy(sp.div[MainTiming]) + fill!(sp.div[MainTiming], 0.0) + end + + return (ppp_timings, evp_timings, mp_timings, sp_timings) +end + get_storagevalues_stoch(subix) = get_local_db().mp[subix].div[StorageValues] function get_output_final(steplength, skipmax) From 15a8272fc412b2f80d42dba3b36a25ce4e3c6597 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 09:39:35 +0200 Subject: [PATCH 03/10] get subsystem info from local database --- src/io.jl | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/io.jl b/src/io.jl index 2d61212..fbef44d 100644 --- a/src/io.jl +++ b/src/io.jl @@ -1504,8 +1504,7 @@ function get_output_timing_local(data, steplength, skipmax) df_evp = DataFrame([name => [] for name in ["scenix", "subix", "update", "solve", "total", "core", "skipmed"]]) for (scenix, subix, core) in db.dist_evp values = dropdims(mean(db.output.timing_evp[(scenix, subix)], dims=1), dims=1) - f = @spawnat core get_skipmed_impact(subix) - push!(df_evp, [scenix, subix, values[1], values[2], values[3], core, fetch(f)]) + push!(df_evp, [scenix, subix, values[1], values[2], values[3], core, get_skipmed_impact(db.subsystems[subix])]) end df_evp[!, :other] = df_evp[!, :total] - df_evp[!, :solve] - df_evp[!, :update] df_evp[df_evp.skipmed.==true, [:update, :solve, :total]] .= df_evp[df_evp.skipmed.==true, [:update, :solve, :total]] .* skipfactor @@ -1531,8 +1530,7 @@ function get_output_timing_local(data, steplength, skipmax) df_mp = DataFrame([name => [] for name in ["subix", "mp_u", "mp_s", "mp_fin", "mp_o", "bend_it", "core", "skipmed"]]) for (subix, core) in db.dist_mp values = dropdims(mean(db.output.timing_mp[(subix)], dims=1), dims=1) - f = @spawnat core get_skipmed_impact(subix) - push!(df_mp, [subix, values[1], values[2], values[3], values[4], values[5], core, fetch(f)]) + push!(df_mp, [subix, values[1], values[2], values[3], values[4], values[5], core, get_skipmed_impact(db.subsystems[subix])]) end df_mp[!, :mp_tot] = df_mp[!, :mp_s] + df_mp[!, :mp_u] + df_mp[!, :mp_fin] + df_mp[!, :mp_o] df_mp[df_mp.skipmed.==true, [:mp_u, :mp_s, :mp_fin, :mp_o, :mp_tot, :bend_it]] .= df_mp[df_mp.skipmed.==true, [:mp_u, :mp_s, :mp_fin, :mp_o, :mp_tot, :bend_it]] .* skipfactor @@ -1552,8 +1550,7 @@ function get_output_timing_local(data, steplength, skipmax) df_sp = DataFrame([name => [] for name in ["scenix", "subix", "update", "solve", "other", "core", "skipmed"]]) for (scenix, subix, core) in db.dist_sp values = dropdims(mean(db.output.timing_sp[(scenix, subix)], dims=1), dims=1) - f = @spawnat core get_skipmed_impact(subix) - push!(df_sp, [scenix, subix, values[1], values[2], values[3], core, fetch(f)]) + push!(df_sp, [scenix, subix, values[1], values[2], values[3], core, get_skipmed_impact(db.subsystems[subix])]) end df_sp[!, :total] = df_sp[!, :solve] + df_sp[!, :update] + df_sp[!, :other] df_sp[df_sp.skipmed.==true, [:update, :solve, :other, :total]] .= df_sp[df_sp.skipmed.==true, [:update, :solve, :other, :total]] .* skipfactor From 3ff79ccba25112add40c3298560c80564dafbff7 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 10:30:32 +0200 Subject: [PATCH 04/10] collect storagevalues in batches per core --- src/io.jl | 101 +++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 88 insertions(+), 13 deletions(-) diff --git a/src/io.jl b/src/io.jl index fbef44d..d25bf3e 100644 --- a/src/io.jl +++ b/src/io.jl @@ -942,21 +942,40 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) end if has_result_storagevalues(settings) + need_sv = has_result_storagevalues_all_problems(settings) || !haskey(settings["problems"], "clearing") + need_cutsids = haskey(settings["problems"], "clearing") + + # Phase 1: Fetch storagevalues and cutsids, one call per unique mp core + sv_futures = Pair{CoreId, Any}[] + @sync for core in unique(last.(db.dist_mp)) + f = @spawnat core get_mp_watervalues_and_cutsids_local(need_sv, need_cutsids) + push!(sv_futures, core => f) + end + all_sv = Dict{CoreId, Any}(core => fetch(f) for (core, f) in sv_futures) + + # Phase 2: Process locally, resolve cuts, build enddual requests grouped by core + enddual_sp_reqs = Dict{CoreId, Vector{Tuple}}() + enddual_evp_reqs = Dict{CoreId, Vector{Tuple}}() + sp_output_map = Dict{Tuple, Tuple}() + evp_output_map = Dict{Tuple, Tuple}() + for (subix, core) in db.dist_mp if has_headlosscost(settings["problems"]["stochastic"]["master"]) dim = get_numscen_stoch(db.input) * 2 + 2 # scenarios + master operative + master operative after headlosscost adjustment else dim = get_numscen_stoch(db.input) * 2 + 1 # scenarios + master operative end - if has_result_storagevalues_all_problems(settings) || !haskey(settings["problems"], "clearing") - f = @spawnat core get_storagevalues_stoch(subix) - storagevalues_stoch = fetch(f) - dim = (size(storagevalues_stoch, 1)) + + (sv_dict, cutsid_dict) = all_sv[core] + + if (has_result_storagevalues_all_problems(settings) || !haskey(settings["problems"], "clearing")) && haskey(sv_dict, subix) + storagevalues_stoch = sv_dict[subix] + dim = size(storagevalues_stoch, 1) db.output.storagevalues[subix][stepnr, 1:dim, :] .= storagevalues_stoch end if haskey(settings["problems"], "clearing") - cutid = fetch(@spawnat core get_cutsid(subix)) + cutid = cutsid_dict[subix] cuts = get_obj_from_id(TuLiPa.getobjects(db.cp.prob), cutid) for (j, statevar) in enumerate(cuts.statevars) # master / operative water values after headlosscost obj = get_obj_from_id(TuLiPa.getobjects(db.cp.prob), first(TuLiPa.getvarout(statevar))) # TODO: OK to assume objid = varoutid? @@ -965,18 +984,19 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) db.output.storagevalues[subix][stepnr, dim+1, j] = TuLiPa.getcondual(db.cp.prob, TuLiPa.getid(balance), TuLiPa.getnumperiods(TuLiPa.gethorizon(balance))) if has_result_storagevalues_all_problems(settings) + objid = first(TuLiPa.getvarout(statevar)) if haskey(settings["problems"], "stochastic") for scenix in 1:get_numscen_stoch(db.input) - core_stoch = get_core_sp(db.dist_sp, scenix, subix) - f = @spawnat core_stoch get_enddual_stoch(scenix, subix, first(TuLiPa.getvarout(statevar))) - db.output.storagevalues[subix][stepnr, dim+3+scenix, j] = fetch(f) + core_sp = get_core_sp(db.dist_sp, scenix, subix) + push!(get!(() -> Tuple[], enddual_sp_reqs, core_sp), (scenix, subix, objid)) + sp_output_map[(scenix, subix, objid)] = (subix, dim+3+scenix, j) end end if haskey(settings["problems"], "endvalue") && is_subsystem_evp(db.subsystems[subix]) for scenix in 1:get_numscen_stoch(db.input) core_evp = get_core_evp(db.dist_evp, scenix, subix) - f = @spawnat core_evp get_enddual_evp(scenix, subix, first(TuLiPa.getvarout(statevar))) - db.output.storagevalues[subix][stepnr, dim+3+get_numscen_stoch(db.input)+scenix, j] = fetch(f) + push!(get!(() -> Tuple[], enddual_evp_reqs, core_evp), (scenix, subix, objid)) + evp_output_map[(scenix, subix, objid)] = (subix, dim+3+get_numscen_stoch(db.input)+scenix, j) end end end @@ -985,13 +1005,38 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) statevars = db.mp[subix].cuts.statevars for scenix in 1:get_numscen_stoch(db.input) for (j, statevar) in enumerate(statevars) - core_stoch = get_core_sp(db.dist_sp, scenix, subix) - f = @spawnat core_stoch get_enddual_stoch(scenix, subix, first(TuLiPa.getvarout(statevar))) - db.output.storagevalues[subix][stepnr, dim+scenix, j] = fetch(f) + objid = first(TuLiPa.getvarout(statevar)) + core_sp = get_core_sp(db.dist_sp, scenix, subix) + push!(get!(() -> Tuple[], enddual_sp_reqs, core_sp), (scenix, subix, objid)) + sp_output_map[(scenix, subix, objid)] = (subix, dim+scenix, j) end end end end + + # Phase 3: Fetch all endduals, one call per unique core + all_cores_enddual = union(keys(enddual_sp_reqs), keys(enddual_evp_reqs)) + if !isempty(all_cores_enddual) + enddual_futures = Pair{CoreId, Any}[] + @sync for core in all_cores_enddual + sp_reqs = get(enddual_sp_reqs, core, Tuple[]) + evp_reqs = get(enddual_evp_reqs, core, Tuple[]) + f = @spawnat core get_endperiod_duals_local(sp_reqs, evp_reqs) + push!(enddual_futures, core => f) + end + + for (core, f) in enddual_futures + (sp_results, evp_results) = fetch(f) + for (key, val) in sp_results + (subix_out, row, j) = sp_output_map[key] + db.output.storagevalues[subix_out][stepnr, row, j] = val + end + for (key, val) in evp_results + (subix_out, row, j) = evp_output_map[key] + db.output.storagevalues[subix_out][stepnr, row, j] = val + end + end + end end if haskey(settings["results"], "mainresults") @@ -1263,6 +1308,36 @@ end get_storagevalues_stoch(subix) = get_local_db().mp[subix].div[StorageValues] +function get_mp_watervalues_and_cutsids_local(need_sv::Bool, need_cutsids::Bool) + db = get_local_db() + sv = Dict{Int, Any}() + cutsids = Dict{Int, Any}() + for (subix, mp) in db.mp + need_sv && (sv[subix] = copy(mp.div[StorageValues])) + need_cutsids && (cutsids[subix] = mp.cuts.id) + end + return (sv, cutsids) +end + +function get_endperiod_duals_local(sp_requests, evp_requests) + db = get_local_db() + sp_results = Dict{Tuple, Float64}() + for (scenix, subix, objid) in sp_requests + sp = db.sp[(scenix, subix)] + obj = get_obj_from_id(TuLiPa.getobjects(sp.prob), objid) + balance = TuLiPa.getbalance(obj) + sp_results[(scenix, subix, objid)] = TuLiPa.getcondual(sp.prob, TuLiPa.getid(balance), TuLiPa.getnumperiods(TuLiPa.gethorizon(balance))) + end + evp_results = Dict{Tuple, Float64}() + for (scenix, subix, objid) in evp_requests + evp = db.evp[(scenix, subix)] + obj = get_obj_from_id(TuLiPa.getobjects(evp.prob), objid) + balance = TuLiPa.getbalance(obj) + evp_results[(scenix, subix, objid)] = TuLiPa.getcondual(evp.prob, TuLiPa.getid(balance), TuLiPa.getnumperiods(TuLiPa.gethorizon(balance))) + end + return (sp_results, evp_results) +end + function get_output_final(steplength, skipmax) output = get_output_main() From 1c96eeb29c35efd2cc8dd8ec4b799337f2cabd73 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 10:45:37 +0200 Subject: [PATCH 05/10] remove unused functions --- src/io.jl | 30 ------------------------------ 1 file changed, 30 deletions(-) diff --git a/src/io.jl b/src/io.jl index d25bf3e..eca530a 100644 --- a/src/io.jl +++ b/src/io.jl @@ -1188,24 +1188,6 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) return end -function get_enddual_stoch(scenix, subix, objid) - db = get_local_db() - sp = db.sp[(scenix, subix)] - - obj = get_obj_from_id(TuLiPa.getobjects(sp.prob), objid) # TODO: OK to assume objid = varoutid? - balance = TuLiPa.getbalance(obj) - return TuLiPa.getcondual(sp.prob, TuLiPa.getid(balance), TuLiPa.getnumperiods(TuLiPa.gethorizon(balance))) -end - -function get_enddual_evp(scenix, subix, objid) - db = get_local_db() - evp = db.evp[(scenix, subix)] - - obj = get_obj_from_id(TuLiPa.getobjects(evp.prob), objid) # TODO: OK to assume objid = varoutid? - balance = TuLiPa.getbalance(obj) - return TuLiPa.getcondual(evp.prob, TuLiPa.getid(balance), TuLiPa.getnumperiods(TuLiPa.gethorizon(balance))) -end - function reset_ppp_prices(scenix) db = get_local_db() ppp = db.ppp[scenix] @@ -1266,16 +1248,6 @@ end get_output_from_input(input::DefaultJulESInput) = DefaultJulESOutput(input) -get_maintiming_ppp(scenix) = get_local_db().ppp[scenix].div[MainTiming] -get_maintiming_evp(scenix, subix) = get_local_db().evp[(scenix, subix)].div[MainTiming] -get_maintiming_mp(subix) = get_local_db().mp[subix].div[MainTiming] -get_maintiming_sp(scenix, subix) = get_local_db().sp[(scenix, subix)].div[MainTiming] - -reset_maintiming_ppp(scenix) = fill!(get_local_db().ppp[scenix].div[MainTiming], 0.0) -reset_maintiming_evp(scenix, subix) = fill!(get_local_db().evp[(scenix, subix)].div[MainTiming], 0.0) -reset_maintiming_mp(subix) = fill!(get_local_db().mp[subix].div[MainTiming], 0.0) -reset_maintiming_sp(scenix, subix) = fill!(get_local_db().sp[(scenix, subix)].div[MainTiming], 0.0) - function collect_and_reset_timings_local() db = get_local_db() @@ -1306,8 +1278,6 @@ function collect_and_reset_timings_local() return (ppp_timings, evp_timings, mp_timings, sp_timings) end -get_storagevalues_stoch(subix) = get_local_db().mp[subix].div[StorageValues] - function get_mp_watervalues_and_cutsids_local(need_sv::Bool, need_cutsids::Bool) db = get_local_db() sv = Dict{Int, Any}() From 4bb0e716ab747b2c135490a10b25a5fc037f40b5 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 10:52:24 +0200 Subject: [PATCH 06/10] batch update cuts per core --- src/prob_cp.jl | 96 +++++++++++++++++++++++++++----------------------- 1 file changed, 52 insertions(+), 44 deletions(-) diff --git a/src/prob_cp.jl b/src/prob_cp.jl index 319a8bb..736a1c4 100644 --- a/src/prob_cp.jl +++ b/src/prob_cp.jl @@ -125,19 +125,13 @@ function update_statedependent_cp(stepnr, t) # Headlosscosts if has_headlosscost(settings["problems"]["clearing"]) - for (_subix, _core) in db.dist_mp - future = @spawnat _core get_headlosscost_data_from_mp(_subix, t) - - ret = fetch(future) - if ret isa RemoteException - throw(ret) - end - headlosscost_data = ret + all_data = fetch_per_core(db.dist_mp) do core, subixs + @spawnat core get_all_headlosscost_data_local(subixs, t) + end - for (resid, headlosscost, T_mp) in headlosscost_data - obj = find_obj_by_id(TuLiPa.getobjects(db.cp.prob), resid) - T = TuLiPa.getnumperiods(TuLiPa.gethorizon(obj)) - + for headlosscost_data in Iterators.flatten(all_data) + for (resid, headlosscost, _) in headlosscost_data + T = TuLiPa.getnumperiods(TuLiPa.gethorizon(find_obj_by_id(TuLiPa.getobjects(db.cp.prob), resid))) TuLiPa.setobjcoeff!(db.cp.prob, resid, T, headlosscost) end end @@ -145,12 +139,14 @@ function update_statedependent_cp(stepnr, t) return end -function get_headlosscost_data_from_mp(subix, t) # TODO: get method from config +function get_all_headlosscost_data_local(subixs, t) db = get_local_db() - - mp = db.mp[subix] - - return TuLiPa.get_headlosscost_data(TuLiPa.ReservoirCurveSlopeMethod(), mp.prob, t) + results = Vector{Any}() + for subix in subixs + mp = db.mp[subix] + push!(results, TuLiPa.get_headlosscost_data(TuLiPa.ReservoirCurveSlopeMethod(), mp.prob, t)) + end + return results end function update_nonstoragestates_cp(dist_ppp, prob_cp) @@ -179,45 +175,57 @@ function get_nonstoragestates_short(scenix) end function update_cuts(dist_mp, prob_cp, skipmed) - for (_subix, _core) in dist_mp - if skipmed_check(_subix, skipmed) - future = @spawnat _core get_cutsdata(_subix) - - ret = fetch(future) - if ret isa RemoteException - throw(ret) - end - - (cutid, constants, slopes) = ret - - cuts_cp = find_obj_by_id(TuLiPa.getobjects(prob_cp), cutid) - cuts_cp.constants = constants - cuts_cp.slopes = slopes + all_cutsdata = fetch_per_core(dist_mp, skipmed) do core, subixs + @spawnat core get_all_cutsdata_local(subixs) + end - TuLiPa.updatecuts!(prob_cp, cuts_cp) - end + for (cutid, constants, slopes) in Iterators.flatten(all_cutsdata) + cuts_cp = find_obj_by_id(TuLiPa.getobjects(prob_cp), cutid) + cuts_cp.constants = constants + cuts_cp.slopes = slopes + TuLiPa.updatecuts!(prob_cp, cuts_cp) end return end -function get_lightcuts(subix) - db = get_local_db() - cuts = db.mp[subix].cuts - return TuLiPa.getlightweightself(cuts) -end +"""Group dist_mp by core, spawn one call per core via spawn_fn, fetch all results. +Optional skipmed filters subsystems via skipmed_check.""" +function fetch_per_core(spawn_fn, dist_mp, skipmed=nothing) + core_subixs = Dict{CoreId, Vector{Int}}() + for (subix, core) in dist_mp + if isnothing(skipmed) || skipmed_check(subix, skipmed) + push!(get!(() -> Int[], core_subixs, core), subix) + end + end -function get_cutsdata(subix) - db = get_local_db() + futures = Pair{CoreId, Any}[] + @sync for (core, subixs) in core_subixs + push!(futures, core => spawn_fn(core, subixs)) + end - cuts = db.mp[subix].cuts - return (cuts.id, cuts.constants, cuts.slopes) + results = Any[] + for (_, f) in futures + ret = fetch(f) + ret isa RemoteException && throw(ret) + push!(results, ret) + end + return results end -function get_cutsid(subix) +function get_all_cutsdata_local(subixs) db = get_local_db() + results = Vector{Tuple}() + for subix in subixs + cuts = db.mp[subix].cuts + push!(results, (cuts.id, cuts.constants, cuts.slopes)) + end + return results +end +function get_lightcuts(subix) + db = get_local_db() cuts = db.mp[subix].cuts - return cuts.id + return TuLiPa.getlightweightself(cuts) end function update_startstates_cp(prob_cp, startstates, stepnr, t) From 8ffac8554665ef16823668f274cb3752d4ede2a5 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 10:59:28 +0200 Subject: [PATCH 07/10] collect ifm results concurrently --- src/io.jl | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/src/io.jl b/src/io.jl index eca530a..8176352 100644 --- a/src/io.jl +++ b/src/io.jl @@ -769,8 +769,13 @@ end function collect_ifm_u0(stepnr) db = get_local_db() d = Dict{String,Vector{Float64}}() - for core in get_cores(db.input) - fetched = fetch(@spawnat core local_collect_ifm_u0(stepnr)) + futures = Pair{CoreId, Any}[] + @sync for core in get_cores(db.input) + f = @spawnat core local_collect_ifm_u0(stepnr) + push!(futures, core => f) + end + for (core, f) in futures + fetched = fetch(f) if fetched isa RemoteException throw(fetched) end @@ -798,8 +803,13 @@ end function collect_ifm_Q(stepnr) db = get_local_db() d = Dict{String,Float64}() - for core in get_cores(db.input) - fetched = fetch(@spawnat core local_collect_ifm_Q(stepnr)) + futures = Pair{CoreId, Any}[] + @sync for core in get_cores(db.input) + f = @spawnat core local_collect_ifm_Q(stepnr) + push!(futures, core => f) + end + for (core, f) in futures + fetched = fetch(f) if fetched isa RemoteException throw(fetched) end From ce2ebfea81f48df2b0248d4ef6d1051765ebac85 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 11:05:24 +0200 Subject: [PATCH 08/10] update_endstate_evp parallellization --- src/prob_evp.jl | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/prob_evp.jl b/src/prob_evp.jl index 17f663e..00a57a3 100644 --- a/src/prob_evp.jl +++ b/src/prob_evp.jl @@ -80,6 +80,10 @@ function update_endstates_evp(input, scenix::ScenarioIx, subix::SubsystemIx, evp elseif endvaluemethod_ev == "ppp" detailedrescopl = get_dataset(input)["detailedrescopl"] enekvglobaldict = get_dataset(input)["enekvglobaldict"] + term_ppp = get_horizonterm_evp(subsystem) + core_ppp = get_core_ppp(get_local_db().dist_ppp, scenix) + + pending = Vector{Tuple{typeof(first(storages)), TuLiPa.Id, Int, Future}}() for obj in storages balance = TuLiPa.getbalance(obj) bid = TuLiPa.getid(balance) @@ -94,13 +98,16 @@ function update_endstates_evp(input, scenix::ScenarioIx, subix::SubsystemIx, evp bid = TuLiPa.Id(bid.conceptname, instancename[1] * "Balance_" * balancename * "_hydro_reservoir") # TODO: This should be in the dataset end endperiod = TuLiPa.gethorizon(TuLiPa.getbalance(obj)).ix_stop - term_ppp = get_horizonterm_evp(subsystem) - core_ppp = get_core_ppp(get_local_db().dist_ppp, scenix) future = @spawnat core_ppp get_balancedual_ppp(scenix, bid, endperiod, term_ppp) + push!(pending, (obj, bid, endperiod, future)) + end + + for (obj, bid, endperiod, future) in pending dual_ppp = fetch(future) if dual_ppp isa RemoteException throw(dual_ppp) end + instancename = split(TuLiPa.getinstancename(TuLiPa.getid(TuLiPa.getbalance(obj))), "Balance_") if haskey(enekvglobaldict, instancename[2]) dual_ppp *= enekvglobaldict[instancename[2]] end From 06058cffc685c382dfcefe75034fa12ffa9c0929 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 11:10:42 +0200 Subject: [PATCH 09/10] parallellization in update_endconditions_sp --- src/prob_stoch.jl | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/src/prob_stoch.jl b/src/prob_stoch.jl index 8c02e2d..6924be5 100644 --- a/src/prob_stoch.jl +++ b/src/prob_stoch.jl @@ -516,23 +516,32 @@ function update_endconditions_sp(scenix::ScenarioIx, subix::SubsystemIx, t::TuLi set_endstates!(sp.prob, storages, db.startstates) elseif endvaluemethod_sp == "evp" # TODO: Store bid and period in sp (or subsystem?) core_evp = get_core_evp(db.dist_evp, parentscenix, subix) + term_evp = get_horizonterm_evp(subsystem) + + pending = Vector{Tuple{typeof(first(storages)), Int, Future}}() for obj in storages commodityname = TuLiPa.getinstancename(TuLiPa.getid(TuLiPa.getcommodity(TuLiPa.getbalance(obj)))) horizon_sp = TuLiPa.gethorizon(TuLiPa.getbalance(obj)) duration_stoch = TuLiPa.getdurationtoend(horizon_sp) - term_evp = get_horizonterm_evp(subsystem) horizon_evp = db.horizons[(parentscenix, term_evp, commodityname)] period_evp = TuLiPa.getendperiodfromduration(horizon_evp, duration_stoch) bid = TuLiPa.getid(TuLiPa.getbalance(obj)) future = @spawnat core_evp get_balancedual_evp(parentscenix, subix, bid, period_evp) - dual_evp = fetch(future) - period_sp = TuLiPa.getnumperiods(horizon_sp) + push!(pending, (obj, period_sp, future)) + end + + for (obj, period_sp, future) in pending + dual_evp = fetch(future) TuLiPa.setobjcoeff!(sp.prob, TuLiPa.getid(obj), period_sp, dual_evp) end elseif endvaluemethod_sp == "ppp" detailedrescopl = get_dataset(db)["detailedrescopl"] enekvglobaldict = get_dataset(db)["enekvglobaldict"] + term_ppp = get_horizonterm_stoch(subsystem) + core_ppp = get_core_ppp(db.dist_ppp, parentscenix) + + pending = Vector{Tuple{typeof(first(storages)), String, Int, Future}}() for obj in storages balance = TuLiPa.getbalance(obj) bid = TuLiPa.getid(balance) @@ -547,15 +556,15 @@ function update_endconditions_sp(scenix::ScenarioIx, subix::SubsystemIx, t::TuLi bid = TuLiPa.Id(bid.conceptname, instancename[1] * "Balance_" * balancename * "_hydro_reservoir") # TODO: This should be in the dataset end endperiod = TuLiPa.getlastperiod(TuLiPa.gethorizon(TuLiPa.getbalance(obj))) - term_ppp = get_horizonterm_stoch(subsystem) - core_ppp = get_core_ppp(db.dist_ppp, parentscenix) future = @spawnat core_ppp get_balancedual_ppp(parentscenix, bid, endperiod, term_ppp) + push!(pending, (obj, instancename[2], TuLiPa.getnumperiods(TuLiPa.gethorizon(TuLiPa.getbalance(obj))), future)) + end + + for (obj, iname, numperiods, future) in pending dual_ppp = fetch(future) - if haskey(enekvglobaldict, instancename[2]) - dual_ppp *= enekvglobaldict[instancename[2]] + if haskey(enekvglobaldict, iname) + dual_ppp *= enekvglobaldict[iname] end - - numperiods = TuLiPa.getnumperiods(TuLiPa.gethorizon(TuLiPa.getbalance(obj))) TuLiPa.setobjcoeff!(sp.prob, TuLiPa.getid(obj), numperiods, dual_ppp) end end From 1a7702b9038878ae3e35967caa3aa74ecbee2ea8 Mon Sep 17 00:00:00 2001 From: cjuli1 Date: Mon, 17 Aug 2026 22:58:28 +0200 Subject: [PATCH 10/10] support onlysubsystemmode --- src/io.jl | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/io.jl b/src/io.jl index 8176352..fba73bd 100644 --- a/src/io.jl +++ b/src/io.jl @@ -938,8 +938,10 @@ function update_output(t::TuLiPa.ProbTime, stepnr::Int) db.output.timing_sp[(scenix, subix)][stepnr, :] .= all_timings[core][4][(scenix, subix)] end - db.output.timing_cp[stepnr, :] .= db.cp.div[MainTiming] - fill!(db.cp.div[MainTiming], 0.0) + if haskey(settings["problems"], "clearing") + db.output.timing_cp[stepnr, :] .= db.cp.div[MainTiming] + fill!(db.cp.div[MainTiming], 0.0) + end end if has_result_scenarios(settings)