ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Настроить conflict resolution между агентами

1. Цель задачи

Реализовать механизм разрешения конфликтов при одновременном доступе нескольких агентов к общим ресурсам. Система должна поддерживать приоритеты агентов, динамическое управление блокировками (locks) и детекцию взаимоблокировок (deadlock). Ключевой результат отсутствие race conditions при конкурентном выполнении агентов – все состояния системы остаются консистентными, deadlock’и обнаруживаются и разрешаются автоматически или с явным уведомлением.

2. Исходные данные

Если нет реального инструмента — симулируем:

1. Создаём директорию src/conflict_resolution/
2. Пишем класс ResourceManager с внутренним словарём блокировок (asyncio.Lock)
3. Эмулируем агентов через asyncio.Task с разными задержками и приоритетами
4. Для deadlock detection реализуем таймер ожидания и граф зависимостей

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование и структура данных (30 мин)

Действия

1. Создайте файловую структуру:

   src/conflict_resolution/
   ├── __init__.py
   ├── resource_manager.py
   ├── agent.py
   ├── deadlock_detector.py
   └── priority_scheduler.py
   

2. Определите перечисление ResourceState: FREE, LOCKED, WAITING
3. Опишите класс Resource с атрибутами: id, name, priority_ceiling, owner_agent_id
4. Спроектируйте класс ResourceManager с методом acquire(agent_id, resource_id, timeout) и release(agent_id, resource_id)
5. Разработайте интерфейс приоритетов: каждый агент имеет базовый приоритет (1 — низкий, 10 — высокий), а ResourceManager может повышать приоритет (priority inheritance) для предотвращения инверсии

Ожидаемый результат этапа

• Псевдокод или диаграмма классов в формате Mermaid (сохранить как docs/class_diagram.md)

Этап 2: Реализация базовых блокировок и приоритетов (1.5 ч)

Действия

1. Реализуйте ResourceManager с использованием asyncio.Lock для каждого ресурса:

   class ResourceManager:
       def __init__(self):
           self._locks = {}
           self._owners = {}
           self._waiting = defaultdict(list)
   
       async def acquire(self, agent_id, resource_id, timeout=5.0):
           lock = self._locks.setdefault(resource_id, asyncio.Lock())
           try:
               await asyncio.wait_for(lock.acquire(), timeout)
               self._owners[resource_id] = agent_id
               return True
           except asyncio.TimeoutError:
               return False
   
       async def release(self, agent_id, resource_id):
           if self._owners.get(resource_id) == agent_id:
               self._locks[resource_id].release()
               del self._owners[resource_id]
   

2. Добавьте в Agent метод run() с последовательностью захвата ресурсов (используйте разные порядки для провоцирования deadlock)
3. Реализуйте приоритетное планирование: при ожидании блокировки, если пришедший агент имеет более высокий приоритет, текущий владелец временно повышается до его приоритета (priority inheritance)
4. Напишите 3 сценария: случай без конфликтов, с конфликтом приоритетов, с потенциальным deadlock

Ожидаемый результат этапа

• Рабочие файлы resource_manager.py, agent.py, priority_scheduler.py

Этап 3: Deadlock detection и разрешение (2 ч)

Действия

1. Создайте класс DeadlockDetector, который хранит граф ожидания (agent -> resource -> agent)
2. Добавьте метод detect_cycle() (обход в глубину с раскраской: WHITE, GRAY, BLACK)
3. Интегрируйте детектор в ResourceManager: после каждого неуспешного acquire запускать проверку цикла
4. Реализуйте стратегию разрешения:
   • Preemption: отобрать ресурс у агента с меньшим приоритетом (уведомить его)
   • Или откат транзакции агента (abort)
5. Для тестов добавьте тайм-аут ожидания (например, 5 секунд) и логируйте события с метками времени

Проверка deadlock detection

# Пример: создаём циклическую зависимость
agent1 = Agent(1, priority=5)
agent2 = Agent(2, priority=3)
await agent1.acquire(resource_A)
await agent2.acquire(resource_B)
# Теперь agent1 ждёт resource_B, agent2 ждёт resource_A
await agent1.acquire(resource_B)  # повиснет
# DeadlockDetector через 2 секунды обнаружит цикл

Ожидаемый результат этапа

• Класс DeadlockDetector и стратегия разрешения (preemption)
• Логи с временными метками, показывающие детекцию

Этап 4: Тестирование и устранение race conditions (1.5 ч)

Действия

1. Напишите pytest-тесты:

   @pytest.mark.asyncio
   async def test_no_race_condition():
       manager = ResourceManager()
       results = []
       async def agent_task(agent_id, resources):
           for res in resources:
               await manager.acquire(agent_id, res)
               results.append((agent_id, res, time.time()))
               await asyncio.sleep(0.01)
               await manager.release(agent_id, res)
       tasks = [agent_task(i, res_list) for i, res_list in enumerate(...)]
       await asyncio.gather(*tasks)
       # Проверяем, что не было одновременного владения одним ресурсом
   

2. Используйте logging с метками времени и threading.Barrier (или asyncio.Barrier) для синхронизации запуска
3. Запустите 100 симуляций с разными последовательностями захвата – проверьте, что ни один ресурс не оказался в состоянии гонки
4. Включите детектор race condition: например, храните set текущих владельцев для каждого ресурса и проверяйте, что он не пуст для двух разных агентов одновременно

Ожидаемый результат этапа

• Пакет тестов test_conflict_resolution.py
• Скрипт benchmark.py, который прогоняет 1000 циклов и выводит статистику deadlock’ов и race conditions (должен быть 0)

Этап 5: Документирование и финальная проверка (1 ч)

Действия

1. Напишите README.md с описанием архитектуры, инструкцией по запуску и примерами
2. Добавьте docstring ко всем публичным методам
3. Проведите код-ревью: проверьте все блокировки на корректность освобождения (finally гарантии)
4. Убедитесь, что detect_cycle() не блокирует систему – вызов асинхронный с таймаутом
5. Зафиксируйте версию кода в репозитории (git commit с тегом v1.0)

Ожидаемый результат этапа

• Файлы README.md, docs/, финальный commit

5. Критерии приемки (Definition of Done)

• Все 5 этапов выполнены, код находится в репозитории
• При запуске pytest test_conflict_resolution.py все тесты проходят
• В логах при запуске 10 агентов с 4 ресурсами за 30 секунд не зафиксировано ни одной race condition
• Deadlock обнаруживается и разрешается не позднее чем через 2 секунды после возникновения
• Агенты с высоким приоритетом не ждут более 1 секунды при наличии блокировки у низкоприоритетного агента (priority ceiling)
• Граф ожидания доступен для экспорта в DOT-формат (функция export_wait_graph())
• В README.md описаны стратегии разрешения конфликтов и пример вызова

6. Ожидаемый результат

Основной артефакт Папка src/conflict_resolution/ с модулями, тестами и документацией.
Содержание

• Рабочая система разрешения конфликтов с приоритетами, блокировками и deadlock detection
• Набор тестов, подтверждающих отсутствие race conditions
• Скрипт для демонстрации (например, demo.py)
• Граф ожидания (при наличии graphviz) или текстовое описание

Дополнительные результаты

• Сравнительный анализ: производительность системы с приоритетами и без
• Логи симуляций в каталоге logs/

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание: При первом выполнении возможно увеличение времени до 8–9 часов из-за отладки асинхронных гонок.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я проверил, что все блокировки освобождаются даже при исключениях (использован try/finally или async with)
• Я запустил тесты не менее 3 раз и не получил ни одного false positive на race condition
• Я импортировал модуль deadlock_detector и протестировал его на искусственном deadlock
• Я добавил документацию к каждому публичному методу в стиле Sphinx
• Я убедился, что приоритеты работают корректно: высокоприоритетный агент не ждёт более 1 секунды при конкуренции с низкоприоритетным