Реализация многопоточных сетевых серверов на Python

import socket
 
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('127.0.0.1', 8888))
server_socket.listen(5)  # максимальная очередь ожидающих соединений
 
while True:
    client_socket, addr = server_socket.accept()
    data = client_socket.recv(1024)
    client_socket.send(data)  # эхо-ответ
    client_socket.close()

while True:
    client_socket, addr = server_socket.accept()
    # Создаём новый поток для каждого клиента
    client_thread = threading.Thread(
        target=handle_client, 
        args=(client_socket,)
    )
    client_thread.start()

import threading
import time
 
def worker(name):
    print(f"Поток {name} начал работу")
    time.sleep(1)  # Имитация I/O операции
    print(f"Поток {name} завершил работу")
 
# Создаем и запускаем 5 потоков
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()
 
# Ждем завершения всех потоков
for t in threads:
    t.join()

from multiprocessing import Process
 
def worker(name):
    print(f"Процесс {name} начал работу")
    # CPU-intensive операция - здесь multiprocessing эффективнее
    result = sum(i * i for i in range(10**7))
    print(f"Процесс {name} завершил работу, результат: {result}")
 
processes = []
for i in range(3):
    p = Process(target=worker, args=(i,))
    processes.append(p)
    p.start()
 
for p in processes:
    p.join()

import socket
import threading
 
def handle_client(client_socket, client_address):
    print(f"Подключение с {client_address}")
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break  # Клиент отключился
            
            # Обрабатываем данные (эхо-сервер)
            response = f"Получено: {data.decode('utf-8')}"
            client_socket.send(response.encode('utf-8'))
    except Exception as e:
        print(f"Ошибка при обработке клиента: {e}")
    finally:
        client_socket.close()
        print(f"Соединение с {client_address} закрыто")
 
def start_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # Привязка к адресу и порту
    server_address = ('0.0.0.0', 9999)
    server.bind(server_address)
    server.listen(5)
    
    print(f"Сервер запущен на {server_address}")
    
    try:
        while True:
            # Ожидание подключения
            client_sock, client_addr = server.accept()
            
            # Создание и запуск потока для обработки клиента
            client_thread = threading.Thread(
                target=handle_client,
                args=(client_sock, client_addr)
            )
            client_thread.daemon = True  # Поток завершится при выходе из программы
            client_thread.start()
            
            print(f"Активных подключений: {threading.active_count() - 1}")
    
    except KeyboardInterrupt:
        print("Завершение работы сервера...")
    finally:
        server.close()
 
if __name__ == "__main__":
    start_server()

import socket
import threading
from concurrent.futures import ThreadPoolExecutor
 
def handle_client(client_data):
    client_socket, client_address = client_data
    # Та же логика обработки, что и раньше...
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            
            response = f"Получено: {data.decode('utf-8')}"
            client_socket.send(response.encode('utf-8'))
    except Exception as e:
        print(f"Ошибка при обработке клиента: {e}")
    finally:
        client_socket.close()
        print(f"Соединение с {client_address} закрыто")
 
def start_server_with_pool():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_address = ('0.0.0.0', 9999)
    server.bind(server_address)
    server.listen(5)
    
    print(f"Сервер запущен на {server_address}")
    
    # Создаем пул потоков
    with ThreadPoolExecutor(max_workers=20) as executor:
        try:
            while True:
                client_sock, client_addr = server.accept()
                executor.submit(handle_client, (client_sock, client_addr))
        except KeyboardInterrupt:
            print("Завершение работы сервера...")
        finally:
            server.close()

import socket
import select
 
def start_reactor_server():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.setblocking(0)  # Неблокирующий режим
    server_address = ('0.0.0.0', 9999)
    server.bind(server_address)
    server.listen(5)
    
    inputs = [server]  # Сокеты, ожидающие чтения
    outputs = []       # Сокеты, ожидающие записи
    message_queues = {} # Очереди сообщений для каждого клиента
    
    print(f"Реактор-сервер запущен на {server_address}")
    
    try:
        while inputs:
            # Ожидание активности на любом сокете
            readable, writable, exceptional = select.select(
                inputs, outputs, inputs)
            
            # Обработка сокетов, готовых к чтению
            for s in readable:
                if s is server:
                    # Новое соединение
                    client_socket, client_address = s.accept()
                    client_socket.setblocking(0)
                    inputs.append(client_socket)
                    message_queues[client_socket] = []
                    print(f"Соединение с {client_address}")
                else:
                    # Данные от существующего клиента
                    data = s.recv(1024)
                    if data:
                        # Есть данные от клиента
                        message_queues[s].append(data)
                        if s not in outputs:
                            outputs.append(s)
                    else:
                        # Клиент отключился
                        if s in outputs:
                            outputs.remove(s)
                        inputs.remove(s)
                        s.close()
                        del message_queues[s]
            
            # Обработка сокетов, готовых к записи
            for s in writable:
                if message_queues[s]:
                    data = message_queues[s].pop(0)
                    response = f"Получено: {data.decode('utf-8')}"
                    s.send(response.encode('utf-8'))
                else:
                    outputs.remove(s)
            
            # Обработка сокетов с ошибками
            for s in exceptional:
                inputs.remove(s)
                if s in outputs:
                    outputs.remove(s)
                s.close()
                del message_queues[s]
    
    except KeyboardInterrupt:
        print("Завершение работы сервера...")
    finally:
        server.close()

import socket
import threading
import queue
 
def worker(task_queue):
    """Функция рабочего потока, обрабатывающего задачи из очереди"""
    while True:
        client_socket, client_address = task_queue.get()
        try:
            print(f"Обработка клиента {client_address}")
            while True:
                data = client_socket.recv(1024)
                if not data:
                    break
                
                response = f"Получено: {data.decode('utf-8')}"
                client_socket.send(response.encode('utf-8'))
        except Exception as e:
            print(f"Ошибка при обработке клиента: {e}")
        finally:
            client_socket.close()
            task_queue.task_done()

def start_server_with_queue():
 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 server_address = ('0.0.0.0', 9999)
 server.bind(server_address)
 server.listen(5)
 
 print(f"Сервер с очередью запущен на {server_address}")
 
 # Создаем очередь задач
 task_queue = queue.Queue()
 
 # Запускаем пул рабочих потоков
 NUM_WORKERS = 10
 workers = []
 for _ in range(NUM_WORKERS):
     t = threading.Thread(target=worker, args=(task_queue,))
     t.daemon = True
     t.start()
     workers.append(t)
 
 # Основной цикл сервера - принимает соединения и добавляет их в очередь
 try:
     while True:
         client_sock, client_addr = server.accept()
         task_queue.put((client_sock, client_addr))
 except KeyboardInterrupt:
     print("Завершение работы сервера...")
 finally:
     server.close()

import threading
 
# Глобальный счетчик и блокировка для него
counter = 0
counter_lock = threading.Lock()
 
def increment_counter():
 global counter
 with counter_lock:
     # Критическая секция - только один поток может выполнять её одновременно
     current = counter
     # Имитируем некоторую работу, которая может привести к переключению контекста
     time.sleep(0.001) 
     counter = current + 1
 
# Запускаем 10 потоков, каждый из которых инкрементирует счетчик 100 раз
threads = []
for _ in range(10):
 t = threading.Thread(target=lambda: [increment_counter() for _ in range(100)])
 threads.append(t)
 t.start()
 
for t in threads:
 t.join()
 
print(f"Финальное значение счетчика: {counter}")  # Должно быть 1000

# Ограничиваем количество одновременных подключений к внешнему ресурсу
max_connections = 5
connection_semaphore = threading.Semaphore(max_connections)
 
def access_external_resource():
 with connection_semaphore:
     # Только max_connections потоков могут быть здесь одновременно
     print("Доступ к внешнему ресурсу получен")
     time.sleep(1)  # Имитация работы с внешним ресурсом
     print("Работа с внешним ресурсом завершена")

# Очередь задач с ограниченным размером
class BoundedQueue:
 def __init__(self, size):
     self.queue = []
     self.size = size
     self.condition = threading.Condition()
 
 def put(self, item):
     with self.condition:
         while len(self.queue) >= self.size:
             print("Очередь заполнена, ожидаем освобождения места")
             self.condition.wait()  # Ждем, пока не освободится место
         
         self.queue.append(item)
         print(f"Добавлен элемент: {item}, размер очереди: {len(self.queue)}")
         self.condition.notify()  # Сообщаем ожидающим потокам, что есть элементы
 
 def get(self):
     with self.condition:
         while not self.queue:
             print("Очередь пуста, ожидаем элементы")
             self.condition.wait()  # Ждем, пока не появятся элементы
         
         item = self.queue.pop(0)
         print(f"Извлечен элемент: {item}, размер очереди: {len(self.queue)}")
         self.condition.notify()  # Сообщаем ожидающим потокам, что есть место
         return item

# Синхронизируем запуск нагрузочного теста
barrier = threading.Barrier(10)  # Ждем 10 потоков
 
def load_test_client(client_id):
 print(f"Клиент {client_id} готовится к тесту")
 time.sleep(random.random())  # Случайная задержка для подготовки
 
 # Все потоки ждут здесь, пока все 10 не будут готовы
 barrier.wait()
 
 print(f"Клиент {client_id} начал нагрузочный тест")
 # Выполняем тест...

# Неоптимальный подход - слишком маленький буфер
data = b""
chunk = client_socket.recv(1)  # Считываем по 1 байту - крайне неэффективно
while chunk != b"\n":
 data += chunk
 chunk = client_socket.recv(1)
 
# Оптимальный подход - буфер разумного размера
data = client_socket.recv(8192)  # 8KB - обычно хороший размер для TCP

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
# Переиспользование адреса (важно при перезапусках сервера)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
# Отключение алгоритма Нейгла для снижения задержки (важно для интерактивных приложений)
server_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 
# Увеличение размера буферов сокета
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 131072)  # 128KB
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 131072)  # 128KB
 
# Настройка keep-alive для обнаружения обрыва соединения
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

client_socket.setblocking(0)
try:
 data = client_socket.recv(8192)
except socket.error as e:
 if e.errno == socket.EAGAIN or e.errno == socket.EWOULDBLOCK:
     # Данных пока нет, но это не ошибка
     pass
 else:
     # Реальная ошибка, обработать
     raise

import psycopg2
from psycopg2 import pool
 
# Создаем пул соединений с БД
db_pool = psycopg2.pool.ThreadedConnectionPool(
 minconn=5,       # Минимальное количество соединений
 maxconn=20,      # Максимальное количество соединений
 host="localhost",
 database="mydb",
 user="user",
 password="password"
)
 
def handle_client(client_socket):
 # Получаем соединение из пула
 conn = db_pool.getconn()
 try:
     # Используем соединение
     with conn.cursor() as cursor:
         cursor.execute("SELECT * FROM users")
         results = cursor.fetchall()
         # Обработка результатов...
 finally:
     # Возвращаем соединение в пул
     db_pool.putconn(conn)

import asyncio
import threading
 
def run_async_in_thread(coro_func, *args):
 """Запускает асинхронную функцию в отдельном потоке с новым циклом событий"""
 async def wrapper():
     result = await coro_func(*args)
     return result
 
 loop = asyncio.new_event_loop()
 asyncio.set_event_loop(loop)
 result = loop.run_until_complete(wrapper())
 loop.close()
 return result
 
# Использование в обработчике клиента
def handle_client(client_socket):
 data = client_socket.recv(8192)
 
 # Предположим, что нам нужно сделать несколько длительных HTTP-запросов
 results = run_async_in_thread(make_multiple_api_calls, data)
 
 client_socket.send(results.encode('utf-8'))
 client_socket.close()
 
# Асинхронная функция для параллельных API-вызовов
async def make_multiple_api_calls(data):
 async with aiohttp.ClientSession() as session:
     tasks = [
         fetch_api_data(session, 'http://api1.example.com'),
         fetch_api_data(session, 'http://api2.example.com'),
         fetch_api_data(session, 'http://api3.example.com')
     ]
     results = await asyncio.gather(*tasks)
     return process_results(results)

import asyncio
 
async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"Подключение с {addr}")
    
    while True:
        data = await reader.read(1024)
        if not data:
            break
        
        message = data.decode('utf-8')
        print(f"Получено от {addr}: {message}")
        
        response = f"Эхо: {message}"
        writer.write(response.encode('utf-8'))
        await writer.drain()
    
    print(f"Закрытие соединения с {addr}")
    writer.close()
    await writer.wait_closed()
 
async def main():
    server = await asyncio.start_server(
        handle_client, '0.0.0.0', 9999)
    
    addr = server.sockets[0].getsockname()
    print(f'Асинхронный сервер запущен на {addr}')
    
    async with server:
        await server.serve_forever()
 
if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import multiprocessing
import os
 
async def handle_client(reader, writer):
    # Асинхронная обработка клиента...
    pass
 
async def run_server():
    server = await asyncio.start_server(
        handle_client, '0.0.0.0', 9999)
    
    async with server:
        await server.serve_forever()
 
def process_worker():
    print(f"Рабочий процесс {os.getpid()} запущен")
    asyncio.run(run_server())
 
if __name__ == "__main__":
    # Запускаем по рабочему процессу на каждое ядро
    num_processes = multiprocessing.cpu_count()
    print(f"Запуск {num_processes} рабочих процессов")
    
    processes = []
    for _ in range(num_processes):
        p = multiprocessing.Process(target=process_worker)
        p.start()
        processes.append(p)
    
    # Ожидаем завершения всех процессов
    for p in processes:
        p.join()

def handle_client(client_socket):
   try:
       while True:
           try:
               data = client_socket.recv(1024)
               if not data:
                   break
               # Обработка данных...
           except socket.timeout:
               print("Тайм-аут чтения, повторная попытка")
               continue
           except ConnectionResetError:
               print("Клиент неожиданно отключился")
               break
   finally:
       client_socket.close()

def watchdog():
   """Мониторит состояние рабочих потоков и перезапускает зависшие"""
   while True:
       time.sleep(30)  # Проверка каждые 30 секунд
       
       with thread_status_lock:
           current_time = time.time()
           for thread_id, status in thread_status.items():
               if status['state'] == 'busy' and (current_time - status['last_active']) > 300:
                   print(f"Поток {thread_id} завис! Обработка...")
                   # Здесь можно добавить логику восстановления

# /etc/supervisor/conf.d/myserver.conf
[program:myserver]
command=python3 /path/to/server.py
directory=/path/to
user=www-data
autostart=true
autorestart=true
startretries=10
redirect_stderr=true
stdout_logfile=/var/log/myserver.log

def reload_server():
   """Перезагружает сервер без отключения клиентов"""
   # Сигнал рабочим потокам о предстоящей перезагрузке
   global should_reload
   should_reload = True
   
   # Ждем завершения всех текущих соединений
   while active_connections > 0:
       time.sleep(0.1)
   
   # Перезагрузка модуля с кодом обработчиков
   importlib.reload(handlers_module)
   
   # Сброс флага перезагрузки
   should_reload = False
   print("Сервер успешно обновлен!")

class RateLimiter:
   def __init__(self, max_calls, time_frame):
       self.max_calls = max_calls  # Максимальное количество запросов
       self.time_frame = time_frame  # Временной интервал в секундах
       self.calls = {}  # IP -> список временных меток
       self.lock = threading.Lock()
   
   def is_allowed(self, ip):
       with self.lock:
           now = time.time()
           if ip not in self.calls:
               self.calls[ip] = [now]
               return True
           
           # Удаляем устаревшие вызовы
           self.calls[ip] = [t for t in self.calls[ip] if now - t < self.time_frame]
           
           # Проверяем лимит
           if len(self.calls[ip]) < self.max_calls:
               self.calls[ip].append(now)
               return True
           return False
 
# Использование
limiter = RateLimiter(max_calls=10, time_frame=60)  # 10 запросов в минуту
 
def handle_client(client_socket, client_address):
   ip = client_address[0]
   if not limiter.is_allowed(ip):
       client_socket.send(b"Too many requests\n")
       client_socket.close()
       return
   
   # Нормальная обработка клиента...

# Установка таймаута сокета
client_socket.settimeout(30)  # 30 секунд на чтение/запись
 
# Ограничение размера запроса
MAX_REQUEST_SIZE = 1024 * 1024  # 1 MB
 
def receive_full_request(sock):
   data = b""
   while True:
       chunk = sock.recv(8192)
       if not chunk:
           break
       
       data += chunk
       if len(data) > MAX_REQUEST_SIZE:
           raise ValueError("Запрос слишком большой")
       
       # Проверяем, получен ли полный запрос
       if b"\r\n\r\n" in data:
           return data

def exception_handler(exc_type, exc_value, exc_traceback):
   """Глобальный обработчик необработанных исключений"""
   logger.error("Необработанное исключение", 
               exc_info=(exc_type, exc_value, exc_traceback))
   # Можно отправить уведомление, например, в Slack или по email
   send_alert("Критическая ошибка на сервере: " + str(exc_value))
 
# Устанавливаем глобальный обработчик
sys.excepthook = exception_handler
 
# Для потоков нужен отдельный хук
threading.excepthook = lambda args: exception_handler(
   args.exc_type, args.exc_value, args.exc_traceback)

import ssl
 
def create_ssl_server():
   context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
   context.load_cert_chain(certfile="server.crt", keyfile="server.key")
   
   server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   server.bind(('0.0.0.0', 8443))
   server.listen(5)
   
   while True:
       client_sock, client_addr = server.accept()
       try:
           # Оборачиваем сокет в SSL
           secure_sock = context.wrap_socket(
               client_sock, server_side=True)
           
           # Запускаем обработку в отдельном потоке
           client_thread = threading.Thread(
               target=handle_client,
               args=(secure_sock, client_addr)
           )
           client_thread.daemon = True
           client_thread.start()
       
       except ssl.SSLError as e:
           print(f"SSL ошибка: {e}")
           client_sock.close()

multithread_server/
│
├── server.py           # Основной файл сервера
├── config.py           # Конфигурация
├── worker_pool.py      # Реализация пула рабочих потоков
├── metrics.py          # Сбор и визуализация метрик
├── client_handler.py   # Обработчики клиентских соединений
├── utils/
│   ├── __init__.py
│   ├── logger.py       # Настройка логирования
│   └── security.py     # Функции безопасности
└── tests/              # Тесты
    ├── __init__.py
    ├── test_server.py
    └── test_load.py

import threading
import queue
import time
import logging
from typing import Callable, Any, Dict, List, Optional
 
class WorkerPool:
    """Пул рабочих потоков с мониторингом состояния и балансировкой нагрузки"""
    
    def __init__(self, num_workers: int, max_queue_size: int = 100):
        self.task_queue = queue.Queue(maxsize=max_queue_size)
        self.workers: List[threading.Thread] = []
        self.active_tasks: Dict[int, Dict[str, Any]] = {}
        self.lock = threading.Lock()
        self.running = True
        self.stats = {
            'processed_tasks': 0,
            'errors': 0,
            'avg_processing_time': 0,
            'queue_size_history': []
        }
        
        # Запуск рабочих потоков
        for i in range(num_workers):
            worker = threading.Thread(
                target=self._worker_loop, 
                args=(i,),
                daemon=True
            )
            worker.start()
            self.workers.append(worker)
        
        # Запуск мониторинга
        self.monitor_thread = threading.Thread(
            target=self._monitor_loop,
            daemon=True
        )
        self.monitor_thread.start()
    
    def _worker_loop(self, worker_id: int) -> None:
        """Основной цикл рабочего потока"""
        logger = logging.getLogger(f'worker-{worker_id}')
        
        while self.running:
            try:
                # Получение задачи из очереди
                task_id, task_func, args, kwargs = self.task_queue.get(timeout=1)
                
                # Регистрация активной задачи
                with self.lock:
                    self.active_tasks[task_id] = {
                        'worker_id': worker_id,
                        'start_time': time.time(),
                        'function': task_func.__name__,
                        'args': args
                    }
                
                try:
                    # Выполнение задачи
                    logger.debug(f"Выполняется задача {task_id}: {task_func.__name__}")
                    start_time = time.time()
                    result = task_func(*args, **kwargs)
                    processing_time = time.time() - start_time
                    
                    # Обновление статистики
                    with self.lock:
                        self.stats['processed_tasks'] += 1
                        n = self.stats['processed_tasks']
                        self.stats['avg_processing_time'] = (
                            (self.stats['avg_processing_time'] * (n-1) + processing_time) / n
                        )
                
                except Exception as e:
                    logger.error(f"Ошибка в задаче {task_id}: {e}")
                    with self.lock:
                        self.stats['errors'] += 1
                
                finally:
                    # Удаление из активных задач
                    with self.lock:
                        if task_id in self.active_tasks:
                            del self.active_tasks[task_id]
                    
                    # Отметка о завершении задачи
                    self.task_queue.task_done()
            
            except queue.Empty:
                # Очередь пуста, продолжаем ожидание
                continue
            except Exception as e:
                logger.error(f"Неожиданная ошибка в рабочем потоке: {e}")
                with self.lock:
                    self.stats['errors'] += 1
    
    def _monitor_loop(self) -> None:
        """Мониторинг состояния пула и сбор метрик"""
        logger = logging.getLogger('pool-monitor')
        
        while self.running:
            try:
                # Сбор текущих метрик
                with self.lock:
                    queue_size = self.task_queue.qsize()
                    self.stats['queue_size_history'].append(queue_size)
                    self.stats['queue_size_history'] = self.stats['queue_size_history'][-100:]
                
                # Логирование состояния пула каждые 60 секунд
                logger.info(
                    f"Статистика пула: задач обработано: {self.stats['processed_tasks']}, "
                    f"ошибок: {self.stats['errors']}, "
                    f"среднее время обработки: {self.stats['avg_processing_time']:.4f}s, "
                    f"размер очереди: {queue_size}, "
                    f"активных задач: {len(self.active_tasks)}"
                )
                
                # Проверка зависших задач
                current_time = time.time()
                with self.lock:
                    for task_id, task_info in list(self.active_tasks.items()):
                        task_duration = current_time - task_info['start_time']
                        # Если задача выполняется более 5 минут, логируем предупреждение
                        if task_duration > 300:
                            logger.warning(
                                f"Задача {task_id} ({task_info['function']}) "
                                f"выполняется слишком долго: {task_duration:.1f}s"
                            )
                
                time.sleep(60)  # Интервал мониторинга
            
            except Exception as e:
                logger.error(f"Ошибка в мониторинге: {e}")
    
    def submit_task(self, task_func: Callable, *args, **kwargs) -> int:
        """Отправка задачи в пул на выполнение"""
        if not self.running:
            raise RuntimeError("Пул остановлен")
        
        # Генерация ID задачи
        task_id = hash(f"{time.time()}-{task_func.__name__}-{args}")
        
        # Добавление задачи в очередь
        self.task_queue.put((task_id, task_func, args, kwargs))
        
        return task_id
    
    def shutdown(self, wait=True) -> None:
        """Остановка пула потоков"""
        self.running = False
        
        if wait:
            # Ожидание завершения всех задач
            self.task_queue.join()

# server.py
import socket
import threading
import logging
import time
import signal
import sys
from typing import Tuple, Optional
 
from worker_pool import WorkerPool
from config import ServerConfig
from client_handler import handle_client
from utils.logger import setup_logger
from metrics import MetricsCollector
 
class MultiThreadServer:
    """Многопоточный сервер с пулом рабочих потоков и мониторингом"""
    
    def __init__(self, config: ServerConfig):
        self.config = config
        self.running = False
        self.server_socket: Optional[socket.socket] = None
        
        # Настройка логирования
        self.logger = setup_logger('server', config.log_level, config.log_file)
        
        # Создание пула потоков
        self.worker_pool = WorkerPool(
            num_workers=config.num_workers,
            max_queue_size=config.task_queue_size
        )
        
        # Создание сборщика метрик
        self.metrics = MetricsCollector(
            collect_interval=config.metrics_interval,
            history_size=config.metrics_history_size
        )
        
        # Регистрация обработчиков сигналов для корректного завершения
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)
    
    def _signal_handler(self, sig, frame) -> None:
        """Обработчик сигналов остановки"""
        self.logger.info(f"Получен сигнал {sig}, завершение работы...")
        self.stop()
    
    def start(self) -> None:
        """Запуск сервера"""
        if self.running:
            return
        
        self.running = True
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        
        # Применяем дополнительные настройки сокета
        if self.config.tcp_nodelay:
            self.server_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        
        if self.config.socket_buffer_size:
            self.server_socket.setsockopt(
                socket.SOL_SOCKET, 
                socket.SO_RCVBUF, 
                self.config.socket_buffer_size
            )
        
        # Привязка к адресу и порту
        self.server_socket.bind((self.config.host, self.config.port))
        self.server_socket.listen(self.config.listen_backlog)
        
        self.logger.info(f"Сервер запущен на {self.config.host}:{self.config.port}")
        
        # Запуск сборщика метрик
        self.metrics.start()
        
        # Запуск основного цикла приема соединений
        self.accept_thread = threading.Thread(target=self._accept_connections)
        self.accept_thread.daemon = True
        self.accept_thread.start()
    
    def _accept_connections(self) -> None:
        """Цикл приема входящих соединений"""
        while self.running:
            try:
                # Принимаем соединение с таймаутом для возможности корректного завершения
                self.server_socket.settimeout(1.0)
                try:
                    client_socket, client_address = self.server_socket.accept()
                except socket.timeout:
                    continue
                
                # Сбрасываем таймаут для клиентского сокета
                client_socket.settimeout(self.config.client_timeout)
                
                # Регистрируем новое соединение в метриках
                self.metrics.record_event('connection_accepted')
                
                # Отправляем задачу в пул потоков
                self.worker_pool.submit_task(
                    handle_client, 
                    client_socket, 
                    client_address,
                    self.config,
                    self.metrics
                )
                
                # Обновляем метрики очереди
                self.metrics.record_gauge(
                    'task_queue_size', 
                    self.worker_pool.task_queue.qsize()
                )
                
            except OSError as e:
                if not self.running:
                    break  # Сервер останавливается
                self.logger.error(f"Ошибка при принятии соединения: {e}")
                
            except Exception as e:
                self.logger.error(f"Неожиданная ошибка: {e}")
                self.metrics.record_event('server_error')
    
    def stop(self) -> None:
        """Остановка сервера"""
        if not self.running:
            return
        
        self.running = False
        
        # Закрытие серверного сокета
        if self.server_socket:
            self.server_socket.close()
        
        # Остановка пула потоков
        self.logger.info("Остановка пула потоков...")
        self.worker_pool.shutdown(wait=True)
        
        # Остановка сборщика метрик
        self.metrics.stop()
        
        self.logger.info("Сервер остановлен")

def _adaptive_scaling(self) -> None:
    """Адаптивное масштабирование пула потоков"""
    logger = logging.getLogger('adaptive-scaler')
    
    min_workers = self.config.min_workers
    max_workers = self.config.max_workers
    current_workers = len(self.workers)
    
    while self.running:
        try:
            with self.lock:
                queue_size = self.task_queue.qsize()
                active_tasks = len(self.active_tasks)
                
                # Расчет нагрузки на пул
                utilization = active_tasks / current_workers if current_workers > 0 else 1.0
                
                # Принятие решения о масштабировании
                if utilization > 0.8 and queue_size > 10 and current_workers < max_workers:
                    # Увеличиваем количество рабочих потоков
                    new_workers = min(
                        max_workers - current_workers,
                        max(1, current_workers // 4)  # Увеличиваем на 25%
                    )
                    
                    logger.info(f"Масштабирование вверх: добавляем {new_workers} потоков")
                    
                    for i in range(new_workers):
                        worker_id = current_workers + i
                        worker = threading.Thread(
                            target=self._worker_loop, 
                            args=(worker_id,),
                            daemon=True
                        )
                        worker.start()
                        self.workers.append(worker)
                    
                    current_workers += new_workers
                
                elif utilization < 0.2 and current_workers > min_workers:
                    # Уменьшаем количество рабочих потоков
                    workers_to_remove = min(
                        current_workers - min_workers,
                        max(1, current_workers // 5)  # Уменьшаем на 20%
                    )
                    
                    logger.info(f"Масштабирование вниз: убираем {workers_to_remove} потоков")
                    
                    # Помечаем потоки для удаления
                    self.workers_to_remove = workers_to_remove
            
            # Проверяем каждые 30 секунд
            time.sleep(30)
            
        except Exception as e:
            logger.error(f"Ошибка в адаптивном масштабировании: {e}")

import time
import threading
import collections
import logging
from typing import Dict, List, Any, Optional
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
 
class MetricsCollector:
    """Сборщик и визуализатор метрик сервера"""
    
    def __init__(self, collect_interval=1.0, history_size=3600):
        self.collect_interval = collect_interval
        self.history_size = history_size
        self.running = False
        
        # Временные ряды для различных метрик
        self.metrics = {
            'connections': collections.deque(maxlen=history_size),
            'active_threads': collections.deque(maxlen=history_size),
            'task_queue_size': collections.deque(maxlen=history_size),
            'response_time': collections.deque(maxlen=history_size),
            'errors': collections.deque(maxlen=history_size),
        }
        
        # Метка времени для каждой точки
        self.timestamps = collections.deque(maxlen=history_size)
        
        # Счетчики событий
        self.counters = {
            'total_connections': 0,
            'completed_requests': 0,
            'errors': 0,
        }
        
        self.lock = threading.Lock()
        self.logger = logging.getLogger('metrics')
    
    def start(self) -> None:
        """Запуск сбора метрик"""
        if self.running:
            return
        
        self.running = True
        self.collector_thread = threading.Thread(target=self._collect_loop)
        self.collector_thread.daemon = True
        self.collector_thread.start()
        
        self.logger.info("Сборщик метрик запущен")
    
    def stop(self) -> None:
        """Остановка сбора метрик"""
        self.running = False
        self.logger.info("Сборщик метрик остановлен")
    
    def _collect_loop(self) -> None:
        """Периодический сбор метрик"""
        while self.running:
            try:
                with self.lock:
                    # Текущее время
                    now = time.time()
                    self.timestamps.append(now)
                    
                    # Собираем текущие значения метрик
                    thread_count = threading.active_count()
                    self.metrics['active_threads'].append(thread_count)
                    
                    # Добавляем пустые значения для метрик, которые обновляются явно
                    for metric in ['connections', 'task_queue_size', 'response_time', 'errors']:
                        if not self.metrics[metric] or len(self.metrics[metric]) < len(self.timestamps):
                            self.metrics[metric].append(0)
                
                # Ждем до следующего интервала сбора
                time.sleep(self.collect_interval)
            except Exception as e:
                self.logger.error(f"Ошибка при сборе метрик: {e}")
    
    def record_event(self, event_type: str) -> None:
        """Запись события"""
        with self.lock:
            if event_type == 'connection_accepted':
                self.counters['total_connections'] += 1
                # Увеличиваем счетчик текущих соединений
                if self.metrics['connections']:
                    self.metrics['connections'][-1] += 1
            
            elif event_type == 'request_completed':
                self.counters['completed_requests'] += 1
            
            elif event_type == 'error':
                self.counters['errors'] += 1
                if self.metrics['errors']:
                    self.metrics['errors'][-1] += 1
    
    def record_gauge(self, metric: str, value: float) -> None:
        """Запись значения метрики"""
        with self.lock:
            if metric in self.metrics and self.metrics[metric]:
                self.metrics[metric][-1] = value
    
    def record_response_time(self, response_time: float) -> None:
        """Запись времени отклика"""
        with self.lock:
            if self.metrics['response_time']:
                # Обновляем как скользящее среднее
                current = self.metrics['response_time'][-1]
                if current == 0:
                    self.metrics['response_time'][-1] = response_time
                else:
                    self.metrics['response_time'][-1] = (current * 0.9 + response_time * 0.1)
    
    def visualize(self, interval=1000) -> None:
        """Визуализация метрик в реальном времени"""
        try:
            import matplotlib
            matplotlib.use('TkAgg')  # Используем TkAgg бэкенд для интерактивных графиков
            
            fig, axs = plt.subplots(2, 2, figsize=(12, 8))
            fig.suptitle('Метрики многопоточного сервера')
            
            conn_line, = axs[0, 0].plot([], [], 'b-', label='Соединения')
            axs[0, 0].set_title('Активные соединения')
            axs[0, 0].set_xlabel('Время (с)')
            axs[0, 0].set_ylabel('Количество')
            
            threads_line, = axs[0, 1].plot([], [], 'g-', label='Потоки')
            queue_line, = axs[0, 1].plot([], [], 'r-', label='Очередь')
            axs[0, 1].set_title('Потоки и очередь задач')
            axs[0, 1].set_xlabel('Время (с)')
            axs[0, 1].set_ylabel('Количество')
            axs[0, 1].legend()
            
            resp_line, = axs[1, 0].plot([], [], 'c-', label='Время отклика')
            axs[1, 0].set_title('Среднее время отклика')
            axs[1, 0].set_xlabel('Время (с)')
            axs[1, 0].set_ylabel('Время (с)')
            
            error_line, = axs[1, 1].plot([], [], 'm-', label='Ошибки')
            axs[1, 1].set_title('Количество ошибок')
            axs[1, 1].set_xlabel('Время (с)')
            axs[1, 1].set_ylabel('Количество')
            
            def update(frame):
                with self.lock:
                    if not self.timestamps:
                        return conn_line, threads_line, queue_line, resp_line, error_line
                    
                    # Преобразуем временные метки в относительное время в секундах
                    start_time = self.timestamps[0]
                    times = [(t - start_time) for t in self.timestamps]
                    
                    # Обновляем данные на графиках
                    conn_line.set_data(times, list(self.metrics['connections']))
                    threads_line.set_data(times, list(self.metrics['active_threads']))
                    queue_line.set_data(times, list(self.metrics['task_queue_size']))
                    resp_line.set_data(times, list(self.metrics['response_time']))
                    error_line.set_data(times, list(self.metrics['errors']))
                    
                    # Автоматически масштабируем оси
                    for ax in axs.flat:
                        ax.relim()
                        ax.autoscale_view()
                    
                    return conn_line, threads_line, queue_line, resp_line, error_line
            
            # Создаем анимацию
            ani = FuncAnimation(fig, update, interval=interval, blit=True)
            plt.tight_layout()
            plt.show()
            
        except ImportError:
            self.logger.error("Не удалось импортировать matplotlib для визуализации")
            print("Для визуализации метрик установите matplotlib: pip install matplotlib")

   # Пример с использованием Numba для обхода GIL
   import numba
   
   @numba.jit(nopython=True, nogil=True)
   def cpu_intensive_task(data):
       result = 0
       for i in range(len(data)):
           for j in range(1000):
               result += (data[i] * j) % 177
       return result

   # Неоптимальный подход - слишком долгое удержание блокировки
   def process_data(self, data):
       with self.lock:
           # Длительная операция с защищенными данными
           result = self.expensive_operation(data)
           return result
   
   # Оптимизированный подход - минимизация времени удержания блокировки
   def process_data_optimized(self, data):
       # Сначала получаем копию защищенных данных 
       with self.lock:
           protected_data = copy.deepcopy(self.shared_data)
       
       # Выполняем длительную обработку без блокировки
       result = self.expensive_operation(protected_data, data)
       
       # Опять захватываем блокировку только для обновления результата
       with self.lock:
           self.shared_results.append(result)
       
       return result

   # Наивная отправка данных - много мелких передач
   def send_data_naive(sock, items):
       for item in items:
           sock.send(item.encode())  # Много мелких системных вызовов
   
   # Оптимизированная отправка - буферизация
   def send_data_optimized(sock, items):
       buffer = b''.join(item.encode() for item in items)
       sock.sendall(buffer)  # Один системный вызов для всех данных

# Пример Dockerfile для многопоточного Python-сервера
FROM python:3.10-slim
 
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
# Используем оптимизированную версию Python
ENV PYTHONOPTIMIZE=2
# Отключаем буферизацию stdout/stderr
ENV PYTHONUNBUFFERED=1
# Устанавливаем максимальный размер пула потоков по умолчанию
ENV PYTHONTHREADSWITCHINTERVAL=10
 
# Открываем порт сервера
EXPOSE 9999
 
# Запуск сервера с ограничением по памяти и CPU
# docker run --memory=2g --cpus=2 my-server-image
CMD ["python", "server.py"]

   from prometheus_client import start_http_server, Counter, Gauge
   
   # Инициализация метрик
   REQUEST_COUNT = Counter('python_server_requests_total', 'Total request count')
   ACTIVE_THREADS = Gauge('python_server_active_threads', 'Number of active threads')
   QUEUE_SIZE = Gauge('python_server_queue_size', 'Current task queue size')
   
   # Экспорт метрик на отдельном порту
   start_http_server(8000)
   
   # Обновление метрик в коде
   def handle_client(client_socket):
       REQUEST_COUNT.inc()
       ACTIVE_THREADS.inc()
       try:
           # Обработка клиента...
       finally:
           ACTIVE_THREADS.dec()

   import opentracing
   from jaeger_client import Config
   
   # Инициализация трассировщика
   def init_tracer():
       config = Config(
           config={
               'sampler': {'type': 'const', 'param': 1},
               'logging': True,
           },
           service_name='python-server',
       )
       return config.initialize_tracer()
   
   tracer = init_tracer()
   
   # Использование в обработчике клиента
   def handle_client(client_socket):
       with tracer.start_active_span('handle_client') as scope:
           client_address = client_socket.getpeername()
           scope.span.set_tag('client_address', str(client_address))
           
           # Обработка запроса...
           data = client_socket.recv(1024)
           
           with tracer.start_active_span('process_data') as child_scope:
               # Более глубокая вложенная операция...
               result = process_data(data)
           
           client_socket.send(result)

   import signal
   import sys
   import threading
   
   def graceful_shutdown(sig, frame):
       print(f"Получен сигнал {sig}, начинаем корректное завершение...")
       
       # Устанавливаем флаг остановки для всех рабочих потоков
       global running
       running = False
       
       # Ждем завершения всех потоков
       print("Ожидание завершения обработки клиентов...")
       for thread in threading.enumerate():
           if thread != threading.current_thread():
               thread.join(timeout=5.0)
       
       # Закрываем все ресурсы
       if server_socket:
           server_socket.close()
       
       print("Сервер остановлен корректно")
       sys.exit(0)
   
   # Регистрируем обработчики сигналов
   signal.signal(signal.SIGTERM, graceful_shutdown)
   signal.signal(signal.SIGINT, graceful_shutdown)

   import pickle
   import os
   import time
   import threading
   
   class StatefulServer:
       def __init__(self, state_file="server_state.pkl"):
           self.state_file = state_file
           self.state = self.load_state() or {}
           self.state_lock = threading.Lock()
           
           # Запускаем периодическое сохранение состояния
           self.save_thread = threading.Thread(target=self.periodic_save)
           self.save_thread.daemon = True
           self.save_thread.start()
       
       def load_state(self):
           if os.path.exists(self.state_file):
               try:
                   with open(self.state_file, 'rb') as f:
                       return pickle.load(f)
               except Exception as e:
                   print(f"Ошибка при загрузке состояния: {e}")
           return None
       
       def save_state(self):
           with self.state_lock:
               with open(self.state_file, 'wb') as f:
                   pickle.dump(self.state, f)
       
       def periodic_save(self):
           while True:
               time.sleep(60)  # Сохраняем состояние каждую минуту
               try:
                   self.save_state()
               except Exception as e:
                   print(f"Ошибка при сохранении состояния: {e}")