diff --git a/services/document-updater/test/stress/coffee/run.js b/services/document-updater/test/stress/coffee/run.js index 2c48583014..351e726bbc 100644 --- a/services/document-updater/test/stress/coffee/run.js +++ b/services/document-updater/test/stress/coffee/run.js @@ -1,209 +1,284 @@ -DocUpdaterClient = require "../../acceptance/coffee/helpers/DocUpdaterClient" -# MockTrackChangesApi = require "../../acceptance/js/helpers/MockTrackChangesApi" -# MockWebApi = require "../../acceptance/js/helpers/MockWebApi" -assert = require "assert" -async = require "async" +/* + * decaffeinate suggestions: + * DS101: Remove unnecessary use of Array.from + * DS102: Remove unnecessary code created because of implicit returns + * DS202: Simplify dynamic range loops + * DS205: Consider reworking code to avoid use of IIFEs + * DS207: Consider shorter variations of null checks + * Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md + */ +const DocUpdaterClient = require("../../acceptance/coffee/helpers/DocUpdaterClient"); +// MockTrackChangesApi = require "../../acceptance/js/helpers/MockTrackChangesApi" +// MockWebApi = require "../../acceptance/js/helpers/MockWebApi" +const assert = require("assert"); +const async = require("async"); -insert = (string, pos, content) -> - result = string.slice(0, pos) + content + string.slice(pos) - return result +const insert = function(string, pos, content) { + const result = string.slice(0, pos) + content + string.slice(pos); + return result; +}; -transform = (op1, op2) -> - if op2.p < op1.p +const transform = function(op1, op2) { + if (op2.p < op1.p) { return { - p: op1.p + op2.i.length + p: op1.p + op2.i.length, i: op1.i - } - else - return op1 + }; + } else { + return op1; + } +}; -class StressTestClient - constructor: (@options = {}) -> - @options.updateDelay ?= 200 - @project_id = @options.project_id or DocUpdaterClient.randomId() - @doc_id = @options.doc_id or DocUpdaterClient.randomId() - @pos = @options.pos or 0 - @content = @options.content or "" +class StressTestClient { + constructor(options) { + if (options == null) { options = {}; } + this.options = options; + if (this.options.updateDelay == null) { this.options.updateDelay = 200; } + this.project_id = this.options.project_id || DocUpdaterClient.randomId(); + this.doc_id = this.options.doc_id || DocUpdaterClient.randomId(); + this.pos = this.options.pos || 0; + this.content = this.options.content || ""; - @client_id = DocUpdaterClient.randomId() - @version = @options.version or 0 - @inflight_op = null - @charCode = 0 + this.client_id = DocUpdaterClient.randomId(); + this.version = this.options.version || 0; + this.inflight_op = null; + this.charCode = 0; - @counts = { - conflicts: 0 - local_updates: 0 - remote_updates: 0 + this.counts = { + conflicts: 0, + local_updates: 0, + remote_updates: 0, max_delay: 0 - } + }; - DocUpdaterClient.subscribeToAppliedOps (channel, update) => - update = JSON.parse(update) - if update.error? - console.error new Error("Error from server: '#{update.error}'") - return - if update.doc_id == @doc_id - @processReply(update) - - sendUpdate: () -> - data = String.fromCharCode(65 + @charCode++ % 26) - @content = insert(@content, @pos, data) - @inflight_op = { - i: data - p: @pos++ - } - @resendUpdate() - @inflight_op_sent = Date.now() - - resendUpdate: () -> - assert(@inflight_op?) - DocUpdaterClient.sendUpdate( - @project_id, @doc_id - { - doc: @doc_id - op: [@inflight_op] - v: @version - meta: - source: @client_id - dupIfSource: [@client_id] + DocUpdaterClient.subscribeToAppliedOps((channel, update) => { + update = JSON.parse(update); + if (update.error != null) { + console.error(new Error(`Error from server: '${update.error}'`)); + return; } - ) - @update_timer = setTimeout () => - console.log "[#{new Date()}] \t[#{@client_id.slice(0,4)}] WARN: Resending update after 5 seconds" - @resendUpdate() - , 5000 + if (update.doc_id === this.doc_id) { + return this.processReply(update); + } + }); + } - processReply: (update) -> - if update.op.v != @version - if update.op.v < @version - console.log "[#{new Date()}] \t[#{@client_id.slice(0,4)}] WARN: Duplicate ack (already seen version)" - return - else - console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] ERROR: Version jumped ahead (client: #{@version}, op: #{update.op.v})" - @version++ - if update.op.meta.source == @client_id - if @inflight_op? - @counts.local_updates++ - @inflight_op = null - clearTimeout @update_timer - delay = Date.now() - @inflight_op_sent - @counts.max_delay = Math.max(@counts.max_delay, delay) - @continue() - else - console.log "[#{new Date()}] \t[#{@client_id.slice(0,4)}] WARN: Duplicate ack" - else - assert(update.op.op.length == 1) - @counts.remote_updates++ - external_op = update.op.op[0] - if @inflight_op? - @counts.conflicts++ - @inflight_op = transform(@inflight_op, external_op) - external_op = transform(external_op, @inflight_op) - if external_op.p < @pos - @pos += external_op.i.length - @content = insert(@content, external_op.p, external_op.i) + sendUpdate() { + const data = String.fromCharCode(65 + (this.charCode++ % 26)); + this.content = insert(this.content, this.pos, data); + this.inflight_op = { + i: data, + p: this.pos++ + }; + this.resendUpdate(); + return this.inflight_op_sent = Date.now(); + } - continue: () -> - if @updateCount > 0 - @updateCount-- - setTimeout () => - @sendUpdate() - , @options.updateDelay * ( 0.5 + Math.random() ) - else - @updateCallback() - - runForNUpdates: (n, callback = (error) ->) -> - @updateCallback = callback - @updateCount = n - @continue() - - check: (callback = (error) ->) -> - DocUpdaterClient.getDoc @project_id, @doc_id, (error, res, body) => - throw error if error? - if !body.lines? - return console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] ERROR: Invalid response from get doc (#{doc_id})", body - content = body.lines.join("\n") - version = body.version - if content != @content - if version == @version - console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Error: Client content does not match server." - console.error "Server: #{content.split('a')}" - console.error "Client: #{@content.split('a')}" - else - console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Error: Version mismatch (Server: '#{version}', Client: '#{@version}')" - - if !@isContentValid(@content) - for chunk, i in @content.split("") - if chunk? and chunk != "a" - console.log chunk, i - throw new Error("bad content") - callback() - - isChunkValid: (chunk) -> - char = 0 - for letter, i in chunk - if letter.charCodeAt(0) != 65 + i % 26 - console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Invalid Chunk:", chunk - return false - return true - - isContentValid: (content) -> - for chunk in content.split('a') - if chunk? and chunk != "" - if !@isChunkValid(chunk) - - console.error "[#{new Date()}] \t[#{@client_id.slice(0,4)}] Invalid content", content - return false - return true - - -checkDocument = (project_id, doc_id, clients, callback = (error) ->) -> - jobs = clients.map (client) -> - (cb) -> client.check cb - async.parallel jobs, callback - -printSummary = (doc_id, clients) -> - slot = require('cluster-key-slot') - now = new Date() - console.log "[#{now}] [#{doc_id.slice(0,4)} (slot: #{slot(doc_id)})] #{clients.length} clients..." - for client in clients - console.log "[#{now}] \t[#{client.client_id.slice(0,4)}] { local: #{client.counts.local_updates }, remote: #{client.counts.remote_updates}, conflicts: #{client.counts.conflicts}, max_delay: #{client.counts.max_delay} }" - client.counts = { - local_updates: 0 - remote_updates: 0 - conflicts: 0 - max_delay: 0 + resendUpdate() { + assert(this.inflight_op != null); + DocUpdaterClient.sendUpdate( + this.project_id, this.doc_id, + { + doc: this.doc_id, + op: [this.inflight_op], + v: this.version, + meta: { + source: this.client_id + }, + dupIfSource: [this.client_id] + } + ); + return this.update_timer = setTimeout(() => { + console.log(`[${new Date()}] \t[${this.client_id.slice(0,4)}] WARN: Resending update after 5 seconds`); + return this.resendUpdate(); } + , 5000); + } + + processReply(update) { + if (update.op.v !== this.version) { + if (update.op.v < this.version) { + console.log(`[${new Date()}] \t[${this.client_id.slice(0,4)}] WARN: Duplicate ack (already seen version)`); + return; + } else { + console.error(`[${new Date()}] \t[${this.client_id.slice(0,4)}] ERROR: Version jumped ahead (client: ${this.version}, op: ${update.op.v})`); + } + } + this.version++; + if (update.op.meta.source === this.client_id) { + if (this.inflight_op != null) { + this.counts.local_updates++; + this.inflight_op = null; + clearTimeout(this.update_timer); + const delay = Date.now() - this.inflight_op_sent; + this.counts.max_delay = Math.max(this.counts.max_delay, delay); + return this.continue(); + } else { + return console.log(`[${new Date()}] \t[${this.client_id.slice(0,4)}] WARN: Duplicate ack`); + } + } else { + assert(update.op.op.length === 1); + this.counts.remote_updates++; + let external_op = update.op.op[0]; + if (this.inflight_op != null) { + this.counts.conflicts++; + this.inflight_op = transform(this.inflight_op, external_op); + external_op = transform(external_op, this.inflight_op); + } + if (external_op.p < this.pos) { + this.pos += external_op.i.length; + } + return this.content = insert(this.content, external_op.p, external_op.i); + } + } + + continue() { + if (this.updateCount > 0) { + this.updateCount--; + return setTimeout(() => { + return this.sendUpdate(); + } + , this.options.updateDelay * ( 0.5 + Math.random() )); + } else { + return this.updateCallback(); + } + } + + runForNUpdates(n, callback) { + if (callback == null) { callback = function(error) {}; } + this.updateCallback = callback; + this.updateCount = n; + return this.continue(); + } + + check(callback) { + if (callback == null) { callback = function(error) {}; } + return DocUpdaterClient.getDoc(this.project_id, this.doc_id, (error, res, body) => { + if (error != null) { throw error; } + if ((body.lines == null)) { + return console.error(`[${new Date()}] \t[${this.client_id.slice(0,4)}] ERROR: Invalid response from get doc (${doc_id})`, body); + } + const content = body.lines.join("\n"); + const { + version + } = body; + if (content !== this.content) { + if (version === this.version) { + console.error(`[${new Date()}] \t[${this.client_id.slice(0,4)}] Error: Client content does not match server.`); + console.error(`Server: ${content.split('a')}`); + console.error(`Client: ${this.content.split('a')}`); + } else { + console.error(`[${new Date()}] \t[${this.client_id.slice(0,4)}] Error: Version mismatch (Server: '${version}', Client: '${this.version}')`); + } + } -CLIENT_COUNT = parseInt(process.argv[2], 10) -UPDATE_DELAY = parseInt(process.argv[3], 10) -SAMPLE_INTERVAL = parseInt(process.argv[4], 10) + if (!this.isContentValid(this.content)) { + const iterable = this.content.split(""); + for (let i = 0; i < iterable.length; i++) { + const chunk = iterable[i]; + if ((chunk != null) && (chunk !== "a")) { + console.log(chunk, i); + } + } + throw new Error("bad content"); + } + return callback(); + }); + } -for doc_and_project_id in process.argv.slice(5) - do (doc_and_project_id) -> - [project_id, doc_id] = doc_and_project_id.split(":") - console.log {project_id, doc_id} - DocUpdaterClient.setDocLines project_id, doc_id, [(new Array(CLIENT_COUNT + 2)).join('a')], null, null, (error) -> - throw error if error? - DocUpdaterClient.getDoc project_id, doc_id, (error, res, body) => - throw error if error? - if !body.lines? - return console.error "[#{new Date()}] ERROR: Invalid response from get doc (#{doc_id})", body - content = body.lines.join("\n") - version = body.version + isChunkValid(chunk) { + const char = 0; + for (let i = 0; i < chunk.length; i++) { + const letter = chunk[i]; + if (letter.charCodeAt(0) !== (65 + (i % 26))) { + console.error(`[${new Date()}] \t[${this.client_id.slice(0,4)}] Invalid Chunk:`, chunk); + return false; + } + } + return true; + } + + isContentValid(content) { + for (let chunk of Array.from(content.split('a'))) { + if ((chunk != null) && (chunk !== "")) { + if (!this.isChunkValid(chunk)) { + + console.error(`[${new Date()}] \t[${this.client_id.slice(0,4)}] Invalid content`, content); + return false; + } + } + } + return true; + } +} + + +const checkDocument = function(project_id, doc_id, clients, callback) { + if (callback == null) { callback = function(error) {}; } + const jobs = clients.map(client => cb => client.check(cb)); + return async.parallel(jobs, callback); +}; + +const printSummary = function(doc_id, clients) { + const slot = require('cluster-key-slot'); + const now = new Date(); + console.log(`[${now}] [${doc_id.slice(0,4)} (slot: ${slot(doc_id)})] ${clients.length} clients...`); + return (() => { + const result = []; + for (let client of Array.from(clients)) { + console.log(`[${now}] \t[${client.client_id.slice(0,4)}] { local: ${client.counts.local_updates }, remote: ${client.counts.remote_updates}, conflicts: ${client.counts.conflicts}, max_delay: ${client.counts.max_delay} }`); + result.push(client.counts = { + local_updates: 0, + remote_updates: 0, + conflicts: 0, + max_delay: 0 + }); + } + return result; + })(); +}; + +const CLIENT_COUNT = parseInt(process.argv[2], 10); +const UPDATE_DELAY = parseInt(process.argv[3], 10); +const SAMPLE_INTERVAL = parseInt(process.argv[4], 10); + +for (let doc_and_project_id of Array.from(process.argv.slice(5))) { + (function(doc_and_project_id) { + const [project_id, doc_id] = Array.from(doc_and_project_id.split(":")); + console.log({project_id, doc_id}); + return DocUpdaterClient.setDocLines(project_id, doc_id, [(new Array(CLIENT_COUNT + 2)).join('a')], null, null, function(error) { + if (error != null) { throw error; } + return DocUpdaterClient.getDoc(project_id, doc_id, (error, res, body) => { + let runBatch; + if (error != null) { throw error; } + if ((body.lines == null)) { + return console.error(`[${new Date()}] ERROR: Invalid response from get doc (${doc_id})`, body); + } + const content = body.lines.join("\n"); + const { + version + } = body; - clients = [] - for pos in [1..CLIENT_COUNT] - do (pos) -> - client = new StressTestClient({doc_id, project_id, content, pos: pos, version: version, updateDelay: UPDATE_DELAY}) - clients.push client + const clients = []; + for (let pos = 1, end = CLIENT_COUNT, asc = 1 <= end; asc ? pos <= end : pos >= end; asc ? pos++ : pos--) { + (function(pos) { + const client = new StressTestClient({doc_id, project_id, content, pos, version, updateDelay: UPDATE_DELAY}); + return clients.push(client); + })(pos); + } - do runBatch = () -> - jobs = clients.map (client) -> - (cb) -> client.runForNUpdates(SAMPLE_INTERVAL / UPDATE_DELAY, cb) - async.parallel jobs, (error) -> - throw error if error? - printSummary(doc_id, clients) - checkDocument project_id, doc_id, clients, (error) -> - throw error if error? - runBatch() + return (runBatch = function() { + const jobs = clients.map(client => cb => client.runForNUpdates(SAMPLE_INTERVAL / UPDATE_DELAY, cb)); + return async.parallel(jobs, function(error) { + if (error != null) { throw error; } + printSummary(doc_id, clients); + return checkDocument(project_id, doc_id, clients, function(error) { + if (error != null) { throw error; } + return runBatch(); + }); + }); + })(); + }); + }); + })(doc_and_project_id); +}