/** SD-UI Backend control and classes. */ (function () { "use strict"; const RETRY_DELAY_IF_BUFFER_IS_EMPTY = 1000 // ms const RETRY_DELAY_IF_SERVER_IS_BUSY = 30 * 1000 // ms, status_code 503, already a task running const RETRY_DELAY_ON_ERROR = 4000 // ms const TASK_STATE_SERVER_UPDATE_DELAY = 1500 // ms const SERVER_STATE_VALIDITY_DURATION = 90 * 1000 // ms - 90 seconds to allow ping to timeout more than once before killing tasks. const HEALTH_PING_INTERVAL = 5000 // ms const IDLE_COOLDOWN = 2500 // ms const CONCURRENT_TASK_INTERVAL = 100 // ms /** Connects to an endpoint and resumes connection after reaching end of stream until all data is received. * Allows closing the connection while the server buffers more data. */ class ChunkedStreamReader { #bufferedString = '' // Data received waiting to be read. #url #fetchOptions #response constructor(url, initialContent='', options={}) { if (typeof url !== 'string' && !(url instanceof String)) { throw new Error('url is not a string.') } if (typeof initialContent !== 'undefined' && typeof initialContent !== 'string') { throw new Error('initialContent is not a string.') } this.#bufferedString = initialContent this.#url = url this.#fetchOptions = Object.assign({ headers: { 'Content-Type': 'application/json' } }, options) this.onNext = undefined } get url() { if (this.#response.redirected) { return this.#response.url } return this.#url } get bufferedString() { return this.#bufferedString } get status() { this.#response?.status } get statusText() { this.#response?.statusText } parse(value) { if (typeof value === 'undefined') { return } if (!isArrayOrTypedArray(value)) { return [value] } if (value.length === 0) { return value } if (typeof this.textDecoder === 'undefined') { this.textDecoder = new TextDecoder() } return [this.textDecoder.decode(value)] } onComplete(value) { return value } onError(response) { throw new Error(response.statusText) } onNext({value, done}, response) { return {value, done} } async *[Symbol.asyncIterator]() { return this.open() } async *open() { let value = undefined let done = undefined do { if (this.#response) { await asyncDelay(RETRY_DELAY_IF_BUFFER_IS_EMPTY) } this.#response = await fetch(this.#url, this.#fetchOptions) if (!this.#response.ok) { if (this.#response.status === 425) { continue } // Request status indicate failure console.warn('Stream %o stopped unexpectedly.', this.#response) value = await Promise.resolve(this.onError(this.#response)) if (typeof value === 'boolean' && value) { continue } return value } const reader = this.#response.body.getReader() done = false do { const readState = await reader.read() value = this.parse(readState.value) if (value) { for(let sVal of value) { ({value: sVal, done} = await Promise.resolve(this.onNext({value:sVal, done:readState.done}))) yield sVal if (done) { return this.onComplete(sVal) } } } if (done) { return } } while(value && !done) } while (!done && (this.#response.ok || this.#response.status === 425)) } *readStreamAsJSON(jsonStr, throwOnError) { if (typeof jsonStr !== 'string') { throw new Error('jsonStr is not a string.') } do { if (this.#bufferedString.length > 0) { // Append new data when required if (jsonStr.length > 0) { jsonStr = this.#bufferedString + jsonStr } else { jsonStr = this.#bufferedString } this.#bufferedString = '' } if (!jsonStr) { return } // Find next delimiter let lastChunkIdx = jsonStr.indexOf('}{') if (lastChunkIdx >= 0) { this.#bufferedString = jsonStr.substring(0, lastChunkIdx + 1) jsonStr = jsonStr.substring(lastChunkIdx + 1) } else { this.#bufferedString = jsonStr jsonStr = '' } if (this.#bufferedString.length <= 0) { return } // hack for a middleman buffering all the streaming updates, and unleashing them on the poor browser in one shot. // this results in having to parse JSON like {"step": 1}{"step": 2}{"step": 3}{"ste... // which is obviously invalid and can happen at any point while rendering. // So we need to extract only the next {} section try { // Try to parse const jsonObj = JSON.parse(this.#bufferedString) this.#bufferedString = jsonStr jsonStr = '' yield jsonObj } catch (e) { if (throwOnError) { console.error(`Parsing: "${this.#bufferedString}", Buffer: "${jsonStr}"`) } this.#bufferedString += jsonStr if (e instanceof SyntaxError && !throwOnError) { return } throw e } } while (this.#bufferedString.length > 0 && this.#bufferedString.indexOf('}') >= 0) } } const EVENT_IDLE = 'idle' const EVENT_STATUS_CHANGED = 'statusChange' const EVENT_UNHANDLED_REJECTION = 'unhandledRejection' const EVENT_TASK_QUEUED = 'taskQueued' const EVENT_TASK_START = 'taskStart' const EVENT_TASK_END = 'taskEnd' const EVENT_TASK_ERROR = 'task_error' const EVENT_UNEXPECTED_RESPONSE = 'unexpectedResponse' const EVENTS_TYPES = [ EVENT_IDLE, EVENT_STATUS_CHANGED, EVENT_UNHANDLED_REJECTION, EVENT_TASK_QUEUED, EVENT_TASK_START, EVENT_TASK_END, EVENT_TASK_ERROR, EVENT_UNEXPECTED_RESPONSE, ] Object.freeze(EVENTS_TYPES) const eventSource = new GenericEventSource(EVENTS_TYPES) function setServerStatus(msgType, msg) { return eventSource.fireEvent(EVENT_STATUS_CHANGED, {type: msgType, message: msg}) } const ServerStates = { init: 'Init', loadingModel: 'LoadingModel', online: 'Online', rendering: 'Rendering', unavailable: 'Unavailable', } Object.freeze(ServerStates) let sessionId = Date.now() let serverState = {'status': ServerStates.unavailable, 'time': Date.now()} async function healthCheck() { if (Date.now() < serverState.time + (HEALTH_PING_INTERVAL / 2) && isServerAvailable()) { // Ping confirmed online less than half of HEALTH_PING_INTERVAL ago. return true } if (Date.now() >= serverState.time + SERVER_STATE_VALIDITY_DURATION) { console.warn('WARNING! SERVER_STATE_VALIDITY_DURATION has elapsed since the last Ping completed.') } try { let res = undefined if (typeof sessionId !== 'undefined') { res = await fetch('/ping?session_id=' + sessionId) } else { res = await fetch('/ping') } serverState = await res.json() if (typeof serverState !== 'object' || typeof serverState.status !== 'string') { console.error(`Server reply didn't contain a state value.`) serverState = {'status': ServerStates.unavailable, 'time': Date.now()} setServerStatus('error', 'offline') return false } // Set status switch(serverState.status) { case ServerStates.init: // Wait for init to complete before updating status. break case ServerStates.online: setServerStatus('online', 'ready') break case ServerStates.loadingModel: setServerStatus('busy', 'loading..') break case ServerStates.rendering: setServerStatus('busy', 'rendering..') break default: // Unavailable console.error('Ping received an unexpected server status. Status: %s', serverState.status) setServerStatus('error', serverState.status.toLowerCase()) break } serverState.time = Date.now() return true } catch (e) { console.error(e) serverState = {'status': ServerStates.unavailable, 'time': Date.now()} setServerStatus('error', 'offline') } return false } function isServerAvailable() { if (typeof serverState !== 'object') { console.error('serverState not set to a value. Connection to server could be lost...') return false } if (Date.now() >= serverState.time + SERVER_STATE_VALIDITY_DURATION) { console.warn('SERVER_STATE_VALIDITY_DURATION elapsed. Connection to server could be lost...') return false } switch (serverState.status) { case ServerStates.loadingModel: case ServerStates.rendering: case ServerStates.online: return true default: console.warn('Unexpected server status. Server could be unavailable... Status: %s', serverState.status) return false } } async function waitUntil(isReadyFn, delay, timeout) { if (typeof delay === 'number') { const msDelay = delay delay = () => asyncDelay(msDelay) } if (typeof delay !== 'function') { throw new Error('delay is not a number or a function.') } if (typeof timeout !== 'undefined' && typeof timeout !== 'number') { throw new Error('timeout is not a number.') } if (typeof timeout === 'undefined' || timeout < 0) { timeout = Number.MAX_SAFE_INTEGER } timeout = Date.now() + timeout while (timeout > Date.now() && Date.now() < serverState.time + SERVER_STATE_VALIDITY_DURATION && !Boolean(await Promise.resolve(isReadyFn())) ) { await delay() if (!isServerAvailable()) { // Can fail if ping got frozen/suspended... if (await healthCheck() && isServerAvailable()) { // Force a recheck of server status before failure... continue // Continue waiting if last healthCheck confirmed the server is still alive. } throw new Error('Connection with server lost.') } } if (Date.now() >= serverState.time + SERVER_STATE_VALIDITY_DURATION) { console.warn('SERVER_STATE_VALIDITY_DURATION elapsed. Released waitUntil on stale server state.') } } const TaskStatus = { init: 'init', pending: 'pending', // Queued locally, not yet posted to server waiting: 'waiting', // Waiting to run on server processing: 'processing', stopped: 'stopped', completed: 'completed', failed: 'failed', } Object.freeze(TaskStatus) const TASK_STATUS_ORDER = [ TaskStatus.init, TaskStatus.pending, TaskStatus.waiting, TaskStatus.processing, //Don't add status that are final. ] const task_queue = new Map() const concurrent_generators = new Map() const weak_results = new WeakMap() class Task { // Private properties... _reqBody = {} // request body of this task. #reader = undefined #status = TaskStatus.init #id = undefined #exception = undefined constructor(options={}) { this._reqBody = Object.assign({}, options) if (typeof this._reqBody.session_id === 'undefined') { this._reqBody.session_id = sessionId } else if (this._reqBody.session_id !== SD.sessionId && String(this._reqBody.session_id) !== String(SD.sessionId)) { throw new Error('Use SD.sessionId to set the request session_id.') } this._reqBody.session_id = String(this._reqBody.session_id) } get id() { return this.#id } _setId(id) { if (typeof this.#id !== 'undefined') { throw new Error('The task ID can only be set once.') } this.#id = id } get exception() { return this.#exception } async abort(exception) { if (this.isCompleted || this.isStopped || this.hasFailed) { return } if (typeof exception !== 'undefined') { if (typeof exception === 'string') { exception = new Error(exception) } if (typeof exception !== 'object') { throw new Error('exception is not an object.') } if (!(exception instanceof Error)) { throw new Error('exception is not an Error or a string.') } } const res = await fetch('/image/stop?task=' + this.id) if (!res.ok) { console.log('Stop response:', res) throw new Error(res.statusText) } task_queue.delete(this) this.#exception = exception this.#status = (exception ? TaskStatus.failed : TaskStatus.stopped) } get reqBody() { if (this.#status === TaskStatus.init) { return this._reqBody } console.warn('Task reqBody cannot be changed after the init state.') return Object.assign({}, this._reqBody) } get isPending() { return TASK_STATUS_ORDER.indexOf(this.#status) >= 0 } get isCompleted() { return this.#status === TaskStatus.completed } get hasFailed() { return this.#status === TaskStatus.failed } get isStopped() { return this.#status === TaskStatus.stopped } get status() { return this.#status } _setStatus(status) { if (status === this.#status) { return } const currentIdx = TASK_STATUS_ORDER.indexOf(this.#status) if (currentIdx < 0) { throw Error(`The task status ${this.#status} is final and can't be changed.`) } const newIdx = TASK_STATUS_ORDER.indexOf(status) if (newIdx >= 0 && newIdx < currentIdx) { throw Error(`The task status ${status} can't replace ${this.#status}.`) } this.#status = status } /** Send current task to server. * @param {*} [timeout=-1] Optional timeout value in ms * @returns the response from the render request. * @memberof Task */ async post(url, timeout=-1) { if(this.status !== TaskStatus.init && this.status !== TaskStatus.pending) { throw new Error(`Task status ${this.status} is not valid for post.`) } this._setStatus(TaskStatus.pending) Object.freeze(this._reqBody) const abortSignal = (timeout >= 0 ? AbortSignal.timeout(timeout) : undefined) let res = undefined try { this.checkReqBody() do { abortSignal?.throwIfAborted() res = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(this._reqBody), signal: abortSignal }) // status_code 503, already a task running. } while (res.status === 503 && await asyncDelay(RETRY_DELAY_IF_SERVER_IS_BUSY)) } catch (err) { this.abort(err) throw err } if (!res.ok) { const err = new Error(`Unexpected response HTTP${res.status}. Details: ${res.statusText}`) this.abort(err) throw err } return await res.json() } static getReader(url) { const reader = new ChunkedStreamReader(url) const parseToString = reader.parse reader.parse = function(value) { value = parseToString.call(this, value) if (!value || value.length <= 0) { return } return reader.readStreamAsJSON(value.join('')) } reader.onNext = function({done, value}) { // By default is completed when the return value has a status defined. if (typeof value === 'object' && 'status' in value) { done = true } return {done, value} } return reader } _setReader(reader) { if (typeof this.#reader !== 'undefined') { throw new Error('The task reader can only be set once.') } this.#reader = reader } get reader() { if (this.#reader) { return this.#reader } if (!this.streamUrl) { throw new Error('The task has no stream Url defined.') } this.#reader = Task.getReader(this.streamUrl) const task = this const onNext = this.#reader.onNext this.#reader.onNext = function({done, value}) { if (value && typeof value === 'object') { if (task.status === TaskStatus.init || task.status === TaskStatus.pending || task.status === TaskStatus.waiting ) { task._setStatus(TaskStatus.processing) } if ('step' in value && 'total_steps' in value) { task.step = value.step task.total_steps = value.total_steps } } return onNext.call(this, {done, value}) } this.#reader.onComplete = function(value) { task.result = value if (task.isPending) { task._setStatus(TaskStatus.completed) } return value } this.#reader.onError = function(response) { const err = new Error(response.statusText) task.abort(err) throw err } return this.#reader } async waitUntil({timeout=-1, callback, status, signal}) { const currentIdx = TASK_STATUS_ORDER.indexOf(this.#status) if (currentIdx <= 0) { return false } const stIdx = (status ? TASK_STATUS_ORDER.indexOf(status) : currentIdx + 1) if (stIdx >= 0 && stIdx <= currentIdx) { return true } if (stIdx < 0 && currentIdx < 0) { return this.#status === (status || TaskStatus.completed) } if (signal?.aborted) { return false } const task = this switch(this.#status) { case TaskStatus.pending: case TaskStatus.waiting: // Wait for server status to include this task. await waitUntil( async () => { if (task.#id && typeof serverState.tasks === 'object' && Object.keys(serverState.tasks).includes(String(task.#id))) { return true } if (await Promise.resolve(callback?.call(task)) || signal?.aborted) { return true } }, TASK_STATE_SERVER_UPDATE_DELAY, timeout, ) if (this.#id && typeof serverState.tasks === 'object' && Object.keys(serverState.tasks).includes(String(task.#id))) { this._setStatus(TaskStatus.waiting) } if (await Promise.resolve(callback?.call(this)) || signal?.aborted) { return false } if (stIdx >= 0 && stIdx <= TASK_STATUS_ORDER.indexOf(TaskStatus.waiting)) { return true } // Wait for task to start on server. await waitUntil( async () => { if (typeof serverState.tasks !== 'object' || serverState.tasks[String(task.#id)] !== 'pending') { return true } if (await Promise.resolve(callback?.call(task)) || signal?.aborted) { return true } }, TASK_STATE_SERVER_UPDATE_DELAY, timeout, ) const state = (typeof serverState.tasks === 'object' ? serverState.tasks[String(task.#id)] : undefined) if (state === 'running' || state === 'buffer' || state === 'completed') { this._setStatus(TaskStatus.processing) } if (await Promise.resolve(callback?.call(task)) || signal?.aborted) { return false } if (stIdx >= 0 && stIdx <= TASK_STATUS_ORDER.indexOf(TaskStatus.processing)) { return true } case TaskStatus.processing: await waitUntil( async () => { if (typeof serverState.tasks !== 'object' || serverState.tasks[String(task.#id)] !== 'running') { return true } if (await Promise.resolve(callback?.call(task)) || signal?.aborted) { return true } }, TASK_STATE_SERVER_UPDATE_DELAY, timeout, ) await Promise.resolve(callback?.call(this)) default: return this.#status === (status || TaskStatus.completed) } } async enqueue(promiseGenerator, ...args) { if (this.status !== TaskStatus.init) { throw new Error(`Task is in an invalid status ${this.status} to add to queue.`) } this._setStatus(TaskStatus.pending) task_queue.set(this, promiseGenerator) await eventSource.fireEvent(EVENT_TASK_QUEUED, {task:this}) await Task.enqueue(promiseGenerator, ...args) await this.waitUntil({status: TaskStatus.completed}) if (this.exception) { throw this.exception } return this.result } static async enqueue(promiseGenerator, ...args) { if (typeof promiseGenerator === 'undefined') { throw new Error('To enqueue a concurrent task, a *Promise Generator is needed but undefined was found.') } //if (Symbol.asyncIterator in result || Symbol.iterator in result) { //concurrent_generators.set(result, Promise.resolve(args)) if (typeof promiseGenerator === 'function') { concurrent_generators.set(asGenerator({callback: promiseGenerator}), Promise.resolve(args)) } else { concurrent_generators.set(promiseGenerator, Promise.resolve(args)) } await waitUntil(() => !concurrent_generators.has(promiseGenerator), CONCURRENT_TASK_INTERVAL) return weak_results.get(promiseGenerator) } static enqueueNew(task, classCtor, progressCallback) { if (task.status !== TaskStatus.init) { throw new Error('Task has an invalid status to add to queue.') } if (!(task instanceof classCtor)) { throw new Error('Task is not a instance of classCtor.') } let promiseGenerator = undefined if (typeof progressCallback === 'undefined') { promiseGenerator = classCtor.start(task) } else if (typeof progressCallback === 'function') { promiseGenerator = classCtor.start(task, progressCallback) } else { throw new Error('progressCallback is not a function.') } return Task.prototype.enqueue.call(task, promiseGenerator) } static async run(promiseGenerator, {callback, signal, timeout=-1}={}) { let value = undefined let done = undefined if (timeout < 0) { timeout = Number.MAX_SAFE_INTEGER } timeout = Date.now() + timeout do { ({value, done} = await Promise.resolve(promiseGenerator.next(value))) if (value instanceof Promise) { value = await value } if (callback) { ({value, done} = await Promise.resolve(callback.call(promiseGenerator, {value, done}))) } if (value instanceof Promise) { value = await value } } while(!done && !signal?.aborted && timeout > Date.now()) return value } static async *asGenerator({callback, generator, signal, timeout=-1}={}) { let value = undefined let done = undefined if (timeout < 0) { timeout = Number.MAX_SAFE_INTEGER } timeout = Date.now() + timeout do { ({value, done} = await Promise.resolve(generator.next(value))) if (value instanceof Promise) { value = await value } if (callback) { ({value, done} = await Promise.resolve(callback.call(generator, {value, done}))) if (value instanceof Promise) { value = await value } } value = yield value } while(!done && !signal?.aborted && timeout > Date.now()) return value } } const TASK_REQUIRED = { "session_id": 'string', "prompt": 'string', "negative_prompt": 'string', "width": 'number', "height": 'number', "seed": 'number', "sampler_name": 'string', "use_stable_diffusion_model": 'string', "num_inference_steps": 'number', "guidance_scale": 'number', "num_outputs": 'number', "stream_progress_updates": 'boolean', "stream_image_progress": 'boolean', "show_only_filtered_image": 'boolean', "output_format": 'string', "output_quality": 'number', } const TASK_DEFAULTS = { "sampler_name": "plms", "use_stable_diffusion_model": "sd-v1-4", "num_inference_steps": 50, "guidance_scale": 7.5, "negative_prompt": "", "num_outputs": 1, "stream_progress_updates": true, "stream_image_progress": true, "show_only_filtered_image": true, "output_format": "png", "output_quality": 75, } const TASK_OPTIONAL = { "device": 'string', "init_image": 'string', "mask": 'string', "save_to_disk_path": 'string', "use_face_correction": 'string', "use_upscale": 'string', "use_vae_model": 'string', "use_hypernetwork_model": 'string', "hypernetwork_strength": 'number', } // Higer values will result in... // pytorch_lightning/utilities/seed.py:60: UserWarning: X is not in bounds, numpy accepts from 0 to 4294967295 const MAX_SEED_VALUE = 4294967295 class RenderTask extends Task { constructor(options={}) { super(options) if (typeof this._reqBody.seed === 'undefined') { this._reqBody.seed = Math.floor(Math.random() * (MAX_SEED_VALUE + 1)) } if (typeof typeof this._reqBody.seed === 'number' && (this._reqBody.seed > MAX_SEED_VALUE || this._reqBody.seed < 0)) { throw new Error(`seed must be in range 0 to ${MAX_SEED_VALUE}.`) } if ('use_cpu' in this._reqBody) { if (this._reqBody.use_cpu) { this._reqBody.device = 'cpu' } delete this._reqBody.use_cpu } if (this._reqBody.init_image) { if (typeof this._reqBody.prompt_strength === 'undefined') { this._reqBody.prompt_strength = 0.8 } else if (typeof this._reqBody.prompt_strength !== 'number') { throw new Error(`prompt_strength need to be of type number but ${typeof this._reqBody.prompt_strength} was found.`) } } if ('modifiers' in this._reqBody) { if (Array.isArray(this._reqBody.modifiers) && this._reqBody.modifiers.length > 0) { this._reqBody.modifiers = this._reqBody.modifiers.filter((val) => val.trim()) if (this._reqBody.modifiers.length > 0) { this._reqBody.prompt = `${this._reqBody.prompt}, ${this._reqBody.modifiers.join(', ')}` } } if (typeof this._reqBody.modifiers === 'string' && this._reqBody.modifiers.length > 0) { this._reqBody.modifiers = this._reqBody.modifiers.trim() if (this._reqBody.modifiers.length > 0) { this._reqBody.prompt = `${this._reqBody.prompt}, ${this._reqBody.modifiers}` } } delete this._reqBody.modifiers } this.checkReqBody() } checkReqBody() { for (const key in TASK_DEFAULTS) { if (typeof this._reqBody[key] === 'undefined') { this._reqBody[key] = TASK_DEFAULTS[key] } } for (const key in TASK_REQUIRED) { if (typeof this._reqBody[key] !== TASK_REQUIRED[key]) { throw new Error(`${key} need to be of type ${TASK_REQUIRED[key]} but ${typeof this._reqBody[key]} was found.`) } } for (const key in this._reqBody) { if (key in TASK_REQUIRED) { continue } if (key in TASK_OPTIONAL) { if (typeof this._reqBody[key] == "undefined") { delete this._reqBody[key] console.warn(`reqBody[${key}] was set to undefined. Removing optional key without value...`) continue } if (typeof this._reqBody[key] !== TASK_OPTIONAL[key]) { throw new Error(`${key} need to be of type ${TASK_OPTIONAL[key]} but ${typeof this._reqBody[key]} was found.`) } } } } /** Send current task to server. * @param {*} [timeout=-1] Optional timeout value in ms * @returns the response from the render request. * @memberof Task */ async post(timeout=-1) { performance.mark('make-render-request') if (performance.getEntriesByName('click-makeImage', 'mark').length > 0) { performance.measure('diff', 'click-makeImage', 'make-render-request') console.log('delay between clicking and making the server request:', performance.getEntriesByName('diff', 'measure')[0].duration + ' ms') } let jsonResponse = await super.post('/render', timeout) if (typeof jsonResponse?.task !== 'number') { console.warn('Endpoint error response: ', jsonResponse) const event = Object.assign({task:this}, jsonResponse) await eventSource.fireEvent(EVENT_UNEXPECTED_RESPONSE, event) if ('continueWith' in event) { jsonResponse = await Promise.resolve(event.continueWith) } if (typeof jsonResponse?.task !== 'number') { const err = new Error(jsonResponse?.detail || 'Endpoint response does not contains a task ID.') this.abort(err) throw err } } this._setId(jsonResponse.task) if (jsonResponse.stream) { this.streamUrl = jsonResponse.stream } this._setStatus(TaskStatus.waiting) return jsonResponse } enqueue(progressCallback) { return Task.enqueueNew(this, RenderTask, progressCallback) } *start(progressCallback) { if (typeof progressCallback !== 'undefined' && typeof progressCallback !== 'function') { throw new Error('progressCallback is not a function. progressCallback type: ' + typeof progressCallback) } if (this.isStopped) { return } this._setStatus(TaskStatus.pending) progressCallback?.call(this, {reqBody: this._reqBody}) Object.freeze(this._reqBody) // Post task request to backend let renderRequest = undefined try { renderRequest = yield this.post() yield progressCallback?.call(this, {renderResponse: renderRequest}) } catch (e) { yield progressCallback?.call(this, { detail: e.message }) throw e } try { // Wait for task to start on server. yield this.waitUntil({ callback: function() { return progressCallback?.call(this, {}) }, status: TaskStatus.processing, }) } catch (e) { this.abort(err) throw e } // Update class status and callback. const taskState = (typeof serverState.tasks === 'object' ? serverState.tasks[String(this.id)] : undefined) switch(taskState) { case 'pending': // Session has pending tasks. console.error('Server %o render request %o is still waiting.', serverState, renderRequest) //Only update status if not already set by waitUntil if (this.status === TaskStatus.init || this.status === TaskStatus.pending ) { // Set status as Waiting in backend. this._setStatus(TaskStatus.waiting) } break case 'running': case 'buffer': // Normal expected messages. this._setStatus(TaskStatus.processing) break case 'completed': if (this.isPending) { // Set state to processing until we read the reply this._setStatus(TaskStatus.processing) } console.warn('Server %o render request %o completed unexpectedly', serverState, renderRequest) break // Continue anyway to try to read cached result. case 'error': this._setStatus(TaskStatus.failed) console.error('Server %o render request %o has failed', serverState, renderRequest) break // Still valid, Update UI with error message case 'stopped': this._setStatus(TaskStatus.stopped) console.log('Server %o render request %o was stopped', serverState, renderRequest) return false default: if (!progressCallback) { const err = new Error('Unexpected server task state: ' + taskState || 'Undefined') this.abort(err) throw err } const response = yield progressCallback.call(this, {}) if (response instanceof Error) { this.abort(response) throw response } if (!response) { return false } } // Task started! // Open the reader. const reader = this.reader const task = this reader.onError = function(response) { if (progressCallback) { task.abort(new Error(response.statusText)) return progressCallback.call(task, { response, reader }) } return Task.prototype.onError.call(task, response) } yield progressCallback?.call(this, { reader }) //Start streaming the results. const streamGenerator = reader.open() let value = undefined let done = undefined yield progressCallback?.call(this, { stream: streamGenerator }) do { ({value, done} = yield streamGenerator.next()) if (typeof value !== 'object') { continue } yield progressCallback?.call(this, { update: value }) } while(!done) return value } static start(task, progressCallback) { if (typeof task !== 'object') { throw new Error ('task is not an object. task type: ' + typeof task) } if (!(task instanceof Task)) { if (task.reqBody) { task = new RenderTask(task.reqBody) } else { task = new RenderTask(task) } } return task.start(progressCallback) } static run(task, progressCallback) { const promiseGenerator = RenderTask.start(task, progressCallback) return Task.run(promiseGenerator) } } class FilterTask extends Task { constructor(options={}) { } /** Send current task to server. * @param {*} [timeout=-1] Optional timeout value in ms * @returns the response from the render request. * @memberof Task */ async post(timeout=-1) { let jsonResponse = await super.post('/filter', timeout) //this._setId(jsonResponse.task) this._setStatus(TaskStatus.waiting) } enqueue(progressCallback) { return Task.enqueueNew(this, FilterTask, progressCallback) } *start(progressCallback) { if (typeof progressCallback !== 'undefined' && typeof progressCallback !== 'function') { throw new Error('progressCallback is not a function. progressCallback type: ' + typeof progressCallback) } if (this.isStopped) { return } } static start(task, progressCallback) { if (typeof task !== 'object') { throw new Error ('task is not an object. task type: ' + typeof task) } if (!(task instanceof Task)) { if (task.reqBody) { task = new FilterTask(task.reqBody) } else { task = new FilterTask(task) } } return task.start(progressCallback) } static run(task, progressCallback) { const promiseGenerator = FilterTask.start(task, progressCallback) return Task.run(promiseGenerator) } } const getSystemInfo = debounce(async function() { let systemInfo = { devices: { all: {}, active: {}, }, hosts: [] } try { const res = await fetch('/get/system_info') if (!res.ok) { console.error('Invalid response fetching devices', res.statusText) return systemInfo } systemInfo = await res.json() } catch (e) { console.error('error fetching system info', e) } return systemInfo }, 250, true) async function getDevices() { let systemInfo = getSystemInfo() return systemInfo.devices } async function getHosts() { let systemInfo = getSystemInfo() return systemInfo.hosts } async function getModels() { let models = { 'stable-diffusion': [], 'vae': [], } try { const res = await fetch('/get/models') if (!res.ok) { console.error('Invalid response fetching models', res.statusText) return models } models = await res.json() console.log('get models response', models) } catch (e) { console.log('get models error', e) } return models } function getServerCapacity() { let activeDevicesCount = Object.keys(serverState?.devices?.active || {}).length if (typeof window === "object" && window.document.visibilityState === 'hidden') { activeDevicesCount = 1 + activeDevicesCount } return activeDevicesCount } let idleEventPromise = undefined function continueTasks() { if (typeof navigator?.scheduling?.isInputPending === 'function') { const inputPendingOptions = { // Report mouse/pointer move events when queue is empty. // Delay idle after mouse moves stops. includeContinuous: Boolean(task_queue.size <= 0 && concurrent_generators.size <= 0) } if (navigator.scheduling.isInputPending(inputPendingOptions)) { // Browser/User still active. return asyncDelay(CONCURRENT_TASK_INTERVAL) } } const serverCapacity = getServerCapacity() if (task_queue.size <= 0 && concurrent_generators.size <= 0) { if (!idleEventPromise?.isPending) { idleEventPromise = makeQuerablePromise(eventSource.fireEvent(EVENT_IDLE, {capacity: serverCapacity, idle: true})) } // Calling idle could result in task being added to queue. // if (task_queue.size <= 0 && concurrent_generators.size <= 0) { // return asyncDelay(IDLE_COOLDOWN).then(() => idleEventPromise) // } } if (task_queue.size < serverCapacity) { if (!idleEventPromise?.isPending) { idleEventPromise = makeQuerablePromise(eventSource.fireEvent(EVENT_IDLE, {capacity: serverCapacity - task_queue.size})) } } const completedTasks = [] for (let [generator, promise] of concurrent_generators.entries()) { if (promise.isPending) { continue } let value = promise.resolvedValue?.value || promise.resolvedValue if (promise.isRejected) { console.error(promise.rejectReason) const event = {generator, reason: promise.rejectReason} eventSource.fireEvent(EVENT_UNHANDLED_REJECTION, event) if ('continueWith' in event) { value = Promise.resolve(event.continueWith) } else { concurrent_generators.delete(generator) completedTasks.push({generator, promise}) continue } } if (value instanceof Promise) { promise = makeQuerablePromise(value.then((val) => ({done: promise.resolvedValue?.done, value: val}))) concurrent_generators.set(generator, promise) continue } weak_results.set(generator, value) if (promise.resolvedValue?.done) { concurrent_generators.delete(generator) completedTasks.push({generator, promise}) continue } promise = generator.next(value) if (!(promise instanceof Promise)) { promise = Promise.resolve(promise) } promise = makeQuerablePromise(promise) concurrent_generators.set(generator, promise) } for (let [task, generator] of task_queue.entries()) { const cTsk = completedTasks.find((item) => item.generator === generator) if (cTsk?.promise?.rejectReason || task.hasFailed) { eventSource.fireEvent(EVENT_TASK_ERROR, {task, generator, reason: cTsk?.promise?.rejectReason || task.exception }) task_queue.delete(task) continue } if (task.isCompleted || task.isStopped || cTsk) { const eventEndArgs = {task, generator} if (task.isStopped) { eventEndArgs.stopped = true } eventSource.fireEvent(EVENT_TASK_END, eventEndArgs) task_queue.delete(task) continue } if (concurrent_generators.size > serverCapacity) { break } if (!generator) { if (typeof task.start === 'function') { generator = task.start() } } else if (concurrent_generators.has(generator)) { continue } const event = {task, generator}; const beforeStart = eventSource.fireEvent(EVENT_TASK_START, event) // optional beforeStart promise to wait on before starting task. const promise = makeQuerablePromise(beforeStart.then(() => Promise.resolve(event.beforeStart))) concurrent_generators.set(event.generator, promise) task_queue.set(task, event.generator) } const promises = Array.from(concurrent_generators.values()) if (promises.length <= 0) { return asyncDelay(CONCURRENT_TASK_INTERVAL) } return Promise.race(promises).finally(continueTasks) } let taskPromise = undefined function startCheck() { if (taskPromise?.isPending) { return } do { if (taskPromise?.resolvedValue instanceof Promise) { taskPromise = makeQuerablePromise(taskPromise.resolvedValue) continue } if (typeof navigator?.scheduling?.isInputPending === 'function' && navigator.scheduling.isInputPending()) { return } const continuePromise = continueTasks().catch(async function(err) { console.error(err) await eventSource.fireEvent(EVENT_UNHANDLED_REJECTION, {reason: err}) await asyncDelay(RETRY_DELAY_ON_ERROR) }) taskPromise = makeQuerablePromise(continuePromise) } while(taskPromise?.isResolved) } const SD = { ChunkedStreamReader, ServerStates, TaskStatus, Task, RenderTask, FilterTask, Events: EVENTS_TYPES, init: async function(options={}) { if ('events' in options) { for (const key in options.events) { eventSource.addEventListener(key, options.events[key]) } } await healthCheck() setInterval(healthCheck, HEALTH_PING_INTERVAL) setInterval(startCheck, CONCURRENT_TASK_INTERVAL) }, /** Add a new event listener */ addEventListener: (...args) => eventSource.addEventListener(...args), /** Remove the event listener */ removeEventListener: (...args) => eventSource.removeEventListener(...args), isServerAvailable, getServerCapacity, getSystemInfo, getDevices, getHosts, getModels, render: (...args) => RenderTask.run(...args), filter: (...args) => FilterTask.run(...args), waitUntil, }; Object.defineProperties(SD, { serverState: { configurable: false, get: () => serverState, }, isAvailable: { configurable: false, get: () => isServerAvailable(), }, serverCapacity: { configurable: false, get: () => getServerCapacity(), }, sessionId: { configurable: false, get: () => sessionId, set: (val) => { if (typeof val === 'undefined') { throw new Error("Can't set sessionId to undefined.") } sessionId = val }, }, MAX_SEED_VALUE: { configurable: false, get: () => MAX_SEED_VALUE, }, activeTasks: { configurable: false, get: () => task_queue, }, }) Object.defineProperties(getGlobal(), { SD: { configurable: false, get: () => SD, }, sessionId: { //TODO Remove in the future in favor of SD.sessionId configurable: false, get: () => { console.warn('Deprecated window.sessionId has been replaced with SD.sessionId.') console.trace() return SD.sessionId }, set: (val) => { console.warn('Deprecated window.sessionId has been replaced with SD.sessionId.') console.trace() SD.sessionId = val } } }) })()