Kontakte
IBSurgeon
Diese Seite wurde maschinell übersetzt. Lesen Sie das englische Original. English

IBSurgeon-Bibliothek

Paralleles Lesen von Daten in Firebird

D.Simonov, V.Horsun

Version 1.0.5 vom 05.12.2023

Dieses Material wird gesponsert und mit der Sponsoring- und Unterstützung von IBSurgeon www.ib-aid.com erstellt, Anbieter von HQbird (erweiterte Distribution von Firebird) sowie Anbieter von Leistungsoptimierungs-, Migrations- und technischen Supportdienstleistungen für Firebird.

Das Material ist unter der Public Documentation License lizenziert https://www.firebirdsql.org/file/documentation/html/en/licenses/pdl/public-documentation-license.html

Verwandte Materialien:

Vorwort

Firebird 5.0 führte die Möglichkeit ein, Parallelität bei der Erstellung eines Backups mit dem gbak-Dienstprogramm zu nutzen, neben anderen parallelen Funktionen. Ursprünglich erschien diese Funktion in HQbird 2.5, dann in HQbird 3.0 und 4.0, und anschließend wurde sie auf Firebird 5.0 portiert.

In diesem Artikel betrachten wir die Funktionen, die bei der Erstellung paralleler Backups innerhalb des gbak-Dienstprogramms verwendet werden. Wir zeigen auch, wie sie in Ihren Anwendungen für das parallele Lesen von Daten verwendet werden können.

Es ist wichtig zu beachten, dass es hier nicht um das parallele Scannen von Tabellen innerhalb der Firebird-Engine bei der Ausführung von SQL-Abfragen geht, sondern um das Lesen von Daten innerhalb Ihrer Anwendung in parallelen Strömen.

Beispielwerkzeug FBCSVExport

Zur Demonstration des parallelen Lesens von Daten aus dem Firebird-DBMS wurde ein Beispielprogramm geschrieben, das Daten aus einer oder mehreren Tabellen in das CSV-Format exportiert.

Seine Beschreibung und sein Open-Source-Code sind hier: https://github.com/IBSurgeon/FBCSVExport.git

Wie Sie in der Beschreibung des Dienstprogramms und aus dem folgenden Artikel sehen können, ist es durch parallele Verarbeitung möglich, Daten zu exportieren und andere parallele Operationen 2-10x schneller durchzuführen als in 1 Thread (abhängig von der Hardware).

Bei Fragen wenden Sie sich bitte an [email protected].

Paralleles Lesen

Lassen Sie uns darüber nachdenken, wie Daten aus mehreren Tabellen parallel gelesen werden können. Wie Sie wissen, erlaubt Firebird die parallele Ausführung von Abfragen nur, wenn jede Abfrage in einer separaten Verbindung ausgeführt wird.

Lassen Sie uns einen Pool von Worker-Threads erstellen. Der Hauptanwendungsthread ist ebenfalls ein Worker-Thread, daher sollte die Anzahl der zusätzlichen Worker-Threads N - 1 betragen, wobei N die Gesamtzahl der parallelen Worker ist. Jeder Worker-Thread führt seine eigene Verbindung und Transaktion aus.

Das erste Problem: Wie kann die Konsistenz der gelesenen Daten sichergestellt werden?

Konsistentes Datenlesen

Da jeder Worker-Thread seine eigene Verbindung und seine eigene Transaktion verwendet, entsteht das Problem inkonsistenter Lesevorgänge - wenn die Tabelle gleichzeitig von anderen Benutzern geändert wird, können die gelesenen Daten inkonsistent sein. Im Einzelthread-Modus verwendet gbak eine Transaktion mit dem SNAPSHOT-Isolationsmodus, was es ermöglicht, konsistente Informationen zum Start der SNAPSHOT-Transaktion zu lesen. Aber hier haben wir mehrere Transaktionen, und es ist notwendig, dass sie denselben “Snapshot” sehen, damit sie dieselben unveränderlichen Daten lesen.

Der Mechanismus zur Erstellung eines gemeinsamen Snapshots für verschiedene Transaktionen mit dem SNAPSHOT-Isolationsmodus wurde in Firebird 4.0 eingeführt (ursprünglich in HQBird 2.5, aber in Firebird 4.0/HQBird 4.0 ist er einfacher und effizienter). Es gibt zwei Möglichkeiten, einen gemeinsamen Snapshot zu erstellen:

  1. Mit SQL
  • Holen Sie die Snapshot-Nummer von der Haupttransaktion (die im Haupt-Worker-Thread gestartet wird).
sql 
SELECT RDB$GET_CONTEXT('SYSTEM', 'SNAPSHOT_NUMBER') FROM RDB$DATABASE
  • Starten Sie andere Transaktionen mit dem folgenden SQL:
sql 
SET TRANSACTION SNAPSHOT AT NUMBER snapshot_number

wobei snapshot_number die Nummer ist, die durch die vorherige Abfrage abgerufen wurde.

  1. Mit API

  • Holen Sie die Snapshot-Nummer von der Haupttransaktion (die im Haupt-Worker-Thread gestartet wird) mit der Funktion isc_transaction_info oder ITransaction.getInfo mit dem Tag fb_info_tra_snapshot_number;

  • Starten Sie andere Transaktionen mit dem Tag isc_tpb_at_snapshot_number mit einer erhaltenen Snapshot-Nummer.

Das Beispielwerkzeug FBCSVExport sowie gbak verwenden den zweiten Ansatz. Diese Ansätze können gemischt werden - zum Beispiel die Snapshot-Nummer mit SQL holen und die erhaltene Snapshot-Nummer verwenden, um andere Transaktionen mit API zu starten, oder umgekehrt.

In FBCSVExport erhalten wir eine Snapshot-Nummer mit dem folgenden 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;
}

Um eine Transaktion mit der Snapshot-Nummer zu starten, verwenden wir den folgenden 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)
    )
);

Jetzt sind die aus verschiedenen Verbindungen gelesenen Daten konsistent, sodass wir die Last auf die Arbeitsthreads verteilen können.

Wie genau soll die Last zwischen den Worker-Threads verteilt werden? Im Falle eines vollständigen Exports aller Tabellen oder einer Sicherungskopie ist die einfachste Option ein Worker-Thread pro Tabelle. Aber bei diesem Ansatz haben wir folgendes Problem: Wenn es viele kleine Tabellen in einer Datenbank gibt und eine große Tabelle, oder sogar nur eine Tabelle, die riesig ist, werden wir keine Verbesserung sehen. In diesem Fall erhält ein Thread eine große Tabelle, und die übrigen Threads bleiben untätig. Um dies zu verhindern, ist es notwendig, eine große Tabelle in Teilen zu verarbeiten.
Hinweis Das folgende Material widmet sich dem vollständigen Lesen von Tabellen. Wenn Sie das parallele Lesen aus einer Abfrage (oder Ansicht) organisieren möchten, ist ein etwas anderer Ansatz erforderlich, der von den tatsächlichen Daten abhängt.

Große Tabelle in Teile aufteilen

Nehmen wir an, wir haben nur eine große Tabelle, die wir vollständig und so schnell wie möglich lesen möchten. Es wird vorgeschlagen, sie in mehrere Teile zu teilen und jeden Teil aus seinem eigenen Strom unabhängig zu lesen. Jeder Thread muss seine eigene Verbindung zur Datenbank haben.

In diesem Fall ergeben sich folgende Fragen:

  • In wie viele Verarbeitungsteile soll die Tabelle aufgeteilt werden?

  • Wie lässt sich die Tabelle im Hinblick auf den Datenzugriff am besten aufteilen?

Lassen Sie uns diese Fragen der Reihe nach beantworten.

In wie viele Verarbeitungsteile soll die Tabelle aufgeteilt werden?

Nehmen wir das ideale Szenario an - Server und Client sind Firebird gewidmet, das heißt, alle CPUs stehen uns vollständig zur Verfügung. Dann wird empfohlen:

a) Als maximale Anzahl paralleler Teile die doppelte Anzahl der CPU-Kerne auf dem Server verwenden. Warum 2x Kerne? Wir wissen sicher, dass es Verzögerungen im Zusammenhang mit IO geben wird, daher können wir eine gewisse zusätzliche CPU-Nutzung zulassen. Diese Zahl sollte jedoch als anfängliche Einstellung betrachtet werden; praktisch hängt sie von den Daten ab.

b) Die Anzahl der Kerne auf dem Client berücksichtigen: Wenn es auf dem Server deutlich mehr gibt (übliche Situation), kann es sinnvoll sein, die Anzahl der Partitionsteile weiter zu begrenzen, um den Client nicht zu überlasten (er kann ohnehin nicht mehr verarbeiten, und die Umschaltkosten für Flüsse verschwinden nicht). Eine genauere Entscheidung kann durch Überwachung der CPU-Last von Client und Server getroffen werden - wenn sie auf dem Client 100% beträgt, aber auf dem Server deutlich weniger, dann ist es sinnvoll, die Anzahl der Teile zu reduzieren.

c) Wenn Client und Server derselbe Host sind, siehe (a).

Wenn Client und/oder Server mit etwas anderem beschäftigt sind, müssen Sie möglicherweise die Anzahl der Teile reduzieren. Dies kann auch durch die Fähigkeit der Festplatten auf dem Server beeinflusst werden, viele IO-Anfragen gleichzeitig zu verarbeiten (überwachen Sie die Warteschlangengröße und Antwortzeit).

Wie lässt sich die Tabelle im Hinblick auf den Datenzugriff am besten aufteilen?

Für eine effektive parallele Verarbeitung ist es wichtig, eine gleichmäßige Verteilung der Aufgaben auf die Handler zu gewährleisten und deren gegenseitige Synchronisation zu minimieren. Darüber hinaus müssen Sie bedenken, dass die Synchronisation von Handlern sowohl auf der Server- als auch auf der Client-Seite erfolgen kann. Zum Beispiel sollten mehrere Handler nicht dieselbe Verbindung zur Datenbank verwenden. Ein weniger offensichtliches Beispiel: Es ist schlecht, wenn verschiedene Handler Datensätze von denselben Datenbankseiten lesen. Zum Beispiel, wenn zwei Handler gerade und ungerade Datensätze lesen - das ist nicht effektiv. Die Synchronisation auf dem Client kann während der Verteilung von Aufgaben, während der Verarbeitung empfangener Daten (Speicherzuweisung für Ergebnisse) usw. auftreten.

Eines der Probleme bei der “fairen” Partitionierung ist, dass der Client nicht weiß, wie Datensätze über Seiten (und über Indexschlüssel) verteilt sind, wie viele Datensätze oder Datenseiten es gibt (bei großen Tabellen wäre es zu langwierig, die Anzahl der Datensätze im Voraus zu zählen).

Lassen Sie uns sehen, wie gbak dieses Problem löst.

Für gbak ist eine Arbeitseinheit eine Menge von Datensätzen aus Datenseiten (DPs), die zur selben Zeigerseite (PP) gehören. Einerseits ist dies eine ziemlich große Anzahl von Datensätzen, um den Handler zu beschäftigen, ohne häufig nach einem neuen Datenstück fragen zu müssen (Synchronisation). Andererseits, selbst wenn solche Datensatzmengen nicht genau dieselbe Größe haben, ermöglicht es eine relativ gleichmäßige Auslastung der Worker. Das heißt, es ist durchaus möglich, dass ein Worker N Datensätze von einer PP liest und der andere M Datensätze, wobei M sich erheblich von N unterscheiden kann. Dieser Ansatz ist nicht ideal, aber er ist ziemlich einfach zu implementieren und in der Regel recht effektiv, zumindest in großem Maßstab (mit Dutzenden oder Hunderten (oder mehr) PPs).

Wie erhält man die Anzahl der PPs (Zeigerseiten) für eine bestimmte Tabelle? Es ist ziemlich einfach und, was am wichtigsten ist, schnell, sie aus der Tabelle RDB$PAGES zu berechnen:

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

Als Nächstes könnten wir einfach die Anzahl der PPs durch die Anzahl der Worker teilen und jedem Worker sein eigenes Stück geben. Das ist für das Szenario in Ordnung, in dem die parallele Verarbeitung vom Entwickler durchgeführt wird, der die Datenverteilung kennt. Aber für das häufigere Szenario gibt es keine Garantie, dass solche “großen” Stücke dieselbe Menge an Arbeit bedeuten. Wir sind nicht daran interessiert, die Situation zu sehen, in der 15 Worker ihre Arbeit beendet haben und untätig stehen, während der 16. seine 10M Datensätze lange liest.

Deshalb macht es gbak anders. Es gibt einen Arbeitskoordinator, der jedem Prozessor jeweils 1 PP ausgibt. Der Koordinator weiß, wie viele PPs es insgesamt gibt und wie viele bereits zur Arbeit ausgegeben wurden. Wenn der Worker das Lesen seiner Datensätze abgeschlossen hat, kontaktiert er den Koordinator für eine neue PP-Nummer. Dies wird fortgesetzt, bis die PPs aufgebraucht sind (oder es aktive Worker gibt). Natürlich erfordert eine solche Interaktion der Worker mit dem Koordinator eine Synchronisation. Die Erfahrung zeigt, dass die Menge an Arbeit, die eine PP bietet, es ermöglicht, nicht zu oft zu synchronisieren. Dieser Ansatz ermöglicht es, alle Worker (und damit CPU-Kerne) praktisch gleichmäßig mit Arbeit zu belasten, unabhängig von der tatsächlichen Anzahl der Datensätze, die zu jeder PP gehören.

Wie liest der Handler Datensätze aus seiner PP? Dazu gibt es, beginnend mit Firebird 4.0 (erstmals erschienen in HQBird 2.5), eine eingebaute Funktion MAKE_DBKEY(). Mit ihrer Hilfe können Sie die RDB$DB_KEY (physische Datensatznummer) für den ersten Datensatz auf der angegebenen PP erhalten.

Und mit Hilfe dieser RDB$DB_KEY werden die benötigten Datensätze ausgewählt:

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)

Wenn Sie zum Beispiel loPP = 0 und hiPP = 1 setzen, werden alle Datensätze mit PP = 0 gelesen, und zwar nur von dieser.

Jetzt, da Sie eine Vorstellung davon haben, wie gbak funktioniert, können Sie zu einer Beschreibung der Implementierung des Dienstprogramms FBCSVExport übergehen.

Implementierung des Dienstprogramms FBCSVExport

Das Dienstprogramm FBCSVExport ist dafür ausgelegt, Daten aus Firebird-Datenbanktabellen in das CSV-Format zu exportieren.

Jede Tabelle wird in eine Datei mit dem Namen .csv exportiert. Im normalen (einzelthreadigen) Modus werden Daten aus Tabellen sequenziell in alphabetischer Reihenfolge der Tabellennamen exportiert.

Im parallelen Modus werden Tabellen parallel exportiert, wobei jede Tabelle in einem separaten Thread verarbeitet wird. Wenn die Tabelle sehr groß ist, wird sie in Teile aufgeteilt, und jeder Teil wird in einem separaten Stream exportiert. Für jeden Teil einer großen Tabelle wird eine separate Datei mit dem Namen .csv.partN erstellt, wobei N die Teilnummer ist.

Wenn alle Teile einer großen Tabelle exportiert sind, werden die Teildateien zu einer Datei mit dem Namen .csv zusammengeführt.

Ein regulärer Ausdruck wird verwendet, um anzugeben, welche Tabellen exportiert werden. Nur reguläre Tabellen können exportiert werden (Systemtabellen, GTT, Ansichten, externe Tabellen werden nicht unterstützt). Reguläre Ausdrücke müssen in SQL-Syntax vorliegen, das heißt, solche, die im SIMILAR TO-Prädikat verwendet werden.

Um eine Liste der exportierten Tabellen sowie eine Liste ihrer PPs im mehrthreadigen Modus auszuwählen, verwenden wir die folgende Abfrage:

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

Im einzelthreadigen Modus kann diese Abfrage vereinfacht werden zu

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

Im einzelthreadigen Modus werden die Werte der Felder PAGE_SEQUENCE und PP_CNT nicht verwendet; sie werden zur Abfrage hinzugefügt, um die Ausgabemeldungen zu vereinheitlichen.

Das Ergebnis dieser Abfrage wird in einen Vektor von Strukturen umgewandelt:

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;
};

Dieser Vektor wird mit einer Funktion gefüllt, die wie folgt deklariert ist:

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

Der letzte Parameter singleWorker schaltet den Füllmodus von std::vector um. Wenn singleWorker = true ist, wird die Abfrage für den einzelthreadigen Modus verwendet; wenn singleWorker = false ist, wird eine aufwändigere und komplexere Abfrage für den mehrthreadigen Modus verwendet. Ich werde die Implementierung selbst nicht angeben, sie ist recht einfach und Sie können sie im Quellcode des Projekts einsehen.

Um eine Tabelle in das CSV-Format zu exportieren, wurde die Klasse CSVExportTable entwickelt, die die folgenden Methoden enthält:

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);

Die Methode prepare dient dazu, eine Abfrage zu erstellen und vorzubereiten, die zum Export einer Tabelle in das CSV-Format verwendet wird. Die innere Abfrage wird je nach Parameter withDbkeyFilter unterschiedlich konstruiert. Wenn withDbkeyFilter = true ist, wird die Abfrage mit einem Filter nach dem Bereich RDB$DB_KEY erstellt:

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

andernfalls wird eine vereinfachte Abfrage verwendet:

sql 
SELECT *
FROM tableName

Der Wert des Parameters withDbkeyFilter wird auf true gesetzt, wenn der mehrthreadige Modus verwendet wird und die Tabelle groß ist. Wir betrachten die Tabelle als groß, wenn pp_cnt > 1 ist.

Die Methode printHeader dient zum Drucken des Headers einer CSV-Datei (Spaltennamen der Tabelle).

Die Methode printData druckt Tabellendaten in eine CSV-Datei ab der PP-Seitennummer ppNum, wenn die Abfrage mit einem Filter nach dem Bereich RDB$DB_KEY vorbereitet wurde, und andernfalls alle Tabellendaten.

Nun schauen wir uns den Code für den einzelthreadigen Modus an

cpp 
...

// Öffnen der Hauptverbindung
Firebird::AutoRelease att(
    provider->attachDatabase(
        &status,
        m_database.c_str(),
        dbpLength,
        dpb
    )
);

// Starten der Haupttransaktion im Isolationsmodus 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)
    )
);
// Abrufen einer Liste von Tabellen mit dem regulären Ausdruck in m_filter.
// m_parallel setzt die Anzahl der parallelen Threads; wenn es gleich 1 ist,
// wird eine vereinfachte Abfrage zum Abrufen der Tabellenliste verwendet,
// andernfalls wird für jede Tabelle eine Liste von PPs und deren Anzahl generiert.
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) {
        // es gibt keinen Grund, hier einen Bereichsfilter RDB$DB_KEY zu verwenden
        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);
    }
}

Hier ist alles recht einfach und bedarf keiner zusätzlichen Erklärung, also gehen wir zum mehrthreadigen Teil über.

Damit der Export im mehrthreadigen Modus erfolgt, müssen zusätzliche m_parallel - 1 Arbeitsthreads erstellt werden. Warum ist die Anzahl der zusätzlichen Threads um 1 geringer? Ja, weil der Hauptthread auch Daten exportiert und den zusätzlichen Threads gleichwertig ist. Verschieben wir den gemeinsamen Teil des Haupt- und des zusätzlichen Flusses in eine separate Funktion:

cpp 
void ExportApp::exportByTableDesc(Firebird::ThrowStatusWrapper* status, FBExport::CSVExportTable& csvExport, const TableDesc& tableDesc)
{
    // Wenn tableDesc pp_cnt > 1 hat, beschreibt es nur einen Teil der Tabelle, und es ist notwendig, eine
    // Abfrage mit einem Filter nach dem Bereich RDB$DB_KEY zu erstellen.

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

    // Wenn dies nicht der erste Teil der Tabelle ist, schreiben Sie diesen Teil in die Datei .csv.part, wobei
    // N - PP-Nummer. Später werden die Tabellenteile zu einer einzigen Datei .csv zusammengeführt
    if (tableDesc.page_sequence > 0) {
        fileName += ".part_" + std::to_string(tableDesc.page_sequence);
    }
    csv::CSVFile csv(m_outputDir / fileName);
    // Der Header der CSV-Datei sollte nur im ersten Teil der Tabelle gedruckt werden.
    if (tableDesc.page_sequence == 0 && m_printHeader) {
        csvExport.printHeader(status, csv);
    }
    csvExport.printData(status, csv, tableDesc.page_sequence);
}

Beschreibungen von Tabellen oder deren Teilen befinden sich in einem gemeinsamen Vektor mit TableDesc-Strukturen. Aus diesem Vektor nimmt jeder Arbeitsthread eine Tabelle oder den nächsten Teil. Um Datenrennen zu verhindern, ist es notwendig, den Zugriff auf die gemeinsame Ressource zu synchronisieren. Aber std::vector selbst ändert sich nicht, sodass nur die gemeinsame Variable synchronisiert werden muss, die der Index in diesem Vektor ist. Dies kann einfach mit std::atomic als solche Variable erfolgen.

cpp 
if (m_parallel == 1) {
    ...
}
else {
    // Bestimmen der Anzahl zusätzlicher Arbeitsthreads
    const auto workerCount = m_parallel - 1;

    // Abrufen der Snapshot-Nummer aus der Haupttransaktion
    auto snapshotNumber = getSnapshotNumber(&status, tra);
    // Variable zum Speichern der Ausnahme innerhalb des Threads
    std::exception_ptr exceptionPointer = nullptr;
    std::mutex m;
	// atomarer Zähler
    // ist der Index der nächsten Tabelle oder eines Teils davon
    std::atomic<size_t> counter = 0;
    // Pool von Arbeitsthreads
    std::vector<std::thread> thread_pool;
    thread_pool.reserve(workerCount);
    for (int i = 0; i < workerCount; i++) {
		// für jeden Thread erstellen wir unsere eigene Verbindung
        Firebird::AutoRelease workerAtt(
            provider->attachDatabase(
                &status,
                m_database.c_str(),
                dbpLength,
                dpb
            )
        );
		// und unsere Transaktion, der wir die Snapshot-Nummer übergeben
        // um einen gemeinsamen Snapshot zu erstellen
        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)
            )
        );
		// einen Thread erstellen
        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) {
                    // den atomaren Zähler erhöhen
                    size_t localCounter = counter++;
					// wenn die Tabellen oder deren Teile zu Ende sind, aus der
                    // Endlosschleife aussteigen und den Thread beenden
                    if (localCounter >= tables.size())
                        break;
                    // eine Beschreibung der Tabelle oder eines Teils davon abrufen
                    const auto& tableDesc = tables[localCounter];
                    // und den Export durchführen
                    exportByTableDesc(&status, csvExport, tableDesc);
                }
                if (tra) {
                    tra->commit(&status);
                    tra.release();
                }

                if (att) {
                    att->detach(&status);
                    att.release();
                }
            }
            catch (...) {
				// wenn eine Ausnahme auftritt, speichern Sie sie für
                // die spätere Freigabe im Hauptthread
                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);
    }
    ...

Es bleibt nur noch, die Dateien, die für Teile der Tabellen erstellt wurden, zu einer einzigen Datei für jede dieser Tabellen zusammenzuführen.

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();
    }
}

Lassen Sie uns die Leistung des Tools im ein- und mehrthreadigen Modus messen.

Benchmark des FBCSVExport-Tools

Zuerst schauen wir uns die Ergebnisse des Vergleichs von mehrthreadigem und einthreadigem Exportmodus auf einem moderaten Heimcomputer an. === Windows

  • Betriebssystem: Windows 10 x64.

  • Prozessor: Intel Core i3 8100, 4 Kerne, 4 Threads.

  • Speicher: 16 GB

  • Festplattensubsystem: NVME SSD (Datenbank), SATA SSD (Ordner zum Speichern von CSV-Dateien).

  • Firebird 4.0.4 x64

Ergebnisse:

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

Aus dem Testergebnis geht hervor, dass bei Verwendung von zwei Threads die Beschleunigung 1,8-fach betrug, was ein gutes Ergebnis ist. Aber die parallele Ausführung des Exports mit 4 Threads zeigte ebenfalls eine 1,8-fache Verbesserung. Warum nicht 3-4-fach? Tatsache ist, dass der Firebird-Server und das Export-Tool auf demselben Computer laufen, der nur 4 Kerne hat. Somit verwendet der Firebird-Server selbst 4 Threads zum Lesen der Tabelle, und das FBCSVExport-Tool verwendet ebenfalls 4 Threads. Offensichtlich ist es in diesem Fall ziemlich schwierig, eine Beschleunigung von mehr als dem 2-fachen zu erreichen. Daher versuchen wir es auf anderer Hardware, wo die Anzahl der Kerne deutlich größer ist.

Linux

  • Betriebssystem: CentOS 8.

  • Prozessor: 2 Intel Xeon E5-2603 v4 Prozessoren, insgesamt 12 Kerne, 12 Threads.

  • Speicher: 32 GB

  • Festplattensubsystem: SAS HDD (RAID 10)

  • Firebird 4.0.4 x64

Ergebnisse:

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 diesem Fall ist die optimale Anzahl von Threads für den Export 6 (6 Threads für Firebird und 6 Threads für das FBCSVExport-Tool). Gleichzeitig gelang es uns, eine 5-fache Beschleunigung zu erreichen, was auf eine ziemlich gute Skalierbarkeit hinweist. Auf dem Linux-Server und dem Windows-Computer haben wir identische Datenbanken verwendet, und Sie haben wahrscheinlich bemerkt, dass der einthreadige Export unter Windows fast 2-mal schneller war: Das liegt an einem schnelleren Festplattensubsystem (NVME-Laufwerk ist viel schneller als SAS-Laufwerke in RAID).

Zusammenfassung

In diesem Artikel haben wir betrachtet, wie man Daten aus Firebird-DBMS-Tabellen effektiv mit Parallelität lesen kann. Außerdem wurde ein Beispiel gezeigt, wie man einige der Fähigkeiten des Firebird-DBMS nutzen kann, um ein solches Lesen in Ihrer Software zu organisieren.

Vielen Dank an Vladislav Khorsun, Firebird-Kernentwickler, für die Hilfe bei diesem Material.

Bei Fragen oder Kommentaren senden Sie bitte eine E-Mail an [email protected].