Недавно встретилась по работе интересная задача, прямо на те самые презираемые на интервью алгоритмы. Очередное доказательство, что, по крайней мере, в Гугле алгоритмы нужны. А значит и интервью по ним вообще-то не оторваны от реальности.

Итак, задача: есть лог видео конференции, состоящий из событий - в такое-то время такой-то пользователь подключился или отключился. И надо посчитать всякую разную статистику. Среди прочего скучного надо для каждого пользователя найти, сколько максимально пользователей было вместе с ним в какой-то момент времени.

Похоже, на интервью спрашивают слишком простые задачки, потому что тут другой программист впендюрил наивное решение за квадрат: перебираем события в порядке возрастания времени, поддерживаем хеш-сет активных пользователей, обновляем значение метрики для всех активных пользователей текущим размером хэш-сета. А квадратичное решение тут действительно плохо, ибо бывают конференции и на тысячи пользователей.

std::unordered_set<UserID> active_users;
for (auto &event: events) {
  if (event.type == Event::kJoin) {
    active_users.insert(event.user_id);
    for (auto &user_id : active_users) {
      stats[user_id].max_concurrent_users = 
        max(stats[user_id].max_concurrent_users, 
            active_users.size());
    }
  } else {
    active_users.erase(event.user_id);
  }
}

Обратите внимание, тут даже оптимизация есть - максимальное количество пользователей считается только при событии типа join. Примерно в 2 раза быстрее. Но это все еще решение за квадрат потому что тут n промежутков времени и в каждом из них может быть активно O(n) пользователей. Тут и далее n - количество событий.

Как всегда, программист просто не понял, что тут вообще есть какие-то эти ваши алгоритмы. У него есть задача - подсчитать метрики. Он эту задачу и решил, переведя требования на язык программирования. То, что тут вообще-то стандартнейшая алгоритмическая задача из книжек и не видно по постановке. Именно это свойство алгоритмических задач и создает у многих ошибочное ощущение, что алгоритмы им в работе вообще не нужны, и кроме литкода и интервью нигде не встречаются.

Задача Range Maximum Query

Эта задача на самом деле - стандартная задача Range Maximum Query: если рассмотреть, сколько пользователей было между двумя соседними событиями, то у нас получается массив из чисел, и для каждого пользователя надо найти максимум на отрезке от его события подключения до его события отключения. Правда, в литературе ставят задачу на минимум (Range Minimum Query), но это вообще не принципиально - абсолютно те же самые алгоритмы можно заставить искать максимум, лишь поменяв пару знаков в сравнениях. Или в худшем случае умножить все числа на -1, найти из них минимум неизмененным алгоритмом, а потом поменять у ответа знак назад.

Вот иллюстрация:

Выше фактически реализовано наивное решение задачи RMQ за , только порядок циклов вывернут, чтобы не хранить все значения в массиве. В наивном решении внешний цикл по запросам, а внутри цикл по всем числам отрезке. А тут, наоборот - происходит проход по всем числам и обновление ответа для всех запросов, включающих это число.

У этой задачи есть множество решений разной степени сложности кода и эффективности. На хабре уже есть множество статей:

• https://habr.com/ru/articles/114980/ предподсчет + на запрос благодаря предподсчету на всех отрезках длины, равной какой-то степени двойки.

• https://habr.com/ru/articles/115026/ предпосдчет + на запрос благодаря использованию структуры данных "дерево отрезков".

• https://habr.com/ru/articles/138946/ предподсчет + на запрос через разбиение на блоки размером .

• На хабре так и не написали об алгоритме Фараха-Колтона и Бендера c пердподсчтетом и на запрос. Но этот алгоритм весьма сложен в реализации и состоит из кучи шагов: Сначала RMQ сводится к задаче Lowest Common Ancestor (LCA) через построение Декартового дерева, потом LCA сводится к задаче +-1 RMQ (тот же RMQ, только числа в массиве отличаются на +1 или -1), и уже эта задача решается за линейную сложность через предподсчет и разбиение на блоки размера .

Тут я приведу другой алгоритм с пердподсчетом и на запрос, который даже проще многих перечисленных выше в реализации. Правда, в отличии от некоторых из них, этот алгоритм offline, т.е. он может обработать все запросы только вместе группой. Этот алгоритм в интернете иногда называют "Arpa's trick".

В итоге этот алгоритм дает решение исходной задачи за , при условии уже отсортированности событий по времени. Но даже с сортировкой он все еще быстрее и проще перечисленных выше алгоритмов.

Линейное решение задачи

Итак, основная идея: давайте для каждого числа найдем следующее большее-равное его справа и сохраним на него указатель. Тогда при запросе на максимум нам надо пройтись по этим ссылкам от левого края запроса, пока мы не выйдем из отрезка. Последний элемент в цепочке будет максимумом на отрезке. Но тут сложно быстро понять, где остановиться.

Тогда давайте обрабатывать запросы слева направо по концам и поддерживать ссылки только среди чисел левее границы. Т.е. мы будем дорисовывать стрелочки ведущие в текущее число, а потом находить все максимумы на запросы, кончающиеся в текущем числе. Тогда задача становится проще: надо просто идти до конца всегда. Мы не выйдем за границу отрезка, потому что ссылки за нее еще не созданы:

Итак, у нас есть структура, где мы от каждого элемента можем провести максимум одну стрелочку к какому-то другому элементу и нам надо уметь проходиться по стрелочкам до конца. Знакомые со структурами данных читатели заметят, что эта структура очень похожа на структуру данных Disjoint Set Union (или Union-Find, или "система непересекающихся множеств"). С той лишь разницей, что в DSU числа куда-то ссылаются всегда, но некоторые ссылаются сами на себя и итерация идет пока не зациклится в таких числах. Давайте тогда проводить ссылку от числа на само себя, если правее нет ничего больше-равного. И вот мы решили задачу стандартной структурой DSU. Прелесть этой структуры данных в том, что она позволяет искать конец цепочки за, где - это обратная функция Аккермана, которая меньше 5 для всех мыслимых объемов данных (), поэтому ее можно считать константой. Об этом свойстве уже есть замечательная статья на хабре.

Вот и алгоритм получается такой: идем слева направо по массиву и объединяем все значения в непересекающиеся множества - элементы принадлежат одному и тому же множеству, если в отрезке, начинающемся с них, максимумом будет один и тот же элемент. Этот элемент будет корнем множества. Чтобы найти максимум на отрезке, заканчивающимся в последнем обработанном числе, надо взять корень множества, которому принадлежит начало отрезка.

Остается вопрос о том, как эти множества поддерживать при добавлении нового элемента. Можно заметить, что множества эти образуют непрерывные отрезки, а сами максимумы будут на концах отрезков и идут в массиве в порядке убывания значений. Действительно, самый большой элемент в массиве будет корнем для всех чисел левее его. Это будет первое множество. Далее можно рассмотреть оставшиеся числа правее, там тоже есть какой-то глобальный максимум и он тоже откусит от этих чисел свое множество и так далее. Когда мы добавляем новый элемент, он "вытеснит" все корни множеств, которые меньше-равны ему, и поглотит их множества. Так что, если хранить эти корни множеств в порядке увеличения индексов, то они будут и в порядке уменьшения значений. И при добавлении нового элемента надо будет убрать все меньшие с конца. Т.е. хранить их надо в стеке и убирать с конца меньшие нового элемента.

Вот и код с "алгоритмами". Несмотря на то, что он работает в тысячи раз быстрее на больших объемах, он не сильно сложнее. Он даже работает не медленнее на маленьких конференциях, так что он всегда лучше наивного решения за квадрат:

// Структура непересекающихся множеств, всегда указывает на больший элемент
// правее данного.
std::vector<int> dsu(events.size());
// Количество акивных пользователей после каждого события.
std::vector<int> user_counts(events.size());
// Индексы максимальных элементов, для которых нет большего правее.
// Они же корни в системе непересекающихся множеств.
// Всегда в порядке возрастания индексов и убывания значений.
std::deque<int> maximums_stack;
// запоминаем для каждого пользователя, когда он подключился
std::unordered_map<UserID, int> join_event_idx;
int active_user_count = 0;

// Основной цикл.
for (int event_idx = 0; event_idx < events.size(); ++event_idx) {
  Event &event = events[event_idx];
  if (event.type == Event::kJoin) {
    ++active_user_count;
    // Запоминаем начало отрезка на будущее.
    join_event_idx[event.user_id] = event_idx;
  } else {
    --active_user_count;
    // Берем известное начало отрезка-запроса.
    int join_idx = join_event_idx[event.user_id];
    // Проходимся до корня от самого левого элемента в отрезке.
    int max_users_count = user_counts[DsuRoot(join_idx, dsu)];
    // Сохраняем ответ.
    stats[event.user_id].max_concurrent_users = max(user_stats.max_concurrent_users, 
                                                    max_users_count);
  }
  // Добавляем новый элемент в структуру.
  user_counts[event_idx] = active_user_count;
  // Правее него пока вообще ничего нет - он сам себе максимум.
  dsu[event_idx] = event_idx;
  // Какие-то из существующих максимумов меньше нового элемента.
  // Но они хранятся в стеке, по убыванию, поэтому все они лежат на верхушке.
  while (!maximums_stack.empty() && 
         user_counts[stack.back()] <= active_user_count) {
    // Новый элемент теперь следующий больший для вытесненного максимума.
    dsu[maximums_stack.back()] = event_idx;
    maximums_stack.pop_back();
  }
  // Новый элемент - максимум, и он должен быть в стеке.
  stack.push_back(event_idx);
}

// Функция для работы с системой непересекающихся множеств.
// Находит корень множества, которому принадлежит idx.
int DsuRoot(int idx, vector<int> &dsu) {
  int root = idx;
  while (dsu[root] != root) {
    root = dsu[root];
  }
  // Эвристика сжатия путей.
  while (idx != root) {
    int nxt = dsu[idx];
    dsu[idx] = root;
    idx = nxt;
  }
  return cur;
}

Замечание по оценке времени работы

Очень продвинутые в алгоритмах читатели возразят, что тут используется DSU лишь с эвристикой сжатия путей, без ранговой эвристики. И вообще DSU работает за а не за Хоть это и можно считать константой на практике. И действительно, если бы у нас была произвольная задача RMQ так все и было бы. И этот код в итоге работал бы за O(n + m log n), где m количество запросов. А чтобы код работал хотя бы за надо было бы реализовывать ранговую эвристику при слиянии множеств, и еще пришлось бы дополнительно хранить значение максимума для каждого множества (ведь максимум уже не обязательно был бы корнем).

Но у нас ведь задача немного ограниченная: числа в массиве не любые, они изменяются только на +-1 от соседних. Плюс, в каждом элементе может начинаться или заканчиваться максимум один запрос. И начало этого запроса всегда выпадает на элемент, который больше предыдущего. Все эти ограничения позволяют доказать строгую оценку O(n) для времени работы этого алгоритма и без ранговой эвристики. А без нее код получается чуть проще и работает чуть быстрее. Но у меня не получилось вывести достаточно простое и короткое доказательство. Так что оно, может быть, появится в следующей статье, если кому-то интересно.

Альтернативные объясниения алгоритма

Можно рассматривать этот алгоритм как смесь алгоритма Фараха-Колтона-Бендера, упомянутого в начале статьи, и алгоритма Тарьяна для LCA. Сначала точно такое же сведение RMQ к LCA в декартовом дереве, но потом применяем алгоритм Тарьяна с DSU для поиска LCA вместо сведения задачи к +-1 RMQ. Если упрощать код и смешивать выполнение LCA вместе с построением декартового дерева, то в итоге получится вот этот алгоритм.

Еще его можно вывести из другой логики, именно так я к нему и пришел. Давайте обрабатывать запросы слева направо в концах. Тогда нам надо уметь выполнять 2 операции: добавить элемент и взять максимум из скольки-то последних. Эта задача чем-то похожа на известную задачу о максимуме в sliding window. И тут применима такая же логика: какие-то элементы никогда не будут ответом ни для какого запроса. Это те, правее котрых есть что-то больше-равное их. Давайте их называть "плохими", а все остальные - "хорошими". Действительно, любой запрос, включающий такой плохой элемент, включает и больший его правее, а значит ответом будет вот тот другой элемент (или что-то еще большее), но точно не сам плохой элемент.

Поэтому плохие элементы можно вообще выкинуть из рассмотрения, тогда оставшиеся хорошие элементы образуют строго убывающую подпоследовательность. Действительно, если где-то они идут по возрастанию, то левый элемент не может быть хорошим, ведь правее его есть что-то большее. При добавлении нового элемента какие-то хорошие с конца станут плохими и будут выкинуты. А новый элемент всегда хороший, ведь правее его вообще ничего нет, не говоря уже о больших элементах. Эти хорошие элементы - это и есть те самые корни множеств и числа в стеке в алгоритме.

При запросе надо найти самый левый хороший элемент, который еще попадает в отрезок-запрос. Ведь в качестве ответа можно взять все хорошие элементы, но самый больший из них - самый левый, они же убывают. Тут можно было бы воспользоваться бинарным поиском по индексу, но можно пойти дальше. Давайте для всех возможных начал запросов следить, где бы они находились в стеке, если бы они там были. Можно их как бы повесить на элемент в стеке, который ближайший к ним справа. При удалении элемента из стека, повешенные на него начала будут перевешаны на новый элемент. Вот и получилось, что нам надо поддерживать непересекающиеся множества и объединять их, а максимум для каждого множества - это какой-то элемент в стеке, соответствующий какому-то множеству. Это можно быстро делать с помощью структуры данных DSU.