Merge pull request #68 from ssb-ngi-pointer/batch-delete

speed up deletes and add onDeletesFlushed()

Author: staltz

README.md

@@ -117,6 +117,14 @@ log.appendTransaction([item1, item2, item3], (err, offset3) => {
 })
 ```
 
+### Wait for all ongoing appends to be flushed to disk
+
+```js
+log.onDrain(() => {
+  // ...
+})
+```
+
 ### Scan all records as a `push-stream`
 
 ```js
@@ -160,10 +168,10 @@ log.del(offset, (err) => {
 })
 ```
 
-### Wait for all ongoing writes to complete
+### Wait for all ongoing deletes to be flushed to disk
 
 ```js
-log.onDrain(() => {
+log.onDeletesFlushed(() => {
   // ...
 })
 ```

index.js

@@ -42,7 +42,7 @@ const DEFAULT_WRITE_TIMEOUT = 250
 const DEFAULT_VALIDATE = () => true
 
 module.exports = function AsyncAppendOnlyLog(filename, opts) {
-  const cache = new Cache(1024) // this is potentially 65mb!
+  const cache = new Cache(1024) // This is potentially 64 MiB!
   const raf = RAF(filename)
   const blockSize = (opts && opts.blockSize) || DEFAULT_BLOCK_SIZE
   const codec = (opts && opts.codec) || DEFAULT_CODEC
@@ -52,7 +52,9 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
 
   const waitingLoad = []
   const waitingDrain = new Map() // blockIndex -> []
+  const waitingFlushDelete = []
   const blocksToBeWritten = new Map() // blockIndex -> { blockBuf, offset }
+  const blocksWithDeletables = new Map() // blockIndex -> blockBuf
   let writingBlockIndex = -1
 
   let latestBlockBuf = null
@@ -244,25 +246,64 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
       cb(delDuringCompactErr())
       return
     }
-    if (blocksToBeWritten.has(getBlockIndex(offset))) {
+    const blockIndex = getBlockIndex(offset)
+    if (blocksToBeWritten.has(blockIndex)) {
       onDrain(function delAfterDrained() {
         del(offset, cb)
       })
       return
     }
-    getBlock(offset, function gotBlockForDelete(err, blockBuf) {
+
+    if (blocksWithDeletables.has(blockIndex)) {
+      const blockBuf = blocksWithDeletables.get(blockIndex)
+      gotBlockForDelete(null, blockBuf)
+    } else {
+      getBlock(offset, gotBlockForDelete)
+    }
+    function gotBlockForDelete(err, blockBuf) {
       if (err) return cb(err)
-      Record.overwriteWithZeroes(blockBuf, getOffsetInBlock(offset))
-      // we write directly here to make normal write simpler
-      const blockStart = getBlockStart(offset)
-      writeWithFSync(blockStart, blockBuf, null, cb)
-    })
+      const actualBlockBuf = blocksWithDeletables.get(blockIndex) || blockBuf
+      Record.overwriteWithZeroes(actualBlockBuf, getOffsetInBlock(offset))
+      blocksWithDeletables.set(blockIndex, actualBlockBuf)
+      scheduleFlushDelete()
+      cb()
+    }
   }
 
   function hasNoSpaceFor(dataBuf, offsetInBlock) {
     return offsetInBlock + Record.size(dataBuf) + EOB.SIZE > blockSize
   }
 
+  const scheduleFlushDelete = debounce(flushDelete, writeTimeout)
+
+  function flushDelete() {
+    if (blocksWithDeletables.size === 0) {
+      for (const cb of waitingFlushDelete) cb()
+      waitingFlushDelete.length = 0
+      return
+    }
+    const blockIndex = blocksWithDeletables.keys().next().value
+    const blockStart = blockIndex * blockSize
+    const blockBuf = blocksWithDeletables.get(blockIndex)
+    blocksWithDeletables.delete(blockIndex)
+    blocksWithDeletables.set(-1, null) // indicate that flush is active
+
+    writeWithFSync(blockStart, blockBuf, null, function flushedDelete(err) {
+      blocksWithDeletables.delete(-1) // indicate that flush is not active
+      if (err) {
+        for (const cb of waitingFlushDelete) cb(err)
+        waitingFlushDelete.length = 0
+        return
+      }
+      flushDelete() // next
+    })
+  }
+
+  function onDeletesFlushed(cb) {
+    if (blocksWithDeletables.size === 0) cb()
+    else waitingFlushDelete.push(cb)
+  }
+
   function appendSingle(data) {
     let encodedData = codec.encode(data)
     if (typeof encodedData === 'string') encodedData = Buffer.from(encodedData)
@@ -439,26 +480,29 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
       return
     }
     onDrain(function startCompactAfterDrain() {
-      compaction = new Compaction(self, (err, sizeDiff) => {
-        compaction = null
-        if (err) return cb(err)
-        compactionProgress.set({ sizeDiff, percent: 1, done: true })
-        for (let i = 0, n = waitingCompaction.length; i < n; ++i) {
-          waitingCompaction[i]()
-        }
-        waitingCompaction.length = 0
-        cb()
-      })
-      compaction.progress((stats) => {
-        compactionProgress.set({ ...stats, done: false })
+      onDeletesFlushed(function startCompactAfterDeletes() {
+        compaction = new Compaction(self, (err, sizeDiff) => {
+          compaction = null
+          if (err) return cb(err)
+          compactionProgress.set({ sizeDiff, percent: 1, done: true })
+          for (const callback of waitingCompaction) callback()
+          waitingCompaction.length = 0
+          cb()
+        })
+        compaction.progress((stats) => {
+          compactionProgress.set({ ...stats, done: false })
+        })
       })
     })
   }
 
   function close(cb) {
     onDrain(function closeAfterHavingDrained() {
-      while (self.streams.length) self.streams.shift().abort(streamClosedErr())
-      raf.close(cb)
+      onDeletesFlushed(function closeAfterDeletesFlushed() {
+        for (const stream of self.streams) stream.abort(streamClosedErr())
+        self.streams = []
+        raf.close(cb)
+      })
     })
   }
 
@@ -500,6 +544,7 @@ module.exports = function AsyncAppendOnlyLog(filename, opts) {
     appendTransaction: onLoad(appendTransaction),
     close: onLoad(close),
     onDrain: onLoad(onDrain),
+    onDeletesFlushed: onLoad(onDeletesFlushed),
     compact: onLoad(compact),
     since,
     compactionProgress,

test/compaction.js

@@ -57,7 +57,7 @@ tape('delete first record, compact, stream', async (t) => {
   t.pass('append two records')
 
   await run(log.del)(offset1)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete first record')
 
   const [err] = await run(log.compact)()
@@ -93,7 +93,7 @@ tape('delete last record, compact, stream', async (t) => {
   t.pass('append three records')
 
   await run(log.del)(offset3)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete third record')
 
   await new Promise((resolve) => {
@@ -150,7 +150,7 @@ tape('shift many blocks', async (t) => {
   await run(log.del)(11 + 3)
   await run(log.del)(11 + 6)
   await run(log.del)(33 + 3)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('deleted some records in the middle')
 
   await new Promise((resolve) => {
@@ -265,7 +265,7 @@ tape('cannot read truncated regions of the log', async (t) => {
   await run(log.del)(11 + 3)
   await run(log.del)(11 + 6)
   await run(log.del)(22 + 0)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete some records')
 
   await new Promise((resolve) => {
@@ -340,7 +340,7 @@ tape('compact handling last deleted record on last block', async (t) => {
 
   await run(log.del)(11 + 3)
   await run(log.del)(22 + 6)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('deleted some records in the middle')
 
   await new Promise((resolve) => {
@@ -413,7 +413,7 @@ tape('compact handling holes of different sizes', async (t) => {
 
   await run(log.del)(3)
   await run(log.del)(14 + 0)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('deleted some records in the middle')
 
   await new Promise((resolve) => {
@@ -606,7 +606,7 @@ tape('append during compaction is postponed', async (t) => {
   t.pass('append two records')
 
   await run(log.del)(offset1)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete first record')
 
   let appendDone = false
@@ -651,7 +651,7 @@ tape('appendTransaction during compaction is postponed', async (t) => {
   t.pass('append two records')
 
   await run(log.del)(offset1)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete first record')
 
   let appendTransactionDone = false
@@ -695,17 +695,27 @@ tape('del during compaction is forbidden', async (t) => {
   t.pass('append two records')
 
   await run(log.del)(offset1)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete first record')
 
+  let compactDone = false
   log.compact((err) => {
     t.error(err, 'no error when compacting')
+    compactDone = true
   })
   const [err, offset3] = await run(log.del)(10)
   t.ok(err, 'del is forbidden')
   t.match(err.message, /Cannot delete/)
   t.notOk(offset3)
-  await run(log.onDrain)()
+
+  await new Promise((resolve) => {
+    const interval = setInterval(() => {
+      if (compactDone) {
+        clearInterval(interval)
+        resolve()
+      }
+    }, 100)
+  })
 
   await new Promise((resolve) => {
     log.stream({ offsets: false }).pipe(
@@ -734,7 +744,7 @@ tape('there can only be one compact at a time', async (t) => {
   t.pass('append two records')
 
   await run(log.del)(offset1)
-  await run(log.onDrain)()
+  await run(log.onDeletesFlushed)()
   t.pass('delete first record')
 
   let compact1Done = false