fcoin/lib/utils/locker.js

/*!
 * locker.js - lock and queue for bcoin
 * Copyright (c) 2014-2015, Fedor Indutny (MIT License)
 * Copyright (c) 2014-2016, Christopher Jeffrey (MIT License).
 * https://github.com/bcoin-org/bcoin
 */

'use strict';

var EventEmitter = require('events').EventEmitter;
var utils = require('../utils/utils');
var assert = utils.assert;

/**
 * Represents a mutex lock for locking asynchronous object methods.
 * @exports Locker
 * @constructor
 * @param {Function} parent - Parent object.
 * @param {Function?} add - `add` method (whichever method is queuing data).
 */

function Locker(parent, add) {
  if (!(this instanceof Locker))
    return Locker.create(parent, add);

  EventEmitter.call(this);

  this.parent = parent;
  this.jobs = [];
  this.busy = false;

  this.pending = [];
  this.pendingMap = {};
  this.add = add;
}

utils.inherits(Locker, EventEmitter);

/**
 * Create a closure scoped locker.
 * @param {Function} parent - Parent object.
 * @param {Function?} add
 * @returns {Function} Lock method.
 */

Locker.create = function create(parent, add) {
  var locker = new Locker(parent, add);
  return function lock(func, args, force) {
    return locker.lock(func, args, force);
  };
};

/**
 * Test whether the locker has a pending
 * object by key (usually a {@link Hash}).
 * @param {Hash|String} key
 * @returns {Boolean}
 */

Locker.prototype.hasPending = function hasPending(key) {
  return this.pendingMap[key] === true;
};

/**
 * Lock the parent object and all its methods
 * which use the locker. Begin to queue calls.
 * @param {Function} func - The method being called.
 * @param {Array} args - Arguments passed to the method.
 * @param {Boolean?} force - Force a call.
 * @returns {Function} Unlocker - must be
 * called once the method finishes executing in order
 * to resolve the queue.
 */

Locker.prototype.lock = function lock(func, args, force) {
  var self = this;
  var callback = args[args.length - 1];
  var obj, called;

  if (typeof callback !== 'function')
    throw new Error(func.name + ' requires a callback.');

  if (force) {
    assert(this.busy);
    return function unlock(err, res1, res2) {
      assert(!called, 'Locked callback executed twice.');
      called = true;
      callback(err, res1, res2);
    };
  }

  if (this.busy) {
    if (this.add && func === this.add) {
      obj = args[0];
      this.pending.push(obj);
      this.pendingMap[obj.hash('hex')] = true;
    }
    this.jobs.push([func, args]);
    return;
  }

  this.busy = true;

  return function unlock(err, res1, res2) {
    var item, obj;

    assert(!called, 'Locked callback executed twice.');
    called = true;

    self.busy = false;

    if (self.add && func === self.add) {
      if (self.pending.length === 0)
        self.emit('drain');
    }

    if (self.jobs.length === 0) {
      callback(err, res1, res2);
      return;
    }

    item = self.jobs.shift();

    if (self.add && item[0] === self.add) {
      obj = item[1][0];
      assert(obj === self.pending.shift());
      delete self.pendingMap[obj.hash('hex')];
    }

    item[0].apply(self.parent, item[1]);

    callback(err, res1, res2);
  };
};

/**
 * Destroy the locker. Purge all pending calls.
 */

Locker.prototype.destroy = function destroy() {
  if (this.add) {
    this.pending.length = 0;
    this.pendingMap = {};
  }
  this.jobs.length = 0;
};

/**
 * Wait for a drain (empty queue).
 * @param {Function} callback
 */

Locker.prototype.onDrain = function onDrain(callback) {
  assert(this.add, 'Cannot wait for drain without add method.');

  if (this.pending.length === 0)
    return callback();

  this.once('drain', callback);
};

/**
 * Represents a mutex lock for locking asynchronous object methods.
 * Locks methods according to passed-in key.
 * @exports MappedLock
 * @constructor
 * @param {Function} parent - Parent object.
 */

function MappedLock(parent) {
  if (!(this instanceof MappedLock))
    return MappedLock.create(parent);

  this.parent = parent;
  this.jobs = [];
  this.busy = {};
}

/**
 * Create a closure scoped locker.
 * @param {Function} parent - Parent object.
 * @returns {Function} Lock method.
 */

MappedLock.create = function create(parent) {
  var locker = new MappedLock(parent);
  return function lock(key, func, args, force) {
    return locker.lock(key, func, args, force);
  };
};

/**
 * Lock the parent object and all its methods
 * which use the locker with a specified key.
 * Begin to queue calls.
 * @param {String|Number} key
 * @param {Function} func - The method being called.
 * @param {Array} args - Arguments passed to the method.
 * @param {Boolean?} force - Force a call.
 * @returns {Function} Unlocker - must be
 * called once the method finishes executing in order
 * to resolve the queue.
 */

MappedLock.prototype.lock = function lock(key, func, args, force) {
  var self = this;
  var callback = args[args.length - 1];
  var called;

  if (typeof callback !== 'function')
    throw new Error(func.name + ' requires a callback.');

  if (force || key == null) {
    assert(key == null || this.busy[key]);
    return function unlock(err, res1, res2) {
      assert(!called, 'Locked callback executed twice.');
      called = true;
      callback(err, res1, res2);
    };
  }

  if (this.busy[key]) {
    this.jobs.push([func, args]);
    return;
  }

  this.busy[key] = true;

  return function unlock(err, res1, res2) {
    var item;

    assert(!called, 'Locked callback executed twice.');
    called = true;

    delete self.busy[key];

    if (self.jobs.length === 0) {
      callback(err, res1, res2);
      return;
    }

    item = self.jobs.shift();

    item[0].apply(self.parent, item[1]);

    callback(err, res1, res2);
  };
};

/*
 * Expose
 */

exports = Locker;
exports.mapped = MappedLock;
module.exports = exports;