Этот код:

# routing.py

import threading
import time
from typing import List, Dict, Optional


class Request:
    def __init__(self, client_id: str, request_id: str, processing_time: float):
        self.client_id = client_id
        self.request_id = request_id
        self.processing_time = processing_time


class Server:
    """
    Represents a server that can process requests.
    """
    def __init__(self, server_id: str, performance_score: int):
        self.server_id = server_id
        self.performance_score = performance_score
        self._alive = True
        self._processed_requests: set[str] = set()
        self._lock = threading.Lock()

    def process_request(self, request: Request) -> None:
        """
        Processes a request. Save the request client_id and request_id.
        Simulates processing time.
        """
        if not self.is_alive():
            raise Exception(f"Server {self.server_id} is not alive.")
        print(f"[Server {self.server_id}] Processing request {request.request_id} from client {request.client_id}")
        # Simulate processing
        time.sleep(request.processing_time)
        with self._lock:
            self._processed_requests.add(request.request_id)
        print(f"[Server {self.server_id}] Finished processing request {request.request_id}")

    def is_alive(self) -> bool:
        """
        Checks if the server is alive.
        """
        with self._lock:
            return self._alive

    def crash(self) -> None:
        """
        Crashes the server.
        """
        with self._lock:
            self._alive = False
        print(f"[Server {self.server_id}] Crashed.")

    def recover(self) -> None:
        """
        Recovers the server.
        """
        with self._lock:
            self._alive = True
        print(f"[Server {self.server_id}] Recovered.")

    def is_processed(self, request_id: str) -> bool:
        """
        Checks if the request is processed. Emulates caching.
        """
        with self._lock:
            return request_id in self._processed_requests


class Router:
    """
    Represents a router that can route requests to servers using Smooth Weighted Round Robin.
    """
    def __init__(self, servers: List[Server], max_load: int):
        self.servers: List[Server] = list(servers)  # Public attribute for testing
        self.max_load = max_load
        self.lock = threading.RLock()  # Reentrant lock for nested locking
        self.load_semaphore = threading.Semaphore(max_load)
        self.client_affinity: Dict[str, str] = dict()  # Maps client_id to server_id

        # Initialize server_semaphores and current_weights based on alive servers
        self._initialize_servers()

        # Condition variable to wait for server availability
        self.server_available = threading.Condition(self.lock)

    def _initialize_servers(self):
        """
        Initializes server_semaphores, current_weights, and total_weight based on alive servers.
        """
        with self.lock:
            self.server_semaphores: Dict[str, threading.Semaphore] = {
                server.server_id: threading.Semaphore(1) for server in self.servers if server.is_alive()
            }

            self.current_weights: Dict[str, int] = {
                server.server_id: 0 for server in self.servers if server.is_alive()
            }

            self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())
            print(f"[Router] Initialized with servers: {[s.server_id for s in self.servers if s.is_alive()]}")
            print(f"[Router] Total weight: {self.total_weight}")

    def _rebuild_server_weights(self):
        """
        Rebuilds the current_weights and total_weight based on alive servers.
        Updates server_semaphores by adding/removing semaphores for servers that have changed state.
        """
        with self.lock:
            alive_servers = {server.server_id for server in self.servers if server.is_alive()}
            current_servers = set(self.server_semaphores.keys())

            dead_servers = current_servers - alive_servers
            for server_id in dead_servers:
                del self.server_semaphores[server_id]
                print(f"[Router] Removed semaphore for dead server {server_id}")
                if server_id in self.current_weights:
                    del self.current_weights[server_id]
                    print(f"[Router] Removed weight for dead server {server_id}")

            newly_alive_servers = alive_servers - current_servers
            for server_id in newly_alive_servers:
                self.server_semaphores[server_id] = threading.Semaphore(1)
                self.current_weights[server_id] = 0
                print(f"[Router] Added semaphore and weight for newly alive server {server_id}")

            self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())
            print(f"[Router] Rebuilt total weight: {self.total_weight}")

            if self.total_weight > 0:
                self.server_available.notify_all()
                print(f"[Router] Notified all waiting threads about available servers.")

    def _get_server_by_id(self, server_id: str) -> Optional[Server]:
        """
        Retrieves a server instance by its server_id.
        """
        for server in self.servers:
            if server.server_id == server_id:
                return server
        return None

    def _get_next_server_id(self) -> Optional[str]:
        if not self.current_weights:
            print("[Router] No available servers to route the request.")
            return None

        # Increment current_weight by server's weight
        for server_id in self.current_weights:
            server = self._get_server_by_id(server_id)
            if server and server.is_alive():
                self.current_weights[server_id] += server.performance_score
                print(f"[Router] Incremented weight of server {server_id} to {self.current_weights[server_id]}")

        available_servers = {
            s_id: weight
            for s_id, weight in self.current_weights.items()
            if (server := self._get_server_by_id(s_id)) is not None and server.is_alive()
        }

        if not available_servers:
            print("[Router] No available servers to route the request.")
            return None

        selected_server_id = max(available_servers, key=lambda s_id: available_servers[s_id])
        print(f"[Router] Selected server {selected_server_id} with weight {self.current_weights[selected_server_id]}")

        # Decrement selected_server's current_weight by total_weight
        new_weight = self.current_weights[selected_server_id] - self.total_weight
        self.current_weights[selected_server_id] = max(0, new_weight)  # Prevent negative weights
        print(
            f"[Router] Decremented weight of server {selected_server_id} to {self.current_weights[selected_server_id]}")

        return selected_server_id

    def route(self, request: Request) -> None:
        """
        Routes a request to an appropriate server using Smooth Weighted Round Robin.
        Waits if no servers are available.
        """
        print(f"[Router] Routing request {request.request_id} from client {request.client_id}")
        self.load_semaphore.acquire()
        try:
            while True:
                with self.lock:
                    server_id_to_use = None

                    # Check client affinity
                    if request.client_id in self.client_affinity:
                        server_id = self.client_affinity[request.client_id]
                        server = self._get_server_by_id(server_id)

                        if server and server.is_alive():
                            semaphore = self.server_semaphores.get(server_id)
                            if semaphore and semaphore.acquire(blocking=False):
                                # Server semaphore acquired
                                server_id_to_use = server_id
                                print(f"[Router] Using affinity: client {request.client_id} -> server {server_id}")
                                break  # Exit loop to process request
                        else:
                            # Server is down, remove affinity
                            del self.client_affinity[request.client_id]
                            print(
                                f"[Router] Affinity removed for client {request.client_id} as server {server_id} is down.")

                    if not server_id_to_use:
                        # Select next server via Smooth Weighted Round Robin
                        selected_server_id = self._get_next_server_id()
                        if selected_server_id:
                            semaphore = self.server_semaphores.get(selected_server_id)
                            if semaphore and semaphore.acquire(blocking=False):
                                # Assign client affinity
                                self.client_affinity[request.client_id] = selected_server_id
                                server_id_to_use = selected_server_id
                                print(f"[Router] Assigned client {request.client_id} to server {selected_server_id}")
                                break  # Exit loop to process request
                            else:
                                selected_server_id = None
                                print(f"[Router] Server {selected_server_id} is busy.")

                if server_id_to_use is None:
                    # Wait until a server becomes available
                    print(f"[Router] Waiting for an available server to process request {request.request_id}")
                    with self.server_available:
                        self.server_available.wait()

            # At this point, semaphore for selected_server_id is acquired
            server = self._get_server_by_id(server_id_to_use)
            assert server is not None, f"Server {server_id_to_use} not found."
            if server and server.is_alive():
                try:
                    server.process_request(request)
                except Exception as e:
                    with self.lock:
                        # Remove client affinity on exception
                        if request.client_id in self.client_affinity:
                            del self.client_affinity[request.client_id]
                            # Rebuild server weights and semaphores
                            self._rebuild_server_weights()
                            print(f"[Router] Exception occurred: {e}. Rebuilt server weights.")
                    with self.server_available:
                        self.server_available.notify_all()
                    raise
                finally:
                    # Release server semaphore after processing
                    self.server_semaphores[server_id_to_use].release()
                    print(f"[Router] Released semaphore for server {server_id_to_use}")
                    # Notify other threads that a server slot is available
                    with self.server_available:
                        self.server_available.notify()
            else:
                # Server is not alive, release semaphore and retry
                if server_id_to_use in self.server_semaphores:
                    self.server_semaphores[server_id_to_use].release()
                    print(f"[Router] Server {server_id_to_use} is not alive. Released semaphore.")
                with self.lock:
                    if request.client_id in self.client_affinity:
                        del self.client_affinity[request.client_id]
                        print(f"[Router] Removed client affinity for {request.client_id}")
                with self.server_available:
                    self.server_available.notify_all()
        finally:
            # Release the load semaphore
            self.load_semaphore.release()
            print(f"[Router] Released load semaphore for request {request.request_id}")

    def add_server(self, server: Server) -> None:
        """
        Adds a new server to the router.
        """
        with self.lock:
            if server.server_id not in [s.server_id for s in self.servers]:
                self.servers.append(server)
                print(f"[Router] Added server {server.server_id}")
                if server.is_alive():
                    self.current_weights[server.server_id] = 0
                    self.total_weight += server.performance_score
                    self.server_semaphores[server.server_id] = threading.Semaphore(1)
                    # Notify all waiting threads that a new server is available
                    with self.server_available:
                        self.server_available.notify_all()
                    print(f"[Router] Server {server.server_id} is alive and added to current weights.")

    def remove_server(self, server: Server) -> None:
        """
        Removes a server from the router.
        """
        with self.lock:
            if server.server_id in [s.server_id for s in self.servers]:
                self.servers = [s for s in self.servers if s.server_id != server.server_id]
                print(f"[Router] Removed server {server.server_id} from server list.")
                # Remove from current_weights and adjust total_weight
                if server.server_id in self.current_weights:
                    self.total_weight -= server.performance_score
                    del self.current_weights[server.server_id]
                    print(f"[Router] Removed server {server.server_id} from current weights.")
                # Remove from server_semaphores
                if server.server_id in self.server_semaphores:
                    del self.server_semaphores[server.server_id]
                    print(f"[Router] Removed semaphore for server {server.server_id}.")
                # Remove client affinities tied to this server
                clients_to_remove = [
                    client for client, srv_id in self.client_affinity.items() if srv_id == server.server_id
                ]
                for client in clients_to_remove:
                    del self.client_affinity[client]
                    print(f"[Router] Removed client affinity for client {client} due to server removal.")
                # Notify all waiting threads that a server slot is available
                with self.server_available:
                    self.server_available.notify_all()

    def crash_server(self, server_id: str) -> None:
        """
        Crashes a server by its ID.
        """
        with self.lock:
            server = self._get_server_by_id(server_id)
            if server:
                server.crash()
                print(f"[Router] Crashed server {server_id}")
            # Remove client affinities tied to this server
            clients_to_remove = [
                client for client, srv_id in self.client_affinity.items() if srv_id == server_id
            ]
            for client in clients_to_remove:
                del self.client_affinity[client]
                print(f"[Router] Removed client affinity for client {client} due to server {server_id} crash.")
            # Rebuild server weights and semaphores
            self._rebuild_server_weights()

    def recover_server(self, server_id: str) -> None:
        """
        Recovers a server by its ID.
        """
        with self.lock:
            server = self._get_server_by_id(server_id)
            if server:
                server.recover()
                print(f"[Router] Recovered server {server_id}")
            # Rebuild server weights and semaphores
            self._rebuild_server_weights()

    def get_servers(self) -> List[Server]:
        """
        Returns the list of current servers.
        """
        with self.lock:
            return list(self.servers)

Решающий эту задачу:

Проблема маршрутизации запросов

alt text Представьте, что вы разрабатываете систему маршрутизации запросов для интернет-магазина (на самом деле это может быть что угодно). У вас есть N серверов, которые могут обрабатывать запросы. Каждый сервер имеет разные параметры производительности.

Вам нужно реализовать систему балансировки нагрузки, которая будет распределять запросы между серверами таким образом, чтобы:

Максимально равномерно распределять нагрузку между серверами при параллельных запросах.
Минимизировать количество переключений между серверами (то есть, если сервер обрабатывает запрос, он должен продолжать обрабатывать запрос, пока не закончит или не произойдёт сбой). Если запросы приходят с одного клиента, они всегда должны идти на один и тот же сервер, это будет позволять приложению кешировать данные на стороне сервера.
Обрабатывать сбои серверов, то есть, если сервер недоступен, он должен быть исключен из распределения нагрузки, а когда он станет доступен, он должен быть добавлен обратно.
Максимальная нагрузка на сервер или router не должна превышать max_load, в противном случае сервис должен ждать освобождения ресурсов, не выкидывая ошибок (воспользуйтесь semaphore).
Подсказка: для решения этой задачи можно использовать алгоритм Round Robin, а именно его продвинутый вариант Weighted Round Robin.

Описание задачи

Здесь представлены классы, которые вам нужно будет реализовать:

class Request:
    def __init__(self, client_id: str, request_id: str, processing_time: float):
        self.client_id = client_id
        self.request_id = request_id
        self.processing_time = processing_time
class Server:
    """
    Represents a server that can process requests.
    """
    def __init__(self, server_id: str, performance_score: int):
        self.server_id = server_id
        self.performance_score = performance_score

    def process_request(self, request: Request) -> None:
        """
        Processes a request. Save the request client_id and request_id.
        """
        pass

    def is_alive(self) -> bool:
        """
        Checks if the server is alive.
        """
        pass
    
    def crash(self) -> None:
        """
        Crashes the server.
        """
        pass

    def recover(self) -> None:
        """
        Recovers the server.
        """
        pass


    def is_processed(self, request_id: str) -> bool:
        """
        Checks if the request is processed. Emulates caching.
        """
        pass
class Router:
    """
    Represents a router that can route requests to servers.
    """
    def __init__(self, servers: List[Server], max_load: int):
        pass

    def route(self, request: Request) -> None:
        pass

    def add_server(self, server: Server) -> None:
        pass

    def remove_server(self, server: Server) -> None:
        pass
Немного полезной информации

Round Robin подразумевает хранение списока серверов и выбирает из него по очереди, при достижении конца списка, возвращайтеся в начало (% len(servers))
Чтобы реализовать Weighted Round Robin используйте список кортежей (server, weight), где weight - это вес сервера. Выбирая сервер, выбирайте его с вероятностью пропорциональной весу.
Помните, что роутер должен обрабатывать не более чем max_load запросов.
Для реализации сбоев и восстановлений можно использовать методы crash() и recover() класса Server. При реализации сбоев и восстановлений помните про перерассчет весов для роутинга
Добавление и удаление должно быть реализовано так, чтобы минимальное число клиентов "сменило" сервер при следующем запросе, однако при этом сервера должны быть нагружны равномерно
Для синхронизации доступа к разделяемым ресурсам используйте средства синхронизации из библиотеки threading. Предполагается что все блокировки будут происходить как операция ввода-вывода, поэтому предполагается что маршрутизатор будет использоваться в многопоточном приложении.

Падает на единственном тесте:

[gw1] linux -- Python 3.12.3 /home/muzaffarsad/myenv/bin/python

create_router = <function create_router.<locals>._create_router at 0x72c06fe80ea0>
num_servers = 3, num_clients = 5, requests_per_client = 10

    @pytest.mark.parametrize("num_servers,num_clients,requests_per_client", [(3, 5, 10), (5, 10, 5), (7, 3, 15)])
    def test_scaled_client_affinity(create_router, num_servers, num_clients, requests_per_client):
        router = create_router(num_servers, max_load=20)
        client_requests = {
            f"client{i}": [Request(f"client{i}", f"request{i}_{j}", 0.1) for j in range(requests_per_client)]
            for i in range(num_clients)
        }
    
        def route_client_requests(client, requests):
            for request in requests:
                router.route(request)
            return client
    
        with ThreadPoolExecutor(max_workers=num_clients) as executor:
            futures = [executor.submit(route_client_requests, client, requests) for client, requests in client_requests.items()]
            results = [future.result() for future in as_completed(futures)]
    
        assert len(results) == num_clients
    
        for client, requests in client_requests.items():
            processed_servers = [server for server in router.servers if any(server.is_processed(request.request_id) for request in requests)]
>           assert len(processed_servers) == 1  # All requests from one client should be processed by the same server
E           assert 2 == 1
E            +  where 2 = len([<routing.Server object at 0x72c074bb8ce0>, <routing.Server object at 0x72c06fe499a0>])

routing/test_private.py:138: AssertionError