Return to Answer

EDIT: I worked on a threaded parser. Simple implementation below. I am far from a concurrency expert, so bear with me. No locks or atomics. Doesn't need it. "Wait-free", is that what they say? Memory locality / bus or L1/L2/L3 cache size for hashmap are the limits to scalability -- not sure.

Pretty happy with sub 400ms? Any feedback on the concurrent code warmly welcome. Is there a cleaner way to pass and/or return from the threads?

#include "flat_hash_map/bytell_hash_map.hpp"
#include "os/bch.hpp"
#include "os/fs.hpp"
#include <cstdint>
#include <iostream>
#include <string>
#include <thread>
#include <vector>

template <typename T>
T yahtzee_upper(const std::string& filename) {
  auto mfile     = os::fs::MemoryMappedFile{filename};
  auto max_total = std::int64_t{0};

  const unsigned n_threads = std::thread::hardware_concurrency();
  auto           threads   = std::vector<std::thread>{};
  auto maps = std::vector<ska::bytell_hash_map<T, T>>{n_threads, ska::bytell_hash_map<T, T>{}};
  std::cout << n_threads << " threads" << "\n";
  {
    auto tim = os::bch::Timer("spawn");
    auto        chunk = std::ptrdiff_t{(mfile.end() - mfile.begin()) / n_threads};
    const char* end   = mfile.begin();
    for (unsigned i = 0; end != mfile.end() && i < n_threads; ++i) {
      const char* begin = end;
      end               = std::min(begin + chunk, mfile.end());

      while (end != mfile.end() && *end != '\n') ++end; // ensure we have a whole line
      if (end != mfile.end()) ++end;                    // one past the end

      threads.push_back(std::thread(
          [&maps](int id, const char* begin, const char* const end) {
            auto& map = maps[id];

            const char* curr = begin;
            auto        val  = std::int64_t{0};
            while (curr != end) {
              if (*curr == '\n') {
                map[val] += val;
                val = 0;
              } else {
                val = val * 10 + (*curr - '0');
              }
              ++curr;
            }
          },
          i, begin, end));
    }
  }
  {
    auto tim = os::bch::Timer("work");
    for (auto&& t: threads) t.join();
  }
  {
    auto tim       = os::bch::Timer("finalise");
    auto final_map = ska::bytell_hash_map<T, T>{};

    for (auto&& m: maps) {
      for (auto p: m) {
        std::int64_t total = final_map[p.first] += p.second;
        if (total > max_total) max_total = total;
      }
    }
  }
  return max_total;
}

int main(int argc, char* argv[]) {
  if (argc < 2) return 1;
  std::cout << yahtzee_upper<std::uint64_t>(argv[1]) << '\n'; // NOLINT
  return 0;
}

EDIT: I worked on a threaded parser. Simple implementation below. I am far from a concurrency expert, so bear with me. No locks or atomics. Doesn't need it: Embarrassingly parallel? Memory locality / bus or L1/L2/L3 cache size for hashmap are the limits to scalability -- not sure.

Pretty happy with sub 400ms? Any feedback on the concurrent code warmly welcome.

#include "flat_hash_map/bytell_hash_map.hpp"
#include "os/bch.hpp"
#include "os/fs.hpp"
#include <cstdint>
#include <iostream>
#include <string>
#include <thread>
#include <vector>

template <typename T>
T yahtzee_upper(const std::string& filename) {
  auto mfile     = os::fs::MemoryMappedFile{filename};
  auto max_total = std::int64_t{0};

  const unsigned n_threads = std::thread::hardware_concurrency();
  auto           threads   = std::vector<std::thread>{};
  auto maps = std::vector<ska::bytell_hash_map<T, T>>{n_threads, ska::bytell_hash_map<T, T>{}};
  std::cout << n_threads << " threads"
            << "\n";
  {
    auto tim = os::bch::Timer("spawn");
    auto        chunk = std::ptrdiff_t{(mfile.end() - mfile.begin()) / n_threads};
    const char* end   = mfile.begin();
    for (unsigned i = 0; end != mfile.end() && i < n_threads; ++i) {
      const char* begin = end;
      end               = std::min(begin + chunk, mfile.end());

      while (end != mfile.end() && *end != '\n') ++end; // ensure we have a whole line
      if (end != mfile.end()) ++end;                    // one past the end

      threads.push_back(std::thread(
          [](const char* begin, const char* const end, ska::bytell_hash_map<T, T>& map) {

            const char* curr = begin;
            auto        val  = std::int64_t{0};
            while (curr != end) {
              if (*curr == '\n') {
                map[val] += val;
                val = 0;
              } else {
                val = val * 10 + (*curr - '0');
              }
              ++curr;
            }
          },
          begin, end, std::ref(maps[i])));
    }
  }
  {
    auto tim = os::bch::Timer("work");
    for (auto&& t: threads) t.join();
  }
  {
    auto tim       = os::bch::Timer("finalise");
    auto final_map = ska::bytell_hash_map<T, T>{};

    for (auto&& m: maps) {
      for (auto p: m) {
        std::int64_t total = final_map[p.first] += p.second;
        if (total > max_total) max_total = total;
      }
    }
  }
  return max_total;
}

int main(int argc, char* argv[]) {
  if (argc < 2) return 1;
  std::cout << yahtzee_upper<std::uint64_t>(argv[1]) << '\n'; // NOLINT
  return 0;
}

EDIT: I worked on a threaded parser. Simple implementation below. I am far from a concurrency expert, so bear with me. No locks or atomics. Doesn't need it. "Wait-free", is that what they say? Memory locality / bus are the limits to scalability.

EDIT: I worked on a threaded parser. Simple implementation below. I am far from a concurrency expert, so bear with me. No locks or atomics. Doesn't need it. "Wait-free", is that what they say? Memory locality / bus or L1/L2/L3 cache size for hashmap are the limits to scalability -- not sure.

Output and simple performance stats below (baseline from above is 1.7s single threaded):

Output and simple performance stats below (baseline from above is 1.7s single threaded for the same work, and 140ms of "overhead" to spin through the mmap'd file with no work):