В этом коде:
# 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()
        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.")
        # Simulate processing
        time.sleep(request.processing_time)
        with self._lock:
            self._processed_requests.add(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

    def recover(self) -> None:
        """
        Recovers the server.
        """
        with self._lock:
            self._alive = True

    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()  # Use 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

        # Semaphore per server to limit concurrency (1 request per server)
        self.server_semaphores: Dict[str, threading.Semaphore] = {
            server.server_id: threading.Semaphore(1) for server in self.servers if server.is_alive()
        }

        # Initialize Smooth Weighted Round Robin variables
        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())

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

    def _rebuild_server_weights(self):
        """
        Rebuilds the current_weights and total_weight based on alive servers.
        Reinitializes semaphores for new servers and removes semaphores for dead servers.
        Notifies all waiting threads if servers are available.
        """
        with self.lock:
            # Reset current_weights
            self.current_weights = {
                server.server_id: 0 for server in self.servers if server.is_alive()
            }
            # Recalculate total_weight
            self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())

            # Rebuild server_semaphores
            self.server_semaphores = {
                server.server_id: threading.Semaphore(1) for server in self.servers if server.is_alive()
            }

            # Notify all waiting threads that servers are available
            if self.total_weight > 0:
                with self.server_available:
                    self.server_available.notify_all()

    def _get_next_server_id(self) -> Optional[str]:
        """
        Gets the next server_id based on Smooth Weighted Round Robin.
        """
        if not self.current_weights:
            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

        # Select server with highest current_weight
        selected_server_id = max(self.current_weights, key=lambda s_id: self.current_weights[s_id])

        # Decrement selected_server's current_weight by total_weight
        self.current_weights[selected_server_id] -= self.total_weight

        return selected_server_id

    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 route(self, request: Request) -> None:
        """
        Routes a request to an appropriate server using Smooth Weighted Round Robin.
        Waits if no servers are available.
        """
        self.load_semaphore.acquire()
        try:
            while True:
                with self.lock:
                    # 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 acquired
                                break
                            else:
                                server = None
                        else:
                            # Remove broken affinity
                            del self.client_affinity[request.client_id]
                            server = None

                    if request.client_id not in self.client_affinity:
                        # Select next server
                        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
                                break
                            else:
                                selected_server_id = None

                if selected_server_id is None:
                    # Wait until a server becomes available
                    with self.server_available:
                        self.server_available.wait()

            # At this point, semaphore for selected_server_id is acquired
            server = self._get_server_by_id(self.client_affinity[request.client_id])
            if server and server.is_alive():
                try:
                    server.process_request(request)
                except Exception as e:
                    # Handle exception (e.g., server crashed during processing)
                    with self.server_available:
                        self.server_available.notify_all()
                    raise e
                finally:
                    # Release server semaphore after processing
                    self.server_semaphores[server.server_id].release()
                    # 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
                self.server_semaphores[self.client_affinity[request.client_id]].release()
                del self.client_affinity[request.client_id]
                with self.lock:
                    self.server_available.notify_all()
        finally:
            # Release the load semaphore
            self.load_semaphore.release()

    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)
                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()

    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]
                # 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]
                # Remove from server_semaphores
                if server.server_id in self.server_semaphores:
                    del self.server_semaphores[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]
                # 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()
            # 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]
            # 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()
            # 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)

Который должен решать задачу:

Представьте, что вы разрабатываете систему маршрутизации запросов для интернет-магазина (на самом деле это может быть что угодно). У вас есть 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. Предполагается что все блокировки будут происходить как операция ввода-вывода, поэтому предполагается что маршрутизатор будет использоваться в многопоточном приложении.

Падает всё на вот этом тесте:

(test_router_server_failure)
router = <routing.Router object at 0x105ca6a80>

    def test_router_server_failure(router):
        router.servers[0].crash()
    
        requests = [Request(f"client{i}", f"request{i}", 0.1) for i in range(6)]
        for request in requests:
            router.route(request)
    
        assert not any(router.servers[0].is_processed(f"request{i}") for i in range(6))
>       assert all(any(server.is_processed(f"request{i}") for server in router.servers[1:]) for i in range(6))
E       assert False
E        +  where False = all(<generator object test_router_server_failure.<locals>.<genexpr> at 0x105b66e30>)

test_public.py:107: AssertionError

Причем в этом коде, который решает эту же задачу:

import threading  
import time  
from typing import List, Dict, Optional  
from concurrent.futures import ThreadPoolExecutor, as_completed  
  
  
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()  
        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.")  
        # Simulate processing  
        time.sleep(request.processing_time)  
        with self._lock:  
            self._processed_requests.add(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  
  
    def recover(self) -> None:  
        """  
        Recovers the server.        """        with self._lock:  
            self._alive = True  
  
    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)  # Open attribute for testing  
        self.max_load = max_load  
        self.lock = threading.RLock()  # Use reentrant lock  
        self.load_semaphore = threading.Semaphore(max_load)  
        self.client_affinity: Dict[str, Server] = dict()  
  
        # Condition variable to wait for available servers  
        self.server_available = threading.Condition(self.lock)  
  
        # Initialize Smooth Weighted Round Robin variables  
        self.current_weights: Dict[Server, int] = {  
            server: 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())  
  
    def _rebuild_server_weights(self):  
        """  
        Rebuilds the current_weights and total_weight based on alive servers.        Notifies all waiting threads if servers are available.        """        with self.lock:  
            # Reset current_weights  
            self.current_weights = {  
                server: 0 for server in self.servers if server.is_alive()  
            }  
            # Recalculate total_weight  
            self.total_weight = sum(server.performance_score for server in self.servers if server.is_alive())  
            # Notify all waiting threads that servers are available  
            if self.total_weight > 0:  
                with self.server_available:  
                    self.server_available.notify_all()  
  
    def _get_next_server(self) -> Optional[Server]:  
        """  
        Gets the next server based on Smooth Weighted Round Robin.        """        with self.lock:  
            alive_servers = [s for s in self.servers if s.is_alive()]  
            if not alive_servers:  
                return None  
  
            # Increment current_weight by server's weight  
            for server in alive_servers:  
                self.current_weights[server] += server.performance_score  
  
            # Select server with highest current_weight  
            selected_server = max(alive_servers, key=lambda s: self.current_weights[s])  
  
            # Decrement selected_server's current_weight by total_weight  
            self.current_weights[selected_server] -= self.total_weight  
  
            return selected_server  
  
    def route(self, request: Request) -> None:  
        """  
        Routes a request to an appropriate server using Smooth Weighted Round Robin.        Waits if no servers are available.        """        self.load_semaphore.acquire()  
        try:  
            while True:  
                with self.lock:  
                    # Check client affinity  
                    if request.client_id in self.client_affinity:  
                        server = self.client_affinity[request.client_id]  
                        if not server.is_alive():  
                            # Remove affinity if server is down  
                            del self.client_affinity[request.client_id]  
                            server = self._get_next_server()  
                            if server:  
                                self.client_affinity[request.client_id] = server  
                    else:  
                        server = self._get_next_server()  
                        if server:  
                            self.client_affinity[request.client_id] = server  
  
                    if server and server.is_alive():  
                        break  # Found an available server  
                    else:  
                        # Wait until a server becomes available  
                        with self.server_available:  
                            self.server_available.wait()  
  
            if server and server.is_alive():  
                server.process_request(request)  
        finally:  
            self.load_semaphore.release()  
  
    def add_server(self, server: Server) -> None:  
        """  
        Adds a new server to the router.        """        with self.lock:  
            if server not in self.servers:  
                self.servers.append(server)  
                if server.is_alive():  
                    self.current_weights[server] = 0  
                    self.total_weight += server.performance_score  
                    self.server_available.notify_all()  
  
    def remove_server(self, server: Server) -> None:  
        """  
        Removes a server from the router.        """        with self.lock:  
            if server in self.servers:  
                self.servers.remove(server)  
                # Remove from current_weights and adjust total_weight  
                if server in self.current_weights:  
                    self.total_weight -= server.performance_score  
                    del self.current_weights[server]  
                # Remove client affinities tied to this server  
                clients_to_remove = [  
                    client for client, srv in self.client_affinity.items() if srv == server  
                ]  
                for client in clients_to_remove:  
                    del self.client_affinity[client]  
                self.server_available.notify_all()  
  
    def crash_server(self, server_id: str) -> None:  
        """  
        Crashes a server by its ID.        """        with self.lock:  
            for server in self.servers:  
                if server.server_id == server_id:  
                    server.crash()  
                    break  
            # Remove client affinities tied to this server  
            clients_to_remove = [  
                client for client, srv in self.client_affinity.items() if srv.server_id == server_id  
            ]  
            for client in clients_to_remove:  
                del self.client_affinity[client]  
            # Rebuild server weights and notify  
            self._rebuild_server_weights()  
  
    def recover_server(self, server_id: str) -> None:  
        """  
        Recovers a server by its ID.        """        with self.lock:  
            for server in self.servers:  
                if server.server_id == server_id:  
                    server.recover()  
                    break  
            # Rebuild server weights and notify  
            self._rebuild_server_weights()  
  
    def get_servers(self) -> List[Server]:  
        """  
        Returns the list of current servers.        """        with self.lock:  
            return list(self.servers)
            
Ничего не падает на этом тесте, сравни два кода и сделай так, чтобы первый не падал на тесте