[Перевод] Новый рекорд производительности FizzBuzz21.02.2024 10:46

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

namespace {

// Вспомогательный Constexpr для вычисления 10^X, потому что std::pow не является constexpr.
constexpr int64_t PowTen(int x) {
  int64_t result = 1;
  for (int i = 0; i < x; ++i) {
    result *= 10;
  }
  return result;
}

// Мы обрабатываем каждую группу параллельно из |kParallelism| потоков. Это число
// должно быть равно или меньше количества доступных ядер CPU. Увеличение числа
// необязательно приводит к повышению производительности из-за дополнительных операций
// по синхронизации между потоками.
constexpr int64_t kParallelism = 7;
// Последние kSuffixDigits цифр каждой числовой строки не затрагиваются
// при итерациях.
constexpr int64_t kSuffixDigits = 6;
// Выполняем инкремент первой правой задействованной цифры за один шаг на эту величину.
// Инкремент должен быть кратен 3. Пока код может обрабатывать инкремент, состоящий только из одной цифры.
constexpr int64_t kIncrementBy = 3;
// Одна группа содержит не больше этого значения строк.
constexpr int64_t kMaxLinesInBatch = PowTen(kSuffixDigits) * kIncrementBy / 3;

constexpr int kFizzLength = 4;
constexpr int kBuzzLength = 4;
constexpr int kNewlineLength = 1;

// Барьер, который в ждущем цикле ожидает, пока его не достигнут все остальные потоки.
template 
class SpinningBarrier {
 public:
  // Создаёт ожидающий барьер с |count| участвующих потоков и 
  // обратный вызов завершения |completion_cb|.
  // После того, как все потоки достигли барьера, последний поток выполняет
  // обратный вызов завершения. Другие потоки заблокированы, пока не будет выполнен
  // возврат обратного вызова завершения.
  SpinningBarrier(int64_t count, Completion completion_cb) :
      count_(count), spaces_(count), generation_(0),
      completion_cb_(completion_cb) {}

  void Wait() {
      int64_t my_generation = generation_;
      if (!--spaces_) {
          spaces_ = count_;
          completion_cb_();
          ++generation_;
      } else {
          while(generation_ == my_generation);
      }
  }

 private:
  int64_t count_;
  std::atomic spaces_;
  std::atomic generation_;
  Completion completion_cb_;
};

// Владеет буферами вывода и следит за тем, какой буфер использовался последним.
class OutputHandler {
 static constexpr size_t kBufferSize = 14 * 1024 * 1024;

 public:
  OutputHandler() {
    for (int i = 0; i < 3; ++i) {
      buffers_[i] =
        static_cast(aligned_alloc(2 * 1024 * 1024, kBufferSize));
      madvise(buffers_[i], kBufferSize, MADV_HUGEPAGE);
    }
  }

  ~OutputHandler() {
    for (int i = 0; i < 3; ++i) {
      free(buffers_[i]);
    }
  }

  void Output(int buffer_id, size_t bytes) {
    // Используется три буфера. Мы должны гарантировать, что пока буфер (или его
    // часть) находится в канале, он не будет модифицироваться. Не существует API, позволяющего узнать,
    // когда downstream-процесс завершил чтение данных из канала,
    // так что нужно выбирать размер канала с умом.
    // Так как канал не может уместить больше двух полных буферов, мы можем гарантировать,
    // что после вывода буфера в порядке 0, 1, 2 канал больше не содержит
    // данные из буфера 0. Однако если сделать канал слишком маленьким,
    // программа будет медленнее. Оптимальный размер канала вычисляется
    // TargetPipeSize. Так как существует минимальный размер канала, ниже которого
    // мы опуститься не можем (4 КБ в Linux), такой подход не сработает при
    // слишком маленьком размере буфера. В таких случаях мы откатываемся к использованию write(),
    // которая копирует содержимое в канал, поэтому нет риска перезаписи
    // памяти, по-прежнему считываемой из downstream-процесса.
    // Однако если в последующем вызове Output() передаётся меньший размер
    // (а значит, исполняется ветвь else), то в канале всё равно могут оказаться
    // какие-то данные из текущей итерации и полные данные
    // из следующей итерации. Мы предполагаем, что Output() будет
    // вызываться с монотонно увеличивающимися размерами (что на практике справедливо,
    // но лучше на такие предположения не опираться).
    SetPipeSize(TargetPipeSize(bytes));
    if (2 * bytes >= pipe_size_) {
      OutputWithVmSplice(buffers_[buffer_id], bytes);
    } else {
      if (write(STDOUT_FILENO, buffers_[buffer_id], bytes) < 0) {
        std::cerr << "write error: " << errno;
        std::abort();
      }
    }
  }

  char* GetBuffer(int buffer_id) {
    return buffers_[buffer_id];
  }

  // Возвращает id следующего буфера, который можно заполнить и вывести.
  // Вызывающие стороны отвечают за сам вывод буфера после его запроса
  // при помощи этого метода.
  int NextBufferId() {
    buffer_id_ = (buffer_id_ + 1) % 3;
    return buffer_id_;
  }

  static constexpr int64_t BufferSize() {
    return kBufferSize;
  }

 private:
  // Вычисляет оптимальный размер канала для вывода |out_bytes|.
  size_t TargetPipeSize(size_t out_bytes) const {
    // Размеры каналов должны быть равны степеням двойки и >= 4 КБ в Linux.
    // Мы хотим, чтобы канал не более чем в два раза был больше вывода (но всё равно
    // максимизировать размер канала), поэтому округляем |out_bytes| вверх до ближайшей
    // степени двойки.
    return std::max(4ul * 1024, std::bit_ceil(out_bytes));
  }

  void OutputWithVmSplice(char* buffer, size_t bytes) const {
    iovec iov;
    iov.iov_base = buffer;
    iov.iov_len = bytes;
    while (true) {
      int64_t ret = vmsplice(STDOUT_FILENO, &iov, 1, SPLICE_F_NONBLOCK);
      if (ret >= 0) {
        iov.iov_len -= ret;
        iov.iov_base = reinterpret_cast(iov.iov_base) + ret;
        if (iov.iov_len == 0) {
          break;
        }
      } else {
        if (errno != EAGAIN) {
          std::cerr << "vmsplice error: " << errno;
          std::abort();
        }
      }
    }
  }

  void SetPipeSize(size_t size) {
    if (pipe_size_ == size) {
      return;
    }
    size_t new_pipe_size = fcntl(STDOUT_FILENO, F_SETPIPE_SZ, size);
    if (new_pipe_size < 0) {
      std::cerr << "Error while calling fcntl F_SETPIPE_SZ " << errno
      << "\nPerhaps you need to update /proc/sys/fs/pipe-max-size or "
         "run the program as sudo";
      std::abort();
    }

    pipe_size_ = new_pipe_size;
  }

  std::array buffers_;
  int buffer_id_ = 0;
  size_t pipe_size_;
};

// Вставляем строку fizzbuzz для номера строки |line| и символ новой строки
// в |out|.
// Возвращаем указатель, указывающий на символ после символа новой строки.
char* InsertFizzBuzzLine(char* out, int64_t line) {
  if (line % 15 == 0) {
    std::memcpy(out, "FizzBuzz\n", 9);
    return out + 9;
  } else if (line % 3 == 0) {
    std::memcpy(out, "Fizz\n", 5);
    return out + 5;
  } else if (line % 5 == 0) {
    std::memcpy(out, "Buzz\n", 5);
    return out + 5;
  } else {
    // Мы поддерживаем числа вплоть до 10^20.
    char* next = std::to_chars(out, out + 20, line).ptr;
    *next = '\n';
    return next + 1;
  }
}

// Прогон (Run) - это все строки, в которых номера строк содержат |DIGITS| цифр.
// Run<1>: [1,9]
// Run<2>: [10,99]
// ...
template
class Run {

  static_assert(DIGITS >= 1);

  static constexpr int FizzBuzzLineLength(int64_t number_mod_15) {
    if (number_mod_15 % 15 == 0) {
      return 9;
    } else if (number_mod_15 % 3 == 0) {
      return 5;
    } else if (number_mod_15 % 5 == 0) {
      return 5;
    } else {
      return DIGITS + 1;
    }
  }

  // Возвращает размер одного fifteener в байтах.
  static constexpr size_t FifteenerBytes() {
    size_t size = 0;
    for (int i = 0; i < 15; ++i) {
      size += FizzBuzzLineLength(i);
    }
    return size;
  }

  // Возвращает количество строк в этом прогоне.
  static constexpr int64_t LinesInRun() {
    return PowTen(DIGITS) - PowTen(DIGITS - 1);
  }

  // Весь вывод fizz-buzz для этого прогона занимает такое количество байтов.
  static constexpr size_t RunBytes() {
    if constexpr(DIGITS == 1) {
      return 5 + 3 * kFizzLength + 1 * kBuzzLength + 9 * kNewlineLength;
    } else {
      return LinesInRun() / 15 * FifteenerBytes();
    }
  }

  // Возвращает количество групп в этом прогоне.
  static constexpr int64_t BatchesInRun() {
    if constexpr (DIGITS > kSuffixDigits) {
      return PowTen(DIGITS - kSuffixDigits - 1) * 9;
    } else {
      return 1;
    }
  }

 public:
  // Выводит все строки для этого прогона, используя буферы из |output_handler|.
  static void Execute(OutputHandler& output_handler) {
    Batch<0> batch0(&output_handler);
    Batch<1> batch1(&output_handler);
    Batch<2> batch2(&output_handler);
    // Заполняем каждую группу начальными значениями. Это довольно медленный
    // процесс, поэтому мы выполняем его только один раз за прогон. В последующих итерациях мы
    // выполняем только инкремент чисел (см. ниже), что намного быстрее.
    batch0.Init();
    batch0.Output();
    if constexpr (BatchesInRun() > 1) {
      batch1.Init();
      batch1.Output();
    }
    if constexpr (BatchesInRun() > 2) {
      batch2.Init();
      batch2.Output();
    }

    if constexpr (BatchesInRun() > 3) {
      int64_t prefix = PowTen(DIGITS - kSuffixDigits - 1);
      // Обновляем группу из |kParallelism| потоков
      // Используем ожидающий барьер для синхронизации между потоками.
      // После того, как все потоки достигли барьера, выполняется функция завершения
      // и печатается вывод. Затем обрабатывается следующая группа
      SpinningBarrier barrier(kParallelism, [&] {
        switch (prefix % 3) {
          // В начале
          //   batch0 соответствует префиксу 10..00 ( ≡ 1 mod 3),
          //   batch1 соответствует префиксу 10..01 ( ≡ 2 mod 3),
          //   batch2 соответствует префиксу 10..02 ( ≡ 0 mod 3).
          // После обработки всех трёх групп выполняется инкремент префикса на 3,
          // поэтому mod не меняются.
          case 0: batch2.Output(); break;
          case 1: batch0.Output(); break;
          case 2: batch1.Output(); break;
        }
        prefix++;
      });

      [&](std::index_sequence) {
        // Запускаем |kParallelism| потоков. Также можно использовать пул потоков,
        // но один прогон занимает так много времени, что запуск новых потоков
        // можно считать несущественной тратой ресурсов.
        (std::jthread([&] {
          for (int64_t batch = 3; batch < BatchesInRun();
               batch += 3) {
            // Каждый поток обрабатывает свой блок в группе.
            Chunk<0, THREAD_ID>(batch0).IncrementNumbers(prefix);
            // На этом этапе все потоки ожидают, пока все остальные потоки достигнут
            // барьера; последний закончивший поток вызывает batch.Output()
            // (см. выше определение |barrier|).
            barrier.Wait();
            Chunk<1, THREAD_ID>(batch1).IncrementNumbers(prefix);
            barrier.Wait();
            Chunk<2, THREAD_ID>(batch2).IncrementNumbers(prefix);
            barrier.Wait();
          }
        }) , ...);
      }(std::make_index_sequence());
    }
  }

  // Группа представляет собой 10^|kSuffixDigits| строк вывода.
  // Это полезно потому, что последние |kSuffixDigits| цифр не нужно обновлять.
  // Более того, номера строк в одной группе имеют одинаковый префикс.
  // BATCH_ID ∈ [0, 1, 2]
  template
  class Batch {
    static_assert(BATCH_ID < 3);
    using PreviousBatch = Batch;

    public:
    Batch(OutputHandler* output_handler) : output_handler_(output_handler) {
      static_assert(OutputHandler::BufferSize() >= BytesInBatch());
    }

    // Инициализирует эту группу, взяв следующий доступный буфер из
    // обработчика вывода и заполняя его начальными значениями.
    void Init() {
      buffer_id_ = output_handler_->NextBufferId();
      char* out = GetBuffer();
      int64_t start = PowTen(DIGITS - 1) + BATCH_ID * LinesInBatch();
      int64_t end = std::min(PowTen(DIGITS), start + LinesInBatch());
      for (int64_t line = start; line < end; ++line) {
        out = InsertFizzBuzzLine(out, line);
      }
    }

    // Возвращает первый номер строки в этом блоке mod 15.
    static constexpr int64_t FirstLineNumberMod15() {
      if constexpr (BATCH_ID == 0) {
        return DIGITS > 1 ? 10 : 1;
      } else {
        return (PreviousBatch::FirstLineNumberMod15() +
          PreviousBatch::LinesInBatch()) % 15;
      }
    }

    // Возвращает количество строк в этой группе.
    static constexpr int64_t LinesInBatch() {
      return std::min(kMaxLinesInBatch, LinesInRun());
    }

    // Возвращает размер этой группы в байтах.
    static constexpr int64_t BytesInBatch() {
      if constexpr (LinesInBatch() < kMaxLinesInBatch) {
        return RunBytes();
      } else {
        size_t size = LinesInBatch() / 15 * FifteenerBytes();
        for (int64_t i = FirstLineNumberMod15() + LinesInBatch() / 15 * 15;
             i < FirstLineNumberMod15() + LinesInBatch(); ++i) {
          size += FizzBuzzLineLength(i);
        }
        return size;
      }
    }

    void Output() {
      output_handler_->Output(buffer_id_, BytesInBatch());
    }

    char* GetBuffer() {
      return output_handler_->GetBuffer(buffer_id_);
    }

    OutputHandler* output_handler_;
    // id буфера, который эта группа должна использовать в |output_handler_|.
    int buffer_id_;
  };

  // Обозначает блок - часть группы, обрабатываемую потоком с id
  // |THREAD_ID|. THREAD_ID ∈ [0, kParallelism)
  // Так как числа в каждом блоке должны инкрементироваться по разным индексам,
  // мы специализируем этот класс для каждого BATCH_ID и THREAD_ID, чтобы индексы
  // можно было предварительно вычислять во время компиляции.
  template
  class Chunk {
    using PreviousChunk = Chunk;

    public:
    // Инициализирует блок, находящийся в |batch|.
    Chunk(Batch batch) : batch_(batch) {}

    // Возвращает номер первой строки этого блока mod 15.
    static constexpr int64_t FirstLineNumberMod15() {
      if constexpr (THREAD_ID == 0) {
        return Batch::FirstLineNumberMod15();
      } else {
        return (PreviousChunk::FirstLineNumberMod15() +
          PreviousChunk::LinesInChunk()) % 15;
      }
    }

    // Возвращает индекс начального байта этого блока в группе.
    static constexpr int64_t StartIndexInBatch() {
      if constexpr (THREAD_ID == 0) {
        return 0;
      } else {
        return PreviousChunk::StartIndexInBatch() +
          PreviousChunk::BytesInChunk();
      }
    }

    // Возвращает количество строк в этом блоке.
    static constexpr int64_t LinesInChunk() {
      int64_t done = THREAD_ID == 0 ? 0 :
        PreviousChunk::CumulativeLinesUpToChunk();
      int64_t remaining_lines = Batch::LinesInBatch() - done;
      int64_t remaining_threads = kParallelism - THREAD_ID;
      // эквивалент ceil(remaining_lines / remaining_threads)
      return (remaining_lines - 1) / remaining_threads + 1;
    }

    // Возвращает количество строк в этом и всех предыдущих блоках группы.
    static constexpr int64_t CumulativeLinesUpToChunk() {
      if constexpr (THREAD_ID < 0) {
        return 0;
      } else {
        return PreviousChunk::CumulativeLinesUpToChunk() + LinesInChunk();
      }
    }

    // Возвращает длину этого блока в байтах.
    static constexpr int64_t BytesInChunk() {
      size_t size = LinesInChunk() / 15 * FifteenerBytes();
      for (int64_t i = FirstLineNumberMod15() + LinesInChunk() / 15 * 15;
           i < FirstLineNumberMod15() + LinesInChunk(); ++i) {
        size += FizzBuzzLineLength(i);
      }
      return size;
    }

    // Выполняет инкремент всех чисел в блоке.
    // Эта функция выполняет сворачивание IncrementNumbersImpl для эффективной диспетчеризации
    // в специализированные версии на основании |prefix|.
    void IncrementNumbers(int64_t prefix) {
      // Если DIGITS < kSuffixDigits, то это значит, что все числа в рамках прогона
      // уместятся в одну группу, поэтому нам не нужно использовать IncrementNumbers().
      // Представленная ниже реализация не будет работать.
      static_assert(DIGITS >= kSuffixDigits);
      constexpr int64_t max_overflow_digits = DIGITS - kSuffixDigits;
      // Содержит специализацию IncrementChunkImpl() для каждого значения
      // в 0..max_overflow_digits. Мы используем её для перехода к нужной специализации.
      constexpr auto increment_chunk_impls = []() {
        std::array res{};
        [&](std::index_sequence) {
          ((res[OVERFLOW_DIGITS] = &IncrementNumbersImpl), ...);
        }(std::make_index_sequence());
        return res;
      }();

      increment_chunk_impls[OverflowDigits(prefix)](batch_.GetBuffer());
    }

    private:
    // Инкрементирует этот блок в |batch|.
    //
    // Каждая числовая строка инкрементируется на |kIncrementBy| * 10^kSuffixDigits.
    // Если OVERFLOW_DIGITS > 0, то мы предполагаем, что операция приведёт к переполнению,
    // поэтому нам нужно заранее выполнить инкремент на это количество цифр.
    // За вычисление количества цифр, которое необходимо обновить в этом блоке,
    // отвечает вызывающая сторона.

    // Например, если kIncrementBy = 3 and kSuffixDigits = 6, то
    // блок [100000000, 100999999] можно инкрементировать до [103000000; 103999999]
    // с  OVERFLOW_DIGITS = 0 (без переполнения).
    // При инкрементировании [108000000, 108999999] до [111000000; 111999999],
    // OVERFLOW_DIGITS = 1 (переполнение на одну цифру).
    // При инкрементировании [198000000, 198999999] до [201000000, 201999999],
    // OVERFLOW_DIGITS = 2 (переполнение на две цифры)
    template
    static void IncrementNumbersImpl(char* batch) {
      char* out = batch;
      constexpr int64_t start_index = StartIndexInBatch();
      constexpr int first_line_number_mod_15 = FirstLineNumberMod15();

      // Инкрементирует |num_lines|, начиная с |out|.
      // |num_lines| должно делиться на 120 (за исключением последней итерации).
      auto increment = [&] (int num_lines) __attribute__((always_inline)) {
        int line_start = 0;
        #pragma GCC unroll 120
        for (int64_t line = 0; line < num_lines; ++line) {
          if (IsFizzBuzzNumber(first_line_number_mod_15 + line)) {
            // Чтобы компилятор генерировал эффективный код,
            // второй и третий параметры должны быть выводимы из констант.
            // Так как цикл развёрнут, значение |line_start|
            // известно на каждой итерации. |start_index| - это constexpr,
            // так что его значение тоже известно.
            IncrementNumber(out, line_start + start_index, OVERFLOW_DIGITS);
          }
          line_start +=
              FizzBuzzLineLength((first_line_number_mod_15 + line) % 15);
        }
        // Так как num_lines кратно 120, правая сторона
        // кратна 8 и это гарантирует, что |out| позже будет
        // выровнен по 8 байтам.
        out += FifteenerBytes() * num_lines / 15;
      };

      for (int64_t i = 0; i < LinesInChunk() / 120; ++i) {
        increment(120);
      }
      increment(LinesInChunk() % 120);
    }

    // Возвращает, нужно ли выводить это число без изменений, то есть оно
    // не кратно 3 и 5.
    static constexpr bool IsFizzBuzzNumber(int64_t number) {
      return number % 3 != 0 && number % 5 != 0;
    }

    // Инкрементирует число, начиная с base[line_start].
    // |base| должна быть выровнена по 8 байтам. Вызывающая сторона должна гарантировать,
    // что количество происходящих переполнений равно |overflow_digits|.
    // Для максимальной производительности |line_start| должна выводиться
    // во время компиляции.
    __attribute__((always_inline))
    static inline void IncrementNumber(char* base,
                                       int64_t line_start,
                                       int overflow_digits) {
      int64_t right_most_digit_to_update_index =
          line_start + DIGITS - 1 - kSuffixDigits;
      // Когда overflow_digits известно во время компиляции, все вызовы IncrementAt,
      // влияющие на одинаковый 8-байтный integer, комбинируются компилятором
      // в одну команду.
      IncrementAt(base, right_most_digit_to_update_index, kIncrementBy);
      #pragma GCC unroll 100
      for (int i = 0; i < overflow_digits; ++i) {
        IncrementAt(base, right_most_digit_to_update_index, -10);
        IncrementAt(base, right_most_digit_to_update_index - 1, 1);
        right_most_digit_to_update_index--;
      }
    }

    // Инкрементирует байт по |index| в |base| на |by|.
    // |base| должна быть выровнена по 8 байтам.
    // Для максимальной производительности |index| и  |by| должны выводиться
    // компилятором из констант.
    __attribute__((always_inline))
    static inline void IncrementAt(char* base, int64_t index, char by) {
      union char_array_int64 {
        char ch[8];
        int64_t int64;
      };
      auto base_as_union = reinterpret_cast(base);
      // Приведённый ниже код работает только в системах little endian.
      static_assert(std::endian::native == std::endian::little);
      // Инкрементируем символ по индексу |index| на |by|. Это работает, потому что
      // мы можем гарантировать, что символ не выполнит переполнение.
      base_as_union[index / 8].int64 +=
        static_cast(by) << ((index % 8) * 8);
    }

    // Возвращает количество цифр, которое будет переполнено при инкременте
    // |prefix| на |kIncrementBy|.
    // Например, если kIncrementBy = 3:
    // OverflowDigits(100) = 0 (нет переполнения цифр)
    // OverflowDigits(108) = 1 (8 переполняется, а 0 инкрементируется на 1)
    // OverflowDigits(198) = 2 (8 переполняется и 9 переполняется)
    static int OverflowDigits(int64_t prefix) {
      int incremented = prefix + kIncrementBy;
      #pragma GCC unroll 2
      for (int i = 0; i < 20; ++i) {
        incremented /= 10;
        prefix /= 10;
        if (incremented == prefix) {
          return i;
        }
      }
      return 20;
    }

    Batch batch_;
  };
};

} // пространство имён

int main() {
  OutputHandler output_handler;

  [&](std::index_sequence){
    (Run::Execute(output_handler), ...);
  }(std::make_index_sequence<18>());

  return 0;
}