fix: multi-repo local dev (file:../ deps, package exports, missing deps)
This commit is contained in:
Raven Scott
+438
@@ -0,0 +1,438 @@
|
||||
const c = require('compact-encoding')
|
||||
const binding = require('../binding')
|
||||
|
||||
const empty = Buffer.alloc(0)
|
||||
const resolved = Promise.resolve()
|
||||
|
||||
class RocksDBBatch {
|
||||
constructor(db, opts = {}) {
|
||||
const { capacity = 8, autoDestroy = false } = opts
|
||||
|
||||
db._ref()
|
||||
|
||||
this._db = db
|
||||
this._destroyed = false
|
||||
this._capacity = capacity
|
||||
this._operations = []
|
||||
this._promises = []
|
||||
|
||||
this._enqueuePromise = this._enqueuePromise.bind(this)
|
||||
|
||||
this._request = null
|
||||
this._resolve = null
|
||||
this._reject = null
|
||||
|
||||
this._handle = null
|
||||
this._buffer = null
|
||||
this._autoDestroy = autoDestroy
|
||||
this._stats = null
|
||||
|
||||
this._resetStats()
|
||||
|
||||
if (db._state.opened === true) this.ready()
|
||||
}
|
||||
|
||||
_reuse(db, opts = {}) {
|
||||
const { autoDestroy = false } = opts
|
||||
|
||||
db._ref()
|
||||
|
||||
this._db = db
|
||||
this._destroyed = false
|
||||
this._autoDestroy = autoDestroy
|
||||
}
|
||||
|
||||
_onfinished(err) {
|
||||
const resolve = this._resolve
|
||||
const reject = this._reject
|
||||
|
||||
if (this._request) this._db._state.io.dec()
|
||||
|
||||
this._operations = []
|
||||
this._promises = []
|
||||
this._request = null
|
||||
this._resolve = null
|
||||
this._reject = null
|
||||
|
||||
this._resetStats()
|
||||
|
||||
if (this._autoDestroy === true) this.destroy()
|
||||
|
||||
if (reject !== null && err) reject(err)
|
||||
else if (resolve !== null) resolve()
|
||||
}
|
||||
|
||||
_resize() {
|
||||
if (this._operations.length <= this._capacity) return false
|
||||
|
||||
while (this._operations.length > this._capacity) {
|
||||
this._capacity *= 2
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
async ready() {
|
||||
if (this._handle !== null) return
|
||||
|
||||
if (this._db._state.opened === false) await this._db._state.ready()
|
||||
|
||||
this._init()
|
||||
}
|
||||
|
||||
destroy() {
|
||||
if (this._request) throw new Error('Request in progress')
|
||||
if (this._destroyed) return
|
||||
|
||||
this._destroyed = true
|
||||
|
||||
if (this._promises.length) this._abort()
|
||||
|
||||
this._db._unref()
|
||||
this._onfree()
|
||||
}
|
||||
|
||||
_onfree() {
|
||||
this._resetStats()
|
||||
this._db._state.freeBatch(this, false)
|
||||
this._db = null
|
||||
}
|
||||
|
||||
_abort() {
|
||||
for (let i = 0; i < this._promises.length; i++) {
|
||||
const promise = this._promises[i]
|
||||
if (promise !== null) promise.reject(new Error('Batch is destroyed'))
|
||||
}
|
||||
|
||||
this._onfinished(new Error('Batch is destroyed'))
|
||||
}
|
||||
|
||||
_resetStats() {}
|
||||
|
||||
async flush() {
|
||||
if (this._request) throw new Error('Request in progress')
|
||||
if (this._destroyed) throw new Error('Batch is destroyed')
|
||||
|
||||
this._request = new Promise((resolve, reject) => {
|
||||
this._resolve = resolve
|
||||
this._reject = reject
|
||||
})
|
||||
|
||||
this._flush()
|
||||
|
||||
return this._request
|
||||
}
|
||||
|
||||
tryFlush() {
|
||||
if (this._request) throw new Error('Request in progress')
|
||||
if (this._destroyed) throw new Error('Batch is destroyed')
|
||||
|
||||
this._request = resolved
|
||||
|
||||
this._flush()
|
||||
}
|
||||
|
||||
async _flush() {
|
||||
if (this._handle === null) await this.ready()
|
||||
|
||||
this._db._state.io.inc()
|
||||
|
||||
if (this._db._state.resumed !== null) {
|
||||
const resumed = await this._db._state.resumed.promise
|
||||
|
||||
if (!resumed) {
|
||||
if (this._destroyed) {
|
||||
this._db._state.io.dec()
|
||||
} else {
|
||||
this._destroyed = true
|
||||
this._abort()
|
||||
this._db._unref()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_enqueuePromise(resolve, reject) {
|
||||
this._promises.push({ resolve, reject })
|
||||
}
|
||||
|
||||
_encodeKey(k) {
|
||||
if (this._db._keyEncoding) return c.encode(this._db._keyEncoding, k)
|
||||
if (typeof k === 'string') return Buffer.from(k)
|
||||
return k
|
||||
}
|
||||
|
||||
_encodeValue(v) {
|
||||
if (this._db._valueEncoding) return c.encode(this._db._valueEncoding, v)
|
||||
if (v === null) return empty
|
||||
if (typeof v === 'string') return Buffer.from(v)
|
||||
return v
|
||||
}
|
||||
|
||||
_decodeValue(b) {
|
||||
if (this._db._valueEncoding) return c.decode(this._db._valueEncoding, b)
|
||||
return b
|
||||
}
|
||||
}
|
||||
|
||||
exports.ReadBatch = class RocksDBReadBatch extends RocksDBBatch {
|
||||
constructor(db, opts = {}) {
|
||||
super(db, opts)
|
||||
|
||||
const { asyncIO = false, fillCache = true } = opts
|
||||
|
||||
this._asyncIO = asyncIO
|
||||
this._fillCache = fillCache
|
||||
}
|
||||
|
||||
_init() {
|
||||
this._handle = binding.readInit()
|
||||
this._buffer = binding.readBuffer(this._handle, this._capacity)
|
||||
}
|
||||
|
||||
_resize() {
|
||||
if (super._resize() && this._handle !== null) {
|
||||
this._buffer = binding.readBuffer(this._handle, this._capacity)
|
||||
}
|
||||
}
|
||||
|
||||
async _flush() {
|
||||
await super._flush()
|
||||
|
||||
if (this._destroyed) return
|
||||
|
||||
try {
|
||||
binding.read(
|
||||
this._db._state._handle,
|
||||
this._handle,
|
||||
this._operations,
|
||||
this._db._snapshot ? this._db._snapshot._handle : undefined,
|
||||
this._asyncIO,
|
||||
this._fillCache,
|
||||
this,
|
||||
this._onread
|
||||
)
|
||||
} catch (err) {
|
||||
this._db._state.io.dec()
|
||||
throw err
|
||||
}
|
||||
|
||||
this._db._state.stats.gets += this._stats.gets
|
||||
this._db._state.stats.readBatches++
|
||||
}
|
||||
|
||||
_onread(errs, values) {
|
||||
let applied = true
|
||||
|
||||
for (let i = 0, n = this._promises.length; i < n; i++) {
|
||||
const err = errs[i]
|
||||
if (err) applied = false
|
||||
|
||||
const promise = this._promises[i]
|
||||
if (promise === null) continue
|
||||
|
||||
if (err) promise.reject(err)
|
||||
else promise.resolve(values[i] ? this._decodeValue(Buffer.from(values[i])) : null)
|
||||
}
|
||||
|
||||
this._onfinished(applied ? null : new AggregateError(errs, 'Batch was not applied'))
|
||||
}
|
||||
|
||||
_resetStats() {
|
||||
this._stats = { gets: 0 }
|
||||
}
|
||||
|
||||
get(key) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.gets++
|
||||
|
||||
const promise = new Promise(this._enqueuePromise)
|
||||
|
||||
this._operations.push(new RocksDBGet(this._encodeKey(key), this._db._columnFamily))
|
||||
|
||||
this._resize()
|
||||
|
||||
return promise
|
||||
}
|
||||
}
|
||||
|
||||
exports.WriteBatch = class RocksDBWriteBatch extends RocksDBBatch {
|
||||
_init() {
|
||||
this._handle = binding.writeInit()
|
||||
this._buffer = binding.writeBuffer(this._handle, this._capacity)
|
||||
}
|
||||
|
||||
_resize() {
|
||||
if (super._resize() && this._handle !== null) {
|
||||
this._buffer = binding.writeBuffer(this._handle, this._capacity)
|
||||
}
|
||||
}
|
||||
|
||||
_onfree() {
|
||||
this._resetStats()
|
||||
this._db._state.freeBatch(this, true)
|
||||
this._db = null
|
||||
}
|
||||
|
||||
async _flush() {
|
||||
await super._flush()
|
||||
|
||||
if (this._destroyed) return
|
||||
|
||||
try {
|
||||
binding.write(this._db._state._handle, this._handle, this._operations, this, this._onwrite)
|
||||
} catch (err) {
|
||||
this._db._state.io.dec()
|
||||
throw err
|
||||
}
|
||||
|
||||
this._db._state.stats.puts += this._stats.puts
|
||||
this._db._state.stats.deletes += this._stats.deletes
|
||||
this._db._state.stats.rangeDeletes += this._stats.rangeDeletes
|
||||
this._db._state.stats.writeBatches++
|
||||
}
|
||||
|
||||
_onwrite(err) {
|
||||
const applied = !err
|
||||
|
||||
for (let i = 0, n = this._promises.length; i < n; i++) {
|
||||
const promise = this._promises[i]
|
||||
if (promise === null) continue
|
||||
|
||||
if (err) promise.reject(err)
|
||||
else promise.resolve()
|
||||
}
|
||||
|
||||
this._onfinished(applied ? null : new Error('Batch was not applied', { cause: err }))
|
||||
}
|
||||
|
||||
_resetStats() {
|
||||
this._stats = { puts: 0, deletes: 0, rangeDeletes: 0 }
|
||||
}
|
||||
|
||||
put(key, value) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.puts++
|
||||
|
||||
const promise = new Promise(this._enqueuePromise)
|
||||
|
||||
this._operations.push(
|
||||
new RocksDBPut(this._encodeKey(key), this._encodeValue(value), this._db._columnFamily)
|
||||
)
|
||||
|
||||
this._resize()
|
||||
|
||||
return promise
|
||||
}
|
||||
|
||||
tryPut(key, value) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.puts++
|
||||
|
||||
this._operations.push(
|
||||
new RocksDBPut(this._encodeKey(key), this._encodeValue(value), this._db._columnFamily)
|
||||
)
|
||||
|
||||
this._promises.push(null)
|
||||
|
||||
this._resize()
|
||||
}
|
||||
|
||||
delete(key) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.deletes++
|
||||
|
||||
const promise = new Promise(this._enqueuePromise)
|
||||
|
||||
this._operations.push(new RocksDBDelete(this._encodeKey(key), this._db._columnFamily))
|
||||
|
||||
this._resize()
|
||||
|
||||
return promise
|
||||
}
|
||||
|
||||
tryDelete(key) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.deletes++
|
||||
|
||||
this._operations.push(new RocksDBDelete(this._encodeKey(key), this._db._columnFamily))
|
||||
|
||||
this._promises.push(null)
|
||||
|
||||
this._resize()
|
||||
}
|
||||
|
||||
deleteRange(start, end) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.rangeDeletes++
|
||||
|
||||
const promise = new Promise(this._enqueuePromise)
|
||||
|
||||
this._operations.push(
|
||||
new RocksDBDeleteRange(this._encodeKey(start), this._encodeKey(end), this._db._columnFamily)
|
||||
)
|
||||
|
||||
this._resize()
|
||||
|
||||
return promise
|
||||
}
|
||||
|
||||
tryDeleteRange(start, end) {
|
||||
if (this._request) throw new Error('Request already in progress')
|
||||
this._stats.rangeDeletes++
|
||||
|
||||
this._operations.push(
|
||||
new RocksDBDeleteRange(this._encodeKey(start), this._encodeKey(end), this._db._columnFamily)
|
||||
)
|
||||
|
||||
this._promises.push(null)
|
||||
|
||||
this._resize()
|
||||
}
|
||||
}
|
||||
|
||||
class RocksDBGet {
|
||||
constructor(key, columnFamily) {
|
||||
this.key = key
|
||||
this.columnFamily = columnFamily._handle
|
||||
}
|
||||
|
||||
get type() {
|
||||
return binding.GET
|
||||
}
|
||||
}
|
||||
|
||||
class RocksDBPut {
|
||||
constructor(key, value, columnFamily) {
|
||||
this.key = key
|
||||
this.value = value
|
||||
this.columnFamily = columnFamily._handle
|
||||
}
|
||||
|
||||
get type() {
|
||||
return binding.PUT
|
||||
}
|
||||
}
|
||||
|
||||
class RocksDBDelete {
|
||||
constructor(key, columnFamily) {
|
||||
this.key = key
|
||||
this.columnFamily = columnFamily._handle
|
||||
}
|
||||
|
||||
get type() {
|
||||
return binding.DELETE
|
||||
}
|
||||
}
|
||||
|
||||
class RocksDBDeleteRange {
|
||||
constructor(start, end, columnFamily) {
|
||||
this.start = start
|
||||
this.end = end
|
||||
this.columnFamily = columnFamily._handle
|
||||
}
|
||||
|
||||
get type() {
|
||||
return binding.DELETE_RANGE
|
||||
}
|
||||
}
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
const binding = require('../binding')
|
||||
const constants = require('./constants')
|
||||
const { BloomFilterPolicy } = require('./filter-policy')
|
||||
|
||||
class RocksDBColumnFamily {
|
||||
constructor(name, opts = {}) {
|
||||
const {
|
||||
// Blob options
|
||||
enableBlobFiles = false,
|
||||
minBlobSize = 0,
|
||||
blobFileSize = 0,
|
||||
enableBlobGarbageCollection = true,
|
||||
// Block table options
|
||||
tableBlockSize = 8192,
|
||||
tableCacheIndexAndFilterBlocks = true,
|
||||
tableFormatVersion = 6,
|
||||
optimizeFiltersForMemory = false,
|
||||
blockCache = true,
|
||||
filterPolicy = new BloomFilterPolicy(10),
|
||||
topLevelIndexPinningTier = constants.pinningTier.ALL,
|
||||
partitionPinningTier = constants.pinningTier.ALL,
|
||||
unpartitionedPinningTier = constants.pinningTier.ALL,
|
||||
optimizeFiltersForHits = false,
|
||||
numLevels = 7,
|
||||
maxWriteBufferNumber = 2,
|
||||
blobGarbageCollectionAgeCutOff = 0.25,
|
||||
blobGarbageCollectionForceThreshold = 1.0
|
||||
} = opts
|
||||
|
||||
this._name = name
|
||||
this._flushing = null
|
||||
this._options = {
|
||||
enableBlobFiles,
|
||||
minBlobSize,
|
||||
blobFileSize,
|
||||
enableBlobGarbageCollection,
|
||||
tableBlockSize,
|
||||
tableCacheIndexAndFilterBlocks,
|
||||
tableFormatVersion,
|
||||
optimizeFiltersForMemory,
|
||||
blockCache,
|
||||
filterPolicy,
|
||||
topLevelIndexPinningTier,
|
||||
partitionPinningTier,
|
||||
unpartitionedPinningTier,
|
||||
optimizeFiltersForHits,
|
||||
numLevels,
|
||||
maxWriteBufferNumber
|
||||
}
|
||||
|
||||
const filterPolicyArguments = [0, 0, 0]
|
||||
|
||||
if (filterPolicy !== null) {
|
||||
filterPolicyArguments[0] = filterPolicy.type
|
||||
|
||||
switch (filterPolicy.type) {
|
||||
case 1: // Bloom filter policy
|
||||
filterPolicyArguments[1] = filterPolicy.bitsPerKey
|
||||
break
|
||||
case 2: // Ribbon filter policy
|
||||
filterPolicyArguments[1] = filterPolicy.bloomEquivalentBitsPerKey
|
||||
filterPolicyArguments[2] = filterPolicy.bloomBeforeLevel
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
this._handle = binding.columnFamilyInit(
|
||||
name,
|
||||
enableBlobFiles,
|
||||
minBlobSize,
|
||||
blobFileSize,
|
||||
enableBlobGarbageCollection,
|
||||
tableBlockSize,
|
||||
tableCacheIndexAndFilterBlocks,
|
||||
tableFormatVersion,
|
||||
optimizeFiltersForMemory,
|
||||
blockCache === false,
|
||||
...filterPolicyArguments,
|
||||
topLevelIndexPinningTier,
|
||||
partitionPinningTier,
|
||||
unpartitionedPinningTier,
|
||||
optimizeFiltersForHits,
|
||||
numLevels,
|
||||
maxWriteBufferNumber,
|
||||
blobGarbageCollectionAgeCutOff,
|
||||
blobGarbageCollectionForceThreshold
|
||||
)
|
||||
}
|
||||
|
||||
cloneSettings(name) {
|
||||
return new RocksDBColumnFamily(name, this._options)
|
||||
}
|
||||
|
||||
get name() {
|
||||
return this._name
|
||||
}
|
||||
|
||||
destroy() {
|
||||
if (this._handle === null) return
|
||||
|
||||
binding.columnFamilyDestroy(this._handle)
|
||||
|
||||
this._handle = null
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = RocksDBColumnFamily
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
module.exports = {
|
||||
pinningTier: {
|
||||
NONE: 0,
|
||||
FLUSHED_AND_SIMILAR: 1,
|
||||
ALL: 2
|
||||
},
|
||||
garbageCollectionPolicy: {
|
||||
DEFAULT: 0,
|
||||
FORCE: 1,
|
||||
DISABLE: 2
|
||||
},
|
||||
bottommostLevelCompaction: {
|
||||
NONE: 0,
|
||||
SKIP: 1,
|
||||
FORCE: 2
|
||||
},
|
||||
walRecoveryMode: {
|
||||
TOLERATE_CORRUPTED_TAIL_RECORDS: 0,
|
||||
ABSOLUTE_CONSISTENCY: 1,
|
||||
POINT_IN_TIME: 2,
|
||||
SKIP_ANY_CORRUPTED_RECORDS: 3
|
||||
}
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
exports.BloomFilterPolicy = class RocksDBBloomFilterPolicy {
|
||||
get type() {
|
||||
return 1
|
||||
}
|
||||
|
||||
constructor(bitsPerKey) {
|
||||
this.bitsPerKey = bitsPerKey
|
||||
}
|
||||
}
|
||||
|
||||
exports.RibbonFilterPolicy = class RocksDBRibbonFilterPolicy {
|
||||
get type() {
|
||||
return 2
|
||||
}
|
||||
|
||||
constructor(bloomEquivalentBitsPerKey, bloomBeforeLevel = 0) {
|
||||
this.bloomEquivalentBitsPerKey = bloomEquivalentBitsPerKey
|
||||
this.bloomBeforeLevel = bloomBeforeLevel
|
||||
}
|
||||
}
|
||||
+208
@@ -0,0 +1,208 @@
|
||||
const { Readable } = require('streamx')
|
||||
const c = require('compact-encoding')
|
||||
const binding = require('../binding')
|
||||
|
||||
const empty = Buffer.alloc(0)
|
||||
|
||||
module.exports = class RocksDBIterator extends Readable {
|
||||
constructor(db, opts = {}) {
|
||||
const {
|
||||
gt = null,
|
||||
gte = null,
|
||||
lt = null,
|
||||
lte = null,
|
||||
reverse = false,
|
||||
values = true,
|
||||
limit = Infinity,
|
||||
capacity = 8
|
||||
} = opts
|
||||
|
||||
super()
|
||||
|
||||
db._ref()
|
||||
|
||||
this._db = db
|
||||
|
||||
this._gt = gt ? this._encodeKey(gt) : empty
|
||||
this._gte = gte ? this._encodeKey(gte) : empty
|
||||
this._lt = lt ? this._encodeKey(lt) : empty
|
||||
this._lte = lte ? this._encodeKey(lte) : empty
|
||||
|
||||
this._reverse = reverse
|
||||
this._values = values
|
||||
this._limit = limit < 0 ? Infinity : limit
|
||||
this._capacity = capacity
|
||||
this._opened = false
|
||||
|
||||
this._pendingOpen = null
|
||||
this._pendingRead = null
|
||||
this._pendingDestroy = null
|
||||
|
||||
this._buffer = null
|
||||
this._handle = null
|
||||
|
||||
if (this._db._state.opened === true) this.ready()
|
||||
}
|
||||
|
||||
_onopen(err) {
|
||||
const cb = this._pendingOpen
|
||||
this._pendingOpen = null
|
||||
this._opened = true
|
||||
this._db._state.io.dec()
|
||||
cb(err)
|
||||
}
|
||||
|
||||
_onread(err, keys, values) {
|
||||
const cb = this._pendingRead
|
||||
this._pendingRead = null
|
||||
this._db._state.io.dec()
|
||||
if (err) return cb(err)
|
||||
|
||||
const n = keys.length
|
||||
|
||||
this._limit -= n
|
||||
|
||||
for (let i = 0; i < n; i++) {
|
||||
this.push({
|
||||
key: this._decodeKey(Buffer.from(keys[i])),
|
||||
value: this._values ? this._decodeValue(Buffer.from(values[i])) : null
|
||||
})
|
||||
}
|
||||
|
||||
if (n < this._capacity) this.push(null)
|
||||
|
||||
cb(null)
|
||||
}
|
||||
|
||||
_onclose(err) {
|
||||
const cb = this._pendingDestroy
|
||||
this._pendingDestroy = null
|
||||
this._db._state.io.dec()
|
||||
this._db._unref()
|
||||
cb(err)
|
||||
}
|
||||
|
||||
_resize() {
|
||||
if (this._handle !== null) {
|
||||
this._buffer = binding.iteratorBuffer(this._handle, this._capacity)
|
||||
}
|
||||
}
|
||||
|
||||
async ready() {
|
||||
if (this._handle !== null) return
|
||||
|
||||
if (this._db._state.opened === false) await this._db._state.ready()
|
||||
|
||||
this._init()
|
||||
}
|
||||
|
||||
_init() {
|
||||
this._handle = binding.iteratorInit()
|
||||
this._buffer = binding.iteratorBuffer(this._handle, this._capacity)
|
||||
}
|
||||
|
||||
async _open(cb) {
|
||||
await this.ready()
|
||||
|
||||
this._db._state.io.inc()
|
||||
|
||||
if (this._db._state.resumed !== null) {
|
||||
const resumed = await this._db._state.resumed.promise
|
||||
|
||||
if (!resumed) {
|
||||
this._db._state.io.dec()
|
||||
|
||||
return cb(new Error('RocksDB session is closed'))
|
||||
}
|
||||
}
|
||||
|
||||
this._pendingOpen = cb
|
||||
|
||||
try {
|
||||
binding.iteratorOpen(
|
||||
this._db._state._handle,
|
||||
this._handle,
|
||||
this._db._columnFamily._handle,
|
||||
this._gt,
|
||||
this._gte,
|
||||
this._lt,
|
||||
this._lte,
|
||||
this._reverse,
|
||||
!this._values, // Keys only
|
||||
this._db._snapshot ? this._db._snapshot._handle : undefined,
|
||||
this,
|
||||
this._onopen,
|
||||
this._onclose,
|
||||
this._onread
|
||||
)
|
||||
} catch (err) {
|
||||
this._db._state.io.dec()
|
||||
|
||||
cb(err)
|
||||
}
|
||||
}
|
||||
|
||||
async _read(cb) {
|
||||
this._db._state.io.inc()
|
||||
|
||||
if (this._db._state.resumed !== null) {
|
||||
const resumed = await this._db._state.resumed.promise
|
||||
|
||||
if (!resumed) {
|
||||
this._db._state.io.dec()
|
||||
|
||||
return cb(new Error('RocksDB session is closed'))
|
||||
}
|
||||
}
|
||||
|
||||
this._pendingRead = cb
|
||||
|
||||
try {
|
||||
binding.iteratorRead(this._handle, Math.min(this._capacity, this._limit))
|
||||
} catch (err) {
|
||||
this._db._state.io.dec()
|
||||
|
||||
cb(err)
|
||||
}
|
||||
}
|
||||
|
||||
async _destroy(cb) {
|
||||
await this.ready()
|
||||
|
||||
this._db._state.io.inc()
|
||||
|
||||
if (this._opened === false) {
|
||||
this._db._state.io.dec()
|
||||
this._db._unref()
|
||||
|
||||
return cb(null)
|
||||
}
|
||||
|
||||
this._pendingDestroy = cb
|
||||
|
||||
try {
|
||||
binding.iteratorClose(this._handle)
|
||||
} catch (err) {
|
||||
this._db._state.io.dec()
|
||||
this._db._unref()
|
||||
|
||||
cb(err)
|
||||
}
|
||||
}
|
||||
|
||||
_encodeKey(k) {
|
||||
if (this._db._keyEncoding !== null) return c.encode(this._db._keyEncoding, k)
|
||||
if (typeof k === 'string') return Buffer.from(k)
|
||||
return k
|
||||
}
|
||||
|
||||
_decodeKey(b) {
|
||||
if (this._db._keyEncoding !== null) return c.decode(this._db._keyEncoding, b)
|
||||
return b
|
||||
}
|
||||
|
||||
_decodeValue(b) {
|
||||
if (this._db._valueEncoding !== null) return c.decode(this._db._valueEncoding, b)
|
||||
return b
|
||||
}
|
||||
}
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
const binding = require('../binding')
|
||||
|
||||
module.exports = class RocksDBSnapshot {
|
||||
constructor(state) {
|
||||
this._state = state
|
||||
|
||||
this._handle = null
|
||||
this._refs = 0
|
||||
|
||||
if (state.deferSnapshotInit === false) this._init()
|
||||
}
|
||||
|
||||
_init() {
|
||||
this._handle = binding.snapshotCreate(this._state._handle)
|
||||
}
|
||||
|
||||
ref() {
|
||||
this._refs++
|
||||
}
|
||||
|
||||
unref() {
|
||||
if (--this._refs > 0) return
|
||||
|
||||
if (this._handle === null) return
|
||||
|
||||
binding.snapshotDestroy(this._handle)
|
||||
|
||||
this._handle = null
|
||||
}
|
||||
}
|
||||
+435
@@ -0,0 +1,435 @@
|
||||
const ReadyResource = require('ready-resource')
|
||||
const RefCounter = require('refcounter')
|
||||
const rrp = require('resolve-reject-promise')
|
||||
const SignalPromise = require('signal-promise')
|
||||
const c = require('compact-encoding')
|
||||
const { ReadBatch, WriteBatch } = require('./batch')
|
||||
const ColumnFamily = require('./column-family')
|
||||
const binding = require('../binding')
|
||||
const constants = require('./constants')
|
||||
|
||||
const MAX_BATCH_REUSE = 64
|
||||
const empty = Buffer.alloc(0)
|
||||
|
||||
module.exports = class RocksDBState extends ReadyResource {
|
||||
constructor(db, path, opts) {
|
||||
super()
|
||||
|
||||
const {
|
||||
columnFamily = new ColumnFamily('default', opts),
|
||||
columnFamilies = [],
|
||||
readOnly = false,
|
||||
createIfMissing = true,
|
||||
createMissingColumnFamilies = true,
|
||||
maxBackgroundJobs = 6,
|
||||
bytesPerSync = 1048576,
|
||||
maxOpenFiles = -1,
|
||||
useDirectReads = false,
|
||||
avoidUnnecessaryBlockingIO = false,
|
||||
skipStatsUpdateOnOpen = false,
|
||||
useDirectIOForFlushAndCompaction = false,
|
||||
maxFileOpeningThreads = 16,
|
||||
lock = null,
|
||||
walRecoveryMode = constants.walRecoveryMode.POINT_IN_TIME,
|
||||
bestEffortsRecovery = false
|
||||
} = opts
|
||||
|
||||
this.path = path
|
||||
this.db = db
|
||||
this.handles = new RefCounter()
|
||||
this.io = new RefCounter()
|
||||
this.sessions = []
|
||||
this.columnFamilies = [columnFamily]
|
||||
this.deferSnapshotInit = true
|
||||
this.resumed = null
|
||||
this.stats = {
|
||||
gets: 0,
|
||||
puts: 0,
|
||||
deletes: 0,
|
||||
rangeDeletes: 0,
|
||||
readBatches: 0,
|
||||
writeBatches: 0
|
||||
}
|
||||
|
||||
this._suspended = false
|
||||
this._suspending = false
|
||||
this._updating = false
|
||||
this._updatingSignal = new SignalPromise()
|
||||
this._columnsFlushed = false
|
||||
this._lock = lock
|
||||
this._readBatches = []
|
||||
this._writeBatches = []
|
||||
|
||||
for (const columnFamily of columnFamilies) {
|
||||
this.columnFamilies.push(
|
||||
typeof columnFamily === 'string' ? new ColumnFamily(columnFamily, opts) : columnFamily
|
||||
)
|
||||
}
|
||||
|
||||
this._handle = binding.init(
|
||||
readOnly,
|
||||
createIfMissing,
|
||||
createMissingColumnFamilies,
|
||||
maxBackgroundJobs,
|
||||
bytesPerSync,
|
||||
maxOpenFiles,
|
||||
useDirectReads,
|
||||
avoidUnnecessaryBlockingIO,
|
||||
skipStatsUpdateOnOpen,
|
||||
useDirectIOForFlushAndCompaction,
|
||||
maxFileOpeningThreads,
|
||||
walRecoveryMode,
|
||||
bestEffortsRecovery
|
||||
)
|
||||
}
|
||||
|
||||
createReadBatch(db, opts) {
|
||||
if (this._readBatches.length === 0) return new ReadBatch(db, opts)
|
||||
const batch = this._readBatches.pop()
|
||||
batch._reuse(db, opts)
|
||||
return batch
|
||||
}
|
||||
|
||||
createWriteBatch(db, opts) {
|
||||
if (this._writeBatches.length === 0) return new WriteBatch(db, opts)
|
||||
const batch = this._writeBatches.pop()
|
||||
batch._reuse(db, opts)
|
||||
return batch
|
||||
}
|
||||
|
||||
freeBatch(batch, writable) {
|
||||
if (batch._capacity > 16) return
|
||||
const queue = writable ? this._writeBatches : this._readBatches
|
||||
if (queue.length >= MAX_BATCH_REUSE) return
|
||||
queue.push(batch)
|
||||
}
|
||||
|
||||
addSession(db) {
|
||||
db._index = this.sessions.push(db) - 1
|
||||
if (db._snapshot) db._snapshot.ref()
|
||||
}
|
||||
|
||||
removeSession(db) {
|
||||
const head = this.sessions.pop()
|
||||
if (head !== db) this.sessions[(head._index = db._index)] = head
|
||||
db._index = -1
|
||||
if (db._snapshot) db._snapshot.unref()
|
||||
}
|
||||
|
||||
upsertColumnFamily(c) {
|
||||
if (typeof c === 'string') {
|
||||
let col = this.getColumnFamilyByName(c)
|
||||
if (col) return col
|
||||
col = this.columnFamilies[0].cloneSettings(c)
|
||||
this.columnFamilies.push(col)
|
||||
return col
|
||||
}
|
||||
|
||||
if (this.columnFamilies.includes(c)) return c
|
||||
this.columnFamilies.push(c)
|
||||
return c
|
||||
}
|
||||
|
||||
getColumnFamily(c) {
|
||||
if (!c) return this.columnFamilies[0]
|
||||
if (!this._columnsFlushed) return this.upsertColumnFamily(c)
|
||||
|
||||
if (typeof c !== 'string') return c
|
||||
|
||||
const col = this.getColumnFamilyByName(c)
|
||||
if (col === null) throw new Error('Unknown column family')
|
||||
return col
|
||||
}
|
||||
|
||||
getColumnFamilyByName(name) {
|
||||
for (const col of this.columnFamilies) {
|
||||
if (col.name === name) return col
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
async _open() {
|
||||
await Promise.resolve() // Allow column families to populate if on-demand
|
||||
|
||||
if (this._lock) await this._lock.ready()
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
this._columnsFlushed = true
|
||||
|
||||
const lock = this._lock === null ? -1 : this._lock.transfer()
|
||||
|
||||
req.handle = binding.open(
|
||||
this._handle,
|
||||
this,
|
||||
this.path,
|
||||
this.columnFamilies.map((c) => c._handle),
|
||||
lock,
|
||||
req,
|
||||
onopen
|
||||
)
|
||||
|
||||
await promise
|
||||
|
||||
this.deferSnapshotInit = false
|
||||
|
||||
for (const session of this.sessions) {
|
||||
if (session._snapshot) session._snapshot._init()
|
||||
}
|
||||
|
||||
function onopen(err) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
async _close() {
|
||||
while (this._updating) await this._updatingSignal.wait()
|
||||
if (this.resumed) this.resumed.resolve(false)
|
||||
|
||||
while (!this.io.isIdle()) await this.io.idle()
|
||||
while (!this.handles.isIdle()) await this.handles.idle()
|
||||
|
||||
while (this.sessions.length > 0) {
|
||||
await this.sessions[this.sessions.length - 1].close()
|
||||
}
|
||||
|
||||
for (const columnFamily of this.columnFamilies) columnFamily.destroy()
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
req.handle = binding.close(this._handle, req, onclose)
|
||||
|
||||
try {
|
||||
await promise
|
||||
} finally {
|
||||
if (this._lock) await this._lock.close()
|
||||
}
|
||||
|
||||
function onclose(err) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
async flush(db, opts) {
|
||||
if (this.opened === false) await this.ready()
|
||||
|
||||
this.io.inc()
|
||||
|
||||
if (this.resumed !== null) {
|
||||
const resumed = await this.resumed.promise
|
||||
|
||||
if (!resumed) {
|
||||
this.io.dec()
|
||||
|
||||
throw new Error('RocksDB session is closed')
|
||||
}
|
||||
}
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
try {
|
||||
req.handle = binding.flush(this._handle, db._columnFamily._handle, req, onflush)
|
||||
|
||||
await promise
|
||||
} finally {
|
||||
this.io.dec()
|
||||
}
|
||||
|
||||
function onflush(err) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
suspend() {
|
||||
this._suspending = true
|
||||
return this.update()
|
||||
}
|
||||
|
||||
resume() {
|
||||
this._suspending = false
|
||||
return this.update()
|
||||
}
|
||||
|
||||
async update() {
|
||||
while (this._updating) await this._updatingSignal.wait()
|
||||
if (this._suspending === this._suspended || this.closing) return
|
||||
this._updating = true
|
||||
try {
|
||||
if (this._suspending) await this._suspend()
|
||||
else await this._resume()
|
||||
} finally {
|
||||
this._updating = false
|
||||
this._updatingSignal.notify()
|
||||
}
|
||||
}
|
||||
|
||||
async _suspend() {
|
||||
if (this._suspended === true) return
|
||||
|
||||
while (!this.io.isIdle()) await this.io.idle()
|
||||
|
||||
this.io.inc()
|
||||
this.resumed = rrp()
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
try {
|
||||
req.handle = binding.suspend(this._handle, req, onsuspend)
|
||||
|
||||
await promise
|
||||
|
||||
this._suspended = true
|
||||
} finally {
|
||||
this.io.dec()
|
||||
}
|
||||
|
||||
function onsuspend(err) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
async _resume() {
|
||||
if (this._suspended === false) return
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
req.handle = binding.resume(this._handle, req, onresume)
|
||||
|
||||
await promise
|
||||
|
||||
this._suspended = false
|
||||
|
||||
const resumed = this.resumed
|
||||
this.resumed = null
|
||||
resumed.resolve(true)
|
||||
|
||||
function onresume(err) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
async compactRange(db, start, end, opts) {
|
||||
if (this.opened === false) await this.ready()
|
||||
|
||||
this.io.inc()
|
||||
|
||||
const {
|
||||
exclusive = false,
|
||||
blobGarbageCollectionPolicy = constants.garbageCollectionPolicy.DEFAULT,
|
||||
blobGarbageCollectionAgeCutoff = 0.25,
|
||||
bottommostLevelCompaction = constants.bottommostLevelCompaction.NONE
|
||||
} = opts
|
||||
|
||||
start = this._encodeKey(start)
|
||||
end = this._encodeKey(end)
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null, start, end }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
try {
|
||||
req.handle = binding.compactRange(
|
||||
this._handle,
|
||||
db._columnFamily._handle,
|
||||
start,
|
||||
end,
|
||||
exclusive,
|
||||
blobGarbageCollectionPolicy,
|
||||
blobGarbageCollectionAgeCutoff,
|
||||
bottommostLevelCompaction,
|
||||
req,
|
||||
oncompactrange
|
||||
)
|
||||
|
||||
await promise
|
||||
} finally {
|
||||
this.io.dec()
|
||||
}
|
||||
|
||||
function oncompactrange(err) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve()
|
||||
}
|
||||
}
|
||||
|
||||
async approximateSize(db, start, end, opts) {
|
||||
if (this.opened === false) await this.ready()
|
||||
|
||||
this.io.inc()
|
||||
|
||||
const { includeMemtables = false, includeFiles = true, filesSizeErrorMargin = -1 } = opts
|
||||
|
||||
start = this._encodeKey(start)
|
||||
end = this._encodeKey(end)
|
||||
|
||||
const req = { resolve: null, reject: null, handle: null, start, end }
|
||||
|
||||
const promise = new Promise((resolve, reject) => {
|
||||
req.resolve = resolve
|
||||
req.reject = reject
|
||||
})
|
||||
|
||||
try {
|
||||
req.handle = binding.approximateSize(
|
||||
this._handle,
|
||||
db._columnFamily._handle,
|
||||
start,
|
||||
end,
|
||||
includeMemtables,
|
||||
includeFiles,
|
||||
filesSizeErrorMargin,
|
||||
req,
|
||||
onapproximatesize
|
||||
)
|
||||
|
||||
return await promise
|
||||
} finally {
|
||||
this.io.dec()
|
||||
}
|
||||
|
||||
function onapproximatesize(err, result) {
|
||||
if (err) req.reject(err)
|
||||
else req.resolve(result)
|
||||
}
|
||||
}
|
||||
|
||||
_encodeKey(k) {
|
||||
if (this.db._keyEncoding) return c.encode(this.db._keyEncoding, k)
|
||||
if (typeof k === 'string') return Buffer.from(k)
|
||||
if (k === null) return empty
|
||||
return k
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user