import { strict as assert } from "assert" ; import { Random } from "best-random" ; import { IChannelServices } from "@fluidframework/datastore-definitions" ; import { MockFluidDataStoreRuntime , MockStorage , MockContainerRuntimeFactoryForReconnection , MockContainerRuntimeForReconnection , } from "@fluidframework/test-runtime-utils" ; import { SharedDelta } from "./delta" ; describe ( "SharedOT" , ( ) => { describe ( "stress" , ( ) => { let containerRuntimeFactory ; let docs ; let runtimes = [ ] ; let trace ; const extract = ( doc ) => { return doc . text ; } ; /**
         * Drains the queue of pending ops for each client and vets that all docs converged on the same state.
         */ const expect = async ( ) => { { for ( let i = 0 ; i < runtimes . length ; i ++ ) { const runtime = runtimes [ i ] ; if ( ! runtime . connected ) { trace ?. push ( `containerRuntime${ i + 1 }.connected = true;` ) ; runtime . connected = true ; } } } trace ?. push ( "await expect();" ) ; containerRuntimeFactory . processAllMessages ( ) ; const doc0 = docs [ 0 ] ; const actual0 = extract ( doc0 ) ; { for ( let i = 1 ; i < docs . length ; i ++ ) { const docN = docs [ i ] ; const actualN = extract ( docN ) ; assert . deepEqual ( actual0 , actualN ) ; } } } ; /**
         * Performs a stress run using the given parameters.
         *
         * 'syncProbability' is the probability that the clients will drain their queue of incoming messages
         * and check for convergence.
         *
         * 'disconnectProbability' is the probability that a client will disconnect, forcing it to regenerate
         * and resubmit any pending local operations on the next sync.
         *
         * 'seed' is the 32-bit integer used to seed the PRNG.
         */ async function stress ( numClients , numOps , syncProbability , disconnectProbability , seed ) { try { docs = [ ] ; runtimes = [ ] ; trace = [ ] ; containerRuntimeFactory = new MockContainerRuntimeFactoryForReconnection ( ) ; for ( let i = 0 ; i < numClients ; i ++ ) { const dataStoreRuntimeN = new MockFluidDataStoreRuntime ( ) ; const containerRuntimeN = containerRuntimeFactory . createContainerRuntime ( dataStoreRuntimeN ) ; const servicesN = { deltaConnection : containerRuntimeN . createDeltaConnection ( ) , objectStorage : new MockStorage ( ) } ; const docN = SharedDelta . getFactory ( ) . create ( dataStoreRuntimeN , `doc-${ i }` ) ; docN . connect ( servicesN ) ; docs . push ( docN ) ; runtimes . push ( containerRuntimeN ) ; } const float64 = new Random ( seed ) . float64 ; const int32 = ( max = 0x7FFFFFFF ) => ( float64 ( ) * max ) | 0 ; const randomText = ( ) => `${ float64 ( ) . toString ( 36 ) . substr ( 0 , int32 ( 12 ) ) }` ; const insert = ( docIndex , position , text ) => { trace ?. push ( `doc${ docIndex + 1 }.insert(/* position: */ ${ position }, /* text: */ ${ JSON . stringify ( text ) });` ) ; docs [ docIndex ] . insert ( position , text ) ; } ; const del = ( docIndex , start , end ) => { trace ?. push ( `doc${ docIndex + 1 }.delete(/* start: */ ${ start }, /* end: */ ${ end });` ) ; docs [ docIndex ] . delete ( start , end ) ; } ; for ( let i = 0 ; i < numOps ; i ++ ) { const docIndex = int32 ( docs . length ) ; const doc = docs [ docIndex ] ; const { length } = doc ; switch ( int32 ( 4 ) ) { case 0 : { if ( length > 0 ) { const start = int32 ( length ) ; const end = int32 ( start - length + 1 ) + start ; del ( docIndex , start , end ) ; } break ; } default : insert ( docIndex , int32 ( length + 1 ) , randomText ( ) ) ; break ; } if ( runtimes [ docIndex ] . connected && float64 ( ) < disconnectProbability ) { trace ?. push ( `containerRuntime${ docIndex + 1 }.connected = false;` ) ; runtimes [ docIndex ] . connected = false ; } if ( float64 ( ) < syncProbability ) { await expect ( ) ; } } await expect ( ) ; } catch ( error ) { for ( const s of trace ) { console . log ( s ) ; } for ( const m of docs ) { console . log ( m . toString ( ) ) ; } throw error ; } } for ( const { numClients , numOps , syncProbability , disconnectProbability , seed } of [ { numClients : 2 , numOps : 1000 , syncProbability : 0.3 , disconnectProbability : 0 , seed : 0x84d43a0a } , { numClients : 3 , numOps : 1000 , syncProbability : 0.1 , disconnectProbability : 0 , seed : 0x655c763b } , { numClients : 5 , numOps : 200 , syncProbability : 0.0 , disconnectProbability : 0 , seed : 0x2f98736d } , { numClients : 2 , numOps : 1000 , syncProbability : 0.3 , disconnectProbability : 0.25 , seed : 0x84d43a0a } , ] ) { it ( `Stress (numClients=${ numClients } numOps=${ numOps } syncProbability=${ syncProbability } disconnectProbability=${ disconnectProbability } seed=0x${ seed . toString ( 16 ) . padStart ( 8 , "0" ) })` , async function ( ) { this . timeout ( 20000 ) ; await stress ( numClients , numOps , syncProbability , disconnectProbability , seed ) ; } ) ; } it . skip ( "stress-loop" , async function ( ) { this . timeout ( 0 ) ; let iterations = 0 ; const start = Date . now ( ) ; let lastStatus = start ; while ( true ) { await stress ( 2 , 1000 , 0.1 , 0 , ( Math . random ( ) * 0x100000000 ) >>> 0 ) ; ++ iterations ; const now = Date . now ( ) ; if ( now - lastStatus > 5000 ) { process . stdout . write ( `Stress loop: ${ iterations } iterations completed - Total Elapsed: ${ ( ( Date . now ( ) - start ) / 1000 ) . toFixed ( 2 ) }s\n` ) ; lastStatus = now ; } } } ) ; } ) ; } ) ;