12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124 |
- import {
- getAddress,
- getBufferAddress,
- write,
- compress,
- lmdbError,
- } from './native.js';
- import { when } from './util/when.js';
- var backpressureArray;
- const WAITING_OPERATION = 0x2000000;
- const BACKPRESSURE_THRESHOLD = 300000;
- const TXN_DELIMITER = 0x8000000;
- const TXN_COMMITTED = 0x10000000;
- const TXN_FLUSHED = 0x20000000;
- const TXN_FAILED = 0x40000000;
- export const FAILED_CONDITION = 0x4000000;
- const REUSE_BUFFER_MODE = 512;
- const RESET_BUFFER_MODE = 1024;
- const NO_RESOLVE = 16;
- const HAS_TXN = 8;
- const CONDITIONAL_VERSION_LESS_THAN = 0x800;
- const CONDITIONAL_ALLOW_NOTFOUND = 0x800;
- const SYNC_PROMISE_SUCCESS = Promise.resolve(true);
- const SYNC_PROMISE_FAIL = Promise.resolve(false);
- SYNC_PROMISE_SUCCESS.isSync = true;
- SYNC_PROMISE_SUCCESS.result = true;
- SYNC_PROMISE_FAIL.isSync = true;
- SYNC_PROMISE_FAIL.result = false;
- const PROMISE_SUCCESS = Promise.resolve(true);
- const arch = process.arch;
- export const ABORT = 4.452694326329068e-106; // random/unguessable numbers, which work across module/versions and native
- export const IF_EXISTS = 3.542694326329068e-103;
- const CALLBACK_THREW = {};
- const LocalSharedArrayBuffer =
- typeof Deno != 'undefined' || // Deno can't handle SharedArrayBuffer as an FFI
- // argument due to https://github.com/denoland/deno/issues/12678
- typeof SharedArrayBuffer == 'undefined' // Sometimes electron doesn't have a SharedArrayBuffer
- ? ArrayBuffer
- : SharedArrayBuffer;
- const ByteArray =
- typeof Buffer != 'undefined'
- ? function (buffer) {
- return Buffer.from(buffer);
- }
- : Uint8Array;
- const queueTask =
- typeof setImmediate != 'undefined' ? setImmediate : setTimeout; // TODO: Or queueMicrotask?
- //let debugLog = []
- const WRITE_BUFFER_SIZE = 0x10000;
- var log = [];
- export function addWriteMethods(
- LMDBStore,
- {
- env,
- fixedBuffer,
- resetReadTxn,
- useWritemap,
- maxKeySize,
- eventTurnBatching,
- txnStartThreshold,
- batchStartThreshold,
- overlappingSync,
- commitDelay,
- separateFlushed,
- maxFlushDelay,
- },
- ) {
- // stands for write instructions
- var dynamicBytes;
- function allocateInstructionBuffer(lastPosition) {
- // Must use a shared buffer on older node in order to use Atomics, and it is also more correct since we are
- // indeed accessing and modifying it from another thread (in C). However, Deno can't handle it for
- // FFI so aliased above
- let buffer = new LocalSharedArrayBuffer(WRITE_BUFFER_SIZE);
- let lastBytes = dynamicBytes;
- dynamicBytes = new ByteArray(buffer);
- let uint32 = (dynamicBytes.uint32 = new Uint32Array(
- buffer,
- 0,
- WRITE_BUFFER_SIZE >> 2,
- ));
- uint32[2] = 0;
- dynamicBytes.float64 = new Float64Array(buffer, 0, WRITE_BUFFER_SIZE >> 3);
- buffer.address = getBufferAddress(dynamicBytes);
- uint32.address = buffer.address + uint32.byteOffset;
- dynamicBytes.position = 1; // we start at position 1 to save space for writing the txn id before the txn delimiter
- if (lastPosition) {
- lastBytes.float64[lastPosition + 1] =
- dynamicBytes.uint32.address + (dynamicBytes.position << 3);
- lastBytes.uint32[lastPosition << 1] = 3; // pointer instruction
- }
- return dynamicBytes;
- }
- var newBufferThreshold = (WRITE_BUFFER_SIZE - maxKeySize - 64) >> 3; // need to reserve more room if we do inline values
- var outstandingWriteCount = 0;
- var startAddress = 0;
- var writeTxn = null;
- var committed;
- var abortedNonChildTransactionWarn;
- var nextTxnCallbacks = [];
- var commitPromise,
- flushPromise,
- flushResolvers = [],
- batchFlushResolvers = [];
- commitDelay = commitDelay || 0;
- eventTurnBatching = eventTurnBatching === false ? false : true;
- var enqueuedCommit;
- var afterCommitCallbacks = [];
- var beforeCommitCallbacks = [];
- var enqueuedEventTurnBatch;
- var batchDepth = 0;
- var lastWritePromise;
- var writeBatchStart,
- outstandingBatchCount,
- lastSyncTxnFlush,
- lastFlushTimeout,
- lastFlushCallback;
- var hasUnresolvedTxns;
- txnStartThreshold = txnStartThreshold || 5;
- batchStartThreshold = batchStartThreshold || 1000;
- maxFlushDelay = maxFlushDelay || 500;
- allocateInstructionBuffer();
- dynamicBytes.uint32[2] = TXN_DELIMITER | TXN_COMMITTED | TXN_FLUSHED;
- var txnResolution,
- nextResolution = {
- uint32: dynamicBytes.uint32,
- flagPosition: 2,
- flag: 0,
- valueBuffer: null,
- next: null,
- meta: null,
- };
- var uncommittedResolution = {
- uint32: null,
- flagPosition: 2,
- flag: 0,
- valueBuffer: null,
- next: nextResolution,
- meta: null,
- };
- var unwrittenResolution = nextResolution;
- var lastPromisedResolution = uncommittedResolution;
- var lastQueuedResolution = uncommittedResolution;
- function writeInstructions(flags, store, key, value, version, ifVersion) {
- let writeStatus;
- let targetBytes, position, encoder;
- let valueSize, valueBuffer, valueBufferStart;
- if (flags & 2) {
- // encode first in case we have to write a shared structure
- encoder = store.encoder;
- if (value && value['\x10binary-data\x02'])
- valueBuffer = value['\x10binary-data\x02'];
- else if (encoder) {
- if (encoder.copyBuffers)
- // use this as indicator for support buffer reuse for now
- valueBuffer = encoder.encode(
- value,
- REUSE_BUFFER_MODE | (writeTxn ? RESET_BUFFER_MODE : 0),
- );
- // in addition, if we are writing sync, after using, we can immediately reset the encoder's position to reuse that space, which can improve performance
- else {
- // various other encoders, including JSON.stringify, that might serialize to a string
- valueBuffer = encoder.encode(value);
- if (typeof valueBuffer == 'string')
- valueBuffer = Buffer.from(valueBuffer); // TODO: Would be nice to write strings inline in the instructions
- }
- } else if (typeof value == 'string') {
- valueBuffer = Buffer.from(value); // TODO: Would be nice to write strings inline in the instructions
- } else if (value instanceof Uint8Array) valueBuffer = value;
- else
- throw new Error(
- 'Invalid value to put in database ' +
- value +
- ' (' +
- typeof value +
- '), consider using encoder',
- );
- valueBufferStart = valueBuffer.start;
- if (valueBufferStart > -1)
- // if we have buffers with start/end position
- valueSize = valueBuffer.end - valueBufferStart; // size
- else valueSize = valueBuffer.length;
- if (store.dupSort && valueSize > maxKeySize)
- throw new Error(
- 'The value is larger than the maximum size (' +
- maxKeySize +
- ') for a value in a dupSort database',
- );
- } else valueSize = 0;
- if (writeTxn) {
- targetBytes = fixedBuffer;
- position = 0;
- } else {
- if (eventTurnBatching && !enqueuedEventTurnBatch && batchDepth == 0) {
- enqueuedEventTurnBatch = queueTask(() => {
- try {
- for (let i = 0, l = beforeCommitCallbacks.length; i < l; i++) {
- try {
- beforeCommitCallbacks[i]();
- } catch (error) {
- console.error('In beforecommit callback', error);
- }
- }
- } catch (error) {
- console.error(error);
- }
- enqueuedEventTurnBatch = null;
- batchDepth--;
- finishBatch();
- if (writeBatchStart) writeBatchStart(); // TODO: When we support delay start of batch, optionally don't delay this
- });
- commitPromise = null; // reset the commit promise, can't know if it is really a new transaction prior to finishWrite being called
- flushPromise = null;
- writeBatchStart = writeInstructions(1, store);
- outstandingBatchCount = 0;
- batchDepth++;
- }
- targetBytes = dynamicBytes;
- position = targetBytes.position;
- }
- let uint32 = targetBytes.uint32,
- float64 = targetBytes.float64;
- let flagPosition = position << 1; // flagPosition is the 32-bit word starting position
- // don't increment position until we are sure we don't have any key writing errors
- if (!uint32) {
- throw new Error('Internal buffers have been corrupted');
- }
- uint32[flagPosition + 1] = store.db.dbi;
- if (flags & 4) {
- let keyStartPosition = (position << 3) + 12;
- let endPosition;
- try {
- endPosition = store.writeKey(key, targetBytes, keyStartPosition);
- if (!(keyStartPosition < endPosition) && (flags & 0xf) != 12)
- throw new Error(
- 'Invalid key or zero length key is not allowed in LMDB ' + key,
- );
- } catch (error) {
- targetBytes.fill(0, keyStartPosition);
- if (error.name == 'RangeError')
- error = new Error(
- 'Key size is larger than the maximum key size (' + maxKeySize + ')',
- );
- throw error;
- }
- let keySize = endPosition - keyStartPosition;
- if (keySize > maxKeySize) {
- targetBytes.fill(0, keyStartPosition); // restore zeros
- throw new Error(
- 'Key size is larger than the maximum key size (' + maxKeySize + ')',
- );
- }
- uint32[flagPosition + 2] = keySize;
- position = (endPosition + 16) >> 3;
- if (flags & 2) {
- let mustCompress;
- if (valueBufferStart > -1) {
- // if we have buffers with start/end position
- // record pointer to value buffer
- float64[position] =
- (valueBuffer.address ||
- (valueBuffer.address = getAddress(valueBuffer.buffer))) +
- valueBufferStart;
- if (store.compression) {
- let compressionFlagIndex =
- valueBufferStart + (store.compression.startingOffset || 0);
- // this is the compression indicator, so we must compress
- mustCompress =
- compressionFlagIndex < valueBuffer.end &&
- valueBuffer[compressionFlagIndex] >= 250;
- }
- } else {
- let valueArrayBuffer = valueBuffer.buffer;
- // record pointer to value buffer
- let address =
- (valueArrayBuffer.address ||
- (valueBuffer.length === 0
- ? 0 // externally allocated buffers of zero-length with the same non-null-pointer can crash node, #161
- : (valueArrayBuffer.address = getAddress(valueArrayBuffer)))) +
- valueBuffer.byteOffset;
- if (address <= 0 && valueBuffer.length > 0)
- console.error('Supplied buffer had an invalid address', address);
- float64[position] = address;
- if (store.compression) {
- let compressionFlagIndex = store.compression.startingOffset || 0;
- // this is the compression indicator, so we must compress
- mustCompress =
- compressionFlagIndex < valueBuffer.length &&
- valueBuffer[compressionFlagIndex] >= 250;
- }
- }
- uint32[(position++ << 1) - 1] = valueSize;
- if (
- store.compression &&
- (valueSize >= store.compression.threshold || mustCompress)
- ) {
- flags |= 0x100000;
- float64[position] = store.compression.address;
- if (!writeTxn)
- compress(env.address, uint32.address + (position << 3), () => {
- // this is never actually called in NodeJS, just use to pin the buffer in memory until it is finished
- // and is a no-op in Deno
- if (!float64) throw new Error('No float64 available');
- });
- position++;
- }
- }
- if (ifVersion !== undefined) {
- if (ifVersion === null)
- flags |= 0x10; // if it does not exist, MDB_NOOVERWRITE
- else {
- flags |= 0x100;
- float64[position++] = ifVersion;
- }
- }
- if (version !== undefined) {
- flags |= 0x200;
- float64[position++] = version || 0;
- }
- } else position++;
- targetBytes.position = position;
- if (writeTxn) {
- uint32[0] = flags;
- write(env.address, uint32.address);
- return () =>
- uint32[0] & FAILED_CONDITION ? SYNC_PROMISE_FAIL : SYNC_PROMISE_SUCCESS;
- }
- // if we ever use buffers that haven't been zero'ed, need to clear out the next slot like this:
- // uint32[position << 1] = 0 // clear out the next slot
- let nextUint32;
- if (position > newBufferThreshold) {
- // make new buffer and make pointer to it
- let lastPosition = position;
- targetBytes = allocateInstructionBuffer(position);
- position = targetBytes.position;
- nextUint32 = targetBytes.uint32;
- } else nextUint32 = uint32;
- let resolution = nextResolution;
- // create the placeholder next resolution
- nextResolution = resolution.next = {
- // we try keep resolutions exactly the same object type
- uint32: nextUint32,
- flagPosition: position << 1,
- flag: 0, // TODO: eventually eliminate this, as we can probably signify HAS_TXN/NO_RESOLVE/FAILED_CONDITION in upper bits
- valueBuffer: fixedBuffer, // these are all just placeholders so that we have the right hidden class initially allocated
- next: null,
- meta: null,
- };
- lastQueuedResolution = resolution;
- let writtenBatchDepth = batchDepth;
- return (callback) => {
- if (writtenBatchDepth) {
- // If we are in a batch, the transaction can't close, so we do the faster,
- // but non-deterministic updates, knowing that the write thread can
- // just poll for the status change if we miss a status update.
- // That is, if we are on x64 architecture...
- if (arch === 'x64') {
- writeStatus = uint32[flagPosition];
- uint32[flagPosition] = flags;
- } else {
- // However, on ARM processors, apparently more radical memory reordering can occur
- // so we need to use the slower atomic operation to ensure that a memory barrier is set
- // and that the value pointer is actually written before the flag is updated
- writeStatus = Atomics.or(uint32, flagPosition, flags);
- }
- if (writeBatchStart && !writeStatus) {
- outstandingBatchCount += 1 + (valueSize >> 12);
- if (outstandingBatchCount > batchStartThreshold) {
- outstandingBatchCount = 0;
- writeBatchStart();
- writeBatchStart = null;
- }
- }
- } // otherwise the transaction could end at any time and we need to know the
- // deterministically if it is ending, so we can reset the commit promise
- // so we use the slower atomic operation
- else writeStatus = Atomics.or(uint32, flagPosition, flags);
- outstandingWriteCount++;
- if (writeStatus & TXN_DELIMITER) {
- commitPromise = null; // TODO: Don't reset these if this comes from the batch start operation on an event turn batch
- flushPromise = null;
- flushResolvers = [];
- queueCommitResolution(resolution);
- if (!startAddress) {
- startAddress = uint32.address + (flagPosition << 2);
- }
- }
- if (!writtenBatchDepth && batchFlushResolvers.length > 0) {
- flushResolvers.push(...batchFlushResolvers);
- batchFlushResolvers = [];
- }
- if (!flushPromise && overlappingSync) {
- flushPromise = new Promise((resolve) => {
- if (writtenBatchDepth) {
- batchFlushResolvers.push(resolve);
- } else {
- flushResolvers.push(resolve);
- }
- });
- }
- if (writeStatus & WAITING_OPERATION) {
- // write thread is waiting
- write(env.address, 0);
- }
- if (outstandingWriteCount > BACKPRESSURE_THRESHOLD && !writeBatchStart) {
- if (!backpressureArray)
- backpressureArray = new Int32Array(new SharedArrayBuffer(4), 0, 1);
- Atomics.wait(
- backpressureArray,
- 0,
- 0,
- Math.round(outstandingWriteCount / BACKPRESSURE_THRESHOLD),
- );
- }
- if (startAddress) {
- if (eventTurnBatching)
- startWriting(); // start writing immediately because this has already been batched/queued
- else if (!enqueuedCommit && txnStartThreshold) {
- enqueuedCommit =
- commitDelay == 0 && typeof setImmediate != 'undefined'
- ? setImmediate(() => startWriting())
- : setTimeout(() => startWriting(), commitDelay);
- } else if (outstandingWriteCount > txnStartThreshold) startWriting();
- }
- if ((outstandingWriteCount & 7) === 0) resolveWrites();
- if (store.cache) {
- resolution.meta = {
- key,
- store,
- valueSize: valueBuffer ? valueBuffer.length : 0,
- };
- }
- resolution.valueBuffer = valueBuffer;
- if (callback) {
- if (callback === IF_EXISTS) ifVersion = IF_EXISTS;
- else {
- let meta = resolution.meta || (resolution.meta = {});
- meta.reject = callback;
- meta.resolve = (value) => callback(null, value);
- return;
- }
- }
- // if it is not conditional because of ifVersion or has any flags that can make the write conditional
- if (ifVersion === undefined && !(flags & 0x22030)) {
- if (writtenBatchDepth > 1) {
- if (!resolution.flag && !store.cache) resolution.flag = NO_RESOLVE;
- return PROMISE_SUCCESS; // or return undefined?
- }
- if (commitPromise) {
- if (!resolution.flag) resolution.flag = NO_RESOLVE;
- } else {
- commitPromise = new Promise((resolve, reject) => {
- let meta = resolution.meta || (resolution.meta = {});
- meta.resolve = resolve;
- resolve.unconditional = true;
- meta.reject = reject;
- });
- if (separateFlushed)
- commitPromise.flushed = overlappingSync
- ? flushPromise
- : commitPromise;
- }
- return commitPromise;
- }
- lastWritePromise = new Promise((resolve, reject) => {
- let meta = resolution.meta || (resolution.meta = {});
- meta.resolve = resolve;
- meta.reject = reject;
- });
- if (separateFlushed)
- lastWritePromise.flushed = overlappingSync
- ? flushPromise
- : lastWritePromise;
- return lastWritePromise;
- };
- }
- let committedFlushResolvers,
- lastSync = Promise.resolve();
- function startWriting() {
- if (enqueuedCommit) {
- clearImmediate(enqueuedCommit);
- enqueuedCommit = null;
- }
- let resolvers = flushResolvers;
- let start = Date.now();
- env.startWriting(startAddress, (status) => {
- if (dynamicBytes.uint32[dynamicBytes.position << 1] & TXN_DELIMITER)
- queueCommitResolution(nextResolution);
- resolveWrites(true);
- switch (status) {
- case 0:
- for (let resolver of resolvers) {
- resolver();
- }
- break;
- case 1:
- break;
- case 2:
- hasUnresolvedTxns = false;
- executeTxnCallbacks();
- return hasUnresolvedTxns;
- break;
- default:
- try {
- lmdbError(status);
- } catch (error) {
- console.error(error);
- if (commitRejectPromise) {
- commitRejectPromise.reject(error);
- commitRejectPromise = null;
- }
- }
- }
- });
- startAddress = 0;
- }
- function queueCommitResolution(resolution) {
- if (!(resolution.flag & HAS_TXN)) {
- resolution.flag = HAS_TXN;
- if (txnResolution) {
- txnResolution.nextTxn = resolution;
- //outstandingWriteCount = 0
- } else txnResolution = resolution;
- }
- }
- var TXN_DONE = TXN_COMMITTED | TXN_FAILED;
- function resolveWrites(async) {
- // clean up finished instructions
- let instructionStatus;
- while (
- (instructionStatus =
- unwrittenResolution.uint32[unwrittenResolution.flagPosition]) &
- 0x1000000
- ) {
- if (unwrittenResolution.callbacks) {
- nextTxnCallbacks.push(unwrittenResolution.callbacks);
- unwrittenResolution.callbacks = null;
- }
- outstandingWriteCount--;
- if (unwrittenResolution.flag !== HAS_TXN) {
- if (
- unwrittenResolution.flag === NO_RESOLVE &&
- !unwrittenResolution.meta
- ) {
- // in this case we can completely remove from the linked list, clearing more memory
- lastPromisedResolution.next = unwrittenResolution =
- unwrittenResolution.next;
- continue;
- }
- unwrittenResolution.uint32 = null;
- }
- unwrittenResolution.valueBuffer = null;
- unwrittenResolution.flag = instructionStatus;
- lastPromisedResolution = unwrittenResolution;
- unwrittenResolution = unwrittenResolution.next;
- }
- while (
- txnResolution &&
- (instructionStatus =
- txnResolution.uint32[txnResolution.flagPosition] & TXN_DONE)
- ) {
- if (instructionStatus & TXN_FAILED) rejectCommit();
- else resolveCommit(async);
- }
- }
- function resolveCommit(async) {
- afterCommit(txnResolution.uint32[txnResolution.flagPosition - 1]);
- if (async) resetReadTxn();
- else queueMicrotask(resetReadTxn); // TODO: only do this if there are actually committed writes?
- do {
- if (uncommittedResolution.meta && uncommittedResolution.meta.resolve) {
- let resolve = uncommittedResolution.meta.resolve;
- if (
- uncommittedResolution.flag & FAILED_CONDITION &&
- !resolve.unconditional
- )
- resolve(false);
- else resolve(true);
- }
- } while (
- (uncommittedResolution = uncommittedResolution.next) &&
- uncommittedResolution != txnResolution
- );
- txnResolution = txnResolution.nextTxn;
- }
- var commitRejectPromise;
- function rejectCommit() {
- afterCommit();
- if (!commitRejectPromise) {
- let rejectFunction;
- commitRejectPromise = new Promise(
- (resolve, reject) => (rejectFunction = reject),
- );
- commitRejectPromise.reject = rejectFunction;
- }
- do {
- if (uncommittedResolution.meta && uncommittedResolution.meta.reject) {
- let flag = uncommittedResolution.flag & 0xf;
- let error = new Error('Commit failed (see commitError for details)');
- error.commitError = commitRejectPromise;
- uncommittedResolution.meta.reject(error);
- }
- } while (
- (uncommittedResolution = uncommittedResolution.next) &&
- uncommittedResolution != txnResolution
- );
- txnResolution = txnResolution.nextTxn;
- }
- function atomicStatus(uint32, flagPosition, newStatus) {
- if (batchDepth) {
- // if we are in a batch, the transaction can't close, so we do the faster,
- // but non-deterministic updates, knowing that the write thread can
- // just poll for the status change if we miss a status update
- let writeStatus = uint32[flagPosition];
- uint32[flagPosition] = newStatus;
- return writeStatus;
- //return Atomics.or(uint32, flagPosition, newStatus)
- } // otherwise the transaction could end at any time and we need to know the
- // deterministically if it is ending, so we can reset the commit promise
- // so we use the slower atomic operation
- else
- try {
- return Atomics.or(uint32, flagPosition, newStatus);
- } catch (error) {
- console.error(error);
- return;
- }
- }
- function afterCommit(txnId) {
- for (let i = 0, l = afterCommitCallbacks.length; i < l; i++) {
- try {
- afterCommitCallbacks[i]({
- next: uncommittedResolution,
- last: txnResolution,
- txnId,
- });
- } catch (error) {
- console.error('In aftercommit callback', error);
- }
- }
- }
- async function executeTxnCallbacks() {
- env.writeTxn = writeTxn = { write: true };
- nextTxnCallbacks.isExecuting = true;
- for (let i = 0; i < nextTxnCallbacks.length; i++) {
- let txnCallbacks = nextTxnCallbacks[i];
- for (let j = 0, l = txnCallbacks.length; j < l; j++) {
- let userTxnCallback = txnCallbacks[j];
- let asChild = userTxnCallback.asChild;
- if (asChild) {
- env.beginTxn(1); // abortable
- let parentTxn = writeTxn;
- env.writeTxn = writeTxn = { write: true };
- try {
- let result = userTxnCallback.callback();
- if (result && result.then) {
- hasUnresolvedTxns = true;
- await result;
- }
- if (result === ABORT) env.abortTxn();
- else env.commitTxn();
- clearWriteTxn(parentTxn);
- txnCallbacks[j] = result;
- } catch (error) {
- clearWriteTxn(parentTxn);
- env.abortTxn();
- txnError(error, txnCallbacks, j);
- }
- } else {
- try {
- let result = userTxnCallback();
- txnCallbacks[j] = result;
- if (result && result.then) {
- hasUnresolvedTxns = true;
- await result;
- }
- } catch (error) {
- txnError(error, txnCallbacks, j);
- }
- }
- }
- }
- nextTxnCallbacks = [];
- clearWriteTxn(null);
- if (hasUnresolvedTxns) {
- env.resumeWriting();
- }
- function txnError(error, txnCallbacks, i) {
- (txnCallbacks.errors || (txnCallbacks.errors = []))[i] = error;
- txnCallbacks[i] = CALLBACK_THREW;
- }
- }
- function finishBatch() {
- let bytes = dynamicBytes;
- let uint32 = bytes.uint32;
- let nextPosition = bytes.position + 1;
- let writeStatus;
- if (nextPosition > newBufferThreshold) {
- allocateInstructionBuffer(nextPosition);
- nextResolution.flagPosition = dynamicBytes.position << 1;
- nextResolution.uint32 = dynamicBytes.uint32;
- writeStatus = atomicStatus(uint32, bytes.position << 1, 2); // atomically write the end block
- } else {
- uint32[nextPosition << 1] = 0; // clear out the next slot
- writeStatus = atomicStatus(uint32, bytes.position++ << 1, 2); // atomically write the end block
- nextResolution.flagPosition += 2;
- }
- if (writeStatus & WAITING_OPERATION) {
- write(env.address, 0);
- }
- }
- function clearWriteTxn(parentTxn) {
- // TODO: We might actually want to track cursors in a write txn and manually
- // close them.
- if (writeTxn && writeTxn.refCount > 0) writeTxn.isDone = true;
- env.writeTxn = writeTxn = parentTxn || null;
- }
- Object.assign(LMDBStore.prototype, {
- put(key, value, versionOrOptions, ifVersion) {
- let callback,
- flags = 15,
- type = typeof versionOrOptions;
- if (type == 'object' && versionOrOptions) {
- if (versionOrOptions.noOverwrite) flags |= 0x10;
- if (versionOrOptions.noDupData) flags |= 0x20;
- if (versionOrOptions.instructedWrite) flags |= 0x2000;
- if (versionOrOptions.append) flags |= 0x20000;
- if (versionOrOptions.ifVersion != undefined)
- ifVersion = versionOrOptions.ifVersion;
- versionOrOptions = versionOrOptions.version;
- if (typeof ifVersion == 'function') callback = ifVersion;
- } else if (type == 'function') {
- callback = versionOrOptions;
- }
- return writeInstructions(
- flags,
- this,
- key,
- value,
- this.useVersions ? versionOrOptions || 0 : undefined,
- ifVersion,
- )(callback);
- },
- remove(key, ifVersionOrValue, callback) {
- let flags = 13;
- let ifVersion, value;
- if (ifVersionOrValue !== undefined) {
- if (typeof ifVersionOrValue == 'function') callback = ifVersionOrValue;
- else if (ifVersionOrValue === IF_EXISTS && !callback)
- // we have a handler for IF_EXISTS in the callback handler for remove
- callback = ifVersionOrValue;
- else if (this.useVersions) ifVersion = ifVersionOrValue;
- else {
- flags = 14;
- value = ifVersionOrValue;
- }
- }
- return writeInstructions(
- flags,
- this,
- key,
- value,
- undefined,
- ifVersion,
- )(callback);
- },
- del(key, options, callback) {
- return this.remove(key, options, callback);
- },
- ifNoExists(key, callback) {
- return this.ifVersion(key, null, callback);
- },
- ifVersion(key, version, callback, options) {
- if (!callback) {
- return new Batch((operations, callback) => {
- let promise = this.ifVersion(key, version, operations, options);
- if (callback) promise.then(callback);
- return promise;
- });
- }
- if (writeTxn) {
- if (version === undefined || this.doesExist(key, version)) {
- callback();
- return SYNC_PROMISE_SUCCESS;
- }
- return SYNC_PROMISE_FAIL;
- }
- let flags = key === undefined || version === undefined ? 1 : 4;
- if (options?.ifLessThan) flags |= CONDITIONAL_VERSION_LESS_THAN;
- if (options?.allowNotFound) flags |= CONDITIONAL_ALLOW_NOTFOUND;
- let finishStartWrite = writeInstructions(
- flags,
- this,
- key,
- undefined,
- undefined,
- version,
- );
- let promise;
- batchDepth += 2;
- if (batchDepth > 2) promise = finishStartWrite();
- else {
- writeBatchStart = () => {
- promise = finishStartWrite();
- };
- outstandingBatchCount = 0;
- }
- try {
- if (typeof callback === 'function') {
- callback();
- } else {
- for (let i = 0, l = callback.length; i < l; i++) {
- let operation = callback[i];
- this[operation.type](operation.key, operation.value);
- }
- }
- } finally {
- if (!promise) {
- finishBatch();
- batchDepth -= 2;
- promise = finishStartWrite(); // finish write once all the operations have been written (and it hasn't been written prematurely)
- writeBatchStart = null;
- } else {
- batchDepth -= 2;
- finishBatch();
- }
- }
- return promise;
- },
- batch(callbackOrOperations) {
- return this.ifVersion(undefined, undefined, callbackOrOperations);
- },
- drop(callback) {
- return writeInstructions(
- 1024 + 12,
- this,
- Buffer.from([]),
- undefined,
- undefined,
- undefined,
- )(callback);
- },
- clearAsync(callback) {
- if (this.encoder) {
- if (this.encoder.clearSharedData) this.encoder.clearSharedData();
- else if (this.encoder.structures) this.encoder.structures = [];
- }
- return writeInstructions(
- 12,
- this,
- Buffer.from([]),
- undefined,
- undefined,
- undefined,
- )(callback);
- },
- _triggerError() {
- finishBatch();
- },
- putSync(key, value, versionOrOptions, ifVersion) {
- if (writeTxn)
- return (
- this.put(key, value, versionOrOptions, ifVersion) ===
- SYNC_PROMISE_SUCCESS
- );
- else
- return this.transactionSync(
- () =>
- this.put(key, value, versionOrOptions, ifVersion) ===
- SYNC_PROMISE_SUCCESS,
- overlappingSync ? 0x10002 : 2,
- ); // non-abortable, async flush
- },
- removeSync(key, ifVersionOrValue) {
- if (writeTxn)
- return this.remove(key, ifVersionOrValue) === SYNC_PROMISE_SUCCESS;
- else
- return this.transactionSync(
- () => this.remove(key, ifVersionOrValue) === SYNC_PROMISE_SUCCESS,
- overlappingSync ? 0x10002 : 2,
- ); // non-abortable, async flush
- },
- transaction(callback) {
- if (writeTxn && !nextTxnCallbacks.isExecuting) {
- // already nested in a transaction, just execute and return
- return callback();
- }
- return this.transactionAsync(callback);
- },
- childTransaction(callback) {
- if (useWritemap)
- throw new Error(
- 'Child transactions are not supported in writemap mode',
- );
- if (writeTxn) {
- let parentTxn = writeTxn;
- let thisTxn = (env.writeTxn = writeTxn = { write: true });
- env.beginTxn(1); // abortable
- let callbackDone, finishTxn;
- try {
- return (writeTxn.childResults = when(
- callback(),
- (finishTxn = (result) => {
- if (writeTxn !== thisTxn)
- // need to wait for child txn to finish asynchronously
- return writeTxn.childResults.then(() => finishTxn(result));
- callbackDone = true;
- if (result === ABORT) env.abortTxn();
- else env.commitTxn();
- clearWriteTxn(parentTxn);
- return result;
- }),
- (error) => {
- env.abortTxn();
- clearWriteTxn(parentTxn);
- throw error;
- },
- ));
- } catch (error) {
- if (!callbackDone) env.abortTxn();
- clearWriteTxn(parentTxn);
- throw error;
- }
- }
- return this.transactionAsync(callback, true);
- },
- transactionAsync(callback, asChild) {
- let txnIndex;
- let txnCallbacks;
- if (lastQueuedResolution.callbacks) {
- txnCallbacks = lastQueuedResolution.callbacks;
- txnIndex =
- txnCallbacks.push(asChild ? { callback, asChild } : callback) - 1;
- } else if (nextTxnCallbacks.isExecuting) {
- txnCallbacks = [asChild ? { callback, asChild } : callback];
- txnCallbacks.results = commitPromise;
- nextTxnCallbacks.push(txnCallbacks);
- txnIndex = 0;
- } else {
- if (writeTxn)
- throw new Error('Can not enqueue transaction during write txn');
- let finishWrite = writeInstructions(
- 8 | (this.strictAsyncOrder ? 0x100000 : 0),
- this,
- );
- txnCallbacks = [asChild ? { callback, asChild } : callback];
- lastQueuedResolution.callbacks = txnCallbacks;
- lastQueuedResolution.id = Math.random();
- txnCallbacks.results = finishWrite();
- txnIndex = 0;
- }
- return txnCallbacks.results.then((results) => {
- let result = txnCallbacks[txnIndex];
- if (result === CALLBACK_THREW) throw txnCallbacks.errors[txnIndex];
- return result;
- });
- },
- transactionSync(callback, flags) {
- if (writeTxn) {
- if (!useWritemap && (flags == undefined || flags & 1))
- // can't use child transactions in write maps
- // already nested in a transaction, execute as child transaction (if possible) and return
- return this.childTransaction(callback);
- let result = callback(); // else just run in current transaction
- if (result == ABORT && !abortedNonChildTransactionWarn) {
- console.warn(
- 'Can not abort a transaction inside another transaction with ' +
- (this.cache ? 'caching enabled' : 'useWritemap enabled'),
- );
- abortedNonChildTransactionWarn = true;
- }
- return result;
- }
- let callbackDone, finishTxn;
- this.transactions++;
- if (!env.address)
- throw new Error(
- 'The database has been closed and you can not transact on it',
- );
- env.beginTxn(flags == undefined ? 3 : flags);
- let thisTxn = (writeTxn = env.writeTxn = { write: true });
- try {
- this.emit('begin-transaction');
- return (writeTxn.childResults = when(
- callback(),
- (finishTxn = (result) => {
- if (writeTxn !== thisTxn)
- // need to wait for child txn to finish asynchronously
- return writeTxn.childResults.then(() => finishTxn(result));
- try {
- callbackDone = true;
- if (result === ABORT) env.abortTxn();
- else {
- env.commitTxn();
- resetReadTxn();
- }
- return result;
- } finally {
- clearWriteTxn(null);
- }
- }),
- (error) => {
- try {
- env.abortTxn();
- } catch (e) {}
- clearWriteTxn(null);
- throw error;
- },
- ));
- } catch (error) {
- if (!callbackDone)
- try {
- env.abortTxn();
- } catch (e) {}
- clearWriteTxn(null);
- throw error;
- }
- },
- getWriteTxnId() {
- return env.getWriteTxnId();
- },
- transactionSyncStart(callback) {
- return this.transactionSync(callback, 0);
- },
- // make the db a thenable/promise-like for when the last commit is committed
- committed: (committed = {
- then(onfulfilled, onrejected) {
- if (commitPromise) return commitPromise.then(onfulfilled, onrejected);
- if (lastWritePromise)
- // always resolve to true
- return lastWritePromise.then(() => onfulfilled(true), onrejected);
- return SYNC_PROMISE_SUCCESS.then(onfulfilled, onrejected);
- },
- }),
- flushed: {
- // make this a thenable for when the commit is flushed to disk
- then(onfulfilled, onrejected) {
- if (flushPromise) flushPromise.hasCallbacks = true;
- return Promise.all([flushPromise || committed, lastSyncTxnFlush]).then(
- onfulfilled,
- onrejected,
- );
- },
- },
- _endWrites(resolvedPromise, resolvedSyncPromise) {
- this.put =
- this.remove =
- this.del =
- this.batch =
- this.removeSync =
- this.putSync =
- this.transactionAsync =
- this.drop =
- this.clearAsync =
- () => {
- throw new Error('Database is closed');
- };
- // wait for all txns to finish, checking again after the current txn is done
- let finalPromise = flushPromise || commitPromise || lastWritePromise;
- if (flushPromise) flushPromise.hasCallbacks = true;
- let finalSyncPromise = lastSyncTxnFlush;
- if (
- (finalPromise && resolvedPromise != finalPromise) ||
- (finalSyncPromise && resolvedSyncPromise != finalSyncPromise)
- ) {
- return Promise.all([finalPromise, finalSyncPromise]).then(
- () => this._endWrites(finalPromise, finalSyncPromise),
- () => this._endWrites(finalPromise, finalSyncPromise),
- );
- }
- Object.defineProperty(env, 'sync', { value: null });
- },
- on(event, callback) {
- if (event == 'beforecommit') {
- eventTurnBatching = true;
- beforeCommitCallbacks.push(callback);
- } else if (event == 'aftercommit') afterCommitCallbacks.push(callback);
- else if (event == 'committed') {
- this.getUserSharedBuffer('__committed__', new ArrayBuffer(0), {
- envKey: true,
- callback,
- });
- } else super.on(event, callback);
- },
- });
- }
- class Batch extends Array {
- constructor(callback) {
- super();
- this.callback = callback;
- }
- put(key, value) {
- this.push({ type: 'put', key, value });
- }
- del(key) {
- this.push({ type: 'del', key });
- }
- clear() {
- this.length = 0;
- }
- write(callback) {
- return this.callback(this, callback);
- }
- }
- export function asBinary(buffer) {
- return {
- ['\x10binary-data\x02']: buffer,
- };
- }
|