" + workText + "...
") + pyodide.globals.torun = msg.data + pyodide.runPythonAsync("exec_work(torun)").then((res) => { + $("#output").append("Did work! 👏
") + c.send(res) + }) + } + c.onopen = function() { + $("#status").text("Status: connected!") + c.send(JSON.stringify({ + status: 2, + error_msg: "WebsocketWorker" + })) + } + c.onclose = function() { + $("#status").text("Status: disconnected") + } +}) }) diff --git a/web/workterms.js b/web/workterms.js new file mode 100644 index 0000000..165531a --- /dev/null +++ b/web/workterms.js @@ -0,0 +1,108 @@ +workTerms = [ +"Adding Hidden Agendas", +"Adjusting Bell Curves", +"Aesthesizing Industrial Areas", +"Aligning Covariance Matrices", +"Applying Feng Shui Shaders", +"Applying Theatre Soda Layer", +"Asserting Packed Exemplars", +"Attempting to Lock Back-Buffer", +"Binding Sapling Root System", +"Breeding Fauna", +"Building Data Trees", +"Bureacritizing Bureaucracies", +"Calculating Inverse Probability Matrices", +"Calculating Llama Expectoration Trajectory", +"Calibrating Blue Skies", +"Charging Ozone Layer", +"Coalescing Cloud Formations", +"Cohorting Exemplars", +"Collecting Meteor Particles", +"Compounding Inert Tessellations", +"Compressing Fish Files", +"Computing Optimal Bin Packing", +"Concatenating Sub-Contractors", +"Containing Existential Buffer", +"Debarking Ark Ramp", +"Debunching Unionized Commercial Services", +"Deciding What Message to Display Next", +"Decomposing Singular Values", +"Decrementing Tectonic Plates", +"Deleting Ferry Routes", +"Depixelating Inner Mountain Surface Back Faces", +"Depositing Slush Funds", +"Destabilizing Economic Indicators", +"Determining Width of Blast Fronts", +"Deunionizing Bulldozers", +"Dicing Models", +"Diluting Livestock Nutrition Variables", +"Downloading Satellite Terrain Data", +"Exposing Flash Variables to Streak System", +"Extracting Resources", +"Factoring Pay Scale", +"Fixing Election Outcome Matrix", +"Flood-Filling Ground Water", +"Flushing Pipe Network", +"Gathering Particle Sources", +"Generating Jobs", +"Gesticulating Mimes", +"Graphing Whale Migration", +"Hiding Willio Webnet Mask", +"Implementing Impeachment Routine", +"Increasing Accuracy of RCI Simulators", +"Increasing Magmafacation", +"Initializing My Sim Tracking Mechanism", +"Initializing Rhinoceros Breeding Timetable", +"Initializing Robotic Click-Path AI", +"Inserting Sublimated Messages", +"Integrating Curves", +"Integrating Illumination Form Factors", +"Integrating Population Graphs", +"Iterating Cellular Automata", +"Lecturing Errant Subsystems", +"Mixing Genetic Pool", +"Modeling Object Components", +"Mopping Occupant Leaks", +"Normalizing Power", +"Obfuscating Quigley Matrix", +"Overconstraining Dirty Industry Calculations", +"Partitioning City Grid Singularities", +"Perturbing Matrices", +"Pixalating Nude Patch", +"Polishing Water Highlights", +"Populating Lot Templates", +"Preparing Sprites for Random Walks", +"Prioritizing Landmarks", +"Projecting Law Enforcement Pastry Intake", +"Realigning Alternate Time Frames", +"Reconfiguring User Mental Processes", +"Relaxing Splines", +"Removing Road Network Speed Bumps", +"Removing Texture Gradients", +"Removing Vehicle Avoidance Behavior", +"Resolving GUID Conflict", +"Reticulating Splines", +"Retracting Phong Shader", +"Retrieving from Back Store", +"Reverse Engineering Image Consultant", +"Routing Neural Network Infanstructure", +"Scattering Rhino Food Sources", +"Scrubbing Terrain", +"Searching for Llamas", +"Seeding Architecture Simulation Parameters", +"Sequencing Particles", +"Setting Advisor Moods", +"Setting Inner Deity Indicators", +"Setting Universal Physical Constants", +"Sonically Enhancing Occupant-Free Timber", +"Speculating Stock Market Indices", +"Splatting Transforms", +"Stratifying Ground Layers", +"Sub-Sampling Water Data", +"Synthesizing Gravity", +"Synthesizing Wavelets", +"Time-Compressing Simulator Clock", +"Unable to Reveal Current Activity", +"Weathering Buildings", +"Zeroing Crime Network" +] diff --git a/web_python/ray/web.py b/web_python/ray/web.py index 2326817..6718079 100644 --- a/web_python/ray/web.py +++ b/web_python/ray/web.py @@ -1,8 +1,18 @@ from js import XMLHttpRequest, Blob +import js import json import base64 import cloudpickle +import ray.webpb as webpb + + +def b64dumps(b): + return base64.standard_b64encode(b).decode() + + +def b64loads(s): + return base64.standard_b64decode(s) def get(get_id): @@ -38,3 +48,46 @@ def schedule(task): out = json.loads(req.response) out_id = base64.standard_b64decode(out["return_id"]) return out_id + + +WORK_COMPLETE = 0 +WORK_ERROR = 1 +WORK_READY = 2 + +FUNCTION = webpb.ClientTaskType.FUNCTION.value + + +def exec_work(work_blob): + work = json.loads(work_blob) + js.console.log("Got work") + task = work["task"] + if "type" in task and task["type"] != FUNCTION: + return json.dumps({ + "status": WORK_ERROR, + "error_msg": "Unsupported", + }) + args = decode_json_args(task) + getresp = get(b64loads(task["payload_id"])) + func = webpb.loads(getresp["data"]) + res = func(*args) + out_data = cloudpickle.dumps(res) + workstatus = { + "status": WORK_COMPLETE, + "complete_data": b64dumps(out_data), + "finished_ticket": work["ticket"], + } + return json.dumps(workstatus) + + +def decode_json_args(task): + out = [] + for arg in task["args"]: + if "local" in arg and arg["local"] == webpb.Locality.REFERENCE.value: + getresp = get(b64loads(arg["reference_id"])) + data = webpb.loads(getresp["data"]) + out.append(data) + elif "local" not in arg or arg["local"] == webpb.Locality.INTERNED.value: + out.append(webpb.loads(arg["data"])) + else: + raise Exception("convert_from_arg: Uncovered locality enum") + return out diff --git a/worker.go b/worker.go index b1af58a..6c90d5f 100644 --- a/worker.go +++ b/worker.go @@ -2,9 +2,9 @@ package main import ( "errors" - "fmt" "github.com/barakmich/tinkerbell/ray_rpc" + "go.uber.org/zap" ) type WorkstreamConnection = ray_rpc.RayletWorkerConnection_WorkstreamServer @@ -13,6 +13,7 @@ type Worker interface { AssignWork(work *ray_rpc.Work) error Run() error Close() error + Schedulable() bool } type SimpleWorker struct { @@ -21,6 +22,10 @@ type SimpleWorker struct { pool WorkerPool } +func (s *SimpleWorker) Schedulable() bool { + return true +} + func (s *SimpleWorker) AssignWork(work *ray_rpc.Work) error { s.workChan <- work return nil @@ -39,13 +44,13 @@ func (w *SimpleWorker) Run() error { if sentinel.Status != ray_rpc.READY { return errors.New("Sent wrong sentinel? Closing...") } - fmt.Println("New worker:", sentinel.ErrorMsg) + zap.S().Info("New worker:", sentinel.ErrorMsg) go func() { for work := range w.workChan { - fmt.Println("sending work") + zap.S().Debug("Sending work") err = w.clientConn.Send(work) if err != nil { - fmt.Println("Error sending:", err) + zap.S().Error("Error sending:", err) return } } @@ -53,12 +58,12 @@ func (w *SimpleWorker) Run() error { for { result, err := w.clientConn.Recv() if err != nil { - fmt.Println("Error on channel:", err) + zap.S().Error("Error on channel:", err) return err } err = w.pool.Finish(result) if err != nil { - fmt.Println("Error finishing:", err) + zap.S().Error("Error finishing:", err) return err } } diff --git a/worker_pool.go b/worker_pool.go index ed1da4a..71cb137 100644 --- a/worker_pool.go +++ b/worker_pool.go @@ -2,7 +2,6 @@ package main import ( "errors" - "fmt" "sync" "github.com/barakmich/tinkerbell/ray_rpc" @@ -43,11 +42,23 @@ func (wp *SimpleRRWorkerPool) Schedule(work *ray_rpc.Work) error { if len(wp.workers) == 0 { return errors.New("No workers available, try again later") } - zap.S().Info("Sending work to worker", wp.offset) - wp.workers[wp.offset].AssignWork(work) - wp.offset++ - if wp.offset == len(wp.workers) { - wp.offset = 0 + origOffset := wp.offset + + done := false + for !done { + worker := wp.workers[wp.offset] + if worker.Schedulable() { + zap.S().Info("Sending work to worker", wp.offset) + wp.workers[wp.offset].AssignWork(work) + done = true + } + wp.offset++ + if wp.offset == len(wp.workers) { + wp.offset = 0 + } + if wp.offset == origOffset && !done { + return errors.New("No workers schedulable") + } } return nil } @@ -72,7 +83,6 @@ func (wp *SimpleRRWorkerPool) Close() error { func (wp *SimpleRRWorkerPool) Deregister(worker Worker) error { wp.Lock() defer wp.Unlock() - fmt.Println("Deregistering worker") found := false for i, w := range wp.workers { if w == worker { @@ -82,6 +92,7 @@ func (wp *SimpleRRWorkerPool) Deregister(worker Worker) error { } worker.Close() found = true + zap.S().Info("Deregistering worker", i) } } if !found {