Lettura Parallela dei Dati in Firebird

D.Simonov, V.Horsun

versione 1.0.5 del 05.12.2023

Questo materiale è sponsorizzato e creato con il patrocinio e il supporto di IBSurgeon www.ib-aid.com, fornitore di HQbird (distribuzione avanzata di Firebird) e fornitore di servizi di ottimizzazione delle prestazioni, migrazione e supporto tecnico per Firebird.

Il materiale è concesso in licenza secondo la Public Documentation License https://www.firebirdsql.org/file/documentation/html/en/licenses/pdl/public-documentation-license.html

Materiali correlati:

Libro gratuito “Detailed New Featuires Of Firebird 5”: disponibile come HTML e come PDF (119 pagine).

Prefazione

Firebird 5.0 ha introdotto la possibilità di utilizzare il parallelismo durante la creazione di un backup tramite l’utilità gbak, tra le altre funzioni parallele. Inizialmente, questa funzione è apparsa in HQbird 2.5, poi in HQbird 3.0 e 4.0, e successivamente è stata portata in Firebird 5.0.

In questo articolo esamineremo le funzioni utilizzate durante la creazione di backup paralleli all’interno dell’utilità gbak. Mostreremo anche come possono essere utilizzate nelle vostre applicazioni per la lettura parallela dei dati.

È importante notare che qui non si parla di scansione parallela delle tabelle all’interno del motore Firebird durante l’esecuzione di query SQL, ma di lettura dei dati all’interno della vostra applicazione tramite flussi paralleli.

Strumento di esempio FBCSVExport

Per dimostrare la lettura parallela dei dati dal DBMS Firebird, è stata scritta un’utilità di esempio che esporta dati da una o più tabelle in formato CSV.

La sua descrizione e il suo codice sorgente aperto sono qui: https://github.com/IBSurgeon/FBCSVExport.git

Come si può vedere nella descrizione dell’utilità e dall’articolo seguente, attraverso l’elaborazione parallela è possibile esportare dati ed eseguire altre operazioni parallele da 2 a 10 volte più velocemente rispetto a un singolo thread (a seconda dell’hardware).

Per qualsiasi domanda, contattare [email protected].

Lettura parallela

Pensiamo a come leggere dati da più tabelle in parallelo. Come è noto, Firebird consente di eseguire query in parallelo solo se ogni query viene eseguita in una connessione separata.

Creiamo un pool di thread di lavoro. Il thread principale dell’applicazione è anche un thread di lavoro, quindi il numero di thread di lavoro aggiuntivi dovrebbe essere N - 1, dove N è il numero totale di worker paralleli. Ogni thread di lavoro eseguirà la propria connessione e transazione.

Il primo problema: come garantire la coerenza dei dati letti?

Lettura coerente dei dati

Poiché ogni thread di lavoro utilizza la propria connessione e la propria transazione, sorge il problema delle letture incoerenti - se la tabella viene modificata simultaneamente da altri utenti, i dati letti potrebbero essere incoerenti. In modalità single-thread, gbak utilizza una transazione con modalità di isolamento SNAPSHOT, che consente di leggere informazioni coerenti all’avvio della transazione SNAPSHOT. Ma qui abbiamo più transazioni ed è necessario che vedano lo stesso “snapshot” affinché leggano gli stessi dati immutabili.

Il meccanismo per creare uno snapshot condiviso per diverse transazioni con modalità di isolamento SNAPSHOT è stato introdotto in Firebird 4.0 (originariamente in HQBird 2.5, ma in Firebird 4.0/HQBird 4.0 è più semplice ed efficiente). Ci sono due modi per creare uno snapshot condiviso:

Con SQL

ottenere il numero di snapshot dalla transazione principale (che viene avviata nel thread di lavoro principale).

sql

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

avviare altre transazioni con il seguente SQL:

sql

SET TRANSACTION SNAPSHOT AT NUMBER snapshot_number

dove snapshot_number è il numero recuperato dalla query precedente.

Con API

ottenere il numero di snapshot dalla transazione principale (che viene avviata nel thread di lavoro principale) con la funzione isc_transaction_info o ITransaction.getInfo con tag fb_info_tra_snapshot_number;
avviare altre transazioni con tag isc_tpb_at_snapshot_number con un numero di snapshot ottenuto.

Lo strumento di esempio FBCSVExport, così come gbak, utilizza il secondo approccio. Questi approcci possono essere mescolati - ad esempio, ottenere il numero di snapshot con SQL e utilizzare il numero di snapshot ottenuto per avviare altre transazioni con API, o viceversa.

In FBCSVExport otteniamo un numero di snapshot con il seguente codice:

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

Per avviare la transazione con il numero di snapshot utilizziamo il seguente codice:

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

Ora i dati letti da diverse connessioni saranno coerenti, quindi possiamo distribuire il carico tra i thread di lavoro.

Come distribuire esattamente il carico tra i thread di lavoro? Nel caso di esportazione completa di tutte le tabelle o di una copia di backup, l’opzione più semplice sarà un thread di lavoro per tabella. Ma con questo approccio abbiamo il seguente problema: se ci sono molte tabelle piccole in un database e una tabella grande, o anche solo una tabella ed è enorme, non vedremo miglioramenti. In questo caso, qualche thread otterrà una tabella grande e i thread rimanenti rimarranno inattivi. Per evitare che ciò accada, è necessario elaborare una tabella grande in parti.


Nota	Il materiale seguente è dedicato alla lettura completa delle tabelle; se si desidera organizzare la lettura parallela da una query (o vista), sarà necessario un approccio leggermente diverso, che dipende dai dati effettivi.

Suddivisione di una tabella grande in parti

Supponiamo di avere solo una tabella grande che vogliamo leggere integralmente e il più rapidamente possibile. Si propone di dividerla in più parti e leggere ciascuna parte dal proprio flusso in modo indipendente. Ogni thread deve avere la propria connessione al database.

In questo caso sorgono le seguenti domande:

In quante parti di elaborazione dovrebbe essere divisa la tabella?
Qual è il modo migliore per dividere la tabella in termini di accesso ai dati?

Rispondiamo a queste domande in ordine.

In quante parti di elaborazione dovrebbe essere divisa la tabella?

Supponiamo lo scenario ideale: il server e il client sono dedicati a Firebird, cioè tutte le CPU sono completamente a nostra disposizione. Allora si raccomanda:

a) Utilizzare come numero massimo di parti parallele il doppio del numero di core CPU sul server. Perché 2x core? Sappiamo per certo che ci saranno ritardi associati all’IO, quindi possiamo consentire un uso extra della CPU. Tuttavia, questo numero dovrebbe essere considerato come un’impostazione iniziale; in pratica dipende dai dati.

b) Tenere conto del numero di core sul client: se sul server ce ne sono molti di più (situazione usuale), potrebbe avere senso limitare ulteriormente il numero di parti della partizione, per non sovraccaricare il client (non sarà comunque in grado di elaborarne di più, e i costi di commutazione dei flussi non vanno da nessuna parte). Si potrà decidere più precisamente monitorando il carico CPU del client e del server - se è al 100% sul client, ma notevolmente inferiore sul server, allora ha senso ridurre il numero di parti.

c) se il client e il server sono lo stesso host, vedere (a).

Se il client e/o il server sono occupati con altro, potrebbe essere necessario ridurre il numero di parti. Ciò potrebbe anche essere influenzato dalla capacità dei dischi sul server di elaborare molte richieste IO simultaneamente (monitorare la dimensione della coda e il tempo di risposta).

Qual è il modo migliore per dividere la tabella in termini di accesso ai dati?

Per implementare un’elaborazione parallela efficace, è importante garantire una distribuzione uniforme dei lavori tra i gestori e minimizzare la loro sincronizzazione reciproca. Inoltre, bisogna ricordare che la sincronizzazione dei gestori può avvenire sia sul lato server che sul lato client. Ad esempio, più gestori non dovrebbero utilizzare la stessa connessione al database. Un esempio meno ovvio: è negativo se diversi gestori leggono record dalle stesse pagine del database. Ad esempio, quando due gestori leggono record pari e dispari - non è efficace. La sincronizzazione sul client può avvenire durante la distribuzione dei compiti, durante l’elaborazione dei dati ricevuti (allocazione di memoria per i risultati) e così via.

Uno dei problemi con la partizione “equa” è che il client non sa come i record sono distribuiti tra le pagine (e tra le chiavi di indice), quanti record o pagine di dati ci sono (per tabelle grandi sarebbe troppo lungo contare in anticipo il numero di record).

Vediamo come gbak risolve questo problema.

Per gbak, un’unità di lavoro è un insieme di record dalle pagine di dati (DP) appartenenti alla stessa pagina puntatore (PP). Da un lato, questo è un numero abbastanza grande di record per mantenere il gestore occupato senza dover richiedere frequentemente un nuovo pezzo di dati (sincronizzazione). Dall’altro lato, anche se tali insiemi di record non hanno esattamente la stessa dimensione, consentirà di caricare i worker in modo relativamente uniforme. Cioè, è abbastanza possibile che un worker legga N record da una PP, e l’altro M record, e M possa essere abbastanza diverso da N. Questo approccio non è ideale, ma è abbastanza semplice da implementare e di solito è abbastanza efficace, almeno su larga scala (con decine o centinaia (o più) PP).

Come ottenere il numero di PP (Pointer Pages) per una data tabella? È abbastanza facile e, soprattutto, veloce calcolarlo dalla tabella RDB$PAGES:

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

Successivamente, potremmo semplicemente dividere il numero di PP per il numero di worker e dare a ciascun worker la propria parte. Va bene per lo scenario in cui l’elaborazione parallela è eseguita dallo sviluppatore che conosce la distribuzione dei dati. Ma, per lo scenario più comune, non c’è garanzia che tali “grandi” pezzi significhino la stessa quantità di lavoro. Non siamo interessati a vedere la situazione in cui 15 worker hanno finito il loro lavoro e restano inattivi, e il 16° legge i suoi 10 milioni di record per molto tempo.

Ecco perché gbak lo fa diversamente. C’è un coordinatore del lavoro che assegna a ogni processore 1 PP alla volta. Il coordinatore sa quante PP ci sono in totale e quante sono già state assegnate per il lavoro. Quando il worker completa la lettura dei suoi record, contatta il coordinatore per un nuovo numero di PP. Continua fino a quando le PP non si esauriscono (o ci sono worker attivi). Naturalmente, tale interazione dei worker con il coordinatore richiede sincronizzazione. L’esperienza mostra che la quantità di lavoro data da una PP consente di non sincronizzarsi troppo spesso. Questo approccio consente di caricare praticamente in modo uniforme tutti i worker (e quindi i core CPU) con il lavoro, indipendentemente dal numero effettivo di record appartenenti a ciascuna PP.

Ecco come l’handler legge i record dal suo PP? A partire da Firebird 4.0 (apparsa per la prima volta in HQBird 2.5) esiste una funzione incorporata MAKE_DBKEY(). Con il suo aiuto, puoi ottenere RDB$DB_KEY (numero fisico del record) per il primo record sul PP specificato.

E con l’aiuto di questi RDB$DB_KEY vengono selezionati i record necessari:

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)

Ad esempio, se imposti loPP = 0 e hiPP = 1, verranno letti tutti i record con PP = 0, e solo da esso.

Ora che hai un’idea di come funziona gbak, puoi passare alla descrizione dell’implementazione dell’utility FBCSVExport.

Implementazione dell’utility `FBCSVExport`

L’utility FBCSVExport è progettata per esportare dati dalle tabelle del database Firebird in formato CSV.

Ogni tabella viene esportata in un file denominato .csv. In modalità normale (single-threaded) i dati delle tabelle vengono esportati sequenzialmente in ordine alfabetico dei nomi delle tabelle.

In modalità parallela, le tabelle vengono esportate in parallelo, ogni tabella in un thread separato. Se la tabella è molto grande, viene suddivisa in parti, e ogni parte viene esportata in un flusso separato. Per ogni parte di una tabella grande, viene creato un file separato con il nome .csv.partN, dove N è il numero della parte.

Quando tutte le parti di una tabella grande sono esportate, i file delle parti vengono uniti in un file chiamato .csv.

Un’espressione regolare viene utilizzata per specificare quali tabelle verranno esportate. Possono essere esportate solo tabelle regolari (le tabelle di sistema, GTT, viste, tabelle esterne non sono supportate). Le espressioni regolari devono essere in sintassi SQL, cioè quelle utilizzate nel predicato SIMILAR TO.

Per selezionare un elenco di tabelle esportate, così come un elenco dei loro PP in modalità multi-threaded, utilizziamo la seguente 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 modalità single-threaded, questa query può essere semplificata a

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 modalità single-threaded, i valori dei campi PAGE_SEQUENCE e PP_CNT non vengono utilizzati; vengono aggiunti alla richiesta per uniformare i messaggi di output.

Il risultato di questa query viene formato in un vettore di strutture:

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

Questo vettore viene riempito utilizzando una funzione dichiarata come:

cpp

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

L’ultimo parametro singleWorker commuta la modalità di riempimento di std::vector; se singleWorker = true, viene utilizzata la richiesta per la modalità single-threaded, se singleWorker = false, viene utilizzata una query più costosa e complessa per la modalità multi-threaded. Non fornirò l’implementazione stessa, è abbastanza semplice e puoi vederla nel codice sorgente del progetto.

Per esportare una tabella in formato CSV, è stata sviluppata la classe CSVExportTable, che contiene i seguenti metodi:

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

Il metodo prepare è destinato a costruire e preparare una query utilizzata per esportare una tabella in formato CSV. La query interna viene costruita in modo diverso a seconda del parametro withDbkeyFilter. Se withDbkeyFilter = true, la query viene costruita con il filtro per intervallo 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, ?)

altrimenti, viene utilizzata una query semplificata:

sql

SELECT *
FROM tableName

Il valore del parametro withDbkeyFilter viene impostato su true se viene utilizzata la modalità multi-threaded e la tabella è grande. Consideriamo la tabella grande se pp_cnt > 1.

Il metodo printHeader è progettato per stampare l’intestazione di un file CSV (nomi delle colonne della tabella).

Il metodo printData stampa i dati della tabella in un file CSV dal numero di pagina PP ppNum, se la richiesta è stata preparata utilizzando un filtro per intervallo RDB$DB_KEY, e tutti i dati della tabella altrimenti.

Ora esaminiamo il codice per la modalità single-threaded

cpp

...

// Apertura della connessione principale
Firebird::AutoRelease att(
    provider->attachDatabase(
        &status,
        m_database.c_str(),
        dbpLength,
        dpb
    )
);

// Avvio della transazione principale in modalità di isolamento 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)
    )
);
// Ottieni un elenco di tabelle utilizzando l'espressione regolare in m_filter.
// m_parallel imposta il numero di thread paralleli; quando è uguale a 1,
// viene utilizzata una query semplificata per ottenere l'elenco delle tabelle,
// altrimenti, per ogni tabella viene generato un elenco di PP e il loro numero.
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) {
        // non ha senso utilizzare un filtro per intervallo RDB$DB_KEY qui
        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);
    }
}

Qui tutto è abbastanza semplice e non richiede ulteriori spiegazioni, quindi passiamo alla parte multi-threaded.

Affinché l’esportazione avvenga in modalità multi-threaded, è necessario creare ulteriori thread di lavoro m_parallel - 1. Perché il numero di thread aggiuntivi è 1 in meno? Sì, perché anche il thread principale esporterà dati ed è uguale ai thread aggiuntivi. Spostiamo la parte comune del flusso principale e di quello aggiuntivo in una funzione separata:

cpp

void ExportApp::exportByTableDesc(Firebird::ThrowStatusWrapper* status, FBExport::CSVExportTable& csvExport, const TableDesc& tableDesc)
{
    // Se tableDesc ha pp_cnt > 1, allora descrive solo una parte della tabella, ed è necessario costruire
    // la query utilizzando un filtro per intervallo 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";

    // Se questa non è la prima parte della tabella, scrivi questa parte nel file .csv.part, dove
    // N - numero PP. Successivamente le parti della tabella verranno combinate in un unico file .csv
    if (tableDesc.page_sequence > 0) {
        fileName += ".part_" + std::to_string(tableDesc.page_sequence);
    }
    csv::CSVFile csv(m_outputDir / fileName);
    // L'intestazione del file CSV deve essere stampata solo nella prima parte della tabella.
    if (tableDesc.page_sequence == 0 && m_printHeader) {
        csvExport.printHeader(status, csv);
    }
    csvExport.printData(status, csv, tableDesc.page_sequence);
}

Le descrizioni delle tabelle o delle loro parti si trovano in un vettore comune con strutture TableDesc. Da questo vettore, ogni thread di lavoro prende una tabella o la parte successiva. Per prevenire race condition sui dati, è necessario sincronizzare l’accesso alla risorsa condivisa. Ma std::vector stesso non cambia, quindi puoi sincronizzare solo la variabile condivisa, che è l’indice in questo vettore. Questo può essere facilmente fatto usando std::atomic come tale variabile.

cpp

if (m_parallel == 1) {
    ...
}
else {
    // Determinazione del numero di thread di lavoro aggiuntivi
    const auto workerCount = m_parallel - 1;

    // Ottenere il numero di snapshot dalla transazione principale
    auto snapshotNumber = getSnapshotNumber(&status, tra);
    // variabile per memorizzare l'eccezione all'interno del thread
    std::exception_ptr exceptionPointer = nullptr;
    std::mutex m;
	// contatore atomico
    // è l'indice della prossima tabella o parte di essa
    std::atomic<size_t> counter = 0;
    // pool di thread di lavoro
    std::vector<std::thread> thread_pool;
    thread_pool.reserve(workerCount);
    for (int i = 0; i < workerCount; i++) {
		// per ogni thread creiamo la nostra connessione
        Firebird::AutoRelease workerAtt(
            provider->attachDatabase(
                &status,
                m_database.c_str(),
                dbpLength,
                dpb
            )
        );
		// e la nostra transazione a cui passiamo il numero di snapshot
        // per creare uno snapshot condiviso
        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)
            )
        );
		// crea un thread
        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) {
                    // incrementa il contatore atomico
                    size_t localCounter = counter++;
					// se le tabelle o le loro parti sono finite, esci
                    // dal ciclo infinito e termina il thread
                    if (localCounter >= tables.size())
                        break;
                    // ottieni una descrizione della tabella o di parte di essa
                    const auto& tableDesc = tables[localCounter];
                    // e fai l'esportazione
                    exportByTableDesc(&status, csvExport, tableDesc);
                }
                if (tra) {
                    tra->commit(&status);
                    tra.release();
                }

                if (att) {
                    att->detach(&status);
                    att.release();
                }
            }
            catch (...) {
				// se si verifica un'eccezione, salvala per
                // il successivo rilascio nel thread principale
                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);
    }
    ...

Non resta che combinare i file creati per le parti delle tabelle in un unico file per ciascuna di queste tabelle.

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

Misuriamo le prestazioni dello strumento in modalità single-threaded e multi-threaded.

Benchmark dello strumento `FBCSVExport`

Innanzitutto, esaminiamo i risultati del confronto tra le modalità di esportazione multi-threaded e single-threaded su un computer domestico di fascia media. === Windows

Sistema operativo: Windows 10 x64.
Processore: Intel Core i3 8100, 4 core, 4 thread.
Memoria: 16 GB
Sottosistema disco: NVME SSD (database), SATA SSD (cartella per i file CSV).
Firebird 4.0.4 x64

Risultati:

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

Dal risultato del test è chiaro che utilizzando due thread, l’accelerazione è stata di 1,8 volte, il che è un buon risultato. Ma l’esecuzione parallela dell’esportazione con 4 thread ha mostrato anch’essa un miglioramento di 1,8 volte. Perché non 3-4? Il fatto è che il server Firebird e l’utilità di esportazione sono in esecuzione sullo stesso computer, che ha solo 4 core. Pertanto, il server Firebird stesso utilizza 4 thread per leggere la tabella, e l’utilità FBCSVExport utilizza anch’essa 4 thread. Ovviamente, in questo caso è abbastanza difficile ottenere un’accelerazione superiore a 2 volte. Quindi, proveremo su un altro hardware, dove il numero di core è significativamente maggiore.

Linux

Sistema operativo: CentOS 8.
Processore: 2 processori Intel Xeon E5-2603 v4, totale 12 core, 12 thread.
Memoria: 32 GB
Sottosistema disco: SAS HDD (RAID 10)
Firebird 4.0.4 x64

Risultati:

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 questo caso, il numero ottimale di thread per l’esportazione è 6 (6 thread per Firebird e 6 thread per l’utilità FBCSVExport). Allo stesso tempo, siamo riusciti a ottenere un’accelerazione di 5 volte, il che indica una scalabilità abbastanza buona. Sul server Linux e sul computer Windows abbiamo utilizzato database identici, e probabilmente avrete notato che l’esportazione single-thread su Windows è stata quasi 2 volte più veloce: ciò è dovuto a un sottosistema disco più veloce (il drive NVME è molto più veloce dei drive SAS combinati in RAID).

Riepilogo

In questo articolo, abbiamo considerato come leggere efficacemente i dati dalle tabelle del DBMS Firebird utilizzando il parallelismo. Inoltre, è stato mostrato un esempio di come utilizzare alcune delle capacità del DBMS Firebird per organizzare tale lettura nel proprio software.

Un ringraziamento speciale a Vladislav Khorsun, sviluppatore core di Firebird, per l’aiuto con questo materiale.

Per qualsiasi domanda o commento, inviare un’email a [email protected].