<commit_before>const fs = require('fs')
const pump = require('pump')
const errors = require('./errors')
const byteStream = require('byte-stream')
const through = require('through2')
const pumpify = require('pumpify')
const messages = require('./messages')

const noop = () => {}

const {
  O_RDONLY,
  O_WRONLY,
  O_RDWR,
  O_CREAT,
  O_TRUNC,
  O_APPEND,
  O_SYNC,
  O_EXCL
} = fs.constants
const O_ACCMODE = 3

class FileDescriptor {
  constructor (drive, path, stat, readable, writable, appending, creating) {
    this.drive = drive
    this.stat = stat
    this.path = path

    this.readable = readable
    this.writable = writable
    this.creating = creating
    this.appending = appending

    this.position = null
    this.blockPosition = stat ? stat.offset : null
    this.blockOffset = 0

    this._err = null
    if (this.writable) {
      if (this.appending) {
        this._appendStream = this.drive.createReadStream(this.path)
        this.position = this.stat.size
      }
    }

    this._batcher = byteStream({ time: 100, limit: 4096 * 16 })

    this._range = null
  }

  read (buffer, offset, len, pos, cb) {
    if (!this.readable) return cb(new errors.BadFileDescriptor('File descriptor not open for reading.'))
    if (this.position !== null && this.position === pos) this._read(buffer, offset, len, cb)
    else this._seekAndRead(buffer, offset, len, pos, cb)
  }

  write (buffer, offset, len, pos, cb) {
    if (!this.writable) return cb(new errors.BadFileDescriptor('File descriptor not open for writing.'))
    if (!this.stat && !this.creating) {
      return cb(new errors.BadFileDescriptor('File descriptor not open in create mode.'))
    }
    if (this.position !== null && pos !== this.position) {
      return cb(new errors.BadFileDescriptor('Random-access writes are not currently supported.'))
    }
    if (this.appending && pos < this.stat.size) {
      return cb(new errors.BadFileDescriptor('Position cannot be less than the file size when appending.'))
    }
    if (!this._writeStream && !this.appending && pos) {
      return cb(new errors.BadFileDescriptor('Random-access writes are not currently supported.'))
    }

    buffer = Buffer.from(buffer)

    const self = this
    if (!this._writeStream) {
      const writeStream = this.drive.createWriteStream(this.path)
      this._writeStream = pumpify(this._batcher, through.obj((chunk, enc, cb) => {
        cb(null, Buffer.concat(chunk)) 
      }), writeStream)
      this._writeStream.on('error', err => {
        return cb(err)
      })
    }

    // TODO: This is a temporary (bad) way of supporting appends.
    if (this._appendStream) {
      this.position = this.stat.size
      // pump does not support the `end` option.
      this._appendStream.pipe(this._writeStream, { end: false })

      this._appendStream.on('error', err => this._writeStream.destroy(err))
      this._writeStream.on('error', err => this._appendStream.destroy(err))

      return this._appendStream.on('end', doWrite)
    }

    return doWrite()

    function doWrite (err) {
      self._appendStream = null
      if (err) return cb(err)
      if (self._err) return cb(self._err)
      if (self._writeStream.destroyed) return cb(new errors.BadFileDescriptor('Write stream was destroyed.'))

      const slice = buffer.slice(offset, len)
      write(self._writeStream, slice, err => {
        if (err) return cb(err)
        self.position += slice.length
        return cb(null, slice.length, buffer)
      })
    }

    function write(stream, data, cb) {
      if (!stream.write(data)) {
        stream.once('drain', cb)
      } else {
        process.nextTick(cb)
      }
    }
  }

  close (cb) {
    // TODO: undownload initial range
    if (this._writeStream) {
      if (this._writeStream.destroyed) {
        this._writeStream = null
      } else {
        return this._writeStream.end(err => {
          if (err) return cb(err)  
          this._writeStream = null
          return cb(null)
        })
      }
    }
    if (this._range) {
      this.drive.contentFeed.undownload(this._range)
      this._range = null
    }
    process.nextTick(cb, null)
  }

  _refreshDownload (start, cb) {
    let end = Math.min(this.stat.blocks + this.stat.offset, start + 15)

    if (this._range) {
      this.drive.contentFeed.undownload(this._range)
    }

    this._range = this.drive.contentFeed.download({ start, end, linear: true }, cb || noop)
  }
<commit_msg>Prefetch 16 blocks in fd + add comment<commit_after>const fs = require('fs')
const pump = require('pump')
const errors = require('./errors')
const byteStream = require('byte-stream')
const through = require('through2')
const pumpify = require('pumpify')
const messages = require('./messages')

const noop = () => {}

const {
  O_RDONLY,
  O_WRONLY,
  O_RDWR,
  O_CREAT,
  O_TRUNC,
  O_APPEND,
  O_SYNC,
  O_EXCL
} = fs.constants
const O_ACCMODE = 3

class FileDescriptor {
  constructor (drive, path, stat, readable, writable, appending, creating) {
    this.drive = drive
    this.stat = stat
    this.path = path

    this.readable = readable
    this.writable = writable
    this.creating = creating
    this.appending = appending

    this.position = null
    this.blockPosition = stat ? stat.offset : null
    this.blockOffset = 0

    this._err = null
    if (this.writable) {
      if (this.appending) {
        this._appendStream = this.drive.createReadStream(this.path)
        this.position = this.stat.size
      }
    }

    this._batcher = byteStream({ time: 100, limit: 4096 * 16 })

    this._range = null
  }

  read (buffer, offset, len, pos, cb) {
    if (!this.readable) return cb(new errors.BadFileDescriptor('File descriptor not open for reading.'))
    if (this.position !== null && this.position === pos) this._read(buffer, offset, len, cb)
    else this._seekAndRead(buffer, offset, len, pos, cb)
  }

  write (buffer, offset, len, pos, cb) {
    if (!this.writable) return cb(new errors.BadFileDescriptor('File descriptor not open for writing.'))
    if (!this.stat && !this.creating) {
      return cb(new errors.BadFileDescriptor('File descriptor not open in create mode.'))
    }
    if (this.position !== null && pos !== this.position) {
      return cb(new errors.BadFileDescriptor('Random-access writes are not currently supported.'))
    }
    if (this.appending && pos < this.stat.size) {
      return cb(new errors.BadFileDescriptor('Position cannot be less than the file size when appending.'))
    }
    if (!this._writeStream && !this.appending && pos) {
      return cb(new errors.BadFileDescriptor('Random-access writes are not currently supported.'))
    }

    const self = this
    if (!this._writeStream) {
      const writeStream = this.drive.createWriteStream(this.path)
      this._writeStream = pumpify(this._batcher, through.obj((chunk, enc, cb) => {
        cb(null, Buffer.concat(chunk)) 
      }), writeStream)
      this._writeStream.on('error', err => {
        return cb(err)
      })
    }

    // TODO: This is a temporary (bad) way of supporting appends.
    if (this._appendStream) {
      this.position = this.stat.size
      // pump does not support the `end` option.
      this._appendStream.pipe(this._writeStream, { end: false })

      this._appendStream.on('error', err => this._writeStream.destroy(err))
      this._writeStream.on('error', err => this._appendStream.destroy(err))

      return this._appendStream.on('end', doWrite)
    }

    return doWrite()

    function doWrite (err) {
      self._appendStream = null
      if (err) return cb(err)
      if (self._err) return cb(self._err)
      if (self._writeStream.destroyed) return cb(new errors.BadFileDescriptor('Write stream was destroyed.'))

      const slice = buffer.slice(offset, len)
      write(self._writeStream, slice, err => {
        if (err) return cb(err)
        self.position += slice.length
        return cb(null, slice.length, buffer)
      })
    }

    function write(stream, data, cb) {
      if (!stream.write(data)) {
        stream.once('drain', cb)
      } else {
        process.nextTick(cb)
      }
    }
  }

  close (cb) {
    // TODO: undownload initial range
    if (this._writeStream) {
      if (this._writeStream.destroyed) {
        this._writeStream = null
      } else {
        return this._writeStream.end(err => {
          if (err) return cb(err)  
          this._writeStream = null
          return cb(null)
        })
      }
    }
    if (this._range) {
      this.drive.contentFeed.undownload(this._range)
      this._range = null
    }
    process.nextTick(cb, null)
  }

  /**
   * Will currently request the next 16 blocks linearly.
   *
   * TODO: This behavior should be more customizable in the future.
   */
  _refreshDownload (start, cb) {
    let end = Math.min(this.stat.blocks + this.stat.offset, start + 16)

    if (this._range) {
      this.drive.contentFeed.undownload(this._range)
    }

    this._range = this.drive.contentFeed.download({ start, end, linear: true }, cb || noop)
  }

{'fraction_non_alphanumeric': 0.12759347609009208, 'fraction_numerical': 0.00355042716076778, 'mean_word_length': 2.7340513670256836, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2577062', 'n_tokens_mistral': 2830, 'n_tokens_neox': 2755, 'n_words': 708}