aggregation algorithms fix

Author: SelimChraibi

Project.toml

@@ -1,7 +1,7 @@
 name = "AsynchronousIterativeAlgorithms"
 uuid = "329f2bf2-1773-4f40-9abf-1830ae341a86"
 authors = ["selim-chraibi"]
-version = "0.1.5"
+version = "0.1.6"
 
 [deps]
 Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"

docs/src/manual.md

@@ -5,7 +5,7 @@ We saw how to run an asynchronous version of the SGD algorithm on a LRMSE proble
   - [Working with a distributed problem](@ref)
   - [Synchronous run](@ref)
   - [Active processes](@ref)
-  - [Recording iterates](@ref)
+  - [Recording iterates](@ref recording_iterated)
   - [Customization of `start`'s execution](@ref custom_execution)
   - [Handling worker failures](@ref)
   - [Algorithm wrappers](@ref algorithm_wrappers)
@@ -262,7 +262,7 @@ $$q_j \longleftarrow \textrm{query}(\underset{i \in \textrm{connected}}{\textrm{
 
 where $q_j$ is computed by the worker upon reception of $\textrm{answer}(q_i)$ from worker $j$ and where $connected$ are the list of workers that have answered.
 
-The [`AggregationAlgorithm`](@ref) in this library requires you to specify three methods: query, answer, and aggregate. Here's an example showing the required signatures of these three methods:
+The [`AggregationAlgorithm`](@ref) in this library requires you to define four methods: `initialize`, `query`, `answer`, and `aggregate`. Here's an example showing the required signatures of these three methods:
 
 ```julia
 @everywhere begin 
@@ -273,10 +273,10 @@ The [`AggregationAlgorithm`](@ref) in this library requires you to specify three
         stepsize::Float64 
     end
 
-    (tba::ToBeAggregatedGD)(problem::Any) = tba.q1
-    (tba::ToBeAggregatedGD)(a::Vector{Vector{Float64}}, connected::Vector{Int64}) = mean(a)            
-    (tba::ToBeAggregatedGD)(a::Vector{Float64}, problem::Any) = a
-    (tba::ToBeAggregatedGD)(q::Vector{Float64}, problem::Any) = q - tba.stepsize * problem.∇f(q)
+    AIA.initialize(tba::ToBeAggregatedGD, problem::Any) = tba.q1
+    AIA.aggregate(tba::ToBeAggregatedGD, a::Vector{Vector{Float64}}, connected::Vector{Int64}) = mean(a)            
+    AIA.query(tba::ToBeAggregatedGD, a::Vector{Float64}, problem::Any) = a
+    AIA.answer(tba::ToBeAggregatedGD, q::Vector{Float64}, problem::Any) = q - tba.stepsize * problem.∇f(q)
 end 
 
 algorithm = AggregationAlgorithm(ToBeAggregatedGD(rand(10), 0.01); pids=workers())
@@ -286,7 +286,7 @@ history = start(algorithm, distributed_problem, (epoch=100,));
 
 **Memory limitation:** At any point in time, the central worker should have access must have access to the latest answers $a_i$ from *all* the connected workers. This means storing a lot of $a_i$ if we use many workers. There is a workaround when the aggregation operation is an *average*. In this case, only the equivalent of one answer needs to be saved on the central node, regardless of the number of workers.
 
-[`AveragingAlgorithm`](@ref) implements this memory optimization. Here you only need to define `query`, the `answer`
+[`AveragingAlgorithm`](@ref) implements this memory optimization. Here you only need to define `initialize`, `query`, the `answer`
 
 ```julia
 @everywhere begin 
@@ -295,9 +295,9 @@ history = start(algorithm, distributed_problem, (epoch=100,));
         stepsize::Float64 
     end
 
-    (tba::ToBeAveragedGD)(problem::Any) = tba.q1
-    (tba::ToBeAveragedGD)(a::Vector{Float64}, problem::Any) = a
-    (tba::ToBeAveragedGD)(q::Vector{Float64}, problem::Any) = q - tba.stepsize * problem.∇f(q)
+    AIA.initialize(tba::ToBeAveragedGD, problem::Any) = tba.q1
+    AIA.query(tba::ToBeAveragedGD, a::Vector{Float64}, problem::Any) = a
+    AIA.answer(tba::ToBeAveragedGD, q::Vector{Float64}, problem::Any) = q - tba.stepsize * problem.∇f(q)
 end 
 
 algorithm = AveragingAlgorithm(ToBeAveragedGD(rand(10), 0.01); pids=workers(), weights=ones(nworkers()))
@@ -311,5 +311,6 @@ Note that you can implement the [custom callbacks](@ref custom_execution) on bot
 report(::ToBeAggregatedGD) = # do something
 ```
 
+---
 
-Hope you find this library helpful and look forward to seeing how you put it to use!
+Wow you read all this! Hope you find this library helpful and look forward to seeing how you put it to use!

src/algorithm_wrappers.jl

@@ -1,18 +1,23 @@
 export AggregationAlgorithm, AveragingAlgorithm
 
+initialize(algorithm::AbstractAlgorithm, problem::Any) = throw(ArgumentError("Method initialize(::$(typeof(algorithm)), problem::Any) not implemented."))
+aggregate(algorithm::AbstractAlgorithm{Q,A}, as::Vector{A}, workers::Vector{Int64}) where {Q,A} = throw(ArgumentError("Method aggregate(::$(typeof(algorithm)), problem::Any) not implemented."))
+query(algorithm::AbstractAlgorithm, agg::AggregatedA, problem::Any) where {AggregatedA} = throw(ArgumentError("Method query(::$(typeof(algorithm)), problem::Any) not implemented."))
+answer(algorithm::AbstractAlgorithm{Q,A}, q::Q, problem::Any) where {Q,A} = throw(ArgumentError("Method answer(::$(typeof(algorithm)), problem::Any) not implemented."))
+
 """
-    AggregationAlgorithm{Q,A,Alg<:AbstractAlgorithm{Q,A}}(arg; kwarg)::AbstractAlgorithm{Q,A} where {Q,A}
+    AggregationAlgorithm(arg; kwarg)::AbstractAlgorithm
 
 Distributed algorithm that writes: `q_j <- query(aggregate([answer(q_i) for i in connected]))`
 Where a "connected" worker is a worker that has answered at least once.
 (Not memory optimized: `length(pids)` answers are stored on the central worker at all times)
 
 # Argument
-- `algorithm<:AbstractAlgorithm{Q,A}` which should define the following
-    - `algorithm(problem::Any)::Q`: the initialization step that create the first query iterate
-    - `algorithm(as::Vector{A}, workers::Vector{Int64})::AggregatedA` where A: the answer aggregarion step performed by the central node when receiving the answers `as::Vector{A}` from the `workers`
-    - `algorithm(agg::AggregatedA, problem::Any)::Q`: the query step producing a query from the aggregated answer `agg::AggregatedA`, performed by the central node
-    - `algorithm(q::Q, problem::Any)::A`: the answer step perfromed by the wokers when they receive a query `q::Q` from the central node
+- `algorithm<:AbstractAlgorithm{Q,A}` which should define the following (where `const AIA = AsynchronousIterativeAlgorithms`)
+    - `AIA.initialize(algorithm, problem::Any)::Q`: step that creates the first query iterate
+    - `AIA.aggregate(algorithm, as::Vector{A}, workers::Vector{Int64})::AggregatedA` where A: step performed by the central node when receiving the answers `as::Vector{A}` from the `workers`
+    - `AIA.query(algorithm, agg::AggregatedA, problem::Any)::Q`: step producing a query from the aggregated answer `agg::AggregatedA`, performed by the central node
+    - `AIA.answer(algorithm, q::Q, problem::Any)::A`: step perfromed by the wokers when they receive a query `q::Q` from the central node
 
 # Keyword
 - `pids=workers()`: `pids` of the active workers
@@ -30,41 +35,41 @@ struct AggregationAlgorithm{Q,A,Alg<:AbstractAlgorithm{Q,A}} <: AbstractAlgorith
 end
 
 """
-    (::AggregationAlgorithm{Q,A,Alg})(problem::Any)::Q where {Q,A,Alg}
+    (::AggregationAlgorithm{Q,A,Alg})(problem::Any)::Q where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 The initialization step that create the first query iterate
 """
 function (agg::AggregationAlgorithm)(problem::Any)
-    agg.algorithm(problem)
+    initialize(agg.algorithm,  problem)
 end
 
 """
-    (::AggregationAlgorithm{Q,A,Alg})(a::A, worker::Int64, problem::Any)::Q where {Q,A,Alg}
+    (::AggregationAlgorithm{Q,A,Alg})(a::A, worker::Int64, problem::Any)::Q where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 Asynchronous step performed by the central node when receiving an answer `a::A` from a worker
 """
-function (agg::AggregationAlgorithm{Q,A,Alg})(a::A, worker::Int64, problem::Any) where {Q,A,Alg}
+function (agg::AggregationAlgorithm{Q,A,Alg})(a::A, worker::Int64, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
     agg.connected[worker] = true
     agg.answers[worker] = a
-    agg.algorithm(agg.algorithm(agg.answers[agg.connected], (1:maximum(agg.pids))[agg.connected]), problem)
+    query(agg.algorithm, aggregate(agg.algorithm, agg.answers[agg.connected], (1:maximum(agg.pids))[agg.connected]), problem)
 end
 
 """
-    (::AggregationAlgorithm{Q,A,Alg})(as::Vector{A}, workers::Vector{Int64}, problem::Any)::Q where {Q,A,Alg}
+    (::AggregationAlgorithm{Q,A,Alg})(as::Vector{A}, workers::Vector{Int64}, problem::Any)::Q where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 Synchronous step performed by the central node when receiving answers `as::Vector{A}` respectively from `workers::Vector{Int64}`
 """
-function (agg::AggregationAlgorithm{Q,A,Alg})(as::Vector{A}, workers::Vector{Int64}, problem::Any) where {Q,A,Alg}
-    agg.algorithm(agg.algorithm(as, workers), problem)
+function (agg::AggregationAlgorithm{Q,A,Alg})(as::Vector{A}, workers::Vector{Int64}, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
+    query(agg.algorithm, aggregate(agg.algorithm,  as, workers), problem)
 end
 
 """
-    (::AggregationAlgorithm{Q,A,Alg})(q::Q, problem::Any)->A where {Q,A,Alg}
+    (::AggregationAlgorithm{Q,A,Alg})(q::Q, problem::Any)->A where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 Steps performed by the workers when they receive a query `q::Q` from the central node
 """
-function (agg::AggregationAlgorithm{Q,A,Alg})(q::Q, problem::Any) where {Q,A,Alg}
-    agg.algorithm(q, problem)
+function (agg::AggregationAlgorithm{Q,A,Alg})(q::Q, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
+    answer(agg.algorithm,  q, problem)
 end
 
 stopnow(agg::AggregationAlgorithm, stopat::NamedTuple) = stopnow(agg.algorithm, stopat)
@@ -76,20 +81,21 @@ savevalues(agg::AggregationAlgorithm) = savevalues(agg.algorithm)
 
 
 """
-    AveragingAlgorithm{Q,A,Alg<:AbstractAlgorithm{Q,A}}(arg; kwarg)::AbstractAlgorithm{Q,A} where {Q,A}
+    AveragingAlgorithm(arg; kwarg)::AbstractAlgorithm
 
 Distributed algorithm that writes: `q_j <- query(weighted_average([answer(q_i) for i in connected]))`
 Where a "connected" worker is a worker that has answered at least once.
 (Memory optimized: only the equivalent of one answer is stored on the central worker at all times)
 
 # Argument
-- `algorithm<:AbstractAlgorithm{Q,A}` which should define the following
-    - `algorithm(problem::Any)::Q`: the initialization step that create the first query iterate
-    - `algorithm(a::A, problem::Any)::Q`: the query step producing a query from the averaged answer, performed by the central node
-    - `algorithm(q::Q, problem::Any)::A`: the answer step perfromed by the wokers when they receive a query `q::Q` from the central node
+- `algorithm<:AbstractAlgorithm{Q,A}` which should define the following (where `const AIA = AsynchronousIterativeAlgorithms`)
+    - `AIA.initialize(algorithm, problem::Any)::Q`: step that creates the first query iterate
+    - `AIA.query(algorithm, a::A, problem::Any)::Q`: step producing a query from the averaged answer, performed by the central node
+    - `AIA.answer(algorithm, q::Q, problem::Any)::A`: step perfromed by the wokers when they receive a query `q::Q` from the central node
 
 # Keyword
 - `pids=workers()`: `pids` of the active workers
+- `weights=ones(length(pids))`: weights of each pid in the weighted average
 """
 mutable struct AveragingAlgorithm{Q,A,Alg<:AbstractAlgorithm{Q,A}} <: AbstractAlgorithm{Q,A}
     pids::Vector{Int64}
@@ -100,7 +106,7 @@ mutable struct AveragingAlgorithm{Q,A,Alg<:AbstractAlgorithm{Q,A}} <: AbstractAl
     last_answers::Vector{A}
     last_average::Union{A,Nothing}
     weights::Vector{Float64}
-    function AveragingAlgorithm(algorithm::Alg; pids=procs(), weights=ones(nprocs())) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
+    function AveragingAlgorithm(algorithm::Alg; pids=procs(), weights=ones(length(pids))) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
         @assert length(pids) == length(weights) "There should be as many weights as there are pids"
         maxpid = maximum(pids)
         connected = BitVector(zeros(maxpid))
@@ -118,44 +124,44 @@ mutable struct AveragingAlgorithm{Q,A,Alg<:AbstractAlgorithm{Q,A}} <: AbstractAl
 end
 
 """
-    (::AveragingAlgorithm{Q,A,Alg})(problem::Any) where {Q,A,Alg}
+    (::AveragingAlgorithm{Q,A,Alg})(problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 The initialization step that create the first query iterate
 """
 function (avg::AveragingAlgorithm)(problem::Any)
-    avg.algorithm(problem)
+    initialize(avg.algorithm, problem)
 end
 
 """
-    (::AveragingAlgorithm{Q,A,Alg})(δas::Vector{A}, workers::Vector{Int64}, problem::Any) where {Q,A,Alg}
+    (::AveragingAlgorithm{Q,A,Alg})(δas::Vector{A}, workers::Vector{Int64}, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 Asynchronous step performed by the central node when receiving an answer `a::A` from a worker.
 """
-function (avg::AveragingAlgorithm{Q,A,Alg})(δa::A, worker::Int64, problem::Any) where {Q,A,Alg}
+function (avg::AveragingAlgorithm{Q,A,Alg})(δa::A, worker::Int64, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
     avg.connected[worker] = true
     normalization = sum(avg.connected .* avg.weights)
     avg.last_average = isnothing(avg.last_average) ? δa : (avg.weights[worker] * δa + avg.last_normalization * avg.last_average) / normalization
     avg.last_normalization = normalization
-    avg.algorithm(avg.last_average, problem)
+    query(avg.algorithm, avg.last_average, problem)
 end
 
 """
-    (::AveragingAlgorithm{Q,A,Alg})(δas::Vector{A}, workers::Vector{Int64}, problem::Any) where {Q,A,Alg}
+    (::AveragingAlgorithm{Q,A,Alg})(δas::Vector{A}, workers::Vector{Int64}, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 Synchronous step performed by the central node when receiving answers `a::Vector{A}` respectively from `workers::Vector{Int64}`
 """
-function (avg::AveragingAlgorithm{Q,A,Alg})(δas::Vector{A}, workers::Vector{Int64}, problem::Any) where {Q,A,Alg}
+function (avg::AveragingAlgorithm{Q,A,Alg})(δas::Vector{A}, workers::Vector{Int64}, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
     avg.last_average = isnothing(avg.last_average) ? δas : sum(avg.weights[workers] * δas) / sum(avg.weights) + avg.last_average
-    avg.algorithm(avg.last_average, problem)
+    query(avg.algorithm,  avg.last_average, problem)
 end
 
 """
- (::AveragingAlgorithm{Q,A,Alg})(q::Q, problem::Any) where {Q,A,Alg}
+ (::AveragingAlgorithm{Q,A,Alg})(q::Q, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
 
 Steps performed by the workers when they receive a query `q::Q` from the central node
 """
-function (avg::AveragingAlgorithm{Q,A,Alg})(q::Q, problem::Any) where {Q,A,Alg}
-    a = avg.algorithm(q, problem)
+function (avg::AveragingAlgorithm{Q,A,Alg})(q::Q, problem::Any) where Alg<:AbstractAlgorithm{Q,A} where {Q,A}
+    a = answer(avg.algorithm,  q, problem)
     δa = isnothing(avg.last_answer) ? a : a - avg.last_answer
     avg.last_answer = a
     return δa