Viewing file:  local_counters.py (3.37 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
import logging
import os
import threading
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Dict, Generator
from xray.internal.constants import request_data_storage
logger = logging.getLogger(__name__)
@dataclass
class TaskCounterStorage:
"""
Typing that has lock and value storage with some syntax sugar.
"""
lock: threading.Lock
next_request_id: int
@property
def processed_requests(self):
return self.next_request_id - 1
_request_id_storage: Dict[str, TaskCounterStorage] = dict()
_global_storage_lock = threading.Lock()
@contextmanager
def open_local_storage(fake_task_id: str, flush=False) -> Generator[TaskCounterStorage, None, None]:
"""
Open local task information storage.
@param fake_task_id:
unique string, usually obtained as task.fake_id
@param flush:
whether to save data to file right after update
"""
logger.debug('Opening storage %s', fake_task_id)
storage = _get_or_create_record(fake_task_id)
with storage.lock:
yield storage
if flush:
logger.info('Updating task %s requests counter in file', fake_task_id)
_save_data_to_file(fake_task_id, storage)
def remove_local_storage(fake_task_id):
"""
Remove local storage record.
@param fake_task_id:
unique string, usually obtained as task.fake_id
"""
logger.info('Removing memory storage for task %s', fake_task_id)
with _global_storage_lock:
if fake_task_id in _request_id_storage:
del _request_id_storage[fake_task_id]
def get_task_ids():
"""
List all fake task ids saved in local storage.
"""
return list(_request_id_storage)
def flush_memory_storage(remove=True):
"""
List all fake task ids saved in local storage.
"""
for fake_task_id in list(_request_id_storage.keys()):
logger.info('Flushing task id %s on disk', fake_task_id)
with open_local_storage(fake_task_id) as storage:
_save_data_to_file(fake_task_id, storage)
if remove:
del _request_id_storage[fake_task_id]
def _save_data_to_file(fake_task_id: str, storage: TaskCounterStorage):
"""
Saves storage data from memory to file.
"""
req_id_file = os.path.join(request_data_storage, fake_task_id)
with open(req_id_file, 'w') as f:
f.write(str(storage.next_request_id))
def _get_or_create_record(fake_task_id) -> TaskCounterStorage:
"""
Takes record from local storage or creates new one and returns object
"""
with _global_storage_lock:
if fake_task_id not in _request_id_storage:
_request_id_storage[fake_task_id] = TaskCounterStorage(
lock=threading.Lock(),
next_request_id=1
)
storage = _request_id_storage[fake_task_id]
req_id_file = os.path.join(request_data_storage, fake_task_id)
try:
with open(req_id_file, 'r') as f:
storage.next_request_id = int(f.read())
except FileNotFoundError:
pass
else:
storage = _request_id_storage[fake_task_id]
return storage
|