<commit_before># Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from collections import defaultdict, deque
from math import sqrt, log, ceil
import atexit
import logging
import queue
import threading
import time
import types
import weakref

from psutil import virtual_memory, Process, NoSuchProcess
from sortedcontainers.sorteddict import SortedDict

from bndl.net.connection import NotConnected
from bndl.util.conf import Float
from bndl.util.funcs import getter
import bndl
import numpy as np


logger = logging.getLogger(__name__)


limit_system = Float(90, desc='The maximum amount of memory to be used by the system in percent (0-100).')
limit = Float(80, desc='The maximum amount of memory to be used by BNDL in percent (0-100).')


class MemorySupervisor(threading.Thread):
    def __init__(self, supervisor_rmi, limit=None, limit_system=None):
        super().__init__(name='memory-monitor', daemon=True)

        self.limit = limit or bndl.conf['bndl.compute.memory.limit']
        self.limit_system = limit_system or bndl.conf['bndl.compute.memory.limit_system']

        supervisor_rmi.peers.listeners.add(self.children_changed)
        self.procs = {}
        self.lock = threading.Lock()

        self._total = virtual_memory().total

        self.running = False
        atexit.register(self.stop)


    def children_changed(self, event, peer):
        pid = int(peer.name.split('.')[-1])
        with self.lock:
            if event == 'added':
                try:
                    proc = Process(pid)
                except NoSuchProcess:
                    pass
                else:
                    self.procs[pid] = (proc, peer.service('memory').release)
            elif event == 'removed':
                del self.procs[pid]


    def stop(self):
        self.running = False


    def run(self):
        self.running = True
        interval = .1
        rates = deque(maxlen=10)

        usage_prev = self.usage_pct()
        time_prev = time.time()

        while self.running:
            usage_system = virtual_memory().percent
            usage = self.usage_pct()

            over = max(usage - self.limit, usage_system - self.limit_system)
            release = (over * 1.1) if over > 0 else 0

            now = time.time()
            rate = (usage - usage_prev) / (now - time_prev)
            time_prev = now

            rates.append(rate)
            if rate > 0:
                rate = np.average(rates, weights=range(len(rates)))

            rate_factor = 0
            if usage > self.limit * .75:
                rate_factor += usage / self.limit
            if usage_system > self.limit_system * .75:
                rate_factor += usage_system / self.limit_system

            if rate > 0 and rate_factor > 0:
                rate_release = rate * log(rate_factor * 2)
                release += rate_release<commit_msg>apply weights 1 .. n instead of 0 .. n-1<commit_after># Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from collections import defaultdict, deque
from math import sqrt, log, ceil
import atexit
import logging
import queue
import threading
import time
import types
import weakref

from psutil import virtual_memory, Process, NoSuchProcess
from sortedcontainers.sorteddict import SortedDict

from bndl.net.connection import NotConnected
from bndl.util.conf import Float
from bndl.util.funcs import getter
import bndl
import numpy as np


logger = logging.getLogger(__name__)


limit_system = Float(90, desc='The maximum amount of memory to be used by the system in percent (0-100).')
limit = Float(80, desc='The maximum amount of memory to be used by BNDL in percent (0-100).')


class MemorySupervisor(threading.Thread):
    def __init__(self, supervisor_rmi, limit=None, limit_system=None):
        super().__init__(name='memory-monitor', daemon=True)

        self.limit = limit or bndl.conf['bndl.compute.memory.limit']
        self.limit_system = limit_system or bndl.conf['bndl.compute.memory.limit_system']

        supervisor_rmi.peers.listeners.add(self.children_changed)
        self.procs = {}
        self.lock = threading.Lock()

        self._total = virtual_memory().total

        self.running = False
        atexit.register(self.stop)


    def children_changed(self, event, peer):
        pid = int(peer.name.split('.')[-1])
        with self.lock:
            if event == 'added':
                try:
                    proc = Process(pid)
                except NoSuchProcess:
                    pass
                else:
                    self.procs[pid] = (proc, peer.service('memory').release)
            elif event == 'removed':
                del self.procs[pid]


    def stop(self):
        self.running = False


    def run(self):
        self.running = True
        interval = .1
        rates = deque(maxlen=10)

        usage_prev = self.usage_pct()
        time_prev = time.time()

        while self.running:
            usage_system = virtual_memory().percent
            usage = self.usage_pct()

            over = max(usage - self.limit, usage_system - self.limit_system)
            release = (over * 1.1) if over > 0 else 0

            now = time.time()
            rate = (usage - usage_prev) / (now - time_prev)
            time_prev = now

            rates.append(rate)
            if rate > 0:
                rate = np.average(rates, weights=range(1, len(rates) + 1))

            rate_factor = 0
            if usage > self.limit * .75:
                rate_factor += usage / self.limit
            if usage_system > self.limit_system * .75:
                rate_factor += usage_system / self.limit_system

            if rate > 0 and rate_factor > 0:
                rate_release = rate * log(rate_factor * 2)
                release += rate_release
{'fraction_non_alphanumeric': 0.09030446349394029, 'fraction_numerical': 0.010493644694058528, 'mean_word_length': 2.3009756097560974, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4082477', 'n_tokens_mistral': 1962, 'n_tokens_neox': 1846, 'n_words': 600}