From a34e514e4ef4faa46af227d45c13ecea8a1d3143 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 18 Oct 2022 23:16:58 +0200 Subject: [PATCH] migrating benchmark to deno client 1.38.5 --- benchmarks/main.ts | 60 ++++++++++++---------------- benchmarks/worker.ts | 94 +++++++++++++++++++++----------------------- 2 files changed, 71 insertions(+), 83 deletions(-) diff --git a/benchmarks/main.ts b/benchmarks/main.ts index 7d81281593..2f3a106ab4 100644 --- a/benchmarks/main.ts +++ b/benchmarks/main.ts @@ -3,17 +3,17 @@ import { Command } from "https://deno.land/x/cliffy@v0.25.2/command/mod.ts"; import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts"; -import * as windmill from "https://deno.land/x/windmill@v1.37.0/mod.ts"; -import * as api from "https://deno.land/x/windmill@v1.37.0/windmill-api/index.ts"; +import * as windmill from "https://deno.land/x/windmill@v1.38.5/mod.ts"; async function login( - config: api.Configuration, email: string, password: string ): Promise { - return await new windmill.UserApi(config).login({ - email: email, - password: password, + return await windmill.UserService.login({ + requestBody: { + email: email, + password: password, + } }); } @@ -131,6 +131,7 @@ await new Command() useFlows, zombieTimeout, }) => { + windmill.setClient("", host); const metrics_worker = new Worker( new URL("./scraper.ts", import.meta.url).href, { @@ -173,20 +174,17 @@ await new Command() zombieTimeout, }, null, 4)); console.log("collecting samples..."); - host = host.endsWith("/") ? host.substring(0, host.length - 1) : host; - host = `${host}/api`; - let config = { - ...api.createConfiguration({ - baseServer: new api.ServerConfiguration(host, {}), - }), + const config = { + token: "", + server: host, workspace_id: workspace, }; let final_token: string; if (!token) { if (email && password) { - final_token = await login(config, email, password); + final_token = await login(email, password); } else { console.error("Token or email with password are required."); return; @@ -195,21 +193,8 @@ await new Command() final_token = token; } - config = { - ...api.createConfiguration({ - baseServer: config.baseServer, - authMethods: { - bearerAuth: { - tokenProvider: { - getToken() { - return final_token; - }, - }, - }, - }, - }), - workspace_id: config.workspace_id, - }; + config.token = final_token; + windmill.setClient(final_token, host); const per_worker_throughput = maximumThroughput / num_workers; const shared_config = { @@ -230,19 +215,21 @@ await new Command() let start: number | undefined = undefined; - let jobSent = Array(num_workers).fill(0); + const jobsSent = Array(num_workers).fill(0); const enc = (s: string) => new TextEncoder().encode(s); - const i = setInterval(async () => { + const updateState = setInterval(async () => { const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0; - await Deno.stdout.write(enc(`elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(jobSent)}\r`)) + const sum = jobsSent.reduce((a, b) => a + b, 0) + const queue_length = (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length + await Deno.stdout.write(enc(`elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(jobsSent)} (sum: ${sum}) | queue: ${queue_length} \r`)) }, 100); workers.forEach((worker, i) => { worker.postMessage({ ...shared_config, i }); worker.addEventListener("message", (evt: MessageEvent) => { if (evt.data.type === "jobs_sent") { - jobSent[i] = evt.data.jobs_sent + jobsSent[i] = evt.data.jobs_sent } }) }); @@ -250,14 +237,17 @@ await new Command() await sleep(seconds); - clearInterval(i); - await Deno.stdout.write(enc(` \rduration: ${seconds} | jobs sent: ${JSON.stringify(jobSent)}\n`)); + clearInterval(updateState); + + await Deno.stdout.write(enc(" ".padStart(30) + `\rduration: ${seconds} | jobs sent: ${JSON.stringify(jobsSent)}\n`)); let zombie_jobs = 0; + let incorrect_results = 0; workers.forEach((worker) => { const l = (evt: MessageEvent) => { if (evt.data.type === "zombie_jobs") { zombie_jobs += evt.data.zombie_jobs; + incorrect_results += evt.data.incorrect_results; worker.removeEventListener("message", l); workers = workers.filter((w) => w != worker); worker.terminate(); @@ -275,6 +265,8 @@ await new Command() } console.log("zombie jobs: ", zombie_jobs); + console.log("incorrect results: ", incorrect_results); + console.log("queue length:", (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length) metrics_worker.postMessage("stop"); console.log("waiting for metrics"); diff --git a/benchmarks/worker.ts b/benchmarks/worker.ts index 78e926e1f5..53b2400b31 100644 --- a/benchmarks/worker.ts +++ b/benchmarks/worker.ts @@ -1,11 +1,11 @@ /// /// import { sleep } from "https://deno.land/x/sleep@v1.2.1/sleep.ts"; -import * as windmill from "https://deno.land/x/windmill@v1.37.0/mod.ts"; -import * as api from "https://deno.land/x/windmill@v1.37.0/windmill-api/index.ts"; +import * as windmill from "https://deno.land/x/windmill@v1.38.5/mod.ts"; +import * as api from "https://deno.land/x/windmill@v1.38.5/windmill-api/index.ts"; const promise = new Promise< - api.Configuration & { + { workspace_id: string; per_worker_throughput: number; useFlows: boolean; @@ -13,19 +13,8 @@ const promise = new Promise< >((resolve, _reject) => { self.onmessage = (evt) => { const sharedConfig = evt.data; + windmill.setClient(sharedConfig.token, sharedConfig.server) const config = { - ...api.createConfiguration({ - baseServer: new api.ServerConfiguration(sharedConfig.server, {}), - authMethods: { - bearerAuth: { - tokenProvider: { - getToken() { - return sharedConfig.token; - }, - }, - }, - }, - }), workspace_id: sharedConfig.workspace_id, per_worker_throughput: sharedConfig.per_worker_throughput, useFlows: sharedConfig.useFlows, @@ -36,13 +25,11 @@ const promise = new Promise< }; }); const config = await promise; -const jobApi = new windmill.JobApi(config); const outstanding: string[] = []; let cont = true; let total_spawned = 0; const start_time = Date.now(); let complete_timeout = Infinity; - self.onmessage = (evt) => { cont = false; complete_timeout = evt.data; @@ -54,9 +41,11 @@ const updateStatusInterval = setInterval(() => { }, 100) + while (cont) { - if ((await jobApi.listQueue(config.workspace_id)).length > 500) { - console.log("queue very long. waiting..."); + const queue_length = (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length + if (queue_length > 500) { + console.log(`queue length: ${queue_length} > 500. waiting...`); await sleep(0.5); continue; } @@ -70,36 +59,40 @@ while (cont) { } let uuid: string; if (config.useFlows) { - uuid = await jobApi.runFlowPreview(config.workspace_id, { - args: {}, - value: { - modules: [ - { - inputTransforms: {}, - value: { - language: "deno", - type: "rawscript", - content: - 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + uuid = await windmill.JobService.runFlowPreview({ + workspace: config.workspace_id, requestBody: { + args: {}, + value: { + modules: [ + { + input_transforms: {}, + value: { + language: api.RawScript.language.DENO, + type: "rawscript", + content: + 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + }, }, - }, - { - inputTransforms: {}, - value: { - language: "deno", - type: "rawscript", - content: - 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + { + input_transforms: {}, + value: { + language: api.RawScript.language.DENO, + type: "rawscript", + content: + 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + }, }, - }, - ], - }, + ], + }, + } }); } else { - uuid = await jobApi.runScriptPreview(config.workspace_id, { - language: "deno", - content: 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', - args: {}, + uuid = await windmill.JobService.runScriptPreview({ + workspace: config.workspace_id, requestBody: { + language: api.Preview.language.DENO, + content: 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + args: {}, + } }); } outstanding.push(uuid); @@ -109,18 +102,21 @@ while (cont) { clearInterval(updateStatusInterval); const end_time = Date.now() + complete_timeout; +let incorrect_results = 0; while (outstanding.length > 0 && Date.now() < end_time) { const uuid = outstanding.shift()!; - const r = await jobApi.getJob(config.workspace_id, uuid); + let r = await windmill.JobService.getJob({ workspace: config.workspace_id, id: uuid }); if (r.type == 'QueuedJob') { outstanding.push(uuid); - console.log(uuid) + console.log(uuid, (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length) } else if (!config.useFlows) { + r = r as api.CompletedJob; try { if (r.result != uuid) { console.log( - "job did not return correct UUID: " + r.result + " != " + uuid + "job did not return correct UUID: " + r.result + " != " + uuid + "job: \n" + JSON.stringify(r, null, 2), ); + incorrect_results++; } } catch (e) { console.log("error during wait: ", e); @@ -129,4 +125,4 @@ while (outstanding.length > 0 && Date.now() < end_time) { } } -self.postMessage({ type: "zombie_jobs", zombie_jobs: outstanding.length }); +self.postMessage({ type: "zombie_jobs", zombie_jobs: outstanding.length, incorrect_results });