NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

notice_routing_cache.py (4692B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\publicGateway\notice\notice_routing_cache.py
      2 from datetime import datetime, timedelta
      3 import collections
      4 from logging import getLogger
      5 EXPIRY_DELAY_SECS = 10
      6 MAX_SIZE_BYTES = 2000000
      7 session_to_target_group = {'solarsystemid2': 'solar_system',
      8  'userid': 'user',
      9  'corpid': 'corporation',
     10  'allianceid': 'alliance',
     11  'charid': 'character'}
     12 
     13 class CacheMemoryTracker(object):
     14 
     15     def __init__(self, max_bytes):
     16         self.current_bytes = 0
     17         self.max_bytes = max_bytes
     18 
     19     def will_fit(self, notice):
     20         notice_size = notice.ByteSize()
     21         if notice_size + self.current_bytes <= self.max_bytes:
     22             return 0
     23         return notice_size + self.current_bytes - self.max_bytes
     24 
     25     def track_bytes(self, bytes):
     26         if bytes + self.current_bytes > self.max_bytes:
     27             raise CacheOOMException()
     28         self.current_bytes += bytes
     29 
     30     def release_bytes(self, bytes):
     31         self.current_bytes = max(0, self.current_bytes - bytes)
     32 
     33 
     34 class NoticeCacheEntry(object):
     35 
     36     def __init__(self, notice, target_group, target_group_id, expiry):
     37         self.notice = notice
     38         self.target_group = target_group
     39         self.target_group_id = target_group_id
     40         self.expiry = expiry
     41         self.bytes = notice.ByteSize()
     42 
     43 
     44 class NoticeCache(object):
     45 
     46     def __init__(self, memory_tracker):
     47         self.deque = collections.deque()
     48         self.memory_tracker = memory_tracker
     49 
     50     def add_notice_entry(self, notice_entry):
     51         self.memory_tracker.track_bytes(notice_entry.bytes)
     52         self.deque.append(notice_entry)
     53 
     54     def evict_for_bytes(self, bytes_needed):
     55         bytes_recovered = 0
     56         while len(self.deque) > 0:
     57             entry = self.deque.popleft()
     58             bytes_recovered += entry.bytes
     59             self.memory_tracker.release_bytes(entry.bytes)
     60             if bytes_recovered >= bytes_needed:
     61                 break
     62 
     63         if bytes_recovered < bytes_needed:
     64             raise CacheOOMException()
     65 
     66     def remove_by_expiry(self, expiry):
     67         while len(self.deque) > 0:
     68             entry = self.deque[0]
     69             if entry.expiry > expiry:
     70                 break
     71             self.memory_tracker.release_bytes(entry.bytes)
     72             self.deque.popleft()
     73 
     74     def get_newly_valid_notices(self, notices_to_dispatch, targets):
     75         requeue_entries = []
     76         while len(self.deque) > 0:
     77             entry = self.deque.popleft()
     78             if (entry.target_group, entry.target_group_id) not in targets:
     79                 requeue_entries.append(entry)
     80                 continue
     81             notices_to_dispatch.append(entry.notice)
     82             self.memory_tracker.release_bytes(entry.bytes)
     83 
     84         requeue_entries.reverse()
     85         self.deque.extendleft(requeue_entries)
     86 
     87     def empty(self):
     88         return len(self.deque) == 0
     89 
     90     def peek_first(self):
     91         return self.deque[0]
     92 
     93     def num_entries(self):
     94         return len(self.deque)
     95 
     96 
     97 class UnhandledTargetGroupException(Exception):
     98     pass
     99 
    100 
    101 class CacheOOMException(Exception):
    102     pass
    103 
    104 
    105 class NoticeRoutingCache(object):
    106 
    107     def __init__(self):
    108         self._memory_tracker = CacheMemoryTracker(MAX_SIZE_BYTES)
    109         self._notice_cache = NoticeCache(self._memory_tracker)
    110         self._logger = getLogger(__name__)
    111 
    112     def remove_expired_entries(self):
    113         now = datetime.utcnow()
    114         self._notice_cache.remove_by_expiry(now)
    115 
    116     def add_notice(self, notice, target_group, target_group_id, override_expiry = None):
    117         expiry = override_expiry if override_expiry is not None else datetime.utcnow()
    118         expiry += timedelta(seconds=EXPIRY_DELAY_SECS)
    119         if target_group not in session_to_target_group.values():
    120             raise UnhandledTargetGroupException
    121         remaining_bytes = self._memory_tracker.will_fit(notice)
    122         if remaining_bytes > 0:
    123             self._notice_cache.evict_for_bytes(remaining_bytes)
    124         self._notice_cache.add_notice_entry(NoticeCacheEntry(notice, target_group, target_group_id, expiry))
    125 
    126     def get_newly_valid_notices(self, session_change):
    127         self.remove_expired_entries()
    128         notices_to_dispatch = []
    129         targets = []
    130         _populate_targets(targets, session_change)
    131         self._notice_cache.get_newly_valid_notices(notices_to_dispatch, targets)
    132         return notices_to_dispatch
    133 
    134 
    135 def _populate_targets(targets, session_change):
    136     for key in session_to_target_group:
    137         if key not in session_change:
    138             continue
    139         target_group = session_to_target_group[key]
    140         _, target_group_id = session_change[key]
    141         targets.append((target_group, target_group_id))