diff --git a/client/bin/app.ts b/client/bin/app.ts index e95b9a4..73417db 100755 --- a/client/bin/app.ts +++ b/client/bin/app.ts @@ -161,9 +161,18 @@ for (;;) { const cursor = floor.byIno(FIXED_STRING).refAt(e.pos); const anchor = floor.byIno(FIXED_STRING).refAt(e.anchor); - floor.update(pack); + const changes = floor.update(pack); + if (!(changes.has(FIXED_STRING) || changes.has(FIXED_CURSORS))) { + continue; + } + + // TODO: it's probably faster to just read the string again, but this 'proves' the change works + const stringChange = changes.get(FIXED_STRING); + if (stringChange) { + const patch = model.string(stringChange.lo, stringChange.lo + stringChange.length); + e.text = e.text.substring(0, stringChange.lo) + patch + e.text.substring(stringChange.hi); + } - // TODO: we assume FIXED_STRING changes (not always true) e.text = model.string(); reconcileOtherCursors(); diff --git a/client/lib/gumnut/floor.test.ts b/client/lib/gumnut/floor.test.ts new file mode 100644 index 0000000..dbb8534 --- /dev/null +++ b/client/lib/gumnut/floor.test.ts @@ -0,0 +1,50 @@ +import test from 'node:test'; +import * as assert from 'node:assert'; +import { GumnutFloor } from './floor.ts'; + +test('floor', () => { + const f = new GumnutFloor(); + + const changes1 = f.update({ + p: { + def: { + r: [-5, 0], + s: [5, 'hello'], + }, + }, + }); + assert.deepStrictEqual(f.byIno('def').string(), 'hello'); + assert.deepStrictEqual(changes1, new Map([['def', { lo: 0, hi: 0, length: 5 }]])); + + const changes2 = f.perform(function* (t) { + const node = t.byIno('def'); + node.stringUpdate(3, 2, 'ls to the yeah'); + }); + assert.deepStrictEqual(f.byIno('def').string(), 'hells to the yeah'); + assert.deepStrictEqual(changes2, new Map([['def', { lo: 3, hi: 5, length: 14 }]])); + + const changes3 = f.update({ + p: { + def: { + r: [0, 2, 5], // delete "llo" + s: [2, 'i'], + }, + }, + }); + assert.deepStrictEqual(f.byIno('def').string(), 'hi'); + assert.deepStrictEqual(changes3, new Map([['def', { lo: 1, hi: 17, length: 1 }]])); +}); + +test('floor local changes', () => { + const f = new GumnutFloor(); + + const changes1 = f.perform(function* (t) { + t.byIno('abc').dataUpdate(0, 0, 1, 2, 3); + }); + assert.deepStrictEqual(changes1, new Map([['abc', { lo: 0, hi: 0, length: 3 }]])); + + const changes2 = f.perform(function* (t) { + t.byIno('abc').set(2, 200); + }); + assert.deepStrictEqual(changes2, new Map([['abc', { lo: 1, hi: 2, length: 1 }]])); +}); diff --git a/client/lib/gumnut/floor.ts b/client/lib/gumnut/floor.ts index 9017851..9bae07f 100644 --- a/client/lib/gumnut/floor.ts +++ b/client/lib/gumnut/floor.ts @@ -12,6 +12,7 @@ import { } from '../ymodel/types.ts'; import { emptySymbol, InoPosRef, InoPosRefImpl, type GumnutData } from './types.ts'; import type { PatchApi } from '../ymodel/work-holder.ts'; +import type { RangeUpdate } from '../ymodel/internal/range.ts'; type WireType = { b?: string[]; @@ -147,18 +148,53 @@ class GumnutFloorApi extends GumnutLow { } } +export type InoChanges = ReadonlyMap>; + export class GumnutFloor { private readonly low = new GumnutFloorApi(); private readonly announceCh = newChannel(); + private readonly listeners = new Set<(x: InoChanges) => void>(); + + /** + * Listen to all changes here. + */ + addListener(signal: AbortSignal, listener: (x: InoChanges) => void) { + if (signal.aborted) { + return; + } + + if (this.listeners.has(listener)) { + const orig = listener; + listener = (x) => orig(x); + } + + this.listeners.add(listener); + signal.addEventListener('abort', () => this.listeners.delete(listener)); + } + + private dispatch(changes: InoChanges) { + this.listeners.forEach((l) => { + try { + l(changes); + } catch (e) { + Promise.resolve().then(() => { + throw e; + }); + } + }); + } + /** * An update arrived from the server. */ - update(w: WireType) { + update(w: WireType): InoChanges { if (w.b) { throw new Error(`panic; can't have barrier in incoming update`); } - this.low.applyServer(w); + const updates = this.low.applyServer(w); + this.dispatch(updates); + return updates; } /** @@ -166,6 +202,12 @@ export class GumnutFloor { */ ackToSeq(seq: number) { this.low.ackToSeq(seq); + + // by acking, we might have unlocked a barrier op + for (const _ of this.low.pendingSeq()) { + this.announceCh.push(true); + break; + } } /** @@ -189,6 +231,9 @@ export class GumnutFloor { return p.size ? p : undefined; } + /** + * Returns a read-only {@link FloorReadApi} for the given ino. + */ byIno(ino: string): FloorReadApi { const internal = this.low.byIno(ino); @@ -217,11 +262,8 @@ export class GumnutFloor { /** * Builds a {@link FloorApi} from the internal {@link PatchApi}. - * - * This just smooths over internal string/internal stuff. - * TODO: can we not? */ - private writeForPatchApi(ino: string, internal: PatchApi): FloorApi { + private writeForPatchApi(ino: string, internal: PatchApi): FloorPerformApi { return { length() { return internal.length(); @@ -276,10 +318,10 @@ export class GumnutFloor { return this.low.byIno(impl.target).posOf(impl.id); } - perform(fn: FloorFn) { + perform(fn: FloorFn): InoChanges { const writeForPatchApi = this.writeForPatchApi.bind(this); - this.low.perform(function* (byIno) { + const updates = this.low.perform(function* (byIno) { const api: FloorInoApi = { byIno(ino) { const int = byIno(ino); @@ -293,17 +335,19 @@ export class GumnutFloor { yield* fn(api); }); this.announceCh.push(true); + this.dispatch(updates); + return updates; } } export type FloorInoApi = { - byIno(ino: string): FloorApi; + byIno(ino: string): FloorPerformApi; resolveRef(r: InoPosRef): number; }; export type FloorFn = (api: FloorInoApi) => Generator, void, void>; -export type FloorReadApi = { +export interface FloorReadApi { /** * Returns the length of this inode. */ @@ -331,9 +375,12 @@ export type FloorReadApi = { * If this is to track an _item_, you should use the index after that item, as it will be consistent over time. */ refAt(at: number): InoPosRef; -}; +} -export type FloorApi = FloorReadApi & { +/** + * This is a work/transactional API. + */ +export interface FloorPerformApi extends FloorReadApi { /** * Set replaces data _before_ the given position (1-indexed). * @@ -350,4 +397,4 @@ export type FloorApi = FloorReadApi & { * Acts like 'splice', merging delete/insert operations for string data. */ stringUpdate(at: number, deleteCount: number, str: string): void; -}; +} diff --git a/client/lib/ymodel/api.ts b/client/lib/ymodel/api.ts index ef5374c..b40c643 100644 --- a/client/lib/ymodel/api.ts +++ b/client/lib/ymodel/api.ts @@ -3,11 +3,13 @@ import { type PatchApi, type ReadApi, WorkHolder } from './work-holder.ts'; import { RopeState, type InternalPatch } from './internal/state.ts'; import type { Op, Patch, SetPart } from './types.ts'; import type { Id } from './internal/shared.ts'; +import { accumulateRangeUpdate, invertRangeUpdate, type RangeUpdate } from './internal/range.ts'; type InoState = { state: RopeState; serverState: RopeState; ro: ReadApi; + update: RangeUpdate | undefined; }; export type WorkFn = ( @@ -74,6 +76,7 @@ export abstract class GumnutLow { return out.state.posOfValid(id); }, }, + update: undefined, }; return out; @@ -126,19 +129,19 @@ export abstract class GumnutLow { * * TODO: we should return a handle which lets the user track application state- e.g. confirm lock. */ - public perform(handler: WorkFn): void { + public perform(handler: WorkFn): Map { if (this.performActive) { throw new Error(`cannot nest perform() calls`); } try { this.performActive = true; - this.internalPerform(handler); + return this.internalPerform(handler); } finally { this.performActive = false; } } - private internalPerform(handler: WorkFn) { + private internalPerform(handler: WorkFn): Map { const seq = ++this.transactionSeq; const pending = new PendingWork(this.patchToServer.bind(this), handler); @@ -150,7 +153,7 @@ export abstract class GumnutLow { this.pendingWork.push({ seq, pending }); // apply changes inline - this.applyAllLocalOps(pending.forInternal()); + return this.applyAllLocalOps(pending.forInternal()); } /** @@ -159,7 +162,7 @@ export abstract class GumnutLow { * * Updates to any ino that are in barrier will cause the work to re-run (otherwise we just reapply ops). */ - public applyServer(update: WireType) { + public applyServer(update: WireType): Map { // this starts by resetting everything, this feels overkill // however, the only optimization I think can occur is to not reset inodes which are: // - unaffected by server changes @@ -168,9 +171,21 @@ export abstract class GumnutLow { // however the filter layer feels awkward, because we just let the `resume()` call have access to inodes in a basically unfettered way. // TODO: care later. + const out = new Map(); + const accumulate = (ino: string, update: RangeUpdate | undefined) => { + if (update) { + out.set(ino, accumulateRangeUpdate(out.get(ino), update)); + } + }; + // #1: reset all inodes to server state - for (const is of this._byIno.values()) { + for (const [ino, is] of this._byIno.entries()) { is.state = is.serverState; + + if (is.update !== undefined) { + out.set(ino, invertRangeUpdate(is.update)); + is.update = undefined; + } } // #2a: enact structural changes @@ -178,7 +193,11 @@ export abstract class GumnutLow { for (const ino in ops) { const is = this._byIno.get(ino); const target = is.serverState; - ops[ino].forEach((op) => target.applyOp(target.coerceOp(op))); + + ops[ino].forEach((op) => { + const update = target.applyOp(target.coerceOp(op)); + accumulate(ino, update); + }); } // #3a: enact sets @@ -189,7 +208,11 @@ export abstract class GumnutLow { for (const ino in sets) { const is = this._byIno.get(ino); const target = is.serverState; - sets[ino].forEach((set) => target.applySet(target.coerceSet(set))); + + sets[ino].forEach((set) => { + const update = target.applySet(target.coerceSet(set)); + accumulate(ino, update); + }); } // #3: reapply all local @@ -205,10 +228,14 @@ export abstract class GumnutLow { this.sentToSeq = Math.min(seq, this.sentToSeq); } - this.applyAllLocalOps(pending.forInternal()); + const localUpdates = this.applyAllLocalOps(pending.forInternal()); + for (const [ino, update] of localUpdates) { + accumulate(ino, update); + } }); - // TODO: return uniquely effected inos + // return all changes + return out; } /** @@ -245,7 +272,9 @@ export abstract class GumnutLow { * * This may fork the server state. */ - private applyAllLocalOps(update: Record>) { + private applyAllLocalOps(update: Record>): Map { + const out = new Map(); + for (const ino in update) { const patch = update[ino]; @@ -255,9 +284,31 @@ export abstract class GumnutLow { } const target = is.state; - patch.run.forEach((op) => target.applyOp(op)); - patch.set.forEach((set) => target.applySet(set)); + let rangeUpdate: RangeUpdate | undefined; + + patch.run.forEach((op) => { + const update = target.applyOp(op); + if (update) { + rangeUpdate = accumulateRangeUpdate(rangeUpdate, update); + } + }); + patch.set.forEach((set) => { + const update = target.applySet(set); + if (update) { + rangeUpdate = accumulateRangeUpdate(rangeUpdate, update); + } + }); + + if (rangeUpdate) { + // set local to this out + out.set(ino, rangeUpdate); + + // but update aggregate too + is.update = accumulateRangeUpdate(is.update, rangeUpdate); + } } + + return out; } /** diff --git a/client/lib/ymodel/internal/range.test.ts b/client/lib/ymodel/internal/range.test.ts new file mode 100644 index 0000000..155ebd7 --- /dev/null +++ b/client/lib/ymodel/internal/range.test.ts @@ -0,0 +1,52 @@ +import test from 'node:test'; +import * as assert from 'node:assert'; +import { accumulateRangeUpdate, type RangeUpdate } from './range.ts'; + +test('accumulateRangeUpdate', async (t) => { + await t.test('initial update', () => { + const update: RangeUpdate = { lo: 5, hi: 10, length: 2 }; + const result = accumulateRangeUpdate(undefined, update); + assert.deepStrictEqual(result, update); + }); + + await t.test('sequential non-overlapping', () => { + const have: RangeUpdate = { lo: 10, hi: 12, length: 5 }; // [10, 12) -> length 5 (delta +3) + const update: RangeUpdate = { lo: 2, hi: 4, length: 1 }; // [2, 4) -> length 1 (delta -1) + const result = accumulateRangeUpdate(have, update); + // Combined affected range: [2, 12) + // acc.length = 10 (orig length) + 3 (delta_have) - 1 (delta_update) = 12 + assert.deepStrictEqual(result, { lo: 2, hi: 12, length: 12 }); + }); + + await t.test('overlapping: update touches have', () => { + const have: RangeUpdate = { lo: 2, hi: 5, length: 2 }; // [2, 5) -> length 2 (delta -1) + const update: RangeUpdate = { lo: 1, hi: 3, length: 4 }; // [1, 3) -> length 4 (delta +2) + const result = accumulateRangeUpdate(have, update); + // Combined affected range: [1, 5) + // acc.length = 4 (orig length) - 1 (delta_have) + 2 (delta_update) = 5 + assert.deepStrictEqual(result, { lo: 1, hi: 5, length: 5 }); + }); + + await t.test('update inside have insertion', () => { + const have: RangeUpdate = { lo: 5, hi: 10, length: 10 }; // [5, 10) -> length 10 (delta +5) + const update: RangeUpdate = { lo: 6, hi: 8, length: 1 }; // [6, 8) -> length 1 (delta -1) + const result = accumulateRangeUpdate(have, update); + // Combined: [5, 10) + // acc.length = 5 (orig length) + 5 - 1 = 9 + assert.deepStrictEqual(result, { lo: 5, hi: 10, length: 9 }); + }); + + await t.test('deletion then insertion at same spot', () => { + const have: RangeUpdate = { lo: 5, hi: 10, length: 0 }; // delete [5, 10) + const update: RangeUpdate = { lo: 5, hi: 5, length: 3 }; // insert 3 at pos 5 + const result = accumulateRangeUpdate(have, update); + assert.deepStrictEqual(result, { lo: 5, hi: 10, length: 3 }); + }); + + await t.test('insertion then deletion at same spot', () => { + const have: RangeUpdate = { lo: 5, hi: 5, length: 10 }; // insert 10 at pos 5 + const update: RangeUpdate = { lo: 5, hi: 6, length: 0 }; // delete 1 from inserted + const result = accumulateRangeUpdate(have, update); + assert.deepStrictEqual(result, { lo: 5, hi: 5, length: 9 }); + }); +}); diff --git a/client/lib/ymodel/internal/range.ts b/client/lib/ymodel/internal/range.ts new file mode 100644 index 0000000..4175d7b --- /dev/null +++ b/client/lib/ymodel/internal/range.ts @@ -0,0 +1,56 @@ +/** + * Describes the way in which we modify an array-like. + * + * The range of data from lo-hi is now length. + * + * For deletions, this will have a length of zero. + * For inserts, lo and hi will likely be the same with a non-zero length. + * + * For merged updates, all three values might be different (range lo-hi is now this length). + */ +export type RangeUpdate = { lo: number; hi: number; length: number }; + +/** + * @param have we have this update + * @param update merge this range with it + * @returns the merged range + */ +export function accumulateRangeUpdate( + have: RangeUpdate | undefined, + update: RangeUpdate, +): RangeUpdate { + if (have === undefined) { + return update; + } + + const delta = have.length - (have.hi - have.lo); + + const mapToOriginal = (pos: number, isHi: boolean) => { + if (pos <= have.lo) { + return pos; + } + if (pos >= have.lo + have.length) { + return pos - delta; + } + return isHi ? have.hi : have.lo; + }; + + const lo = Math.min(have.lo, mapToOriginal(update.lo, false)); + const hi = Math.max(have.hi, mapToOriginal(update.hi, true)); + const length = hi - lo + delta + (update.length - (update.hi - update.lo)); + + if (isNaN(length)) { + console.info({ have, update }); + throw new Error('got NAN'); + } + + return { lo, hi, length }; +} + +export function invertRangeUpdate(have: RangeUpdate): RangeUpdate { + return { + lo: have.lo, + hi: have.lo + have.length, + length: have.hi - have.lo, + }; +} diff --git a/client/lib/ymodel/internal/state.ts b/client/lib/ymodel/internal/state.ts index 0fdb932..3ffd845 100644 --- a/client/lib/ymodel/internal/state.ts +++ b/client/lib/ymodel/internal/state.ts @@ -1,6 +1,7 @@ import { AATree, Rope } from 'thorish'; import type { Op, SetPart } from '../types.ts'; import { zeroId, allocLocalId, type Id } from './shared.ts'; +import { accumulateRangeUpdate, type RangeUpdate } from './range.ts'; type Entry = number | string | X[]; @@ -199,9 +200,9 @@ export class RopeState { * * Returns if a visible change occured. */ - private internalInsert(at: Id, newId: Id, length: number): boolean { + private internalInsert(at: Id, newId: Id, length: number): RangeUpdate | undefined { if (length <= 0) { - return false; + return; } // remember: the rope doesn't really care that we think long entries take up "rope space" @@ -215,10 +216,14 @@ export class RopeState { } const exists = this.ensureIdEdge(at); + this.rope.insertAfter(at, newId, exists ? length : 0, length); this.tree.insert(newId); - return exists; + if (exists) { + const pos = this.posOf(at); + return { lo: pos, hi: pos, length }; + } } /** @@ -226,9 +231,9 @@ export class RopeState { * * Returns if a visible change occured. */ - private internalDelete(lo: Id, hi: Id): boolean { + private internalDelete(lo: Id, hi: Id): RangeUpdate | undefined { if (lo === hi) { - return false; + return; } this.ensureIdEdge(lo); @@ -246,8 +251,12 @@ export class RopeState { cleared += prev.length; } } + if (cleared === 0) { + return; + } - return cleared !== 0; + const pos = this.posOf(lo); + return { lo: pos, hi: pos + cleared, length: 0 }; } public coerceOp(op: Op): InternalOp { @@ -273,14 +282,21 @@ export class RopeState { return int; } - public applyOp(int: InternalOp): boolean { + /** + * Applies this operation here. + * Returns the 'real-space' {@link RangeUpdate} for this work. + */ + public applyOp(int: InternalOp): RangeUpdate | undefined { switch (int.op.r.length) { case 1: return this.internalInsert(int.op.r[0] as Id, int.id, int.op.length ?? 0); + case 2: return this.internalDelete(int.op.r[0] as Id, int.op.r[1] as Id); + case 3: throw new Error(`TODO: support flip`); + default: throw new Error(`unsupported Op`); } @@ -292,6 +308,9 @@ export class RopeState { length += part.length; } const start = set.before - length; + if (start < 0) { + throw new Error(`can't set index<0`); + } // TODO: right now, InteralSet splays out every entry or char with a target ID! const out: InternalSet = { ids: [], data: set.body }; @@ -310,6 +329,10 @@ export class RopeState { * Applies a single chunk of a {@link InteralSet}, i.e., a string or array chunk. */ private internalApplySet(chunk: InternalSetChunk): boolean { + if (chunk.data.length === 0) { + return false; + } + let any = false; let i = 0; // fast-path array inserts @@ -323,23 +346,32 @@ export class RopeState { } const e = this.rope.lookup(hostId); - if (!Array.isArray(e.data)) { - break; // give up + if (e.length === 0) { + ++i; // couldn't set, already deleted + continue; + } + + if (typeof e.data === 'number') { + // create new and adjust (e.data isn't live) + e.data = Array(e.data).fill(this.zeroValue); + this.rope.adjust(hostId, e.data, e.data.length); + any = true; + } else if (!Array.isArray(e.data)) { + break; // give up, is a string so we need to split it } const index = e.length - 1 - (hostId - id); e.data[index] = chunk.data[i]; + any = true; ++i; } if (i === chunk.ids.length) { - return true; + return any; } } // consume as much contiguous data and set, rinse and repeat - let lastId = this.leftOf(chunk.ids[0]); - this.ensureIdEdge(lastId); while (i < chunk.ids.length) { let id = chunk.ids[i]; @@ -352,36 +384,40 @@ export class RopeState { } id = nextId; } + let localData = chunk.data.slice(lo, i); - const length = i - lo; - const localData = chunk.data.slice(lo, i); - - this.ensureIdEdge(id); - const info = this.rope.lookup(id); - - if (info.length === length) { - this.rope.adjust(id, localData, length); - } else { - // remove all tree entries >lastId ): boolean { + public applySet(int: InternalSet): RangeUpdate | undefined { let any = false; // consume chunks (typically just one!) @@ -390,7 +426,24 @@ export class RopeState { const change = this.internalApplySet(chunk); any ||= change; } - return any; + + if (!any) { + return undefined; + } + + // otherwise, because IDs can be anywhere, ... find the lo/hi + // this is stupidly slow because a set might apply in literally any placement after xform + + const pos = int.ids.map((id) => this.posOfValid(id)).filter((pos) => pos !== -1); + if (!pos.length) { + throw new Error(`panic; got change set without valid IDs`); + } + pos.sort((a, b) => a - b); + + const lo = pos[0] - 1; + const hi = pos.at(-1)!; + + return { lo, hi, length: hi - lo }; } /** diff --git a/client/lib/ymodel/work-holder.ts b/client/lib/ymodel/work-holder.ts index 37f951b..1625549 100644 --- a/client/lib/ymodel/work-holder.ts +++ b/client/lib/ymodel/work-holder.ts @@ -7,6 +7,7 @@ import { rollbackForOp, type RopeState, } from './internal/state.ts'; +import type { RangeUpdate } from './internal/range.ts'; export interface ReadApi { length(): number; @@ -25,7 +26,7 @@ export interface PatchApi extends ReadApi { export type ApplyResult = { rollback(): boolean; - visible: boolean; + range: RangeUpdate | undefined; }; /** @@ -194,7 +195,10 @@ export class WorkHolder implements PatchApi { applyOp(op: Op): ApplyResult { const target = this.mustDerive(); const int = target.coerceOp(op); - const visible = target.applyOp(int); + const range = target.applyOp(int); + if (!range) { + return { rollback: () => false, range: undefined }; + } // store for later this.patch.run.push(int); @@ -223,14 +227,18 @@ export class WorkHolder implements PatchApi { } return true; }, - visible, + range, }; } applySet(part: SetPart): ApplyResult { const target = this.mustDerive(); const ipart = target.coerceSet(part); - const visible = target.applySet(ipart); + const range = target.applySet(ipart); + + if (!range) { + return { rollback: () => false, range: undefined }; + } // store for later this.patch.set.push(ipart); @@ -247,7 +255,7 @@ export class WorkHolder implements PatchApi { this.derivative = undefined; return true; }, - visible, + range, }; } }