function scheduler_loop(queue, time_slice) {
    while (true) {
        process = queue.dequeue()
        reductions_executed = 0
        
        while (process.has_messages() && reductions_executed < time_slice) {
            message = process.next_message()
            result = process.execute(message)
            reductions_executed += calculate_reductions(result)
            
            if (process.crashed()) {
                notify_supervisor(process)
                break
            }
        }
        
        if (process.has_messages()) {
            queue.enqueue(process) // Вернуть в конец очереди
        }
    }
}

Frame Structure:
+----------+----------+----------+----------+----------+----------+
|   Type   | Channel  |   Size   |   Payload         | Frame End |
| (1 byte) | (2 bytes)| (4 bytes)| (size bytes)      | (1 byte)  |
+----------+----------+----------+----------+----------+----------+

Frame Types:
- METHOD (1)    : Вызов метода AMQP (declare, publish, consume)
- HEADER (2)    : Заголовки сообщения (properties, headers)
- BODY (3)      : Часть тела сообщения (может быть несколько фреймов)
- HEARTBEAT (8) : Keep-alive фрейм

[METHOD] Basic.Publish(exchange="amq.direct", routing_key="queue1")
[HEADER] properties={content_type: "text/plain"}, body_size=1024
[BODY]   chunk 1 of 1024 bytes
[BODY]   chunk 2 of 1024 bytes (если сообщение больше frame_max)

// Псевдокод обработки publish на стороне брокера
handle_publish(frame) {
    // 1. Парсинг и валидация фреймов
    method_frame = parse_method_frame(frame)
    header_frame = read_next_frame() // Ожидаем HEADER фрейм
    
    // 2. Поиск exchange по имени
    exchange = lookup_exchange(method_frame.exchange)
    if (!exchange) {
        if (method_frame.mandatory) {
            send_basic_return() // Сообщение возвращается отправителю
        }
        return
    }
    
    // 3. Маршрутизация через exchange
    routes = exchange.route(method_frame.routing_key, header_frame.properties)
    
    // 4. Для каждого получателя (очереди)
    for (queue in routes.queues) {
        // 5. Проверка TTL сообщения
        if (header_frame.properties.expiration && is_expired(header_frame)) {
            continue
        }
        
        // 6. Сохранение в очередь
        message = {
            id: generate_message_id(),
            properties: header_frame.properties,
            body: read_body_frames() // Чтение всех BODY фреймов
        }
        
        // 7. В зависимости от типа очереди
        if (queue.type == "quorum") {
            quorum_queue_append(queue, message)
        } else if (queue.type == "stream") {
            stream_append(queue, message)
        } else {
            classic_queue_append(queue, message)
        }
    }
    
    // 8. Подтверждение издателю (если включены publisher confirms)
    if (connection.publisher_confirms) {
        send_basic_ack(delivery_tag)
    }
}

Quorum Queue Architecture:
+-------------------+      +-------------------+      +-------------------+
|   Лидер           |      |   Последователь   |      |   Последователь   |
|   (Leader)        |<---->|   (Follower)      |<---->|   (Follower)      |
|                   |      |                   |      |                   |
| • Принимает запись|      | • Реплицирует     |      | • Реплицирует     |
| • Отвечает клиентам|      |   данные         |      |   данные          |
| • Управляет       |      | • Голосует за     |      | • Голосует за     |
|   логом Raft      |      |   выборы лидера   |      |   выборы лидера   |
+-------------------+      +-------------------+      +-------------------+
         |                         |                         |
         |       Кворум (N/2 + 1) узлов согласны            |
         +---------------------------------------------------+

// Псевдокод обработки записи в Quorum Queue
quorum_queue_append(queue, message) {
    // 1. Лидер добавляет запись в свой лог
    log_entry = {
        term: current_term,
        index: next_index++,
        command: "ADD_MESSAGE",
        data: message
    }
    
    leader_log.append(log_entry)
    
    // 2. Репликация на последователей
    followers_acked = 1 // Лидер уже записал
    
    for (follower in queue.followers) {
        send_append_entries(follower, log_entry)
        
        // 3. Ожидание подтверждения от большинства
        if (wait_for_ack(follower, timeout)) {
            followers_acked++
            
            if (followers_acked >= quorum_size(queue)) {
                // 4. Коммит записи (становится видимой для чтения)
                log_entry.committed = true
                apply_to_state_machine(queue, log_entry)
                return SUCCESS
            }
        }
    }
    
    // 5. Если кворум не достигнут
    if (followers_acked < quorum_size(queue)) {
        // Возврат к предыдущему индексу
        next_index--
        return FAILURE
    }
}

// Псевдокод работы Stream Queue
stream_append(queue, message) {
    // 1. Проверка дедупликации
    if (queue.deduplication_enabled && message.id in seen_ids) {
        return DUPLICATE
    }
    
    // 2. Добавление в текущий активный сегмент
    segment = get_active_segment(queue)
    
    // 3. Если сегмент заполнен, ротация
    if (segment.size + message.size > segment.max_size) {
        segment.close()
        segment = create_new_segment(queue)
        queue.active_segment = segment
        
        // 4. Компрессия закрытых сегментов (асинхронно)
        schedule_compression(closed_segments)
    }
    
    // 5. Запись в сегмент
    write_result = segment.append(
        offset: segment.next_offset++,
        timestamp: current_time(),
        message: message
    )
    
    // 6. Обновление индекса для быстрого поиска
    update_index(queue, message.id, segment.id, write_result.position)
    
    return write_result.offset
}

// Упрощенная схема работы журнала
wal_append(message) {
    // 1. Запись в журнал (последовательная запись)
    log_entry = serialize(message)
    wal_file.append(log_entry)
    fsync(wal_file) // Синхронизация с диском
    
    // 2. Добавление в in-memory структуры для быстрого доступа
    index_entry = {
        message_id: message.id,
        wal_position: current_position,
        queue: message.queue,
        status: "PERSISTED"
    }
    
    memory_index.add(index_entry)
    
    // 3. Периодическая очистка устаревших записей
    if (wal_file.size > max_wal_size) {
        rotate_wal_file()
    }
}

// Механизм flow control на основе памяти
check_memory_pressure() {
    memory_used = get_memory_usage()
    
    if (memory_used > memory_limit_high_watermark) {
        // 1. Приостановка публикаций
        block_publishers()
        
        // 2. Принудительная выгрузка сообщений на диск
        for (queue in queues) {
            if (queue.messages_in_memory > threshold) {
                page_to_disk(queue)
            }
        }
        
        // 3. Если не помогло, отключение соединений
        if (memory_used > memory_limit_critical) {
            disconnect_clients_by_memory_usage()
        }
    }
}

// Современная обработка dead letter
dead_letter_strategy {
    // 1. Автоматический DLQ для Quorum Queues
    policy: {
        "dead-letter-strategy": "at-most-once",
        "max-deliveries": 5,
        "dead-letter-exchange": "dlx.global"
    }
    
    // 2. Мониторинг и алертинг DLQ
    monitor_dlq_size(threshold=1000)
    
    // 3. Автоматическая обработка или архивация
    if (dlq.size > critical_threshold) {
        archive_to_object_storage(dlq.messages)
        dlq.purge()
    }
}

if (требуется_гарантированная_доставка && кластер >= 3_узлов) {
    использовать Quorum Queue
} else if (потоковая_обработка && хранение_истории) {
    использовать Stream Queue
} else if (временные_данные && высокая_производительность) {
    использовать Classic Queue in memory
} else {
    // По умолчанию в 2025
    использовать Quorum Queue
}

import java.util.HashSet;

public class Task291225 {
    public static void main(String[] args) {
        HashSet<String> set1 = new HashSet<>();
        HashSet<String> set2 = new HashSet<>();
        
        set1.add("hello");
        set2.add("hello");
        
        String s1 = new String("hello");
        String s2 = new String("hello");
        
        set1.add(s1);
        set2.add(s2);
        
        System.out.println(set1.equals(set2));
        System.out.println(set1.contains(s1));
        System.out.println(set2.contains(s1));
        System.out.println(set1.size());
    }
}

// Упрощенная концептуальная реализация
public static <T> List<T> synchronizedList(List<T> list) {
    return (list instanceof RandomAccess ?
        new SynchronizedRandomAccessList<>(list) :
        new SynchronizedList<>(list));
}

static class SynchronizedList<E> implements List<E> {
    final List<E> list;   // Оборачиваемый список
    final Object mutex;   // Объект для синхронизации
    
    SynchronizedList(List<E> list) {
        this.list = list;
        this.mutex = this; // По умолчанию синхронизируемся на обертке
    }
    
    public E get(int index) {
        synchronized (mutex) { return list.get(index); }
    }
    
    public void add(int index, E element) {
        synchronized (mutex) { list.add(index, element); }
    }
    
    // Все методы синхронизированы аналогично
}

// ОПАСНО: неатомарная составная операция
List<String> syncList = Collections.synchronizedList(new ArrayList<>());

if (!syncList.contains("item")) {    // Операция 1
    syncList.add("item");            // Операция 2
}
// Между проверкой и добавлением другой поток может добавить элемент

List<String> syncList = Collections.synchronizedList(new ArrayList<>());

// ОПАСНО: ConcurrentModificationException все еще возможен
for (String item : syncList) {
    // Другой поток может модифицировать список
    syncList.remove("someItem"); // Из другого потока
}

List<String> syncList = Collections.synchronizedList(new ArrayList<>());

// Безопасная итерация
synchronized (syncList) {
    Iterator<String> it = syncList.iterator();
    while (it.hasNext()) {
        String item = it.next();
        process(item);
    }
}

// Концептуальная структура с Java 8
public class ConcurrentHashMap<K,V> {
    volatile Node<K,V>[] table;
    
    static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        volatile V val;
        volatile Node<K,V> next;
    }
    
    static final class TreeNode<K,V> extends Node<K,V> {
        TreeNode<K,V> parent;
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;
        boolean red;
    }
}

ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

// Поток 1
map.put("key", 42); // Запись с memory barrier

// Путок 2
Integer value = map.get("key"); // Чтение с happens-before гарантиями
// Гарантированно увидит 42, если нет перезаписи

// Атомарные операции
map.putIfAbsent(key, value);    // Вставить если отсутствует
map.replace(key, oldValue, newValue); // Заменить если совпадает
map.compute(key, (k, v) -> v == null ? 1 : v + 1); // Атомарное вычисление

ConcurrentHashMap<String, Long> wordCounts = new ConcurrentHashMap<>();

// Параллельный forEach
wordCounts.forEach(1, // Параллелизм
    (key, value) -> System.out.println(key + ":" + value));

// Поиск
String result = wordCounts.search(1, 
    (key, value) -> value > 1000 ? key : null);

// Свертка
long total = wordCounts.reduceValues(1, Long::sum);

ConcurrentHashMap<String, Data> map = new ConcurrentHashMap<>(
    16,      // initial capacity
    0.75f,   // load factor
    8        // concurrency level (оценочное количество потоков)
);

ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();

// Приближенный размер (O(1), но может быть неточным)
int approximateSize = map.size();

// Более точный (но дорогой) подсчет
int exactSize = map.mappingCount(); // Java 8+

// Проверка пустоты (эффективная)
boolean isEmpty = map.isEmpty();

public class CopyOnWriteArrayList<E> {
    private transient volatile Object[] array;
    
    final Object[] getArray() {
        return array;
    }
    
    final void setArray(Object[] a) {
        array = a;
    }
}

public class EventPublisher {
    private final CopyOnWriteArrayList<EventListener> listeners =
        new CopyOnWriteArrayList<>();
    
    public void addListener(EventListener listener) {
        listeners.add(listener); // Безопасно даже во время уведомлений
    }
    
    public void publish(Event event) {
        for (EventListener listener : listeners) {
            // Итерация по snapshot - безопасна
            listener.onEvent(event);
        }
    }
}

public class ConfigurationCache {
    private volatile CopyOnWriteArrayList<Config> cache;
    
    public ConfigurationCache() {
        cache = new CopyOnWriteArrayList<>();
    }
    
    public void refresh() {
        List<Config> newConfigs = loadConfigs();
        // Атомарная замена всего кэша
        cache = new CopyOnWriteArrayList<>(newConfigs);
    }
    
    public List<Config> getConfigs() {
        return cache; // Безопасное чтение
    }
}

// Для частых модификаций
List<String> frequentWrites = Collections.synchronizedList(new ArrayList<>());

// Для mixed workloads
ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();

// Для сценариев с преобладанием чтения
List<String> readMostly = new CopyOnWriteArrayList<>();

// Внутренний механизм ArrayList
protected transient int modCount = 0;

// В итераторе
int expectedModCount = modCount;

void checkForComodification() {
    if (modCount != expectedModCount)
        throw new ConcurrentModificationException();
}

List<String> list = new ArrayList<>(Arrays.asList("A", "B", "C"));

for (String item : list) {          // Создается итератор
    if (item.equals("B")) {
        list.remove(item);           // modCount++ → исключение!
    }
}

// Поток 1
for (String item : sharedList) {
    process(item);                   // Итерация
}

// Поток 2
sharedList.add("new");              // ConcurrentModificationException в потоке 1

public class UnsafeCache {
    private Map<String, Data> cache = new HashMap<>();
    
    public Data get(String key) {
        Data data = cache.get(key);
        if (data == null) {
            data = loadData(key);         // Дорогая операция
            cache.put(key, data);         // Race condition!
        }
        return data;                      // Может вернуть null
    }
}

public class BrokenSingleton {
    private static Data instance;
    
    public static Data getInstance() {
        if (instance == null) {                    // Первая проверка (без синхронизации)
            synchronized (BrokenSingleton.class) {
                if (instance == null) {            // Вторая проверка
                    instance = new Data();         // Небезопасная публикация!
                }
            }
        }
        return instance;                           // Может вернуть частично инициализированный объект
    }
}

// ConcurrentHashMap: запрещает null
ConcurrentHashMap<String, String> chm = new ConcurrentHashMap<>();
chm.put("key", null);               // NullPointerException

// CopyOnWriteArrayList: разрешает null
CopyOnWriteArrayList<String> cowal = new CopyOnWriteArrayList<>();
cowal.add(null);                    // Допустимо

// Collections.synchronizedList: зависит от оборачиваемой коллекции
List<String> syncList = Collections.synchronizedList(new ArrayList<>());
syncList.add(null);                 // Допустимо для ArrayList

public class Counter {
    private int count;
    
    public void increment() {
        count++; // Неатомарная операция: read-modify-write
    }
}

// Два потока вызывают increment() 1000 раз каждый
// Ожидаемый результат: 2000
// Фактический результат: что угодно между 1000 и 2000

public class VisibilityProblem {
    private boolean ready = false;
    private int value;
    
    // Поток 1
    public void writer() {
        value = 42;
        ready = true; // Без happens-before!
    }
    
    // Путок 2
    public void reader() {
        if (ready) {
            System.out.println(value); // Может увидеть 0 вместо 42!
        }
    }
}

// Классический deadlock с synchronizedList
List<String> list1 = Collections.synchronizedList(new ArrayList<>());
List<String> list2 = Collections.synchronizedList(new ArrayList<>());

// Поток 1
synchronized (list1) {
    synchronized (list2) {  // Ждет list2
        // Критическая секция
    }
}

// Путок 2 (обратный порядок)
synchronized (list2) {
    synchronized (list1) {  // Ждет list1 → DEADLOCK!
        // Критическая секция
    }
}

public class LivelockExample {
    private final ConcurrentHashMap<String, Boolean> locks = 
        new ConcurrentHashMap<>();
    
    public void process(String key) {
        // Бесконечные попытки захвата "локера"
        while (!locks.putIfAbsent(key, true)) {
            Thread.yield(); // Livelock: постоянно уступаем, но не прогрессируем
        }
        
        try {
            // Работа с ресурсом
        } finally {
            locks.remove(key);
        }
    }
}

// Поток, постоянно читающий
synchronized (sharedList) {
    // Долгая операция чтения
    processAllElements(sharedList);
}

// Другие потоки не могут получить доступ для записи
// → Starvation писателей

import java.util.concurrent.*;

public class Task301225 {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
        map.put("a", 1);
        
        Integer result = map.computeIfAbsent("b", k -> {
            map.put("b", 2);
            return map.get("b");
        });
        
        System.out.println(result);
        System.out.println(map.get("b"));
    }
}