Optimerad 1BRC i C++
Det här är en direkt fortsättning på förra artikeln, där jag implementerade en rättfram enkel lösning till 1BRC i Modern C++. I denna artikel, implementerar jag en betydligt mer optimerad lösning, för att se hur vi kan minska ned totaltiden. Programmet realiserar följande optimerings-steg:
- Memory-mapping of the weather data file
- Multithreaded aggregation of the in-memory weather data
- Thread private heap storage for the hash-map of each worker thread
- Fast parsing from text to floating-point value
(1) Memory-mapping of the weather data file
Det första optimeringssteget är att flytta in hela väderdata-filen i minnet. Detta kallas
för minnes-mappad fil (memory-mapped file). Det här en kolossalt vanligt teknik, och används
av kompilatorer och andra system. T.ex., laddas en Android applikation från en *.apk på
detta sätt.
Hur det fungerar rent tekniskt är i korthet följande. Virtuell-minnes system har till uppgift att avbilda ett stort adressrum på tillgängligt fysiskt minne (RAM). Adressrummet är indelat i ett stort antal segment (VM pages), vanligtvis 4KB.
$ getconf PAGESIZE
4096Varje (Linux) process har en tabell (page table) över vilka segment som finns. För varje segment håller tabellen reda på om den är i bruk eller ej, om den finns i minnet eller ej, om den har modifierats eller ej, om den har sparats på disk (swap partition) eller ej, om den bara är läsbar eller både läs- och skrivbar, med mera.
Varje access av data översätts från 64-bit till ett segment index i tabellen plus ett offset inuti segmentet. Detta kallas för address translation. Resultatet cachas, för att snabba upp återkommande accesser till samma adress.
Om segmentet inte finns i minnet, så måste detta läsas in. Är det första gången, så kan segmentet skapas och fyllas med nollor, men vanligtvis har det sparats ut på swap partitionen och måste då läsas in. Ett segment sparas till swap:en om det har modifierats och inte längre får plats i minnet, för att ett annat segment behöver dess plats.
En läsning från swap är givetvis extremt optimerad och snabb, vilket är vad man vill ha även för
en vanlig fil. Via system-anropet mmap så ändrar man i segment tabellen, så att ett antal sammanhängande
segment pekar på filen i fråga, i stället för in i swap partitionen.
Först öppnar man filen med system-anropet open, samt läser av hur stor den är, sedan anropas mmap.
Efter det kan filen stängas, eftersom den finns mappad i adressrummet.
void load(std::string const& filename) {
auto fd = ::open(filename.c_str(), O_RDONLY);
if (fd == -1) throw std::invalid_argument{"cannot open "s + filename};
num_bytes = std::filesystem::file_size(filename);
payload = ::mmap(nullptr, num_bytes, PROT_READ, MAP_PRIVATE, fd, 0);
::close(fd);
}Argumenten till mmap är i tur och ordning:
- Startadress, eller 0 (
nullptr) vilket innebär att operativet väljer - Storlek på minnesarean
- Flaggor som anger om arean ska vara läsbär, skrivbar och/eller exekverbar
- Flagga som anger om arean är privat eller ska delas med andra processer (shared memory)
- Filindex (file descriptor)
- Startposition i filen
I min lösning har jag paketerat in detta i en klass, samt exponerar arean som en std::string_view.
auto file = io::MemoryMappedFile{filename};
auto const data = file.data();
cout << "loaded " << data.size() * 1E-6 << " MB\n";När första byte:en läses, uppstår ett segment-fel (page fault), vilket får virtuell-minnes systemet att läsa in det första segmentet från filen. I praktiken används pre-fetching, vilket innebär att fler än ett segment läses in, vilket ju är ännu bättre.
Här har du hela koden, i form av en inkluderings-fil.
memory-mapped-file.hxx
#pragma once
#include <string>
#include <string_view>
#include <filesystem>
#include <fcntl.h>
#include <unistd.h>
#include <sys/mman.h>
namespace ribomation::io {
using namespace std::string_literals;
class MemoryMappedFile {
void* payload;
size_t num_bytes;
void load(std::string const& filename) {
auto fd = ::open(filename.c_str(), O_RDONLY);
if (fd == -1) throw std::invalid_argument{"cannot open "s + filename};
num_bytes = std::filesystem::file_size(filename);
payload = ::mmap(nullptr, num_bytes, PROT_READ, MAP_PRIVATE, fd, 0);
::close(fd);
}
public:
MemoryMappedFile(std::string const& filename) {
load(filename);
}
~MemoryMappedFile() {
::munmap(payload, num_bytes);
}
auto data() const -> std::string_view {
return {reinterpret_cast<const char*>(payload), num_bytes};
}
};
}(2) Multithreaded aggregation of weather data
När väl data-filen finns i adressrummet, i praktiken som en lång array av bytes, blir det också enkelt att dela upp analysarbetet i flera oberoende fragment. Först tar vi reda på hur många (virtuella) processorer det finns (processing units).
auto const num_workers = std::thread::hardware_concurrency();Efter det, kan vi stycka upp och fördela fragment. En något förenklad kod ser ut som följer nedan. Notera att vi behöver justera snittets ändpunkt så att det slutar på ett nyrads-tecknen.
Stockholm;10.5<NL>
Uppsala;2.3<NL>auto file = io::MemoryMappedFile{filename};
auto const data = file.data();
auto chunk_size = data.size() / num_workers;
auto start = 0UL;
for (auto id = 0U; id < num_workers; ++id) {
auto size = chunk_size;
while (data[start + size] != '\n') ++size; //divide at newline
auto chunk = data.substr(start, size);
start += size + 1; //start at next row
//...initialize worker...
}En aggregerings-tråd (worker thread) representeras som en klass, eftersom den måste kunna behålla
resultatet tills efter själva OS tråden har terminerat. Här visar jag en något förenklad kod, och
återkommer i nästa sektion med de utelämnade delarna. Kom ihåg att class och struct är samma
sak i C++, med den enda skillnaden att default visibility är private i class och public i struct.
struct Worker {
string_view chunk;
pm::unordered_map<string_view, br::Aggregation> data;
Worker(string_view chunk_) : chunk{chunk_}, data{} {
data.reserve(500);
}
void run() {
try {
auto start = 0UL;
while (start < chunk.size()) {
auto m = br::extract(chunk, start);
data[m.station] += m.temperature;
}
} catch (std::exception const& x) {
std::cerr << "[WORKER] err: " << x.what() << "\n";
}
}
};Funktionen extract extraherar nästa par av stationsnamn och temperatur, från trådens eget fragment av
hela minnesarean, som ett objekt av klassen Measurement. Temperaturvärdet aggregeras (+=) i ett
Aggregation objekt, som utgör värde-objekt till ett tillhörande stations-namn i en hash-map
(unordered_map<string_view, Aggregation>).
Här kan det vara på sin plats att påpeka att ett std::string_view objekt består av
en pekare till startpositionen plus ett heltal som anger antalet bytes från den positionen. Detta gör
denna datatyp sällsynt användbar för uppgifter då man behöver stycka upp en minnesarea i mindre bitar
utan att behöva allokera något extra minne. Med andra ord, varje stations-namn pekar ut ett litet utsnitt
av data-filen som lästes in via memory-mapped I/O.
Så här ser klassen Aggregation ut med tillhörande aggregerings-operatorer plus utskrifts-operator.
struct Aggregation {
unsigned count = 0;
double sum = 0;
double min = std::numeric_limits<double>::max();
double max = std::numeric_limits<double>::min();
void operator+=(double t) {
++count;
sum += t;
min = std::min(min, t);
max = std::max(max, t);
}
void operator+=(Aggregation const& a) {
count += a.count;
sum += a.sum;
min = std::min(min, a.min);
max = std::max(max, a.max);
}
};
inline auto operator<<(std::ostream& os, Aggregation const& a) -> std::ostream& {
return os << std::format("{:+.2f}C, {:+.1f}/{:+.1f} ({:Ld})",
(a.sum / a.count), a.min, a.max, a.count);
}Så här ser klassen Measurement ut med tillhörande funktioner. Konverteringen av temperatur till flyttal,
kommer jag att ändra i näst-nästa sektion av denna artikel.
struct Measurement {
string_view station{};
double temperature{};
};
inline auto extract(string_view chunk, unsigned long& start) -> Measurement {
auto semi_colon = chunk.find(';', start);
auto station_size = semi_colon - start;
auto station = chunk.substr(start, station_size);
start += station_size + 1;
auto nl = chunk.find('\n', start);
auto temp_text = chunk.substr(start, nl - start);
start += temp_text.size() + 1;
double temperature = std::stod(temp_text.data()); //stod = string-to-double
return {station, temperature};
}
inline auto operator<<(ostream& os, Measurement const& m) -> ostream& {
return os << "Measurement{" << m.station << ", " << m.temperature << "}";
}Samtliga worker objekt huserar i en lista (std::vector). Vilket också gäller för alla
tråd-objekt. Här använder jag klassen std::jthread, som infördes i C++20, vars destruktor
väntar på att OS tråden terminerar. Genom att deklarera en lista av trådar i ett inre block
så väntar huvudprogrammet vid blockets slut på att alla trådar terminerat. Så här kan det
i princip se ut:
auto const T = std::thread::hardware_concurrency();
auto workers = vector<br::Worker>{};
workers.reserve(T);
//...split payload into chunks...
{
auto threads = std::vector<std::jthread>{};
threads.reserve(T);
for (auto id = 0U; id < T; ++id) {
threads.emplace_back(&br::Worker::run, &workers[id]);
}
}
//all done, do post-processing...Funktionen emplace_back, tar konstruktor-argumenten till element-klassen och initierar objektet
direkt på det interna minnesblocket. I exemplet ovan skapas ett std::jthread, vars konstruktor är
kolossalt flexibel. Här tar denna konstruktor först pekaren till en medlemsfunktion (run) i klassen
Worker, plus pekaren till objektet självt i fråga. I och med detta, startas en OS thread, vilken
sen kommer att anropa run funktionen, med objektet ifråga som this.
(3) Thread-private heap-storage
I princip, har vi nu en lösning som fungerar. Emellertid, finns det ett underliggande problem som inte är helt självklart att förstå sig på. Varje worker thread har en egen hash-map.
struct Worker {
string_view chunk;
unordered_map<string_view, Aggregation> data;
//...
};
//...
auto workers = vector<Worker>{};
workers.reserve(T);En dylik består internt av en array (buckets), vilken tekniskt ligger i den interna array
som utgörs av en std::vector. Så långt inga problem. Emellertid, när ett element stoppas in
så behöver det allokeras ett objekt. Varje bucket i en hash-tabell utgör en länkad lista. Nu
kan man förvisso hoppas att listan bara består av ett objekt. Givet att hash-funktionen sprider
ut nycklarna jämt över alla buckets.
Var allokeras då länk-objektet? Jo, på den globala heap:en. Om flera trådar gör detta samtidigt, måste de vänta in varandra, så att bara en tråd i taget kan allokera ett minnes-block. Detta realiseras i system-heap:en med ett mutex-lock.
Är det många trådar som samtidigt vill allokera (eller lämna tillbaka) minnesblock så kan vi få en så kallad synchronization hot-spot, också kallad lock contention. Det innebär att trådarna spenderar mer tid på att vänta på ett minnesblock än att göra nyttigt arbete.
Standardbiblioteket för GCC (glibc), har förvisso en smart relaxering i form av allocation arenas.
Varje arena har sitt eget mutex-lock. Första gången en tråd allokerar minne med malloc
(det är vad operator new gör), väljs första lediga arena, vilken sen används fortsättningsvis
av tråden. På detta vis, sprids allokeringar ut på flera mutex-lock så att man mildrar eventuell
lock contention.
Ett bättre sätt är att låta tråden ha sin egen privata heap. Att implementera en fullfjädrad
minneshanterare är inte gjort på en kafferast. Emellertid, finns std::pmr sedan C++17, vilket gör
att man kan skaffa en heap på en kafferast. 😉
PMR står för Polymorphic Memory Resource, typ mångformig minnes resurs. Det stora fördelen med PMR
är att det fungerar med container-typerna i biblioteket, såsom vector och unordered_map. Det
är två saker man måste göra något annorlunda. (1) använda namespace std::pmr, (2) skapa en egen
heap och skicka in den när man skapar ett nytt container objekt.
using std::pmr::unordered_map;
auto heap = ...;
auto data = unordered_map<KeyType, ValueType>{&heap};Det vi skulle vilja; är att lägga en private heap som medlems-variabel i klassen Worker. Så att
vi på nästa rad kan initiera hash-ma:en.
struct Worker {
string_view chunk;
Heap heap{...stuff...}
pm::unordered_map<string_view, br::Aggregation> data{&heap};
};Dessvärre går detta inte, eftersom en delkomponent av heap objektet har sin copy constructor raderad,
vilket också medför att dess move constructor inte finns. Med andra ord, kan vi varken flytta eller
kopiera ett heap-objekt. Denna egenskap propageras sen till klasser som har en dylik som medlem.
Vi får kompileringsfel, när vi försöker skapa Worker objekt. Trist ☹️
Därför har jag lyft ut heap:en och representerat den med en egen klass. Själva minnesarean är på
100,000 bytes och hanteras av en monotonic_buffer_resource. Om minnet tar slut, så kastas ett
exception null_memory_resource. Som alternativ, kan man välja att allokera från system-heap:en,
genom skicka in new_delete_resource. Själva heap-algoritmen hanteras av en pool_resource. Vi behöver
inte en trådsäker variant, eftersom varje tråd har sin egen heap, därför använder vi
unsynchronized_pool_resource. Alternativet är en synchronized_pool_resource.
namespace pm = std::pmr;
using std::array;
struct MapHeap {
array<unsigned char, 100'000> storage{};
pm::monotonic_buffer_resource buffer{storage.data(), storage.size(),
pm::null_memory_resource()};
pm::unsynchronized_pool_resource heap{&buffer};
};I huvudprogrammet ser det då ut (ungefär) på följande sätt.
auto const T = std::thread::hardware_concurrency();
auto map_heap = vector<br::MapHeap>(T);
auto workers = vector<br::Worker>{};
workers.reserve(T);
//...
for (auto id = 0U; id < T; ++id) {
//...
auto chunk = data.substr(start, size);
//...
workers.emplace_back(chunk, map_heap[id]);
}
{
auto threads = std::vector<std::jthread>{};
threads.reserve(T);
for (auto id = 0U; id < num_workers; ++id) {
threads.emplace_back(&br::Worker::run, &workers[id]);
}
}
//...post processing...Samt klassen Worker ser ut på följande sätt. Notera att andra konstruktor-argumentet
utgörs av ett heap-objekt. Detta skickas in som en referens i koden ovan
(workers.emplace_back(chunk, map_heap[id]);).
struct Worker {
string_view chunk;
pm::unordered_map<string_view, br::Aggregation> data;
Worker(string_view chunk_, MapHeap& mapHeap) : chunk{chunk_}, data{&mapHeap.heap} {
data.reserve(500);
}
void run() {...}
};Hur stor heap?
I klassen MapHeap finns en minnesarea på 100.000 bytes. Hur vet vi att detta räcker?
struct MapHeap {
array<unsigned char, 100'000> storage{};
...
};Svaret är att det vet vi inte, om vi inte undersöker detta närmare med lite experimentell programmering. Tanken är då att läsa in väderdata till en hash-map med privat heap och använda speciella versioner av PMR klasserna, som håller reda på hur mycket minne som allokeras.
Först skapar vi en sub-klass till std::pmr::monotonic_buffer_resource.
struct SpyBuffer : std::pmr::monotonic_buffer_resource {
using super = std::pmr::monotonic_buffer_resource;
unsigned long currentBytes{};
unsigned long maxBytes{};
explicit SpyBuffer(void* data, size_t size, memory_resource* upstream)
: super(data, size, upstream) {}
~SpyBuffer() override { cout << *this << "\n"; }
friend auto operator<<(std::ostream& os, SpyBuffer const& h) -> std::ostream& {
return os << "[SPY-BUF] current=" << h.currentBytes << " bytes, max="
<< h.maxBytes << " bytes";
}
protected:
void* do_allocate(size_t numBytes, size_t alignment) override {
currentBytes += numBytes;
maxBytes += numBytes;
return super::do_allocate(numBytes, alignment);
}
void do_deallocate(void* ptr, size_t numBytes, size_t alignment) override {
currentBytes -= numBytes;
super::do_deallocate(ptr, numBytes, alignment);
}
};Sen, på liknande sätt skapar vi en sub-klass till std::pmr::unsynchronized_pool_resource.
struct SpyHeap : std::pmr::unsynchronized_pool_resource {
using super = std::pmr::unsynchronized_pool_resource;
unsigned long currentBytes{};
unsigned long maxBytes{};
explicit SpyHeap(memory_resource* upstream) : super(upstream) {}
~SpyHeap() override { cout << *this << "\n"; }
friend auto operator<<(std::ostream& os, SpyHeap const& h) -> std::ostream& {
return os << "[SPY-HEAP] current=" << h.currentBytes << " bytes, max="
<< h.maxBytes << " bytes";
}
protected:
void* do_allocate(size_t numBytes, size_t alignment) override {
currentBytes += numBytes;
maxBytes += numBytes;
return super::do_allocate(numBytes, alignment);
}
void do_deallocate(void* ptr, size_t numBytes, size_t alignment) override {
currentBytes -= numBytes;
super::do_deallocate(ptr, numBytes, alignment);
}
};Estimerings-programmet ser ut så här. Vi tar in filnamnet från kommandoraden, läser in hela innehållet i minnet och sen skapar en privat heap för vår aggregerings hash-map. När det är klart skriver våra spy PMR objekt ut hur det gick.
int main(int argc, char** argv) {
auto filename = rm::getFilename(argc, argv);
cout << "filename: " << filename << "\n----\n";
auto file = io::MemoryMappedFile{filename};
auto chunk = file.data();
cout << "loaded " << chunk.size() * 1E-6 << " MB, from " << filename << "\n";
using Result = std::pmr::unordered_map<std::string_view, br::Aggregation>;
{
constexpr auto SIZE = 100'000;
cout << "heap size: " << SIZE << " bytes\n";
auto storage = std::array<unsigned char, SIZE>{};
auto buffer = pm::SpyBuffer{storage.data(), storage.size(),
std::pmr::null_memory_resource()};
auto heap = pm::SpyHeap{&buffer};
auto result = Result{&heap};
result.reserve(500);
auto start = 0UL;
while (start < chunk.size()) {
auto m = br::extract(chunk, start);
result[m.station] += m.temperature;
}
}
}Kör vi programmet med för liten heap (10.000), så kraschar det.
oaded 13.8008 MB, from data/weather-data-1M.csv
heap size: 10000 bytes
terminate called after throwing an instance of 'std::bad_alloc'
what(): std::bad_alloc
Process finished with exit code 6Ökar vi på storleken på heap:en (100.000), så går det vägen.
loaded 13.8008 MB, from data/weather-data-1M.csv
heap size: 100000 bytes
[SPY-HEAP] current=0 bytes, max=30456 bytes
[SPY-BUF] current=0 bytes, max=92192 bytes
Process finished with exit code 0Storleken på heap:en är oberoende av data-mängden, eftersom det styrs av antalet väderstationer. Här "vet" vi att de är 413 stycken. Det här syns tydligt då vi kör samma program med 100 miljoner rader.
loaded 1379.6 MB, from data/weather-data-100M.csv
heap size: 100000 bytes
[SPY-HEAP] current=0 bytes, max=30456 bytes
[SPY-BUF] current=0 bytes, max=92192 bytes
Process finished with exit code 0(4) Fast parsing of floating-point text
Efter att ha googla litet, hittade jag en implementation för snabb parsning av text till flyttal.
Biblioteket är header-only, vilket innebär att det är en fil att kopiera och inkludera. Jag länkar
till bibliotekets GitHub repo, i länksamlingen sist i denna artikel.
Funktionen används på en plats i programmet och det är i funktionen extract. Så här ser denna ut.
#include "fast_double_parser.h"
inline auto extract(string_view chunk, unsigned long& start) -> Measurement {
auto semi_colon = chunk.find(';', start);
auto station_size = semi_colon - start;
auto station = chunk.substr(start, station_size);
start += station_size + 1;
auto nl = chunk.find('\n', start);
auto temp_text = chunk.substr(start, nl - start);
start += temp_text.size() + 1;
double temperature{};
[[maybe_unused]] auto ptr = fast_double_parser::parse_number(temp_text.data(), &temperature);
return {station, temperature};
}Sammanställning av delresultaten och utskrift
Efter att alla Worker threads har terminerat, återstår två uppgifter. Dels, att aggregera ihop all delresultat, samt dels att skriva ut dessa sorterade på stationsnamnen.
Eftersom listan med Worker objekt överlever efter att respektive thread terminerat, så kan vi loop:a igenom dessa och flytta data till en enda container.
auto result = unordered_map<string_view, br::Aggregation>{};
for (auto&& w: workers) {
for (auto&& [station, aggr]: w.data) {
result[station] += aggr;
}
}Som vi såg i förra artikeln, så är det snabbare att hälla över data till en vector och sortera denna,
är att flytta data till en tree-map (std::map). Så här ser sista steget ut
auto sorted = vector<pair<string_view, br::Aggregation>>{result.begin(), result.end()};
std::sort(sorted.begin(), sorted.end(), [](auto&& lhs, auto&& rhs) {
return lhs.first < rhs.first;
});
for (auto&& [station, aggr]: sorted) {
cout << station << ": " << aggr << "\n";
}Huvudprogrammet
Huvudprogrammet main, är uppdelad i olika funktionsanrop, så att det enkelt framgår hur det är strukturerat.
inline void
split_chunks(unsigned num_workers, string_view& data,
vector<br::Worker>& workers, vector<br::MapHeap>& map_heap);
inline void
launch_workers(unsigned num_workers, vector<br::Worker>& workers);
inline auto
collect_result(vector<br::Worker>& workers) -> unordered_map<string_view, br::Aggregation>;
inline void
sort_print(unordered_map<string_view, br::Aggregation>& result);
int main(int argc, char** argv) {
auto filename = rm::getFilename(argc, argv);
cout << "filename: " << filename << "\n";
rm::elapsed([filename]() {
auto file = io::MemoryMappedFile{filename};
auto data = file.data();
cout << "loaded " << data.size() * 1E-6 << " MB\n";
auto const T = std::thread::hardware_concurrency();
auto mapHeap = vector<br::MapHeap>(T);
auto workers = vector<br::Worker>{};
workers.reserve(T);
split_chunks(T, data, workers, mapHeap);
launch_workers(T, workers);
auto result = collect_result(workers);
sort_print(result);
});
}Exekvering
OK, om du hängt med hela vägen tills hit; så är du säkert kolossalt spänd på att få veta om och hur mycket snabbare blev detta program, jämfört med det tämligen korta programmet i förra artikeln.
Först kompilerar vi
$ g++ --version
g++ (Ubuntu 13.2.0-4ubuntu3) 13.2.0
$ g++ -std=c++23 -O3 -Wall -I src/cxx/util -I src/cxx/lib -I src/cxx/fast \
-o binary/calculate-average-fast \
src/cxx/util/util.cxx src/cxx/fast/calculate-average-fast.cxxSen kör vi programmet
$ ./binary/calculate-average-fast -f data/weather-data-10M.csv
filename: data/weather-data-10M.csv
loaded 137.967 MB
chunk size: 17.2459 MB
launching 8 worker threads...
collecting results
------
Abha: +18.06C, -22.1/+52.9 (24124)
Abidjan: +25.92C, -14.3/+66.6 (24234)
...
Zürich: +9.27C, -34.3/+51.8 (24223)
Ürümqi: +7.35C, -40.5/+46.4 (24012)
İzmir: +17.92C, -20.6/+56.8 (24241)
------
Elapsed time: 1.352 secondsOm vi kör det icke optimerade programmet får vi följande sluttid
$ ./binary/calculate-average -f data/weather-data-10M.csv
filename: data/weather-data-10M.csv
----
Abha: +18.06C, -22.1/+52.9 (24124)
...
İzmir: +17.92C, -20.6/+56.8 (24241)
------
# records: 10000000
Elapsed time: 3.687 secondsEn förbättring med 2,3 sekunder, för en måttlig datamängd på 10 miljoner textrader. Inte illa. 😀
Sammanställning
Av tabellen nedan, framgår att vi med vår optimerade version kunnat dubblera hastigheten, dvs förfluten tid halverad. Utvecklingstiden för den optimerade versionen, var långt mer än dubbla tiden jämfört med den enkla versionen. Alltså; nerd sniped! 😵💫
| Rows | Simple | Optimized | Speedup |
|---|---|---|---|
| 1M | 0.56 s | 0.26 s | 215% |
| 10M | 5.45 s | 2.26 s | 241% |
| 100M | 59.81 s | 29.84 s | 200% |
| 1000M | 655.99 s | 329.68 s | 199% |