<commit_before>/*jshint latedef:false */

define('kademlia/dht', [
    './id',
    './route_table',
    './bus',
    './message',
    'lodash',
    'when/when',
    'Bacon'
], function(Id, RouteTable, Bus, m, _, when, Bacon) {
    'use strict';

    function DHT(opts) {
        opts = _.assign({}, defaults, opts);

        var idSelf = opts.id || Id.random(opts.idSize)
          , alpha  = opts.alpha
          , routeTable = new RouteTable(idSelf, opts.idSize, opts.bucketSize)
          , bus        = new Bus(idSelf, opts.brokerInfo)
        ;

        bus.messages.onValue(react);
        bus.connectEvents.onValue(function(peer) {
            routeTable.insert(peer);
        });
        bus.closeEvents.onValue(function(id) {
            // TODO: attempt to reconnect?  handle that in bus?
            routeTable.remove(id);
        });

        setInterval(function() {
            routeTable.getNodes().forEach(function(node) {
                checkHealth(node, 3).then(function() {
                    // connection is good
                }, function(err) {
                    console.log('bad health check, disconnecting', node.id, err);
                    routeTable.remove(node.id);
                    bus.disconnect(node.id);
                });
            });
        }, 60000);

        function bootstrap(peers) {
            return when.any(peers.map(connect)).then(function() {
                findNode(idSelf).onValue(connect);
            });
        }

        function destroy() {
            return bus.destroy();
        }

        function connect(peer) {
            return bus.connect(peer).then(function() {
                // TODO: return promise from routeTable.insert,
                // disconnect on fail
                routeTable.insert(peer);
                return peer;
            });
        }

        function ping(peer) {
            return bus.query(peer, m.ping(idSelf));
        }

        function findNode(target) {
            return new Bacon.EventStream(function(subscriber) {
                var startPeers = routeTable.closest(target).slice(0, alpha);
                for (var i = 0; i < alpha; i += 1) {
                    findNode_(target, subscriber, startPeers.slice(i, i+1));
                }
                return function unsubscribe() {};
            });
        }

        function query(peer, type, params) {
            return bus.query(peer, m.query(type, params));
        }

        function findNode_(target, subscriber, peers, lastDist) {
            if (peers.length < 1) {
                return when.reject('out of peers');
            }

            return bus.query(peers[0], m.find_node(idSelf, target)).then(
            function(resp) {
                var nodes = resp.r.nodes.sort(function(a, b) {
                    return Id.compare(Id.dist(target, a.id), Id.dist(target, b.id));
                }).filter(function(node) {
                    return !lastDist || Id.compare(Id.dist(node.id, target), lastDist) < 0;
                });
                var closest = nodes.length && Id.dist(nodes[0].id, target);
                if (nodes.length && Id.equals(nodes[0].id, target)) {
                    subscriber(new Bacon.Next(nodes[0]));
                    subscriber(new Bacon.End());
                    return nodes[0];
                }
                else {
                    nodes.forEach(function(node) {
                        subscriber(new Bacon.Next(node));
                    });
                    return findNode_(target, subscriber, nodes, closest || lastDist);
                }
            },
            function() {
                // backtrack
                return findNode_(target, subscriber, peers.slice(1), lastDist);
            });
        }

        function checkHealth(peer, n) {
            return ping(peer).then(function() {
                // got response
            }, function() {
                if (n > 0) {
                    return when.delay(1000).then(function() {
                        return checkHealth(peer, n - 1);<commit_msg>increases health check timeouts<commit_after>/*jshint latedef:false */

define('kademlia/dht', [
    './id',
    './route_table',
    './bus',
    './message',
    'lodash',
    'when/when',
    'Bacon'
], function(Id, RouteTable, Bus, m, _, when, Bacon) {
    'use strict';

    function DHT(opts) {
        opts = _.assign({}, defaults, opts);

        var idSelf = opts.id || Id.random(opts.idSize)
          , alpha  = opts.alpha
          , routeTable = new RouteTable(idSelf, opts.idSize, opts.bucketSize)
          , bus        = new Bus(idSelf, opts.brokerInfo)
        ;

        bus.messages.onValue(react);
        bus.connectEvents.onValue(function(peer) {
            routeTable.insert(peer);
        });
        bus.closeEvents.onValue(function(id) {
            // TODO: attempt to reconnect?  handle that in bus?
            routeTable.remove(id);
        });

        setInterval(function() {
            routeTable.getNodes().forEach(function(node) {
                checkHealth(node, 3).then(function() {
                    // connection is good
                }, function(err) {
                    console.log('bad health check, disconnecting', node.id, err);
                    routeTable.remove(node.id);
                    bus.disconnect(node.id);
                });
            });
        }, 300000);

        function bootstrap(peers) {
            return when.any(peers.map(connect)).then(function() {
                findNode(idSelf).onValue(connect);
            });
        }

        function destroy() {
            return bus.destroy();
        }

        function connect(peer) {
            return bus.connect(peer).then(function() {
                // TODO: return promise from routeTable.insert,
                // disconnect on fail
                routeTable.insert(peer);
                return peer;
            });
        }

        function ping(peer) {
            return bus.query(peer, m.ping(idSelf));
        }

        function findNode(target) {
            return new Bacon.EventStream(function(subscriber) {
                var startPeers = routeTable.closest(target).slice(0, alpha);
                for (var i = 0; i < alpha; i += 1) {
                    findNode_(target, subscriber, startPeers.slice(i, i+1));
                }
                return function unsubscribe() {};
            });
        }

        function query(peer, type, params) {
            return bus.query(peer, m.query(type, params));
        }

        function findNode_(target, subscriber, peers, lastDist) {
            if (peers.length < 1) {
                return when.reject('out of peers');
            }

            return bus.query(peers[0], m.find_node(idSelf, target)).then(
            function(resp) {
                var nodes = resp.r.nodes.sort(function(a, b) {
                    return Id.compare(Id.dist(target, a.id), Id.dist(target, b.id));
                }).filter(function(node) {
                    return !lastDist || Id.compare(Id.dist(node.id, target), lastDist) < 0;
                });
                var closest = nodes.length && Id.dist(nodes[0].id, target);
                if (nodes.length && Id.equals(nodes[0].id, target)) {
                    subscriber(new Bacon.Next(nodes[0]));
                    subscriber(new Bacon.End());
                    return nodes[0];
                }
                else {
                    nodes.forEach(function(node) {
                        subscriber(new Bacon.Next(node));
                    });
                    return findNode_(target, subscriber, nodes, closest || lastDist);
                }
            },
            function() {
                // backtrack
                return findNode_(target, subscriber, peers.slice(1), lastDist);
            });
        }

        function checkHealth(peer, n) {
            return ping(peer).then(function() {
                // got response
            }, function() {
                if (n > 0) {
                    return when.delay(90000).then(function() {
                        return checkHealth(peer, n - 1);
{'fraction_non_alphanumeric': 0.12790413303986306, 'fraction_numerical': 0.006113964294448521, 'mean_word_length': 1.4112617924528301, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '6106277', 'n_tokens_mistral': 2279, 'n_tokens_neox': 2118, 'n_words': 426}