|
| 1 | +export SimpleCarrier, ActorContainer, cid, start_distributed_optimization, start_coordinated_optimization |
| 2 | + |
| 3 | + |
| 4 | +using UUIDs |
| 5 | + |
| 6 | +abstract type AbstractSimpleCarrier <: Carrier end |
| 7 | + |
| 8 | +""" |
| 9 | + ActorContainer |
| 10 | +
|
| 11 | + A container to manage multiple `SimpleCarrier`. |
| 12 | +""" |
| 13 | +struct ActorContainer |
| 14 | + actors::Vector{AbstractSimpleCarrier} |
| 15 | + function ActorContainer() |
| 16 | + return new(Vector{AbstractSimpleCarrier}()) |
| 17 | + end |
| 18 | +end |
| 19 | + |
| 20 | +function register(actor_container::ActorContainer, carrier::AbstractSimpleCarrier) |
| 21 | + push!(actor_container.actors, carrier) |
| 22 | + carrier.aid = length(actor_container.actors) |
| 23 | +end |
| 24 | + |
| 25 | +""" |
| 26 | + SimpleCarrier |
| 27 | +
|
| 28 | +A concrete implementation of the `Carrier` type representing a simple carrier for distributed resource optimization. |
| 29 | +This carrier facilitates communication and scheduling among distributed algorithms within an `ActorContainer`. |
| 30 | +""" |
| 31 | +mutable struct SimpleCarrier <: AbstractSimpleCarrier |
| 32 | + container::ActorContainer |
| 33 | + actor::Union{<:Coordinator, <:DistributedAlgorithm} |
| 34 | + aid::Real |
| 35 | + uuid_to_handler::Dict{UUID, Function} |
| 36 | + function SimpleCarrier(container::ActorContainer, actor::Union{<:Coordinator, <:DistributedAlgorithm}) |
| 37 | + carrier = new(container, actor, -1, Dict{UUID, Function}()) |
| 38 | + register(container, carrier) |
| 39 | + return carrier |
| 40 | + end |
| 41 | +end |
| 42 | + |
| 43 | +""" |
| 44 | + cid(carrier::SimpleCarrier) |
| 45 | + return the carrier id |
| 46 | +""" |
| 47 | +function cid(carrier::SimpleCarrier) |
| 48 | + return carrier.aid |
| 49 | +end |
| 50 | + |
| 51 | +function Base.wait(carrier::SimpleCarrier, event::EventWithValue) |
| 52 | + wait(event.event) |
| 53 | + return event.value |
| 54 | +end |
| 55 | + |
| 56 | +function _dispatch_to(carrier, content, meta) |
| 57 | + if haskey(meta, :message_id) && haskey(carrier.uuid_to_handler, meta[:message_id]) |
| 58 | + carrier.uuid_to_handler[meta[:message_id]](carrier, content, meta) |
| 59 | + else |
| 60 | + on_exchange_message(carrier.actor, carrier, content, meta) |
| 61 | + end |
| 62 | +end |
| 63 | + |
| 64 | +function send_to_other(carrier::SimpleCarrier, content::Any, receiver::Real; meta::Dict{Symbol,Any}=Dict{Symbol,Any}()) |
| 65 | + other_carrier::Carrier = carrier.container.actors[receiver] |
| 66 | + main_meta = Dict(:sender => carrier.aid, :message_id => uuid4()) |
| 67 | + union_meta = merge(main_meta, meta) # important: meta can override main_meta entries |
| 68 | + return @spawnlog begin |
| 69 | + _dispatch_to(other_carrier, content, union_meta) |
| 70 | + end |
| 71 | +end |
| 72 | + |
| 73 | +function reply_to_other(carrier::SimpleCarrier, content_data::Any, meta) |
| 74 | + return send_to_other(carrier, content_data, meta[:sender], meta=merge(meta, Dict(:reply => true))) |
| 75 | +end |
| 76 | + |
| 77 | +function send_awaitable(carrier::SimpleCarrier, content::Any, receiver::Real; meta::Dict{Symbol,Any}=Dict{Symbol,Any}()) |
| 78 | + other_carrier::Carrier = carrier.container.actors[receiver] |
| 79 | + main_meta = Dict(:sender => carrier.aid, :message_id => uuid4()) |
| 80 | + union_meta = merge(main_meta, meta) # important: meta can override main_meta entries |
| 81 | + event = EventWithValue(Base.Event(), nothing) |
| 82 | + carrier.uuid_to_handler[union_meta[:message_id]] = function(other_carrier::Carrier, content::Any, union_meta::Dict{Symbol,Any}) |
| 83 | + event.value = content |
| 84 | + notify(event.event) |
| 85 | + end |
| 86 | + @spawnlog begin |
| 87 | + _dispatch_to(other_carrier, content, union_meta) |
| 88 | + end |
| 89 | + return event |
| 90 | +end |
| 91 | + |
| 92 | +function schedule_using(carrier::SimpleCarrier, to_be_scheduled::Function, delay_s::Float64) |
| 93 | + @spawnlog begin |
| 94 | + sleep(delay_s) |
| 95 | + to_be_scheduled() |
| 96 | + end |
| 97 | +end |
| 98 | + |
| 99 | +function others(carrier::SimpleCarrier, id::String) |
| 100 | + return setdiff!(collect(range(1, length(carrier.container.actors))), [cid(carrier)]) |
| 101 | +end |
| 102 | + |
| 103 | +""" |
| 104 | + start_distributed_optimization(actors::Vector{<:DistributedAlgorithm}, start_message::Any) |
| 105 | +
|
| 106 | +Start a distributed optimization process among the provided `actors` using the given `start_message`. |
| 107 | +Return a waitable object that can be used to monitor the progress of the optimization. |
| 108 | +""" |
| 109 | +function start_distributed_optimization(actors::Vector{<:DistributedAlgorithm}, start_message::Any) |
| 110 | + actor_container = ActorContainer() |
| 111 | + carriers = [SimpleCarrier(actor_container, actor) for actor in actors] |
| 112 | + return send_to_other(carriers[1], start_message, cid(carriers[2])) |
| 113 | +end |
| 114 | + |
| 115 | +""" |
| 116 | + start_coordinated_optimization(actors::Vector{<:DistributedAlgorithm}, coordinator::Coordinator, start_message::Any) |
| 117 | +
|
| 118 | +Start a coordinated optimization process among the provided `actors` and a `coordinator` using the given `start_message`. |
| 119 | +Return the result of the optimization process. The coordinator manages the overall optimization flow. |
| 120 | +""" |
| 121 | +function start_coordinated_optimization(actors::Vector{<:DistributedAlgorithm}, coordinator::Coordinator, start_message::Any) |
| 122 | + actor_container = ActorContainer() |
| 123 | + carriers = [SimpleCarrier(actor_container, actor) for actor in actors] |
| 124 | + coordinator_carrier = SimpleCarrier(actor_container, coordinator) |
| 125 | + return start_optimization(coordinator, coordinator_carrier, start_message, Dict()) |
| 126 | +end |
0 commit comments