import { assert } from "@fluidframework/common-utils" ; import { ISequencedDocumentMessage } from "@fluidframework/protocol-definitions" ; import { IFluidDataStoreRuntime , IChannelStorageService , Serializable , IChannelAttributes , } from "@fluidframework/datastore-definitions" ; import { IFluidSerializer , makeHandlesSerializable , parseHandles , SharedObject , SummarySerializer , } from "@fluidframework/shared-object-base" ; import { ISummaryTreeWithStats } from "@fluidframework/runtime-definitions" ; import { ObjectStoragePartition , SummaryTreeBuilder } from "@fluidframework/runtime-utils" ; import { IMatrixProducer , IMatrixConsumer , IMatrixReader , IMatrixWriter , } from "@tiny-calc/nano" ; import { MergeTreeDeltaType , IMergeTreeOp , SegmentGroup , ISegment } from "@fluidframework/merge-tree" ; import { MatrixOp } from "./ops" ; import { PermutationVector , PermutationSegment } from "./permutationvector" ; import { SparseArray2D } from "./sparsearray2d" ; import { SharedMatrixFactory } from "./runtime" ; import { Handle , isHandleValid } from "./handletable" ; import { deserializeBlob } from "./serialization" ; import { ensureRange } from "./range" ; import { IUndoConsumer } from "./types" ; import { MatrixUndoProvider } from "./undoprovider" ; const enum SnapshotPath { rows = "rows" , cols = "cols" , cells = "cells" } interface ISetOp < T > { type : MatrixOp . set ; row : number ; col : number ; value : MatrixItem < T > ; } interface ISetOpMetadata { rowHandle : Handle ; colHandle : Handle ; localSeq : number ; } /**
 * A matrix cell value may be undefined (indicating an empty cell) or any serializable type,
 * excluding null.  (However, nulls may be embedded inside objects and arrays.)
 */ export type MatrixItem < T > = Serializable < Exclude < T , null > > | undefined ; /**
 * A SharedMatrix holds a rectangular 2D array of values.  Supported operations
 * include setting values and inserting/removing rows and columns.
 *
 * Matrix values may be any Fluid serializable type, which is the set of JSON
 * serializable types extended to include IFluidHandles.
 *
 * Fluid's SharedMatrix implementation works equally well for dense and sparse
 * matrix data and physically stores data in Z-order to leverage CPU caches and
 * prefetching when reading in either row or column major order.  (See README.md
 * for more details.)
 */ export class SharedMatrix < T = any > extends SharedObject implements IMatrixProducer < MatrixItem < T > > , IMatrixReader < MatrixItem < T > > , IMatrixWriter < MatrixItem < T > > { private readonly consumers = new Set < IMatrixConsumer < MatrixItem < T > > > ( ) ; public static getFactory ( ) { return new SharedMatrixFactory ( ) ; } private readonly rows : PermutationVector ; private readonly cols : PermutationVector ; private cells = new SparseArray2D < MatrixItem < T > > ( ) ; private pending = new SparseArray2D < number > ( ) ; constructor ( runtime , public id , attributes ) { super ( id , runtime , attributes ) ; this . rows = new PermutationVector ( SnapshotPath . rows , this . logger , runtime , this . onRowDelta , this . onRowHandlesRecycled ) ; this . cols = new PermutationVector ( SnapshotPath . cols , this . logger , runtime , this . onColDelta , this . onColHandlesRecycled ) ; } private undo ? : MatrixUndoProvider < T > ; /**
     * Subscribes the given IUndoConsumer to the matrix.
     */ public openUndo ( consumer ) { assert ( this . undo === undefined , 0x019 ) ; this . undo = new MatrixUndoProvider ( consumer , this , this . rows , this . cols ) ; } private get rowHandles ( ) { return this . rows . handleCache ; } private get colHandles ( ) { return this . cols . handleCache ; } /**
     * {@inheritDoc @fluidframework/datastore-definitions#IChannelFactory.create}
     */ public static create < T > ( runtime , id ? ) { return runtime . createChannel ( id , SharedMatrixFactory . Type ) as SharedMatrix < T > ; } openMatrix ( consumer ) { this . consumers . add ( consumer ) ; return this ; } closeMatrix ( consumer ) { this . consumers . delete ( consumer ) ; } public get rowCount ( ) { return this . rows . getLength ( ) ; } public get colCount ( ) { return this . cols . getLength ( ) ; } public getCell ( row , col ) { const rowHandle = this . rowHandles . getHandle ( row ) ; if ( isHandleValid ( rowHandle ) ) { const colHandle = this . colHandles . getHandle ( col ) ; if ( isHandleValid ( colHandle ) ) { return this . cells . getCell ( rowHandle , colHandle ) ; } } else { ensureRange ( col , this . cols . getLength ( ) ) ; } return undefined ; } public get matrixProducer ( ) : IMatrixProducer < MatrixItem < T > > { return this ; } public setCell ( row , col , value ) { assert ( 0 <= row && row < this . rowCount && 0 <= col && col < this . colCount , 0x01a ) ; this . setCellCore ( row , col , value ) ; for ( const consumer of this . consumers . values ( ) ) { consumer . cellsChanged ( row , col , 1 , 1 , this ) ; } } public setCells ( rowStart , colStart , colCount , values ) { const rowCount = Math . ceil ( values . length / colCount ) ; assert ( ( 0 <= rowStart && rowStart < this . rowCount ) && ( 0 <= colStart && colStart < this . colCount ) && ( 1 <= colCount && colCount <= ( this . colCount - colStart ) ) && ( rowCount <= ( this . rowCount - rowStart ) ) , 0x01b ) ; const endCol = colStart + colCount ; let r = rowStart ; let c = colStart ; for ( const value of values ) { this . setCellCore ( r , c , value ) ; if ( ++ c === endCol ) { c = colStart ; r ++ ; } } for ( const consumer of this . consumers . values ( ) ) { consumer . cellsChanged ( rowStart , colStart , rowCount , colCount , this ) ; } } private setCellCore ( row , col , value , rowHandle = this . rows . getAllocatedHandle ( row ) , colHandle = this . cols . getAllocatedHandle ( col ) ) { if ( this . undo !== undefined ) { let oldValue = this . cells . getCell ( rowHandle , colHandle ) ; if ( oldValue === null ) { oldValue = undefined ; } this . undo . cellSet ( rowHandle , colHandle , oldValue ) ; } this . cells . setCell ( rowHandle , colHandle , value ) ; if ( this . isAttached ( ) ) { this . sendSetCellOp ( row , col , value , rowHandle , colHandle ) ; } } private sendSetCellOp ( row , col , value , rowHandle , colHandle , localSeq = this . nextLocalSeq ( ) ) { assert ( this . isAttached ( ) , 0x1e2 ) ; const op = { type : MatrixOp . set , row , col , value } ; const metadata = { rowHandle , colHandle , localSeq } ; this . submitLocalMessage ( op , metadata ) ; this . pending . setCell ( rowHandle , colHandle , localSeq ) ; } private submitVectorMessage ( currentVector , oppositeVector , dimension , message ) { const localSeq = currentVector . getCollabWindow ( ) . localSeq ; const oppositeWindow = oppositeVector . getCollabWindow ( ) ; assert ( localSeq >= oppositeWindow . localSeq , 0x01c ) ; oppositeWindow . localSeq = localSeq ; if ( this . isAttached ( ) ) { message . target = dimension ; this . submitLocalMessage ( message , currentVector . peekPendingSegmentGroups ( message . type === MergeTreeDeltaType . GROUP ? message . ops . length : 1 ) ) ; } } private submitColMessage ( message ) { this . submitVectorMessage ( this . cols , this . rows , SnapshotPath . cols , message ) ; } public insertCols ( colStart , count ) { this . submitColMessage ( this . cols . insert ( colStart , count ) ) ; } public removeCols ( colStart , count ) { this . submitColMessage ( this . cols . remove ( colStart , count ) ) ; } private submitRowMessage ( message ) { this . submitVectorMessage ( this . rows , this . cols , SnapshotPath . rows , message ) ; } public insertRows ( rowStart , count ) { this . submitRowMessage ( this . rows . insert ( rowStart , count ) ) ; } public removeRows ( rowStart , count ) { this . submitRowMessage ( this . rows . remove ( rowStart , count ) ) ; } internal public _undoRemoveRows ( segment ) { const original = segment as PermutationSegment ; const { op , inserted } = this . rows . insertRelative ( original , original . cachedLength ) ; this . submitRowMessage ( op ) ; original . transferToReplacement ( inserted ) ; const rowStart = this . rows . getPosition ( inserted ) ; this . rows . handleCache . itemsChanged ( rowStart , 0 , inserted . cachedLength ) ; let rowHandle = inserted . start ; const rowCount = inserted . cachedLength ; for ( let row = rowStart ; row < rowStart + rowCount ; row ++ , rowHandle ++ ) { for ( let col = 0 ; col < this . colCount ; col ++ ) { const colHandle = this . colHandles . getHandle ( col ) ; const value = this . cells . getCell ( rowHandle , colHandle ) ; if ( this . isAttached ( ) && value !== undefined && value !== null ) { this . sendSetCellOp ( row , col , value , rowHandle , colHandle ) ; } } } for ( const consumer of this . consumers . values ( ) ) { consumer . cellsChanged ( rowStart , 0 , rowCount , this . colCount , this ) ; } } internal public _undoRemoveCols ( segment ) { const original = segment as PermutationSegment ; const { op , inserted } = this . cols . insertRelative ( original , original . cachedLength ) ; this . submitColMessage ( op ) ; original . transferToReplacement ( inserted ) ; const colStart = this . cols . getPosition ( inserted ) ; this . cols . handleCache . itemsChanged ( colStart , 0 , inserted . cachedLength ) ; let colHandle = inserted . start ; const colCount = inserted . cachedLength ; for ( let col = colStart ; col < colStart + colCount ; col ++ , colHandle ++ ) { for ( let row = 0 ; row < this . rowCount ; row ++ ) { const rowHandle = this . rowHandles . getHandle ( row ) ; const value = this . cells . getCell ( rowHandle , colHandle ) ; if ( this . isAttached ( ) && value !== undefined && value !== null ) { this . sendSetCellOp ( row , col , value , rowHandle , colHandle ) ; } } } for ( const consumer of this . consumers . values ( ) ) { consumer . cellsChanged ( 0 , colStart , this . rowCount , colCount , this ) ; } } protected summarizeCore ( serializer ) { const builder = new SummaryTreeBuilder ( ) ; builder . addWithStats ( SnapshotPath . rows , this . rows . summarize ( this . runtime , this . handle , serializer ) ) ; builder . addWithStats ( SnapshotPath . cols , this . cols . summarize ( this . runtime , this . handle , serializer ) ) ; builder . addBlob ( SnapshotPath . cells , serializer . stringify ( [ this . cells . snapshot ( ) , this . pending . snapshot ( ) , ] , this . handle ) ) ; return builder . getSummaryTree ( ) ; } /**
     * Runs serializer on the GC data for this SharedMatrix.
     * All the IFluidHandle's stored in the cells represent routes to other objects.
     */ protected processGCDataCore ( serializer ) { for ( let row = 0 ; row < this . rowCount ; row ++ ) { for ( let col = 0 ; col < this . colCount ; col ++ ) { serializer . stringify ( this . getCell ( row , col ) , this . handle ) ; } } } /**
     * Advances the 'localSeq' counter for the cell data operation currently being queued.
     *
     * Do not use with 'submitColMessage()/submitRowMessage()' as these helpers + the MergeTree will
     * automatically advance 'localSeq'.
     */ private nextLocalSeq ( ) { this . cols . getCollabWindow ( ) . localSeq ++ ; return ++ this . rows . getCollabWindow ( ) . localSeq ; } protected submitLocalMessage ( message , localOpMetadata ? ) { assert ( this . isAttached ( ) === true , 0x01d ) ; super . submitLocalMessage ( makeHandlesSerializable ( message , this . serializer , this . handle ) , localOpMetadata ) ; assert ( this . rows . getCollabWindow ( ) . localSeq === this . cols . getCollabWindow ( ) . localSeq , 0x01e ) ; } protected didAttach ( ) { if ( this . isAttached ( ) ) { this . rows . startOrUpdateCollaboration ( this . runtime . clientId ?? "attached" ) ; this . cols . startOrUpdateCollaboration ( this . runtime . clientId ?? "attached" ) ; } } protected onConnect ( ) { assert ( this . rows . getCollabWindow ( ) . collaborating === this . cols . getCollabWindow ( ) . collaborating , 0x01f ) ; this . rows . startOrUpdateCollaboration ( this . runtime . clientId as string ) ; this . cols . startOrUpdateCollaboration ( this . runtime . clientId as string ) ; } protected reSubmitCore ( content , localOpMetadata ) { switch ( content . target ) { case SnapshotPath . cols : this . submitColMessage ( this . cols . regeneratePendingOp ( content as IMergeTreeOp , localOpMetadata as SegmentGroup | SegmentGroup [ ] ) ) ; break ; case SnapshotPath . rows : this . submitRowMessage ( this . rows . regeneratePendingOp ( content as IMergeTreeOp , localOpMetadata as SegmentGroup | SegmentGroup [ ] ) ) ; break ; default : { assert ( content . type === MatrixOp . set , 0x020 ) ; const setOp = content as ISetOp < Serializable < T > > ; const { rowHandle , colHandle , localSeq } = localOpMetadata as ISetOpMetadata ; if ( this . isLatestPendingWrite ( rowHandle , colHandle , localSeq ) ) { const row = this . rows . handleToPosition ( rowHandle , localSeq ) ; const col = this . cols . handleToPosition ( colHandle , localSeq ) ; if ( row >= 0 && col >= 0 ) { this . sendSetCellOp ( row , col , setOp . value , rowHandle , colHandle , localSeq ) ; } } break ; } } } protected onDisconnect ( ) { } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.loadCore}
     */ protected async loadCore ( storage ) { try { await this . rows . load ( this . runtime , new ObjectStoragePartition ( storage , SnapshotPath . rows ) , this . serializer ) ; await this . cols . load ( this . runtime , new ObjectStoragePartition ( storage , SnapshotPath . cols ) , this . serializer ) ; const [ cellData , pendingCliSeqData ] = await deserializeBlob ( storage , SnapshotPath . cells , this . serializer ) ; this . cells = SparseArray2D . load ( cellData ) ; this . pending = SparseArray2D . load ( pendingCliSeqData ) ; } catch ( error ) { this . logger . sendErrorEvent ( { eventName : "MatrixLoadFailed" } , error ) ; } } protected processCore ( rawMessage , local , localOpMetadata ) { const msg = parseHandles ( rawMessage , this . serializer ) ; const contents = msg . contents ; switch ( contents . target ) { case SnapshotPath . cols : this . cols . applyMsg ( msg ) ; break ; case SnapshotPath . rows : this . rows . applyMsg ( msg ) ; break ; default : { assert ( contents . type === MatrixOp . set , 0x021 ) ; const { referenceSequenceNumber : refSeq , clientId } = rawMessage ; const { row , col } = contents ; if ( local ) { const { rowHandle , colHandle , localSeq } = localOpMetadata as ISetOpMetadata ; if ( this . isLatestPendingWrite ( rowHandle , colHandle , localSeq ) ) { this . pending . setCell ( rowHandle , colHandle , undefined ) ; } } else { const rowClientId = this . rows . getOrAddShortClientId ( clientId ) ; const adjustedRow = this . rows . adjustPosition ( row , refSeq , rowClientId ) ; if ( adjustedRow !== undefined ) { const colClientId = this . cols . getOrAddShortClientId ( clientId ) ; const adjustedCol = this . cols . adjustPosition ( col , refSeq , colClientId ) ; if ( adjustedCol !== undefined ) { const rowHandle = this . rows . getAllocatedHandle ( adjustedRow ) ; const colHandle = this . cols . getAllocatedHandle ( adjustedCol ) ; assert ( isHandleValid ( rowHandle ) && isHandleValid ( colHandle ) , 0x022 ) ; if ( this . pending . getCell ( rowHandle , colHandle ) === undefined ) { const { value } = contents ; this . cells . setCell ( rowHandle , colHandle , value ) ; for ( const consumer of this . consumers . values ( ) ) { consumer . cellsChanged ( adjustedRow , adjustedCol , 1 , 1 , this ) ; } } } } } } } } protected registerCore ( ) { this . rows . startOrUpdateCollaboration ( this . runtime . clientId , 0 ) ; this . cols . startOrUpdateCollaboration ( this . runtime . clientId , 0 ) ; } private readonly onRowDelta = ( position , removedCount , insertedCount ) => { for ( const consumer of this . consumers ) { consumer . rowsChanged ( position , removedCount , insertedCount , this ) ; } } ; private readonly onColDelta = ( position , removedCount , insertedCount ) => { for ( const consumer of this . consumers ) { consumer . colsChanged ( position , removedCount , insertedCount , this ) ; } } ; private readonly onRowHandlesRecycled = ( rowHandles ) => { for ( const rowHandle of rowHandles ) { this . cells . clearRows ( rowHandle , 1 ) ; this . pending . clearRows ( rowHandle , 1 ) ; } } ; private readonly onColHandlesRecycled = ( colHandles ) => { for ( const colHandle of colHandles ) { this . cells . clearCols ( colHandle , 1 ) ; this . pending . clearCols ( colHandle , 1 ) ; } } ; /**
     * Returns true if the latest pending write to the cell indicated by the given row/col handles
     * matches the given 'localSeq'.
     *
     * A return value of `true` indicates that there are no later local operations queued that will
     * clobber the write op at the given 'localSeq'.  This includes later ops that overwrite the cell
     * with a different value as well as row/col removals that might recycled the given row/col handles.
     */ private isLatestPendingWrite ( rowHandle , colHandle , localSeq ) { const pendingLocalSeq = this . pending . getCell ( rowHandle , colHandle ) ! ; assert ( ! ( pendingLocalSeq < localSeq ) , 0x023 ) ; return pendingLocalSeq === localSeq ; } public toString ( ) { let s = `client:${ this . runtime . clientId }\nrows: ${ this . rows . toString ( ) }\ncols: ${ this . cols . toString ( ) }\n\n` ; for ( let r = 0 ; r < this . rowCount ; r ++ ) { s += `  [` ; for ( let c = 0 ; c < this . colCount ; c ++ ) { if ( c > 0 ) { s += ", " ; } s += `${ this . serializer . stringify ( this . getCell ( r , c ) , this . handle ) }` ; } s += "]\n" ; } return `${ s }\n` ; } protected applyStashedOp ( ) { throw new Error ( "not implemented" ) ; } }