import { ILoader } from "@fluidframework/container-definitions" ; import { ISharedMap } from "@fluidframework/map" ; import * as MergeTree from "@fluidframework/merge-tree" ; import { IFluidDataStoreRuntime } from "@fluidframework/datastore-definitions" ; import { ISharedString , SequenceDeltaEvent } from "@fluidframework/sequence" ; import queue from "async/queue" ; import clone from "lodash/clone" ; import { Histogram , RateCounter } from "./counters" ; let play = false ; let metrics ; const ackCounter = new RateCounter ( ) ; const latencyCounter = new RateCounter ( ) ; const pingCounter = new RateCounter ( ) ; const processCounter = new RateCounter ( ) ; const typingCounter = new RateCounter ( ) ; const serverOrderCounter = new RateCounter ( ) ; export interface IAuthor { ackCounter : RateCounter ; latencyCounter : RateCounter ; typingCounter : RateCounter ; pingCounter : RateCounter ; metrics : IScribeMetrics ; ss : ISharedString ; } /**
 * Toggle between play and pause.
 */ export function toggleAuthorPlay ( ) { play = ! play ; } /**
 * Processes the input text into a normalized form for the shared string
 */ export function normalizeText ( input ) { let result = "" ; const segments = MergeTree . loadSegments ( input , 0 ) ; for ( const segment of segments ) { result += ( segment as MergeTree . TextSegment ) . text ; } return result ; } export interface IScribeMetrics { latencyAverage ? : number ; latencyStdDev ? : number ; latencyMinimum ? : number ; latencyMaximum ? : number ; ackRate ? : number ; typingRate ? : number ; serverAverage ? : number ; totalOps : number ; typingProgress ? : number ; ackProgress ? : number ; time : number ; textLength : number ; pingAverage ? : number ; pingMaximum ? : number ; processAverage ? : number ; typingInterval : number ; writers : number ; } export declare type ScribeMetricsCallback = ( metrics : IScribeMetrics ) => void ; export declare type QueueCallback = ( metrics : IScribeMetrics , author : IAuthor ) => void ; export async function requestSharedString ( loader , urlBase ) { const response = await loader . request ( { url : `${ urlBase }/sharedstring` } ) ; if ( response . status !== 200 || response . mimeType !== "fluid/sharedstring" ) { return Promise . reject ( new Error ( "Invalid document" ) ) ; } return response . value as ISharedString ; } export async function typeFile ( loader , urlBase , runtime , ss , chunkMap , fileText , intervalTime , writers , scribeCallback ) { const metricsArray = [ ] ; let q ; metrics = { ackProgress : undefined , ackRate : undefined , latencyAverage : undefined , latencyMaximum : undefined , latencyMinimum : undefined , latencyStdDev : undefined , pingAverage : undefined , pingMaximum : undefined , processAverage : undefined , serverAverage : undefined , textLength : fileText . length , time : 0 , totalOps : 0 , typingInterval : intervalTime , typingProgress : undefined , typingRate : undefined , writers } ; const m = { ackProgress : undefined , ackRate : undefined , latencyAverage : undefined , latencyMaximum : undefined , latencyMinimum : undefined , latencyStdDev : undefined , pingAverage : undefined , pingMaximum : undefined , processAverage : undefined , serverAverage : undefined , textLength : fileText . length , time : 0 , totalOps : 0 , typingInterval : intervalTime , typingProgress : undefined , typingRate : undefined , writers } ; let author = { ackCounter : new RateCounter ( ) , latencyCounter : new RateCounter ( ) , metrics : clone ( m ) , pingCounter : new RateCounter ( ) , ss , typingCounter : new RateCounter ( ) } ; const authors = [ author ] ; for ( let i = 1 ; i < writers ; i ++ ) { const sharedString = await requestSharedString ( loader , urlBase ) ; author = { ackCounter : new RateCounter ( ) , latencyCounter : new RateCounter ( ) , metrics : clone ( m ) , pingCounter : new RateCounter ( ) , ss : sharedString , typingCounter : new RateCounter ( ) } ; authors . push ( author ) ; } if ( writers === 1 ) { const startTime = Date . now ( ) ; typingCounter . reset ( ) ; ackCounter . reset ( ) ; latencyCounter . reset ( ) ; pingCounter . reset ( ) ; await new Promise < void > ( ( resolve ) => setTimeout ( ( ) => resolve ( ) , 1000 ) ) ; const metric = await typeChunk ( authors [ 0 ] , runtime , "p-0" , fileText , intervalTime , scribeCallback , scribeCallback ) ; metric . time = Date . now ( ) - startTime ; return metric ; } else { let totalKeys = 0 ; let curKey = 0 ; const startTime = Date . now ( ) ; typingCounter . reset ( ) ; ackCounter . reset ( ) ; latencyCounter . reset ( ) ; pingCounter . reset ( ) ; return new Promise ( ( resolve , reject ) => { q = queue ( async ( chunkKey , queueCallback ) => { const chunk = chunkMap . get ( chunkKey ) ; const a = authors . shift ( ) ! ; curKey ++ ; metrics . typingProgress = curKey / totalKeys ; typeChunk ( a , runtime , chunkKey , chunk , intervalTime , scribeCallback , queueCallback ) ; } , writers ) ; for ( const chunkKey of chunkMap . keys ( ) ) { totalKeys ++ ; q . push ( chunkKey , ( metric , a ) => { authors . push ( a ) ; metricsArray . push ( metric ) ; } ) ; } q . drain ( ( ) => { const now = Date . now ( ) ; metrics . time = now - startTime ; resolve ( metricsArray [ 0 ] ) ; } ) ; } ) ; } } /**
 * Types the given file into the shared string - starting at the end of the string
 */ export async function typeChunk ( a , runtime , chunkKey , chunk , intervalTime , scribeCallback , queueCallback ) { return new Promise < IScribeMetrics > ( ( resolve , reject ) => { let readPosition = 0 ; let totalOps = 0 ; const histogramRange = 5 ; const histogram = new Histogram ( histogramRange ) ; const samplingRate = 1000 ; let mean = 0 ; let stdDev = 0 ; runtime . deltaManager . on ( "pong" , ( latency ) => { pingCounter . increment ( latency ) ; } ) ; runtime . deltaManager . on ( "op" , ( _ , time ) => { processCounter . increment ( time ) ; } ) ; a . ss . on ( "sequenceDelta" , ( ev ) => { const message = ev . opArgs . sequencedMessage ; totalOps ++ ; if ( message !== undefined && message . traces && message . clientSequenceNumber && message . clientSequenceNumber > 100 && ev . isLocal ) { ackCounter . increment ( 1 ) ; if ( ackCounter . elapsed ( ) > samplingRate ) { const rate = ackCounter . getRate ( ) * 1000 ; metrics . ackRate = rate ; } let clientStart = 0 ; let clientEnd = 0 ; let orderBegin = 0 ; let orderEnd = 0 ; for ( const trace of message . traces ) { if ( trace . service === "alfred" && trace . action === "start" ) { orderBegin = trace . timestamp ; } else if ( trace . service === "scriptorium" && trace . action === "end" ) { orderEnd = trace . timestamp ; } else if ( trace . service === "browser" && trace . action === "start" ) { clientStart = trace . timestamp ; } else if ( trace . service === "browser" && trace . action === "end" ) { clientEnd = trace . timestamp ; } } const roundTrip = clientEnd - clientStart ; latencyCounter . increment ( roundTrip ) ; serverOrderCounter . increment ( orderEnd - orderBegin ) ; metrics . pingAverage = pingCounter . getValue ( ) / pingCounter . getSamples ( ) ; metrics . pingMaximum = pingCounter . getMaximum ( ) ; metrics . processAverage = processCounter . getValue ( ) / processCounter . getSamples ( ) ; histogram . add ( roundTrip ) ; const samples = latencyCounter . getSamples ( ) ; metrics . latencyMinimum = latencyCounter . getMinimum ( ) ; metrics . latencyMaximum = latencyCounter . getMaximum ( ) ; metrics . latencyAverage = latencyCounter . getValue ( ) / samples ; metrics . totalOps = totalOps ; metrics . serverAverage = serverOrderCounter . getValue ( ) / serverOrderCounter . getSamples ( ) ; stdDev = stdDev + ( roundTrip - metrics . latencyAverage ) * ( roundTrip - mean ) ; metrics . latencyStdDev = samples > 1 ? Math . sqrt ( stdDev / ( samples - 1 ) ) : 0 ; mean = metrics . latencyAverage ; scribeCallback ( clone ( metrics ) ) ; } } ) ; function trackOperation ( fn ) { typingCounter . increment ( 1 ) ; fn ( ) ; } function type ( ) { if ( readPosition === chunk . length ) { const rate = typingCounter . getRate ( ) * 1000 ; metrics . typingRate = rate ; resolve ( metrics ) ; queueCallback ( metrics , a ) ; return false ; } if ( ! play ) { return true ; } const relPosit = { before : true , id : chunkKey , offset : 0 } ; const pos = a . ss . posFromRelativePos ( relPosit ) ; let code = chunk . charCodeAt ( readPosition ) ; if ( code === 13 ) { readPosition ++ ; code = chunk . charCodeAt ( readPosition ) ; } trackOperation ( ( ) => { const char = chunk . charAt ( readPosition ) ; if ( code === 10 ) { a . ss . insertMarker ( pos , MergeTree . ReferenceType . Tile , { [ MergeTree . reservedTileLabelsKey ] : [ "pg" ] } ) ; readPosition ++ ; metrics . typingProgress = readPosition / chunk . length ; } else { a . ss . insertText ( pos , char ) ; readPosition ++ ; } } ) ; return true ; } function typeFast ( ) { setImmediate ( ( ) => { if ( type ( ) ) { typeFast ( ) ; } } ) ; } if ( intervalTime === 0 && typeof setImmediate === "function" ) { typeFast ( ) ; } else { const interval = setInterval ( ( ) => { for ( let i = 0 ; i < 1 ; i ++ ) { if ( ! type ( ) ) { clearInterval ( interval ) ; break ; } } } , intervalTime ) ; } } ) ; }