Add building, auxiliary types and scheduler

Project.toml

@@ -0,0 +1,8 @@
+name = "ParameterServer"
+uuid = "8f7ad238-d247-11e8-1dbf-45e5d8f9d635"
+authors = ["Arda Aytekin <[email protected]>"]
+version = "0.1.0"
+
+[deps]
+BinaryProvider = "b99e7846-7c00-51b0-8f62-c81ae34c0232"
+Libdl = "8f399da3-3557-5675-b5ff-fb832c97cbdb"

deps/build.jl

@@ -0,0 +1,13 @@
+using BinaryProvider
+
+products = Product[
+  LibraryProduct(
+    joinpath(expanduser("~"), "usr", "local", "lib"), "libpsapi", :libps
+  ),
+]
+
+if any(!satisfied(p) for p in products)
+  error("POLO PS-API is not installed properly on your system.")
+end
+
+write_deps_file(joinpath(@__DIR__, "deps.jl"), products)

src/ParameterServer.jl

@@ -0,0 +1,31 @@
+module ParameterServer
+
+using Libdl
+
+import Base: show
+
+export
+  PSScheduler
+
+# Load in `deps.jl`, complaining if it does not exist
+const depsjl_path = joinpath(dirname(@__FILE__), "..", "deps", "deps.jl")
+if !isfile(depsjl_path)
+  error("ParameterServer not installed properly, run Pkg.build(\"ParameterServer\"), restart Julia and try again")
+end
+include(depsjl_path)
+
+function __init__()
+  check_deps()
+
+  global ps_lib = Libdl.dlopen(libps)
+
+  global paramserver_m = Libdl.dlsym(ps_lib, :paramserver_m)
+  global paramserver_s = Libdl.dlsym(ps_lib, :paramserver_s)
+  global paramserver_w = Libdl.dlsym(ps_lib, :paramserver_w)
+end
+
+include(joinpath("aux", "types.jl"))
+
+include(joinpath("executors", "scheduler.jl"))
+
+end # module

src/aux/types.jl

@@ -0,0 +1,41 @@
+struct _error_t
+  id::Cint
+  what::NTuple{256,Cchar}
+end
+
+struct _options_t
+  linger::Cint
+  mtimeout::Clong
+  wtimeout::Clong
+  stimeout::Clong
+  num_masters::Int32
+  saddress::NTuple{256,Cchar}
+  maddress::NTuple{256,Cchar}
+  sports::NTuple{3,UInt16}
+  mworker::UInt16
+end
+
+function show(io::IO, options::_options_t)
+  println(io, "Parameter Server Options with $(Int(options.num_masters)) total master(s)")
+  println(io, "  - linger: $(Int(options.linger)) [ms]")
+  println(io, "  - master timeout: $(Int(options.mtimeout)) [ms]")
+  println(io, "  - scheduler timeout: $(Int(options.stimeout)) [ms]")
+  println(io, "  - worker timeout: $(Int(options.wtimeout)) [ms]")
+  println(io, "  - scheduler listens on ", String([UInt8(c) for c in options.saddress]),
+          " to ports ", map(Int, options.sports))
+  print(io, "  - master listens on ", String([UInt8(c) for c in options.maddress]),
+          " to port ", Int(options.mworker))
+end
+
+function _psoptions(; linger::Integer = 1000, tm::Integer = 5000,
+  tw::Integer = 5000, ts::Integer = 5000, M::Integer = 1,
+  scheduler::String = "localhost", master::String = "localhost",
+  scheduler_ports::Tuple{<:Integer,<:Integer,<:Integer} = (40000, 40001, 40002),
+  worker_port::Integer = 50000)
+
+  return _options_t(Cint(linger), Clong(tm), Clong(tw), Clong(ts), Int32(M),
+  ntuple(i->(i <= length(scheduler) ? Cchar(scheduler[i]) : Cchar('\0')), 256),
+  ntuple(i->(i <= length(master) ? Cchar(master[i]) : Cchar('\0')), 256),
+  ntuple(i->UInt16(scheduler_ports[i]), 3), UInt16(worker_port)
+  )
+end

src/executors/scheduler.jl

@@ -0,0 +1,18 @@
+struct PSScheduler
+  options::_options_t
+
+  function PSScheduler(; linger::Integer, timeout::Integer, M::Integer,
+    address::String, ports::Tuple{<:Integer,<:Integer,<:Integer})
+    new(_psoptions(linger = linger, ts = timeout, M = M, scheduler = address,
+      scheduler_ports = ports))
+  end
+end
+
+function show(io::IO, agent::PSScheduler)
+  println(io, "Scheduler Agent")
+  println(io, "  Expects      : $(agent.options.num_masters) masters")
+  println(io, "  Linger time  : $(Int(agent.options.linger)) [ms]")
+  println(io, "  Timeout      : $(Int(agent.options.stimeout)) [ms]")
+  print(io, "  Binds        : ", String([UInt8(c) for c in agent.options.saddress]),
+    " on ", map(Int, agent.options.sports))
+end