Merge pull request #377 from JuliaParallel/jps/update-viz-docs

Update scheduler visualization docs

Author: jpsamaroo

docs/src/logging.md

@@ -32,7 +32,7 @@ called. Let's construct one:
 
 ```julia
 ctx = Context()
-ml = TimspanLogging.MultiEventLog()
+ml = TimespanLogging.MultiEventLog()
 
 # Add the BytesAllocd consumer to the log as `:bytes`
 ml[:bytes] = Dagger.Events.BytesAllocd()

docs/src/scheduler-visualization.md

@@ -16,22 +16,24 @@ easiest way to get started with the web dashboard for new users.
 For manual usage, the following snippet of code will suffice:
 
 ```julia
+using Dagger, DaggerWebDash, TimespanLogging
+
 ctx = Context() # or `ctx = Dagger.Sch.eager_context()` for eager API usage
-ml = Dagger.MultiEventLog()
+ml = TimespanLogging.MultiEventLog()
 
 ## Add some logging events of interest
 
-ml[:core] = Dagger.Events.CoreMetrics()
-ml[:id] = Dagger.Events.IDMetrics()
-ml[:timeline] = Dagger.Events.TimelineMetrics()
+ml[:core] = TimespanLogging.Events.CoreMetrics()
+ml[:id] = TimespanLogging.Events.IDMetrics()
+ml[:timeline] = TimespanLogging.Events.TimelineMetrics()
 # ...
 
 # (Optional) Enable profile flamegraph generation with ProfileSVG
 ml[:profile] = DaggerWebDash.ProfileMetrics()
 ctx.profile = true
 
 # Create a LogWindow; necessary for real-time event updates
-lw = Dagger.Events.LogWindow(20*10^9, :core)
+lw = TimespanLogging.Events.LogWindow(20*10^9, :core)
 ml.aggregators[:logwindow] = lw
 
 # Create the D3Renderer server on port 8080
@@ -40,20 +42,20 @@ d3r = DaggerWebDash.D3Renderer(8080)
 ## Add some plots! Rendered top-down in order
 
 # Show an overview of all generated events as a Gantt chart
-push!(d3r, GanttPlot(:core, :id, :esat, :psat; title="Overview"))
+push!(d3r, DaggerWebDash.GanttPlot(:core, :id, :esat, :psat; title="Overview"))
 
 # Show various numerical events as line plots over time
-push!(d3r, LinePlot(:core, :wsat, "Worker Saturation", "Running Tasks"))
-push!(d3r, LinePlot(:core, :loadavg, "CPU Load Average", "Average Running Threads"))
-push!(d3r, LinePlot(:core, :bytes, "Allocated Bytes", "Bytes"))
-push!(d3r, LinePlot(:core, :mem, "Available Memory", "% Free"))
+push!(d3r, DaggerWebDash.LinePlot(:core, :wsat, "Worker Saturation", "Running Tasks"))
+push!(d3r, DaggerWebDash.LinePlot(:core, :loadavg, "CPU Load Average", "Average Running Threads"))
+push!(d3r, DaggerWebDash.LinePlot(:core, :bytes, "Allocated Bytes", "Bytes"))
+push!(d3r, DaggerWebDash.LinePlot(:core, :mem, "Available Memory", "% Free"))
 
 # Show a graph rendering of compute tasks and data movement between them
 # Note: Profile events are ignored if absent from the log
-push!(d3r, GraphPlot(:core, :id, :timeline, :profile, "DAG"))
+push!(d3r, DaggerWebDash.GraphPlot(:core, :id, :timeline, :profile, "DAG"))
 
 # TODO: Not yet functional
-#push!(d3r, ProfileViewer(:core, :profile, "Profile Viewer"))
+#push!(d3r, DaggerWebDash.ProfileViewer(:core, :profile, "Profile Viewer"))
 
 # Add the D3Renderer as a consumer of special events generated by LogWindow
 push!(lw.creation_handlers, d3r)

lib/DaggerWebDash/Project.toml

@@ -6,6 +6,7 @@ version = "0.1.2"
 [deps]
 Dagger = "d58978e5-989f-55fb-8d15-ea34adc7bf54"
 Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"
+HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3"
 JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1"
 MemPool = "f9f48841-c794-520a-933b-121f7ba6ed94"
 Mux = "a975b10e-0019-58db-a62f-e48ff68538c9"
@@ -17,6 +18,7 @@ TimespanLogging = "a526e669-04d3-4846-9525-c66122c55f63"
 
 [compat]
 Dagger = "0.16"
+HTTP = "1.7"
 JSON3 = "1"
 MemPool = "0.3, 0.4"
 Mux = "0.7, 1"

lib/DaggerWebDash/src/d3.jl

@@ -1,4 +1,7 @@
-using Mux, MemPool, Distributed, Sockets
+using Mux, Sockets
+import HTTP
+import HTTP: WebSockets
+using MemPool, Distributed
 
 struct LinePlot
     core_key::Symbol
@@ -172,22 +175,22 @@ function client_handler(sock, id, port, port_range, config_updated, config, seek
     if id != myid()
         fsock_ready = Base.Event()
         worker_host, worker_port = worker_host_port(id, port, port_range)
-        Mux.HTTP.WebSockets.open("ws://$worker_host:$worker_port/data_feed") do _fsock
+        WebSockets.open("ws://$worker_host:$worker_port/data_feed") do _fsock
             fsock = _fsock
             @debug "D3R forwarder for $id ready"
             notify(fsock_ready)
-            while !eof(fsock) && isopen(fsock)
+            while true
                 try
-                    bytes = readavailable(fsock)
+                    bytes = WebSockets.receive(fsock)
                     if length(bytes) == 0
                         sleep(0.1)
                         continue
                     end
                     data = String(bytes)
                     #@info "D3R forwarder for $id received data"
-                    write(sock, data)
+                    WebSockets.send(sock, data)
                 catch err
-                    if err isa Mux.WebSockets.WebSocketClosedError || err isa Base.IOError
+                    if err isa WebSockets.WebSocketError && err.message isa WebSockets.CloseFrameBody
                         # Force-close client and forwarder
                         @async close(sock)
                         @async close(fsock)
@@ -203,25 +206,25 @@ function client_handler(sock, id, port, port_range, config_updated, config, seek
     end
     if id == myid()
         @debug "D3R client for $id sending initial config"
-        write(sock, JSON3.write((;cmd="data", payload=sanitize(D3R_LOGS[port]))))
+        WebSockets.send(sock, JSON3.write((;cmd="data", payload=sanitize(D3R_LOGS[port]))))
         _workers = workers()
         if !(myid() in _workers)
             # FIXME: Get this from the Context
             _workers = vcat(myid(), _workers)
         end
-        write(sock, JSON3.write((;cmd="config", payload=sanitize((;myid=myid(),workers=_workers,ctxs=config)))))
+        WebSockets.send(sock, JSON3.write((;cmd="config", payload=sanitize((;myid=myid(),workers=_workers,ctxs=config)))))
     end
     push!(get!(()->[], D3R_CLIENT_SOCKETS[port], id), sock)
     @debug "D3R client for $id ready"
-    while !eof(sock) && isopen(sock)
+    while true
         try
-            data = String(read(sock))
+            data = String(WebSockets.receive(sock))
             @debug "D3R client for $id received: $data"
             if id == myid()
                 #= FIXME
                 if config_updated[]
                     config_updated[] = false
-                    write(sock, JSON3.write((;cmd="config", payload=sanitize(config))))
+                    WebSockets.send(sock, JSON3.write((;cmd="config", payload=sanitize(config))))
                 end
                 =#
                 if seek_store !== nothing
@@ -231,7 +234,7 @@ function client_handler(sock, id, port, port_range, config_updated, config, seek
                         for (idx,key) in enumerate(Tables.columnnames(raw_logs))
                             logs[key] = Tables.columns(raw_logs)[idx]
                         end
-                        write(sock, JSON3.write((;cmd="data", payload=sanitize(logs))))
+                        WebSockets.send(sock, JSON3.write((;cmd="data", payload=sanitize(logs))))
                         continue
                     end
                     m = match(r"seek\(([0-9]*),([0-9]*)\)", data)
@@ -242,19 +245,19 @@ function client_handler(sock, id, port, port_range, config_updated, config, seek
                         for (idx,key) in enumerate(Tables.columnnames(raw_logs))
                             logs[key] = Tables.columns(raw_logs)[idx]
                         end
-                        write(sock, JSON3.write((;cmd="data", payload=sanitize(logs))))
+                        WebSockets.send(sock, JSON3.write((;cmd="data", payload=sanitize(logs))))
                         continue
                     end
                 end
                 if data == "data"
-                    write(sock, JSON3.write((;cmd="data", payload=sanitize(D3R_LOGS[port]))))
+                    WebSockets.send(sock, JSON3.write((;cmd="data", payload=sanitize(D3R_LOGS[port]))))
                 end
             else
                 @debug "D3R client sending to forwarder: $data"
-                write(fsock, data)
+                WebSockets.send(fsock, data)
             end
         catch err
-            if err isa Mux.WebSockets.WebSocketClosedError || err isa Base.IOError
+            if err isa WebSockets.WebSocketError && err.message isa WebSockets.CloseFrameBody
                 idx = findfirst(x->x==sock, D3R_CLIENT_SOCKETS[port][id])
                 if idx !== nothing
                     deleteat!(D3R_CLIENT_SOCKETS[port][id], idx)
@@ -273,11 +276,9 @@ end
 function TimespanLogging.Events.creation_hook(d3r::D3Renderer, log)
     for sock in get!(()->[], get!(()->Dict{Int,Vector{Any}}(), D3R_CLIENT_SOCKETS, d3r.port), myid())
         try
-            if isopen(sock)
-                write(sock, JSON3.write((;cmd="add", payload=sanitize(log))))
-            end
+            WebSockets.send(sock, JSON3.write((;cmd="add", payload=sanitize(log))))
         catch err
-            if err isa Mux.WebSockets.WebSocketClosedError
+            if err isa WebSockets.WebSocketError && err.message isa WebSockets.CloseFrameBody
                 continue
             end
             rethrow(err)
@@ -287,11 +288,9 @@ end
 function TimespanLogging.Events.deletion_hook(d3r::D3Renderer, idx)
     for sock in get!(()->[], get!(()->Dict{Int,Vector{Any}}(), D3R_CLIENT_SOCKETS, d3r.port), myid())
         try
-            if isopen(sock)
-                write(sock, JSON3.write((;cmd="delete", payload=idx)))
-            end
+            WebSockets.send(sock, JSON3.write((;cmd="delete", payload=idx)))
         catch err
-            if err isa Mux.WebSockets.WebSocketClosedError
+            if err isa WebSockets.WebSocketError && err.message isa WebSockets.CloseFrameBody
                 continue
             end
             rethrow(err)