import { close , createReadStream , createWriteStream , fdatasync , mkdir , open , stat , unlink } from "fs" ; import multistream = require ( "multistream" ) ; import { join } from "path" ; import { Writable } from "stream" ; import { promisify } from "util" ; import uuid = require ( "uuid" ) ; import { ZERO_EXTENT_ID } from "../../blob/persistence/IBlobMetadataStore" ; import ILogger from "../ILogger" ; import BufferStream from "../utils/BufferStream" ; import { DEFAULT_MAX_EXTENT_SIZE , DEFAULT_READ_CONCURRENCY } from "../utils/constants" ; import { rimrafAsync } from "../utils/utils" ; import ZeroBytesStream from "../ZeroBytesStream" ; import IExtentMetadataStore , { IExtentModel } from "./IExtentMetadataStore" ; import IExtentStore , { IExtentChunk , StoreDestinationArray } from "./IExtentStore" ; import IOperationQueue from "./IOperationQueue" ; import OperationQueue from "./OperationQueue" ; const statAsync = promisify ( stat ) ; const mkdirAsync = promisify ( mkdir ) ; const unlinkAsync = promisify ( unlink ) ; const MAX_EXTENT_SIZE = DEFAULT_MAX_EXTENT_SIZE ; enum AppendStatusCode { Idle , Appending } interface IAppendExtent { id : string ; offset : number ; appendStatus : AppendStatusCode ; locationId : string ; fd ? : number ; } const openAsync = promisify ( open ) ; const closeAsync = promisify ( close ) ; export class implements IExtentStore export default class FSExtentStore implements IExtentStore { private readonly metadataStore : IExtentMetadataStore ; private readonly appendQueue : IOperationQueue ; private readonly readQueue : IOperationQueue ; private initialized : boolean = false ; private closed : boolean = true ; private activeWriteExtents : IAppendExtent [ ] ; private activeWriteExtentsNumber : number ; private persistencyPath : Map < string , string > ; public constructor ( metadata , private readonly persistencyConfiguration , private readonly logger ) { this . activeWriteExtents = [ ] ; this . persistencyPath = new Map < string , string > ( ) ; for ( const storeDestination of persistencyConfiguration ) { this . persistencyPath . set ( storeDestination . locationId , storeDestination . locationPath ) ; for ( let i = 0 ; i < storeDestination . maxConcurrency ; i ++ ) { const appendExtent = this . createAppendExtent ( storeDestination . locationId ) ; this . activeWriteExtents . push ( appendExtent ) ; } } this . activeWriteExtentsNumber = this . activeWriteExtents . length ; this . metadataStore = metadata ; this . appendQueue = new OperationQueue ( this . activeWriteExtentsNumber , logger ) ; this . readQueue = new OperationQueue ( DEFAULT_READ_CONCURRENCY , logger ) ; } public isInitialized ( ) { return this . initialized ; } public isClosed ( ) { return this . closed ; } public async init ( ) { for ( const storeDestination of this . persistencyConfiguration ) { try { await statAsync ( storeDestination . locationPath ) ; } catch { await mkdirAsync ( storeDestination . locationPath ) ; } } if ( ! this . metadataStore . isInitialized ( ) ) { await this . metadataStore . init ( ) ; } this . initialized = true ; this . closed = false ; } public async close ( ) { if ( ! this . metadataStore . isClosed ( ) ) { await this . metadataStore . close ( ) ; } this . closed = true ; } public async clean ( ) { if ( this . isClosed ( ) ) { for ( const path of this . persistencyConfiguration ) { try { await rimrafAsync ( path . locationPath ) ; } catch { } } return ; } throw new Error ( `Cannot clean FSExtentStore, it's not closed.` ) ; } param { ( NodeJS . ReadableStream | Buffer ) } data param { string } contextId returns { Promise < IExtentChunk > } memberof public async appendExtent ( data , contextId ? ) { const op = ( ) => new Promise < IExtentChunk > ( ( resolve , reject ) => { ( async ( ) : Promise < IExtentChunk > => { let appendExtentIdx = 0 ; for ( let i = 1 ; i < this . activeWriteExtentsNumber ; i ++ ) { if ( this . activeWriteExtents [ i ] . appendStatus === AppendStatusCode . Idle ) { appendExtentIdx = i ; break ; } } this . activeWriteExtents [ appendExtentIdx ] . appendStatus = AppendStatusCode . Appending ; this . logger . info ( `FSExtentStore:appendExtent() Select extent from idle location for extent append operation. LocationId:${ appendExtentIdx } extentId:${ this . activeWriteExtents [ appendExtentIdx ] . id } offset:${ this . activeWriteExtents [ appendExtentIdx ] . offset } MAX_EXTENT_SIZE:${ MAX_EXTENT_SIZE } ` , contextId ) ; if ( this . activeWriteExtents [ appendExtentIdx ] . offset >= MAX_EXTENT_SIZE ) { this . logger . info ( `FSExtentStore:appendExtent() Size of selected extent offset is larger than maximum extent size ${ MAX_EXTENT_SIZE } bytes, try appending to new extent.` , contextId ) ; const selectedFd = this . activeWriteExtents [ appendExtentIdx ] . fd ; if ( selectedFd ) { this . logger . info ( `FSExtentStore:appendExtent() Close unused fd:${ selectedFd }.` , contextId ) ; try { await closeAsync ( selectedFd ) ; } catch ( err ) { this . logger . error ( `FSExtentStore:appendExtent() Close unused fd:${ selectedFd } error:${ JSON . stringify ( err ) }.` , contextId ) ; } } await this . getNewExtent ( this . activeWriteExtents [ appendExtentIdx ] ) ; this . logger . info ( `FSExtentStore:appendExtent() Allocated new extent LocationID:${ appendExtentIdx } extentId:${ this . activeWriteExtents [ appendExtentIdx ] . id } offset:${ this . activeWriteExtents [ appendExtentIdx ] . offset } MAX_EXTENT_SIZE:${ MAX_EXTENT_SIZE } ` , contextId ) ; } let rs ; if ( data instanceof Buffer ) { rs = new BufferStream ( data ) ; } else { rs = data ; } const appendExtent = this . activeWriteExtents [ appendExtentIdx ] ; const id = appendExtent . id ; const path = this . generateExtentPath ( appendExtent . locationId , id ) ; let fd = appendExtent . fd ; this . logger . debug ( `FSExtentStore:appendExtent() Get fd:${ fd } for extent:${ id } from cache.` , contextId ) ; if ( fd === undefined ) { fd = await openAsync ( path , "a" ) ; appendExtent . fd = fd ; this . logger . debug ( `FSExtentStore:appendExtent() Open file:${ path } for extent:${ id }, get new fd:${ fd }` , contextId ) ; } const ws = createWriteStream ( path , { fd , autoClose : false } ) ; this . logger . debug ( `FSExtentStore:appendExtent() Created write stream for fd:${ fd }` , contextId ) ; let count = 0 ; this . logger . debug ( `FSExtentStore:appendExtent() Start writing to extent ${ id }` , contextId ) ; try { count = await this . streamPipe ( rs , ws , fd , contextId ) ; const offset = appendExtent . offset ; appendExtent . offset += count ; const extent = { id , locationId : appendExtent . locationId , path : id , size : count + offset , lastModifiedInMS : Date . now ( ) } ; this . logger . debug ( `FSExtentStore:appendExtent() Write finish, start updating extent metadata. extent:${ JSON . stringify ( extent ) }` , contextId ) ; await this . metadataStore . updateExtent ( extent ) ; this . logger . debug ( `FSExtentStore:appendExtent() Update extent metadata done. Resolve()` , contextId ) ; appendExtent . appendStatus = AppendStatusCode . Idle ; return { id , offset , count } ; } catch ( err ) { appendExtent . appendStatus = AppendStatusCode . Idle ; throw err ; } } ) ( ) . then ( resolve ) . catch ( reject ) ; } ) ; return this . appendQueue . operate ( op , contextId ) ; } param { IExtentChunk } extentChunk returns { Promise < NodeJS . ReadableStream > } memberof public async readExtent ( extentChunk ? , contextId ? ) { if ( extentChunk === undefined || extentChunk . count === 0 ) { return new ZeroBytesStream ( 0 ) ; } if ( extentChunk . id === ZERO_EXTENT_ID ) { const subRangeCount = Math . min ( extentChunk . count ) ; return new ZeroBytesStream ( subRangeCount ) ; } const persistencyId = await this . metadataStore . getExtentLocationId ( extentChunk . id ) ; const path = this . generateExtentPath ( persistencyId , extentChunk . id ) ; const op = ( ) => new Promise < NodeJS . ReadableStream > ( ( resolve , reject ) => { this . logger . verbose ( `FSExtentStore:readExtent() Creating read stream. LocationId:${ persistencyId } extentId:${ extentChunk . id } path:${ path } offset:${ extentChunk . offset } count:${ extentChunk . count } end:${ extentChunk . offset + extentChunk . count - 1 }` , contextId ) ; const stream = createReadStream ( path , { start : extentChunk . offset , end : extentChunk . offset + extentChunk . count - 1 } ) . on ( "close" , ( ) => { this . logger . verbose ( `FSExtentStore:readExtent() Read stream closed. LocationId:${ persistencyId } extentId:${ extentChunk . id } path:${ path } offset:${ extentChunk . offset } count:${ extentChunk . count } end:${ extentChunk . offset + extentChunk . count - 1 }` , contextId ) ; } ) ; resolve ( stream ) ; } ) ; return this . readQueue . operate ( op , contextId ) ; } param { ( IExtentChunk ) [ ] } extentChunkArray param { number } offset param { number } count param { string } contextId returns { Promise < NodeJS . ReadableStream > } memberof public async readExtents ( extentChunkArray , offset = 0 , count = Infinity , contextId ? ) { this . logger . verbose ( `FSExtentStore:readExtents() Start read from multi extents...` , contextId ) ; if ( count === 0 ) { return new ZeroBytesStream ( 0 ) ; } const start = offset ; const end = offset + count ; const streams = [ ] ; let accumulatedOffset = 0 ; for ( const chunk of extentChunkArray ) { const nextOffset = accumulatedOffset + chunk . count ; if ( nextOffset <= start ) { accumulatedOffset = nextOffset ; continue ; } else if ( end <= accumulatedOffset ) { break ; } else { let chunkStart = chunk . offset ; let chunkEnd = chunk . offset + chunk . count ; if ( start > accumulatedOffset ) { chunkStart = chunkStart + start - accumulatedOffset ; } if ( end <= nextOffset ) { chunkEnd = chunkEnd - ( nextOffset - end ) ; } streams . push ( await this . readExtent ( { id : chunk . id , offset : chunkStart , count : chunkEnd - chunkStart } , contextId ) ) ; accumulatedOffset = nextOffset ; } } if ( end !== Infinity && accumulatedOffset < end ) { throw new RangeError ( `Not enough payload data error. Total length of payloads is ${ accumulatedOffset }, while required data offset is ${ offset }, count is ${ count }.` ) ; } return multistream ( streams ) ; } param { Iterable < string > } persistency returns { Promise < number > } memberof public async deleteExtents ( extents ) { let count = 0 ; for ( const id of extents ) { if ( this . isActiveExtent ( id ) ) { this . logger . debug ( `FSExtentStore:deleteExtents() Skip deleting active extent:${ id }` ) ; continue ; } const locationId = await this . metadataStore . getExtentLocationId ( id ) ; const path = this . generateExtentPath ( locationId , id ) ; this . logger . debug ( `FSExtentStore:deleteExtents() Delete extent:${ id } location:${ locationId } path:${ path }` ) ; try { await unlinkAsync ( path ) ; await this . metadataStore . deleteExtent ( id ) ; } catch ( err ) { if ( err . code === "ENOENT" ) { await this . metadataStore . deleteExtent ( id ) ; } } count ++ ; } return count ; } returns { IExtentMetadataStore } memberof public getMetadataStore ( ) { return this . metadataStore ; } private async streamPipe ( rs , ws , fd ? , contextId ? ) { return new Promise < number > ( ( resolve , reject ) => { this . logger . debug ( `FSExtentStore:streamPipe() Start piping data to write stream` , contextId ) ; let count = 0 ; let wsEnd = false ; rs . on ( "data" , data => { count += data . length ; if ( ! ws . write ( data ) ) { rs . pause ( ) ; } } ) . on ( "end" , ( ) => { this . logger . debug ( `FSExtentStore:streamPipe() Readable stream triggers close event, ${ count } bytes piped` , contextId ) ; if ( ! wsEnd ) { this . logger . debug ( `FSExtentStore:streamPipe() Invoke write stream end()` , contextId ) ; ws . end ( ) ; wsEnd = true ; } } ) . on ( "close" , ( ) => { this . logger . debug ( `FSExtentStore:streamPipe() Readable stream triggers close event, ${ count } bytes piped` , contextId ) ; if ( ! wsEnd ) { this . logger . debug ( `FSExtentStore:streamPipe() Invoke write stream end()` , contextId ) ; ws . end ( ) ; wsEnd = true ; } } ) . on ( "error" , err => { this . logger . debug ( `FSExtentStore:streamPipe() Readable stream triggers error event, error:${ JSON . stringify ( err ) }, after ${ count } bytes piped. Invoke write stream destroy() method.` , contextId ) ; ws . destroy ( err ) ; } ) ; ws . on ( "drain" , ( ) => { rs . resume ( ) ; } ) . on ( "finish" , ( ) => { if ( typeof fd === "number" ) { this . logger . debug ( `FSExtentStore:streamPipe() Writable stream triggers finish event, after ${ count } bytes piped. Flush data to fd:${ fd }.` , contextId ) ; fdatasync ( fd , err => { if ( err ) { this . logger . debug ( `FSExtentStore:streamPipe() Flush data to fd:${ fd } failed with error:${ JSON . stringify ( err ) }. Reject streamPipe().` , contextId ) ; reject ( err ) ; } else { this . logger . debug ( `FSExtentStore:streamPipe() Flush data to fd:${ fd } successfully. Resolve streamPipe().` , contextId ) ; resolve ( count ) ; } } ) ; } else { this . logger . debug ( `FSExtentStore:streamPipe() Resolve streamPipe() without flushing data.` , contextId ) ; resolve ( count ) ; } } ) . on ( "error" , err => { this . logger . debug ( `FSExtentStore:streamPipe() Writable stream triggers error event, error:${ JSON . stringify ( err ) }, after ${ count } bytes piped. Reject streamPipe().` , contextId ) ; reject ( err ) ; } ) ; } ) ; } private param { string } id returns { boolean } memberof private isActiveExtent ( id ) { for ( const extent of this . activeWriteExtents ) { if ( extent . id === id ) { return true ; } } return false ; } private param { string } persistencyPath returns { IAppendExtent } memberof private createAppendExtent ( persistencyId ) { return { id : uuid ( ) , offset : 0 , appendStatus : AppendStatusCode . Idle , locationId : persistencyId } ; } private param { IAppendExtent } appendExtent memberof private getNewExtent ( appendExtent ) { appendExtent . id = uuid ( ) ; appendExtent . offset = 0 ; appendExtent . fd = undefined ; } private param { string } extentId returns { string } memberof private generateExtentPath ( persistencyId , extentId ) { const directoryPath = this . persistencyPath . get ( persistencyId ) ; if ( ! directoryPath ) { } return join ( directoryPath ! , extentId ) ; } }