<commit_before>module.exports = Sevnup;

var _ = require('lodash');
var farmhash = require('farmhash');
var async = require('async');
var CacheStore = require('./cache_store');

var DEFAULT_TOTAL_VNODES = 1024;
var DEFAULT_CALM_THRESHOLD = 500;
var MAX_PARALLEL_TASKS = 10;

/**
 * Params are:
 *   hashRing
 *   hashRingLookup (optional)
 *   store
 *   recoverKey
 *   releaseKey
 *   totalVNodes
 *   addOnLookup
 *   logger
 *   watchMode
 */
function Sevnup(params) {
    this.hashRing = params.hashRing;
    this.hashRingLookup = params.hashRingLookup || this.hashRing.lookup.bind(this.hashRing);
    this.store = new CacheStore(params.store);
    this.recoverKeyCallback = params.recoverKey;
    this.releaseKeyCallback = params.releaseKey;
    this.totalVNodes = params.totalVNodes || DEFAULT_TOTAL_VNODES;
    this.logger = params.logger;
    this.calmThreshold = params.calmThreshold || DEFAULT_CALM_THRESHOLD;
    this.calmTimeout = null;
    this.watchMode = params.watchMode;
    this.running = true;

    this.ownedVNodes = [];

    this.stateChangeQueue = async.queue(this._handleRingStateChange.bind(this), 1);

    this.eventHandler = this._onRingStateChange.bind(this);
    this._attachToRing(params.addOnLookup);
}

/**
 * Adds a key to sevnup if it belongs to the current worker.
 * @param {string} key The key you want to add.
 * @param {function} done Optional callback if you want to listen to completion
 */
Sevnup.prototype.addKey = function addKey(key, done) {
    var vnode = this.getVNodeForKey(key);
    return this.addKeyToVNode(key, vnode, done);
};

Sevnup.prototype.addKeyToVNode = function addKeyToVNode(key, vnode, done) {
    var self = this;
    var node = this.hashRingLookup(vnode);
    if (this.hashRing.whoami() === node) {
        this.store.addKey(vnode, key, function(err) {
            if (err) {
                self.logger.error("Sevnup.sevnupLookup failed to persist key", {
                    vnode: vnode,
                    key: key,
                    error: err
                });
            }
            if (done) {
                done(err);
            }
        });
    } else if (done) {
        done();
    }
};

/**
 * When you are done working on a key, or no longer want it within bookkeeping
 * you can alert sevnup to forget it.  This notifies the ring that it doesn't
 * need attention in the event this node goes down or hands off ownership.
 * We want the service to be ignorant of vnodes so we rediscover the vnode.
 * @param {string} key The key you have finished work on.
 * @param {function} done Optional callback if you want to listen to completion
 */
Sevnup.prototype.workCompleteOnKey = function workCompleteOnKey(key, done) {
    var vnode = this.getVNodeForKey(key);
    this.store.removeKey(vnode, key, done);
};

Sevnup.prototype.getOwnedKeys = function getOwnedKeys(done) {
    async.waterfall([
        async.mapLimit.bind(async, this._getOwnedVNodes(), MAX_PARALLEL_TASKS, this.store.loadKeys.bind(this.store)),
        function(keys, next) {
            next(null, _.flatten(keys));
        }
    ], done);<commit_msg>Also remove key from vnode<commit_after>module.exports = Sevnup;

var _ = require('lodash');
var farmhash = require('farmhash');
var async = require('async');
var CacheStore = require('./cache_store');

var DEFAULT_TOTAL_VNODES = 1024;
var DEFAULT_CALM_THRESHOLD = 500;
var MAX_PARALLEL_TASKS = 10;

/**
 * Params are:
 *   hashRing
 *   hashRingLookup (optional)
 *   store
 *   recoverKey
 *   releaseKey
 *   totalVNodes
 *   addOnLookup
 *   logger
 *   watchMode
 */
function Sevnup(params) {
    this.hashRing = params.hashRing;
    this.hashRingLookup = params.hashRingLookup || this.hashRing.lookup.bind(this.hashRing);
    this.store = new CacheStore(params.store);
    this.recoverKeyCallback = params.recoverKey;
    this.releaseKeyCallback = params.releaseKey;
    this.totalVNodes = params.totalVNodes || DEFAULT_TOTAL_VNODES;
    this.logger = params.logger;
    this.calmThreshold = params.calmThreshold || DEFAULT_CALM_THRESHOLD;
    this.calmTimeout = null;
    this.watchMode = params.watchMode;
    this.running = true;

    this.ownedVNodes = [];

    this.stateChangeQueue = async.queue(this._handleRingStateChange.bind(this), 1);

    this.eventHandler = this._onRingStateChange.bind(this);
    this._attachToRing(params.addOnLookup);
}

/**
 * Adds a key to sevnup if it belongs to the current worker.
 * @param {string} key The key you want to add.
 * @param {function} done Optional callback if you want to listen to completion
 */
Sevnup.prototype.addKey = function addKey(key, done) {
    var vnode = this.getVNodeForKey(key);
    return this.addKeyToVNode(key, vnode, done);
};

Sevnup.prototype.addKeyToVNode = function addKeyToVNode(key, vnode, done) {
    var self = this;
    var node = this.hashRingLookup(vnode);
    if (this.hashRing.whoami() === node) {
        this.store.addKey(vnode, key, function(err) {
            if (err) {
                self.logger.error("Sevnup.sevnupLookup failed to persist key", {
                    vnode: vnode,
                    key: key,
                    error: err
                });
            }
            if (done) {
                done(err);
            }
        });
    } else if (done) {
        done();
    }
};

/**
 * When you are done working on a key, or no longer want it within bookkeeping
 * you can alert sevnup to forget it.  This notifies the ring that it doesn't
 * need attention in the event this node goes down or hands off ownership.
 * We want the service to be ignorant of vnodes so we rediscover the vnode.
 * @param {string} key The key you have finished work on.
 * @param {function} done Optional callback if you want to listen to completion
 */
Sevnup.prototype.workCompleteOnKey = function workCompleteOnKey(key, done) {
    var vnode = this.getVNodeForKey(key);
    this.workCompleteOnKeyInVNode(key, vnode, done);
};

Sevnup.prototype.workCompleteOnKeyInVNode = function workCompleteOnKeyInVNode(key, vnode, done) {
    this.store.removeKey(vnode, key, done);
};

Sevnup.prototype.getOwnedKeys = function getOwnedKeys(done) {
    async.waterfall([
        async.mapLimit.bind(async, this._getOwnedVNodes(), MAX_PARALLEL_TASKS, this.store.loadKeys.bind(this.store)),
        function(keys, next) {
            next(null, _.flatten(keys));
        }
    ], done);
{'fraction_non_alphanumeric': 0.1087092731829574, 'fraction_numerical': 0.003132832080200501, 'mean_word_length': 3.00564617314931, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '150373', 'n_tokens_mistral': 2082, 'n_tokens_neox': 2006, 'n_words': 513}