Fixing issues on averaging consensus.

Author: rcschrg

src/algorithm/consensus/averaging.jl

@@ -10,10 +10,12 @@ end
 mutable struct NoConsensusActor <: ConsensusActor
 end
 
-struct AveragingConsensusMessage 
+struct AveragingConsensusMessage <: OptimizationMessage
     λ::Vector{Float64}
     k::Int
     data::Any
+    initial::Bool
+    AveragingConsensusMessage(λ, k, data, initial=false) = new(λ, k, data, initial)
 end
 
 struct ConsensusFinishedMessage
@@ -27,7 +29,7 @@ end
     first_message = true
     k::Int = 0
     max_iter::Int = 50
-    λ::Vector{Float64} = Vector{Real}()
+    λ::Vector{Float64} = [1]
 
     initial_λ::Real
     α::Real
@@ -37,14 +39,22 @@ end
 end
 
 function on_exchange_message(algorithm_data::AveragingConsensusAlgorithm, carrier::Carrier, message::AveragingConsensusMessage, meta::Any)
+    #@info "Doing something" algorithm_data.first_message message.k algorithm_data.k
     if message.k >= algorithm_data.max_iter
-        # abort if iteration count is reached
+        if algorithm_data.first_message 
+            # negotiation is over, only new with k=0 is expected
+            return
+        end
+        # finish if iteration count is reached, reset all state data
         algorithm_data.finish_callback(algorithm_data, carrier)
+        algorithm_data.first_message = true
+        empty!(algorithm_data.message_queue)
         return
     end
-
-    if algorithm_data.first_message
+    
+    if algorithm_data.first_message || message.initial
         algorithm_data.first_message = false
+        algorithm_data.k = 0
         algorithm_data.λ = ones(length(message.λ)) .* algorithm_data.initial_λ
 
         for addr in others(carrier, "")
@@ -55,11 +65,11 @@ function on_exchange_message(algorithm_data::AveragingConsensusAlgorithm, carrie
 
     push!(queue, message)
 
-    if length(queue) == length(others(carrier, ""))
+    if length(queue) == length(others(carrier, "")) || algorithm_data.k < message.k
         avgλ = sum(m.λ for m in queue) ./ length(queue)
         algorithm_data.λ .+= algorithm_data.α .* (avgλ .- algorithm_data.λ) .+ gradient_term(algorithm_data.actor, algorithm_data.λ, message.data)
+        algorithm_data.k = message.k + 1
 
-        algorithm_data.k += message.k + 1
         delete!(algorithm_data.message_queue, message.k)
 
         for addr in others(carrier, "")

src/algorithm/consensus/economic_dispatch.jl

@@ -3,11 +3,11 @@ export LinearCostEconomicDispatchConsensusActor
 @kwdef mutable struct LinearCostEconomicDispatchConsensusActor <: ConsensusActor
     cost::Real
     P_max::Real
-    ρ::Real = 0.01
+    ρ::Real = 0.05
     ϵ::Real = 0.1
     P_min::Real = 0
     N_guess::Int = 10
-    P::Vector{Float64} = Vector{Real}()
+    P::Vector{Float64} = [0]
 end
 
 function DistributedResourceOptimization.gradient_term(actor::LinearCostEconomicDispatchConsensusActor, λ::Vector{<:Real}, P_target::Vector{<:Real})

src/algorithm/core.jl

@@ -8,6 +8,10 @@ Subtypes should implement specific distributed algorithm logic for resource opti
 """
 abstract type DistributedAlgorithm end
 
+"""
+Supertype for any optimization message. Used to detect whether the message is part of an optimization
+"""
+abstract type OptimizationMessage end
 
 """
     on_exchange_message(algorithm::DistributedAlgorithm, carrier::Carrier, message_data::Any, meta::Any)
@@ -27,6 +31,7 @@ Processes the received exchange message, updating the algorithm's state or trigg
 May return a result or perform side effects depending on the implementation.
 """
 function on_exchange_message(algorithm::DistributedAlgorithm, carrier::Carrier, message_data::Any, meta::Any)
+    @error algorithm carrier message_data meta
     throw("NotImplementedOrWrongArgumentTypes")
 end
 

src/carrier/mango.jl

@@ -103,7 +103,7 @@ function DistributedOptimizationRole(algorithm::DistributedAlgorithm; tid::Symbo
 end
 
 function Mango.handle_message(role::DistributedOptimizationRole, message::Any, meta::Any)
-    if haskey(meta, "optimization_message") && meta["optimization_message"]
+    if haskey(meta, "optimization_message") && meta["optimization_message"] || message isa OptimizationMessage
         on_exchange_message(role.algorithm, role.carrier, message, meta)
     end
 end