Convert to Cadence.

flatheadmill · flatheadmill · commit b2d82729fd19 · 2020-11-02T15:17:45.000-06:00
Since we're implementing an error-first callback interface, let's do so. I've added promise support to Cadence, and error-first callback support to Reciprocate. Now there ought to be no unnecessary escapes to the Node.js event loop or the promise event loop. Anything that can be run synchrnously will be run synchronously using the trampoline in Cadence or the trampoline in Reciprocate. Closes #249.
diff --git a/locket.js b/locket.js
@@ -33,8 +33,7 @@ const Destructible = require('destructible')
 const Amalgamator = require('amalgamate')
 const Locker = require('amalgamate/locker')
 
-const callbackify = require('prospective/callbackify')
-
+const cadence = require('cadence')
 const packet = require('./packet')
 
 // Inheritence.
@@ -54,7 +53,8 @@ const rescue = require('rescue')
 const ascension = require('ascension')
 
 const mvcc = {
-    riffle: require('riffle')
+    riffle: require('riffle'),
+    satiate: require('satiate')
 }
 
 // TODO Let's see if we can get throught his without having to worry about
@@ -63,54 +63,50 @@ function encode (buffer) {
     return Buffer.isBuffer(buffer) ? buffer : Buffer.from(buffer)
 }
 
-class Paginator {
-    constructor (iterator, constraint) {
-        this._iterator = iterator
-        this._constraint = constraint
-        this._keyAsBuffer = constraint.options.keyAsBuffer
-        this._valueAsBuffer = constraint.options.valueAsBuffer
-        this._keys = constraint.options.keys
-        this._values = constraint.options.values
-        this._items = []
-        this._index = 0
-    }
+function Paginator (iterator, constraint) {
+    this._iterator = mvcc.satiate(iterator, 1)
+    this._constraint = constraint
+    this._keyAsBuffer = constraint.options.keyAsBuffer
+    this._valueAsBuffer = constraint.options.valueAsBuffer
+    this._keys = constraint.options.keys
+    this._values = constraint.options.values
+    this._items = []
+    this._index = 0
+}
 
-    async next () {
-        for (;;) {
-            if (this._items.length != this._index) {
-                const item = this._items[this._index++]
-                if (this._constraint.included(item)) {
-                    const result = new Array(2)
-                    if (this._keys) {
-                        result[0] = this._keyAsBuffer ? item.parts[1] : item.parts[1].toString()
-                    }
-                    if (this._values) {
-                        result[1] = this._valueAsBuffer ? item.parts[2] : item.parts[2].toString()
-                    }
-                    return result
-                } else {
-                    return []
-                }
-            } else {
-                let items = null
-                const trampoline = new Trampoline
-                this._iterator.next(trampoline, $items => items = $items)
-                while (trampoline.seek()) {
-                    await trampoline.shift()
-                }
-                if (this._iterator.done) {
-                    return []
-                } else {
-                    this._items = items
-                    this._index = 0
-                }
+Paginator.prototype.next = cadence(function (step) {
+    if (this._items.length != this._index) {
+        const item = this._items[this._index++]
+        if (this._constraint.included(item)) {
+            const result = new Array(2)
+            if (this._keys) {
+                result[0] = this._keyAsBuffer ? item.parts[1] : item.parts[1].toString()
+            }
+            if (this._values) {
+                result[1] = this._valueAsBuffer ? item.parts[2] : item.parts[2].toString()
             }
+            return result
+        } else {
+            return []
         }
     }
+    let items = null
+    step(function () {
+        const trampoline = new Trampoline
+        this._iterator.next(trampoline, $items => items = $items)
+        trampoline.bounce(step())
+    }, function () {
+        if (this._iterator.done) {
+            return []
+        } else {
+            this._items = items
+            this._index = 0
+            this.next(step())
+        }
+    })
+})
 
-    release () {
-        this._iterator['return']()
-    }
+Paginator.prototype.release = function () {
 }
 
 // An implementation of the LevelDOWN `Iterator` object.
@@ -137,9 +133,9 @@ util.inherits(Iterator, AbstractIterator)
 // iterator modules.
 
 //
-Iterator.prototype._next = callbackify(function () {
-    return this._paginator.next()
-})
+Iterator.prototype._next = function (callback) {
+    this._paginator.next(callback)
+}
 
 Iterator.prototype._seek = function (target) {
     const paginator = this._paginator
@@ -168,7 +164,7 @@ Locket.prototype._serializeKey = encode
 
 Locket.prototype._serializeValue = encode
 
-Locket.prototype._open = callbackify(async function (options) {
+Locket.prototype._open = cadence(function (step, options) {
     this._locker = new Locker({ heft: coalesce(options.heft, 1024 * 1024) })
     // TODO What is the behavior if you close while opening, or open while
     // closing?
@@ -209,10 +205,15 @@ Locket.prototype._open = callbackify(async function (options) {
         ...this._options,
         ...options
     })
-    await this._amalgamator.ready
-    await this._amalgamator.count()
-    await this._amalgamator.locker.rotate()
-    return []
+    step(function () {
+        return this._amalgamator.ready
+    }, function () {
+        return this._amalgamator.count()
+    }, function () {
+        return this._amalgamator.locker.rotate()
+    }, function () {
+        return []
+    })
 })
 
 // Iteration of the database requires merging the results from the deep storage
@@ -244,20 +245,21 @@ Locket.prototype._iterator = function (options) {
 }
 
 // TODO Maybe just leave this?
-Locket.prototype._get = callbackify(async function (key, options) {
+Locket.prototype._get = cadence(function (step, key, options) {
     const constraint = constrain(Buffer.compare, encode, {
         gte: key, keys: true, values: true, keyAsBuffer: true, valueAsBuffer: true
     })
     const snapshot = this._locker.snapshot()
     const paginator = this._paginator(constraint, snapshot)
-    // TODO How do I reuse Cursor.found out of Riffle et. al.? Eh, no good way
-    // since we have to advance, merge, dilute, etc. anyway.
-    const next = await paginator.next()
-    this._locker.release(snapshot)
-    if (next.length != 0 && Buffer.compare(next[0], key) == 0) {
-        return [ options.asBuffer ? next[1] : next[1].toString() ]
-    }
-    throw new Error('NotFoundError: not found')
+    step(function () {
+        paginator.next(step())
+    }, [], function (next) {
+        this._locker.release(snapshot)
+        if (next.length != 0 && Buffer.compare(next[0], key) == 0) {
+            return [ options.asBuffer ? next[1] : next[1].toString() ]
+        }
+        throw new Error('NotFoundError: not found')
+    })
 })
 
 Locket.prototype._put = function (key, value, options, callback) {
@@ -273,16 +275,19 @@ Locket.prototype._del = function (key, options, callback) {
 // filter it. It does however mean at least two writes for every `put` or `del`
 // and I suspect that common usage is ingle `put` or `del`, so going to include
 // the count in ever record, it is only 32-bits.
-Locket.prototype._batch = callbackify(async function (batch, options) {
+Locket.prototype._batch = cadence(function (step, batch, options) {
     const version = ++this._version
     this._versions[version] = true
     const mutator = this._locker.mutator()
-    await this._amalgamator.merge(mutator, batch, true)
-    this._locker.commit(mutator)
-    return []
+    step(function () {
+        return this._amalgamator.merge(mutator, batch, true)
+    }, function () {
+        this._locker.commit(mutator)
+        return []
+    })
 })
 
-Locket.prototype._approximateSize = callbackify(async function (from, to) {
+Locket.prototype._approximateSize = cadence(function (from, to) {
     const constraint = constrain(Buffer.compare, encode, { gte: from, lte: to })
     let approximateSize = 0
     for (const items of this._whatever()) {
@@ -294,7 +299,10 @@ Locket.prototype._approximateSize = callbackify(async function (from, to) {
 })
 
 // TODO Countdown through the write queue.
-Locket.prototype._close = callbackify(async function () {
-    await this._amalgamator.destructible.destroy().rejected
-    return []
+Locket.prototype._close = cadence(function (step) {
+    step(function () {
+        return this._amalgamator.destructible.destroy().rejected
+    }, function () {
+        return []
+    })
 })
diff --git a/package.json b/package.json
@@ -24,14 +24,16 @@
                             "abstract-leveldown":           "6.3.0",
                             "amalgamate":                   "0.1.36",
                             "b-tree":                       "2.0.0-alpha.63",
+                            "cadence":                      "4.0.0-alpha.0",
                             "constrain":                    "2.0.0-alpha.0",
                             "destructible":                 "7.0.0-alpha.14",
                             "extant":                       "^1.0.20",
                             "riffle":                       "2.0.0-alpha.40",
                             "prospective":                  "0.3.0",
                             "reciprocate":                  "0.0.2",
                             "rescue":                       "7.0.0-alpha.1",
-                            "rimraf":                       "3.0.2"
+                            "rimraf":                       "3.0.2",
+                            "satiate":                      "0.0.0"
     },
     "devDependencies":
     {
diff --git a/test/locket.t.js b/test/locket.t.js
