Parallel lezen van gegevens in Firebird

D.Simonov, V.Horsun

versie 1.0.5 van 05.12.2023

Dit materiaal is gesponsord en gemaakt met de sponsoring en ondersteuning van IBSurgeon www.ib-aid.com, leverancier van HQbird (geavanceerde distributie van Firebird) en leverancier van prestatieoptimalisatie, migratie en technische ondersteuningsdiensten voor Firebird.

Het materiaal is gelicenseerd onder de Public Documentation License https://www.firebirdsql.org/file/documentation/html/en/licenses/pdl/public-documentation-license.html

Gerelateerde materialen:

Gratis boek “Detailed New Featuires Of Firebird 5”: het is beschikbaar als HTML en als PDF (119 pagina’s).

Voorwoord

Firebird 5.0 introduceerde de mogelijkheid om parallellisme te gebruiken bij het maken van een back-up met behulp van het gbak-hulpprogramma, naast andere parallelle functies. Aanvankelijk verscheen deze functie in HQbird 2.5, daarna in HQbird 3.0 en 4.0, en vervolgens werd deze overgezet naar Firebird 5.0.

In dit artikel bespreken we de functies die worden gebruikt bij het maken van parallelle back-ups binnen het gbak-hulpprogramma. We laten ook zien hoe ze kunnen worden gebruikt in uw toepassingen voor het parallel lezen van gegevens.

Het is belangrijk op te merken dat we het hier niet hebben over het parallel scannen van tabellen binnen de Firebird-engine bij het uitvoeren van SQL-query’s, maar over het lezen van gegevens binnen uw toepassing in parallelle stromen.

Voorbeeldhulpprogramma FBCSVExport

Om parallel lezen van gegevens uit het Firebird DBMS te demonstreren, is een voorbeeldhulpprogramma geschreven dat gegevens uit een of meer tabellen exporteert naar CSV-formaat.

De beschrijving en de open source-code zijn hier te vinden: https://github.com/IBSurgeon/FBCSVExport.git

Zoals u kunt zien in de beschrijving van het hulpprogramma en in het onderstaande artikel, is het door parallelle verwerking mogelijk om gegevens te exporteren en andere parallelle bewerkingen 2-10x sneller uit te voeren dan in 1 thread (afhankelijk van de hardware).

Voor vragen kunt u contact opnemen met [email protected].

Parallel lezen

Laten we nadenken over hoe we gegevens uit meerdere tabellen parallel kunnen lezen. Zoals u weet, staat Firebird toe om query’s alleen parallel uit te voeren als elke query in een afzonderlijke verbinding wordt uitgevoerd.

Laten we een pool van werkthreads maken. De hoofdthread van de toepassing is ook een werkthread, dus het aantal extra werkthreads moet N - 1 zijn, waarbij N het totale aantal parallelle werkers is. Elke werkthread zal zijn eigen verbinding en transactie gebruiken.

Het eerste probleem: hoe zorgen we voor consistentie van de gelezen gegevens?

Consistent gegevens lezen

Aangezien elke werkthread zijn eigen verbinding en zijn eigen transactie gebruikt, ontstaat het probleem van inconsistente reads - als de tabel gelijktijdig door andere gebruikers wordt gewijzigd, kunnen de gelezen gegevens inconsistent zijn. In single-threaded modus gebruikt gbak een transactie met SNAPSHOT-isolatiemodus, waardoor het mogelijk is om consistente informatie te lezen bij de start van de SNAPSHOT-transactie. Maar hier hebben we meerdere transacties en het is noodzakelijk dat ze dezelfde “snapshot” zien om dezelfde onveranderlijke gegevens te lezen.

Het mechanisme voor het creëren van een gedeelde snapshot voor verschillende transacties met de SNAPSHOT-isolatiemodus werd geïntroduceerd in Firebird 4.0 (oorspronkelijk in HQBird 2.5, maar in Firebird 4.0/HQBird 4.0 is het eenvoudiger en efficiënter). Er zijn twee manieren om een gedeelde snapshot te creëren:

Met SQL

haal het snapshotnummer op van de hoofdtransactie (die is gestart in de hoofdwerkthread).

sql

SELECT RDB$GET_CONTEXT('SYSTEM', 'SNAPSHOT_NUMBER') FROM RDB$DATABASE

start andere transacties met de volgende SQL:

sql

SET TRANSACTION SNAPSHOT AT NUMBER snapshot_number

waarbij snapshot_number het nummer is dat door de vorige query is opgehaald.

Met API

haal het snapshotnummer op van de hoofdtransactie (die is gestart in de hoofdwerkthread) met de functie isc_transaction_info of ITransaction.getInfo met tag fb_info_tra_snapshot_number;
start andere transacties met tag isc_tpb_at_snapshot_number met een verkregen snapshotnummer.

Het voorbeeldhulpprogramma FBCSVExport, evenals gbak, gebruikt de tweede benadering. Deze benaderingen kunnen worden gemengd - bijvoorbeeld, haal het snapshotnummer op met SQL en gebruik het verkregen snapshotnummer om andere transacties te starten met API, of vice versa.

In FBCSVExport verkrijgen we een snapshotnummer met de volgende code:

cpp

ISC_INT64 getSnapshotNumber(Firebird::ThrowStatusWrapper* status, Firebird::ITransaction* tra)
{
    ISC_INT64 ret = 0;
    unsigned char in_buf[] = { fb_info_tra_snapshot_number, isc_info_end };
    unsigned char out_buf[16] = { 0 };

    tra->getInfo(status, sizeof(in_buf), in_buf, sizeof(out_buf), out_buf);

    unsigned char* p = out_buf, * e = out_buf + sizeof(out_buf);
    while (p < e)
    {
        short len = 0;
        switch (*p++)
        {
        case isc_info_error:
        case isc_info_end:
            p = e;
            break;

        case fb_info_tra_snapshot_number:
            len = static_cast(isc_vax_integer(reinterpret_cast<char*>(p), 2));
            p += 2;
            ret = isc_portable_integer(p, len);
            p += len;
            break;
        }
    }
    return ret;
}

Om een transactie te starten met het snapshotnummer gebruiken we de volgende code:

sql

Firebird::AutoDispose tpbWorkerBuilder(fbUtil->getXpbBuilder(&status, Firebird::IXpbBuilder::TPB, nullptr, 0));
tpbWorkerBuilder->insertTag(&status, isc_tpb_concurrency);
tpbWorkerBuilder->insertBigInt(&status, isc_tpb_at_snapshot_number, snapshotNumber);

Firebird::AutoRelease workerTra(
    workerAtt->startTransaction(
        &status,
        tpbWorkerBuilder->getBufferLength(&status),
        tpbWorkerBuilder->getBuffer(&status)
    )
);

Nu zullen de gegevens die uit verschillende verbindingen worden gelezen consistent zijn, zodat we de belasting over werkthreads kunnen verdelen.

Hoe verdelen we de belasting precies over werkthreads? In het geval van een volledige export van alle tabellen of een back-upkopie, is de eenvoudigste optie één werkthread per tabel. Maar met deze benadering hebben we het volgende probleem: als er veel kleine tabellen in een database zijn en één grote tabel, of zelfs slechts één tabel die enorm is, zullen we geen verbetering zien. In dat geval krijgt een thread een grote tabel en blijven de overige threads inactief. Om dit te voorkomen, is het noodzakelijk om een grote tabel in delen te verwerken.


Opmerking	Het onderstaande materiaal is gewijd aan het volledig lezen van tabellen. Als u parallel lezen vanuit een query (of view) wilt organiseren, is een iets andere benadering nodig, die afhangt van de werkelijke gegevens.

Grote tabel splitsen in delen

Stel dat we slechts één grote tabel hebben die we volledig en zo snel mogelijk willen lezen. Het wordt voorgesteld om deze in meerdere delen te splitsen en elk deel vanuit zijn eigen stroom onafhankelijk te lezen. Elke thread moet zijn eigen verbinding met de database hebben.

In dit geval rijzen de volgende vragen:

In hoeveel verwerkingsdelen moet de tabel worden verdeeld?
Wat is de beste manier om de tabel te verdelen wat betreft gegevenstoegang?

Laten we deze vragen in volgorde beantwoorden.

In hoeveel verwerkingsdelen moet de tabel worden verdeeld?

Laten we het ideale scenario aannemen - de server en client zijn toegewijd aan Firebird, dat wil zeggen dat alle CPU’s volledig tot onze beschikking staan. Dan wordt aanbevolen:

a) Gebruik als maximaal aantal parallelle delen het dubbele aantal CPU-kernen op de server. Waarom 2x kernen? We weten zeker dat er vertragingen zullen zijn die verband houden met IO, dus we kunnen wat extra CPU-gebruik toestaan. Dit aantal moet echter als een initiële instelling worden beschouwd; in de praktijk hangt het af van de gegevens.

b) Houd rekening met het aantal kernen op de client: als er op de server veel meer zijn (de gebruikelijke situatie), dan kan het zinvol zijn om het aantal delen van de partitie verder te beperken, om de client niet te overbelasten (deze kan toch niet meer verwerken, en schakelkosten voor stromen verdwijnen niet). Het zal mogelijk zijn om nauwkeuriger te beslissen door de CPU-belasting van de client en server te bewaken - als deze 100% op de client is, maar merkbaar minder op de server, dan is het zinvol om het aantal delen te verminderen.

c) als de client en server dezelfde host zijn, zie dan (a).

Als de client en/of server met iets anders bezig zijn, moet u mogelijk het aantal delen verminderen. Dit kan ook worden beïnvloed door het vermogen van de schijven op de server om veel IO-verzoeken tegelijkertijd te verwerken (bewaak de wachtrijgrootte en responstijd).

Wat is de beste manier om de tabel te verdelen wat betreft gegevenstoegang?

Om een effectieve parallelle verwerking te implementeren, is het belangrijk om een gelijkmatige verdeling van taken over handlers te garanderen en hun onderlinge synchronisatie te minimaliseren. Bovendien moet u onthouden dat synchronisatie van handlers zowel aan de serverzijde als aan de clientzijde kan plaatsvinden. Verschillende handlers moeten bijvoorbeeld niet dezelfde verbinding met de database gebruiken. Een minder voor de hand liggend voorbeeld: het is slecht als verschillende handlers records van dezelfde databasepagina’s lezen. Bijvoorbeeld, wanneer twee handlers even en oneven records lezen - dat is niet effectief. De synchronisatie aan de clientzijde kan optreden tijdens de verdeling van taken, tijdens de verwerking van ontvangen gegevens (toewijzen van geheugen voor resultaten), enzovoort.

Een van de problemen met “eerlijke” partitionering is dat de client niet weet hoe records over pagina’s (en over indexsleutels) zijn verdeeld, hoeveel records of gegevenspagina’s er zijn (voor grote tabellen zou het te lang duren om het aantal records vooraf te tellen).

Laten we kijken hoe gbak dit probleem oplost.

Voor gbak is een werkeenheid een set records van gegevenspagina’s (DP’s) die tot dezelfde pointerpagina (PP) behoren. Aan de ene kant is dit een vrij groot aantal records om de handler bezig te houden zonder dat deze vaak om een nieuw stuk gegevens moet vragen (synchronisatie). Aan de andere kant, zelfs als dergelijke recordsets niet exact dezelfde grootte hebben, zal het mogelijk zijn om werkers relatief gelijkmatig te belasten. Dat wil zeggen, het is heel goed mogelijk dat de ene werker N records van één PP leest en de andere M records, en M kan behoorlijk verschillen van N. Deze benadering is niet ideaal, maar is vrij eenvoudig te implementeren en meestal behoorlijk effectief, tenminste op grote schaal (met tientallen of honderden (of meer) PP’s).

Hoe krijgen we het aantal PP’s (pointerpagina’s) voor een bepaalde tabel? Het is vrij eenvoudig en, belangrijker nog, snel te berekenen uit de RDB$PAGES-tabel:

sql

SELECT RDB$PAGE_SEQUENCE
FROM RDB$PAGES
WHERE RDB$RELATION_ID = ? AND RDB$PAGE_TYPE = 4
ORDER BY RDB$PAGE_SEQUENCE DESC ROWS 1

Vervolgens zouden we eenvoudigweg het aantal PP’s kunnen delen door het aantal werkers en elke werker zijn eigen stuk kunnen geven. Dat is prima voor het scenario waarin parallelle verwerking wordt gedaan door de ontwikkelaar die de gegevensverdeling kent. Maar voor een meer algemeen scenario is er geen garantie dat dergelijke “grote” stukken dezelfde hoeveelheid werk betekenen. We zijn niet geïnteresseerd in de situatie waarin 15 werkers hun werk hebben afgerond en inactief staan, terwijl de 16e zijn 10M records lang leest.

Daarom doet gbak het anders. Er is een werkcoördinator die elke processor 1 PP tegelijk uitgeeft. De coördinator weet hoeveel PP’s er in totaal zijn en hoeveel er al voor werk zijn uitgegeven. Wanneer de werker het lezen van zijn records voltooit, neemt deze contact op met de coördinator voor een nieuw PP-nummer. Dit gaat door totdat de PP’s op zijn (of er actieve werkers zijn). Natuurlijk vereist een dergelijke interactie van werkers met de coördinator synchronisatie. De ervaring leert dat de hoeveelheid werk die één PP geeft, het mogelijk maakt om niet te vaak te synchroniseren. Deze benadering maakt het mogelijk om alle werkers (en dus CPU-kernen) praktisch gelijkmatig met werk te belasten, ongeacht het werkelijke aantal records dat bij elke PP hoort.

Hoe leest de handler records uit zijn PP? Om dit te doen is er, vanaf Firebird 4.0 (voor het eerst verschenen in HQBird 2.5), een ingebouwde functie MAKE_DBKEY(). Met behulp hiervan kun je de RDB$DB_KEY (fysiek recordnummer) verkrijgen voor het eerste record op de opgegeven PP.

En met behulp van deze RDB$DB_KEY worden de benodigde records geselecteerd:

sql

SELECT *
FROM relation
WHERE RDB$DB_KEY >= MAKE_DBKEY(:rel_id, 0, 0, :loPP)
    AND RDB$DB_KEY < MAKE_DBKEY(:rel_id, 0, 0, :hiPP)

Als je bijvoorbeeld loPP = 0 en hiPP = 1 instelt, worden alle records met PP = 0 gelezen, en alleen daarvan.

Nu je een idee hebt van hoe gbak werkt, kun je overgaan naar een beschrijving van de implementatie van het FBCSVExport-hulpprogramma.

Implementatie van het `FBCSVExport`-hulpprogramma

Het FBCSVExport-hulpprogramma is ontworpen om gegevens uit Firebird-databasetabellen te exporteren naar CSV-formaat.

Elke tabel wordt geëxporteerd naar een bestand met de naam .csv. In normale (single-threaded) modus worden gegevens uit tabellen sequentieel geëxporteerd in alfabetische volgorde van tabelnamen.

In parallelle modus worden tabellen parallel geëxporteerd, elke tabel in een aparte thread. Als de tabel erg groot is, wordt deze opgesplitst in delen, en elk deel wordt in een aparte stream geëxporteerd. Voor elk deel van een grote tabel wordt een apart bestand aangemaakt met de naam .csv.partN, waarbij N het deelnummer is.

Wanneer alle delen van een grote tabel zijn geëxporteerd, worden de deeltjesbestanden samengevoegd tot een bestand met de naam .csv.

Er wordt een reguliere expressie gebruikt om te specificeren welke tabellen worden geëxporteerd. Alleen reguliere tabellen kunnen worden geëxporteerd (systeemtabellen, GTT, views, externe tabellen worden niet ondersteund). Reguliere expressies moeten in SQL-syntaxis zijn, dat wil zeggen die welke worden gebruikt in het SIMILAR TO-predikaat.

Om een lijst van geëxporteerde tabellen te selecteren, evenals een lijst van hun PP’s in multi-threaded modus, gebruiken we de volgende query:

sql

SELECT
    R.RDB$RELATION_ID AS RELATION_ID,
    TRIM(R.RDB$RELATION_NAME) AS RELATION_NAME,
    P.RDB$PAGE_SEQUENCE AS PAGE_SEQUENCE,
    COUNT(P.RDB$PAGE_SEQUENCE) OVER(PARTITION BY R.RDB$RELATION_NAME) AS PP_CNT
FROM RDB$RELATIONS R
JOIN RDB$PAGES P ON P.RDB$RELATION_ID = R.RDB$RELATION_ID
WHERE R.RDB$SYSTEM_FLAG = 0 AND
      R.RDB$RELATION_TYPE = 0 AND
      P.RDB$PAGE_TYPE = 4 AND
      TRIM(R.RDB$RELATION_NAME) SIMILAR TO CAST(? AS VARCHAR(8191))
ORDER BY R.RDB$RELATION_NAME, P.RDB$PAGE_SEQUENCE

In single-threaded modus kan deze query worden vereenvoudigd tot

sql

SELECT
    R.RDB$RELATION_ID AS RELATION_ID,
    TRIM(R.RDB$RELATION_NAME) AS RELATION_NAME,
    0 AS PAGE_SEQUENCE,
    1 AS PP_CNT
FROM RDB$RELATIONS R
WHERE R.RDB$SYSTEM_FLAG = 0 AND
      R.RDB$RELATION_TYPE = 0 AND
      TRIM(R.RDB$RELATION_NAME) SIMILAR TO CAST(? AS VARCHAR(8191))
ORDER BY R.RDB$RELATION_NAME

In single-threaded modus worden de waarden van de velden PAGE_SEQUENCE en PP_CNT niet gebruikt; ze zijn toegevoegd aan de query om de uitvoerberichten te uniformeren.

Het resultaat van deze query wordt gevormd tot een vector van structuren:

cpp

struct TableDesc
{
    TableDesc() = default;
    TableDesc(const OutputRecord& rec)
        : releation_id(rec->releation_id)
        , relation_name(rec->relation_name.str, rec->relation_name.length)
        , page_sequence(rec->page_sequence)
        , pp_cnt(rec->pp_cnt)
    {}

    short releation_id;
    std::string relation_name;
    int32_t page_sequence;
    int64_t pp_cnt;
};

Deze vector wordt gevuld met behulp van een functie die is gedeclareerd als:

cpp

std::vector getTablesDesc(
    Firebird::ThrowStatusWrapper* status,
    Firebird::IAttachment* att,
    Firebird::ITransaction* tra,
    unsigned int sqlDialect,
    const std::string& tableIncludeFilter,
    bool singleWorker = true);

De laatste parameter singleWorker schakelt de vulmodus van std::vector om; als singleWorker = true, wordt de query voor single-threaded modus gebruikt, als singleWorker = false, wordt een duurdere en complexere query voor multi-threaded modus gebruikt. Ik zal de implementatie zelf niet geven; deze is vrij eenvoudig en je kunt deze in de broncode van het project zien.

Om een tabel naar CSV-formaat te exporteren, is de klasse CSVExportTable ontwikkeld, die de volgende methoden bevat:

cpp

    void prepare(Firebird::ThrowStatusWrapper* status, const std::string& tableName,
                 unsigned int sqlDialect, bool withDbkeyFilter = false);

    void printHeader(Firebird::ThrowStatusWrapper* status, csv::CSVFile& csv);

    void printData(Firebird::ThrowStatusWrapper* status, csv::CSVFile& csv, int64_t ppNum = 0);

De prepare-methode is bedoeld om een query te bouwen en voor te bereiden die wordt gebruikt om een tabel naar CSV-formaat te exporteren. De innerlijke query wordt anders geconstrueerd, afhankelijk van de parameter withDbkeyFilter. Als withDbkeyFilter = true, wordt de query gebouwd met filtering op het bereik RDB$DB_KEY:

sql

SELECT *
FROM tableName
WHERE RDB$DB_KEY >= MAKE_DBKEY('tableName', 0, 0, ?)
  AND RDB$DB_KEY < MAKE_DBKEY('tableName', 0, 0, ?)

anders wordt een vereenvoudigde query gebruikt:

sql

SELECT *
FROM tableName

De waarde van de parameter withDbkeyFilter wordt ingesteld op true als multi-threaded modus wordt gebruikt en de tabel groot is. We beschouwen de tabel als groot als pp_cnt > 1.

De printHeader-methode is bedoeld om de kop van een CSV-bestand af te drukken (kolomnamen van de tabel).

De printData-methode drukt tabelgegevens af naar een CSV-bestand vanaf PP-paginanummer ppNum, als de query is voorbereid met behulp van een filter op het bereik RDB$DB_KEY, en anders alle tabelgegevens.

Laten we nu kijken naar de code voor een single-threaded modus

cpp

...

// De hoofdverbinding openen
Firebird::AutoRelease att(
    provider->attachDatabase(
        &status,
        m_database.c_str(),
        dbpLength,
        dpb
    )
);

// De hoofdtransactie starten in de isolatiemodus SNAPSHOT
Firebird::AutoDispose tpbBuilder(fbUtil->getXpbBuilder(&status, Firebird::IXpbBuilder::TPB, nullptr, 0));
tpbBuilder->insertTag(&status, isc_tpb_concurrency);

Firebird::AutoRelease tra(
    att->startTransaction(
        &status,
        tpbBuilder->getBufferLength(&status),
        tpbBuilder->getBuffer(&status)
    )
);
// Een lijst van tabellen verkrijgen met behulp van de reguliere expressie in m_filter.
// m_parallel stelt het aantal parallelle threads in; wanneer het gelijk is aan 1,
// wordt een vereenvoudigde query gebruikt om een lijst van tabellen te verkrijgen,
// anders wordt voor elke tabel een lijst van PP's en hun aantal gegenereerd.
auto tables = getTablesDesc(&status, att, tra, m_sqlDialect, m_filter, m_parallel == 1);

if (m_parallel == 1) {
    FBExport::CSVExportTable csvExport(att, tra, fb_master);
    for (const auto& tableDesc : tables) {
        // er is geen reden om hier een bereikfilter RDB$DB_KEY te gebruiken
        csvExport.prepare(&status, tableDesc.relation_name, m_sqlDialect, false);
        const std::string fileName = tableDesc.relation_name + ".csv";
        csv::CSVFile csv(m_outputDir / fileName);
        if (m_printHeader) {
            csvExport.printHeader(&status, csv);
        }
        csvExport.printData(&status, csv);
    }
}

Alles hier is vrij eenvoudig en vereist geen extra uitleg, dus laten we overgaan naar het multi-threaded deel.

Om de export in multi-threaded modus te laten plaatsvinden, is het noodzakelijk om extra m_parallel - 1 werkthreads te creëren. Waarom is het aantal extra threads 1 minder? Ja, omdat de hoofdthread ook gegevens exporteert en gelijk is aan de extra threads. Laten we het gemeenschappelijke deel van de hoofd- en extra stroom verplaatsen naar een aparte functie:

cpp

void ExportApp::exportByTableDesc(Firebird::ThrowStatusWrapper* status, FBExport::CSVExportTable& csvExport, const TableDesc& tableDesc)
{
    // Als tableDesc pp_cnt > 1 heeft, beschrijft het slechts een deel van de tabel, en is het noodzakelijk om een
    // query te bouwen met behulp van een filter op het bereik RDB$DB_KEY.

    bool withDbKeyFilter = tableDesc.pp_cnt > 1;
    csvExport.prepare(status, tableDesc.relation_name, m_sqlDialect, withDbKeyFilter);
    std::string fileName = tableDesc.relation_name + ".csv";

    // Als dit niet het eerste deel van de tabel is, schrijf dit deel dan naar het bestand .csv.part, waarbij
    // N - PP-nummer. Later worden de tabeldelen gecombineerd tot een enkel bestand .csv
    if (tableDesc.page_sequence > 0) {
        fileName += ".part_" + std::to_string(tableDesc.page_sequence);
    }
    csv::CSVFile csv(m_outputDir / fileName);
    // De kop van het CSV-bestand mag alleen in het eerste deel van de tabel worden afgedrukt.
    if (tableDesc.page_sequence == 0 && m_printHeader) {
        csvExport.printHeader(status, csv);
    }
    csvExport.printData(status, csv, tableDesc.page_sequence);
}

Beschrijvingen van tabellen of hun delen bevinden zich in een gemeenschappelijke vector met TableDesc-structuren. Uit deze vector haalt elke werkthread een tabel of het volgende deel. Om dataraces te voorkomen, is het noodzakelijk om de toegang tot de gedeelde bron te synchroniseren. Maar std::vector zelf verandert niet, dus je kunt alleen de gedeelde variabele synchroniseren, die de index in deze vector is. Dit kan eenvoudig worden gedaan met std::atomic als zo’n variabele.

cpp

if (m_parallel == 1) {
    ...
}
else {
    // Het aantal extra werkthreads bepalen
    const auto workerCount = m_parallel - 1;

    // Het snapshotnummer uit de hoofdtransactie verkrijgen
    auto snapshotNumber = getSnapshotNumber(&status, tra);
    // variabele om de uitzondering binnen de thread op te slaan
    std::exception_ptr exceptionPointer = nullptr;
    std::mutex m;
	// atomaire teller
    // is de index van de volgende tabel of een deel ervan
    std::atomic<size_t> counter = 0;
    // pool van werkthreads
    std::vector<std::thread> thread_pool;
    thread_pool.reserve(workerCount);
    for (int i = 0; i < workerCount; i++) {
		// voor elke thread creëren we onze eigen verbinding
        Firebird::AutoRelease workerAtt(
            provider->attachDatabase(
                &status,
                m_database.c_str(),
                dbpLength,
                dpb
            )
        );
		// en onze transactie waaraan we het snapshotnummer doorgeven
        // om een gedeeld snapshot te creëren
        Firebird::AutoDispose tpbWorkerBuilder(fbUtil->getXpbBuilder(&status, Firebird::IXpbBuilder::TPB, nullptr, 0));
        tpbWorkerBuilder->insertTag(&status, isc_tpb_concurrency);
        tpbWorkerBuilder->insertBigInt(&status, isc_tpb_at_snapshot_number, snapshotNumber);

        Firebird::AutoRelease workerTra(
            workerAtt->startTransaction(
                &status,
                tpbWorkerBuilder->getBufferLength(&status),
                tpbWorkerBuilder->getBuffer(&status)
            )
        );
		// een thread creëren
        std::thread t([att = std::move(workerAtt), tra = std::move(workerTra), this,
                       &m, &tables, &counter, &exceptionPointer]() mutable {

            Firebird::ThrowStatusWrapper status(fb_master->getStatus());
            try {
                FBExport::CSVExportTable csvExport(att, tra, fb_master);
                while (true) {
                    // de atomaire teller verhogen
                    size_t localCounter = counter++;
					// als de tabellen of hun delen op zijn, verlaat dan
                    // de oneindige lus en beëindig de thread
                    if (localCounter >= tables.size())
                        break;
                    // een beschrijving van de tabel of een deel ervan verkrijgen
                    const auto& tableDesc = tables[localCounter];
                    // en de export uitvoeren
                    exportByTableDesc(&status, csvExport, tableDesc);
                }
                if (tra) {
                    tra->commit(&status);
                    tra.release();
                }

                if (att) {
                    att->detach(&status);
                    att.release();
                }
            }
            catch (...) {
				// als er een uitzondering optreedt, sla deze dan op voor
                // latere vrijgave in de hoofdthread
                std::unique_lock<std::mutex> lock(m);

exceptionPointer = std::current_exception();
            }
        });
        thread_pool.push_back(std::move(t));
    }

    // export in the main thread
    FBExport::CSVExportTable csvExport(att, tra, fb_master);
    while (true) {
        // increment the atomic counter
        size_t localCounter = counter++;
        if (localCounter >= tables.size())
            break;
        // if the tables or their parts are over, exit
        // from an endless loop
        const auto& tableDesc = tables[localCounter];
        exportByTableDesc(&status, csvExport, tableDesc);
    }
    // wait for the worker threads to complete
    for (auto& th : thread_pool) {
        th.join();
    }
    // if there was an exception in the worker threads, throw it again
    if (exceptionPointer) {
        std::rethrow_exception(exceptionPointer);
    }
    ...

Het enige dat nog rest, is het combineren van de bestanden die zijn gemaakt voor delen van de tabellen tot één bestand voor elk van deze tabellen.

cpp

for (size_t i = 0; i < tables.size(); i++) {
    const auto& tableDesc = tables[i];
    // if the number of PP is greater than 1,
    // then the table is large and there were several parts for it
    if (tableDesc.pp_cnt > 1) {
        // main file for the table
        std::string fileName = tableDesc.relation_name + ".csv";
        std::ofstream ofile(m_outputDir / fileName, std::ios::out | std::ios::app);
        i++;
        for (int64_t j = 1; j < tableDesc.pp_cnt; j++, i++) {
            // files of table parts
            std::string partFileName = fileName + ".part_" + std::to_string(j);
            auto partFilePath = m_outputDir / partFileName;
            std::ifstream ifile(partFilePath, std::ios::in);
            ofile << ifile.rdbuf();
            ifile.close();
            fs::remove(partFilePath);
        }
        ofile.close();
    }
}

Laten we de prestaties van het hulpprogramma meten in single-threaded en multi-threaded modus.

Benchmark van het `FBCSVExport`-hulpprogramma

Laten we eerst kijken naar de resultaten van het vergelijken van multi-threaded en single-threaded exportmodi op de gemiddelde thuiscomputer. === Windows

Besturingssysteem: Windows 10 x64.
Processor: Intel Core i3 8100, 4 kernen, 4 threads.
Geheugen: 16 GB
Schijfsubsysteem: NVME SSD (database), SATA SSD (map voor het opslaan van CSV-bestanden).
Firebird 4.0.4 x64

Resultaten:

bash

CSVExport.exe -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=1 \
  -d inet://localhost:3054/horses -u SYSDBA -p masterkey --charset=WIN1251 -o ./single

Elapsed time in milliseconds parallel_part: 35894 ms
Elapsed time in milliseconds: 36317 ms

CSVExport.exe -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=2 \
  -d inet://localhost:3054/horses -u SYSDBA -p masterkey --charset=WIN1251 -o ./multi

Elapsed time in milliseconds parallel_part: 19259 ms
Elapsed time in milliseconds: 20760 ms

CSVExport.exe -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=4 \
  -d inet://localhost:3054/horses -u SYSDBA -p masterkey --charset=WIN1251 -o ./multi

Elapsed time in milliseconds parallel_part: 19600 ms
Elapsed time in milliseconds: 21137 ms

Uit het testresultaat blijkt duidelijk dat bij gebruik van twee threads de versnelling 1,8 keer was, wat een goed resultaat is. Maar parallelle uitvoering van de export met 4 threads liet ook een verbetering van 1,8x zien. Waarom niet 3-4? Het feit is dat de Firebird-server en het export-hulpprogramma op dezelfde computer draaien, die slechts 4 kernen heeft. De Firebird-server zelf gebruikt dus 4 threads om de tabel te lezen, en het FBCSVExport-hulpprogramma gebruikt ook 4 threads. Het is duidelijk dat het in dit geval vrij moeilijk is om een versnelling van meer dan 2 keer te bereiken. Daarom proberen we het op andere hardware, waar het aantal kernen aanzienlijk groter is.

Linux

Besturingssysteem: CentOS 8.
Processor: 2 Intel Xeon E5-2603 v4-processors, in totaal 12 kernen, 12 threads.
Geheugen: 32 GB
Schijfsubsysteem: SAS HDD (RAID 10)
Firebird 4.0.4 x64

Resultaten:

bash

[denis@copyserver build]$ ./CSVExport -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=1 \
  -d inet://localhost/horses -u SYSDBA -p masterkey --charset=UTF8 -o ./single

Elapsed time in milliseconds parallel_part: 57547 ms
Elapsed time in milliseconds: 57595 ms

[denis@copyserver build]$ ./CSVExport -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=4 \
  -d inet://localhost/horses -u SYSDBA -p masterkey --charset=UTF8 -o ./multi

Elapsed time in milliseconds parallel_part: 17755 ms
Elapsed time in milliseconds: 18148 ms

[denis@copyserver build]$ ./CSVExport -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=6 \
  -d inet://localhost/horses -u SYSDBA -p masterkey --charset=UTF8 -o ./multi

Elapsed time in milliseconds parallel_part: 13243 ms
Elapsed time in milliseconds: 13624 ms

[denis@copyserver build]$ ./CSVExport -H --table-filter="COLOR|BREED|HORSE|COVER|MEASURE|LAB_LINE|SEX" --parallel=12 \
  -d inet://localhost/horses -u SYSDBA -p masterkey --charset=UTF8 -o ./multi

Elapsed time in milliseconds parallel_part: 12712 ms
Elapsed time in milliseconds: 13140 ms

In dit geval is het optimale aantal threads voor export 6 (6 threads voor Firebird en 6 threads voor het FBCSVExport-hulpprogramma). Tegelijkertijd slaagden we erin een versnelling van 5x te bereiken, wat duidt op redelijk goede schaalbaarheid. Op de Linux-server en de Windows-computer hebben we identieke databases gebruikt, en u heeft waarschijnlijk gemerkt dat de single-threaded export op Windows bijna 2 keer sneller was: dit komt door een sneller schijfsubsysteem (NVME-schijf is veel sneller dan SAS-schijven in RAID).

Samenvatting

In dit artikel hebben we bekeken hoe u effectief gegevens uit Firebird DBMS-tabellen kunt lezen met behulp van parallellisme. Ook werd het voorbeeld getoond van hoe u enkele mogelijkheden van het Firebird DBMS kunt gebruiken om dergelijk lezen in uw software te organiseren.

Veel dank aan Vladislav Khorsun, Firebird-kernontwikkelaar, voor hulp bij dit materiaal.

Voor vragen of opmerkingen kunt u een e-mail sturen naar [email protected].