<commit_before>from __future__ import print_function, division, absolute_import

from collections import defaultdict
import logging
from functools import partial
import socket

from tornado import gen
from tornado.gen import Return
from tornado.iostream import StreamClosedError

from .core import Server, read, write, rpc
from .utils import ignoring, ignore_exceptions


logger = logging.getLogger(__name__)


class Center(Server):
    """ Central metadata storage

    A Center serves as central point of metadata storage among workers.  It
    maintains dictionaries of which worker has which keys and which keys are
    owned by which workers.

    *  who_has:   {key: {set of workers}}
    *  has_what:  {worker: {set of keys}}

    Workers and clients check in with the Center to discover available resources

    You can start a center with the ``dcenter`` command line application::

       $ dcenter
       Start center at 127.0.0.1:8787

    Examples
    --------
    >>> c = Center('192.168.0.123', 8000)

    See Also
    --------
    distributed.worker.Worker:
    """
    def __init__(self, ip, port, **kwargs):
        self.ip = ip
        self.port = port
        self.who_has = defaultdict(set)
        self.has_what = defaultdict(set)
        self.ncores = dict()
        self.status = None

        d = {func.__name__: func
             for func in [self.add_keys, self.remove_keys, self.get_who_has,
                          self.get_has_what, self.register, self.get_ncores,
                          self.unregister, self.delete_data, self.terminate]}
        d = {k[len('get_'):] if k.startswith('get_') else k: v for k, v in
                d.items()}

        super(Center, self).__init__(d, **kwargs)

    @gen.coroutine
    def terminate(self, stream):
        self.stop()
        return b'OK'

    def register(self, stream, address=None, keys=(), ncores=None):
        self.has_what[address] = set(keys)
        self.ncores[address] = ncores
        logger.info("Register %s", str(address))
        return b'OK'

    def unregister(self, stream, address=None):
        if address not in self.has_what:
            return b'Address not found: ' + str(address).encode()
        keys = self.has_what.pop(address)
        with ignoring(KeyError):
            del self.ncores[address]
        for key in keys:
            self.who_has[key].remove(address)
        logger.info("Unregister %s", str(address))
        return b'OK'

    def add_keys(self, stream, address=None, keys=()):
        self.has_what[address].update(keys)
        for key in keys:
            self.who_has[key].add(address)
        return b'OK'

    def remove_keys(self, stream, keys=(), address=None):
        for key in keys:
            if key in self.has_what[address]:
                self.has_what[address].remove(key)
            with ignoring(KeyError):
                self.who_has[key].remove(address)
        return b'OK'

    def get_who_has(self, stream, keys=None):<commit_msg>remove key from who_has if empty in center<commit_after>from __future__ import print_function, division, absolute_import

from collections import defaultdict
import logging
from functools import partial
import socket

from tornado import gen
from tornado.gen import Return
from tornado.iostream import StreamClosedError

from .core import Server, read, write, rpc
from .utils import ignoring, ignore_exceptions


logger = logging.getLogger(__name__)


class Center(Server):
    """ Central metadata storage

    A Center serves as central point of metadata storage among workers.  It
    maintains dictionaries of which worker has which keys and which keys are
    owned by which workers.

    *  who_has:   {key: {set of workers}}
    *  has_what:  {worker: {set of keys}}

    Workers and clients check in with the Center to discover available resources

    You can start a center with the ``dcenter`` command line application::

       $ dcenter
       Start center at 127.0.0.1:8787

    Examples
    --------
    >>> c = Center('192.168.0.123', 8000)

    See Also
    --------
    distributed.worker.Worker:
    """
    def __init__(self, ip, port, **kwargs):
        self.ip = ip
        self.port = port
        self.who_has = defaultdict(set)
        self.has_what = defaultdict(set)
        self.ncores = dict()
        self.status = None

        d = {func.__name__: func
             for func in [self.add_keys, self.remove_keys, self.get_who_has,
                          self.get_has_what, self.register, self.get_ncores,
                          self.unregister, self.delete_data, self.terminate]}
        d = {k[len('get_'):] if k.startswith('get_') else k: v for k, v in
                d.items()}

        super(Center, self).__init__(d, **kwargs)

    @gen.coroutine
    def terminate(self, stream):
        self.stop()
        return b'OK'

    def register(self, stream, address=None, keys=(), ncores=None):
        self.has_what[address] = set(keys)
        self.ncores[address] = ncores
        logger.info("Register %s", str(address))
        return b'OK'

    def unregister(self, stream, address=None):
        if address not in self.has_what:
            return b'Address not found: ' + str(address).encode()
        keys = self.has_what.pop(address)
        with ignoring(KeyError):
            del self.ncores[address]
        for key in keys:
            s = self.who_has[key]
            s.remove(address)
            if not s:
                del self.who_has[key]
        logger.info("Unregister %s", str(address))
        return b'OK'

    def add_keys(self, stream, address=None, keys=()):
        self.has_what[address].update(keys)
        for key in keys:
            self.who_has[key].add(address)
        return b'OK'

    def remove_keys(self, stream, keys=(), address=None):
        for key in keys:
            if key in self.has_what[address]:
                self.has_what[address].remove(key)
            with ignoring(KeyError):
                self.who_has[key].remove(address)
        return b'OK'

    def get_who_has(self, stream, keys=None):
{'fraction_non_alphanumeric': 0.10468364831553, 'fraction_numerical': 0.007888249794576828, 'mean_word_length': 2.3568670711527853, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '7181532', 'n_tokens_mistral': 1842, 'n_tokens_neox': 1752, 'n_words': 527}