Leitura Paralela de Dados no Firebird

D.Simonov, V.Horsun

versão 1.0.5 de 05.12.2023

Este material é patrocinado e criado com o patrocínio e apoio da IBSurgeon www.ib-aid.com, fornecedora do HQbird (distribuição avançada do Firebird) e fornecedora de serviços de otimização de desempenho, migração e suporte técnico para Firebird.

O material é licenciado sob a Public Documentation License https://www.firebirdsql.org/file/documentation/html/en/licenses/pdl/public-documentation-license.html

Materiais relacionados:

• Livro gratuito “Detailed New Featuires Of Firebird 5”: disponível como HTML e como PDF (119 páginas).

Prefácio

O Firebird 5.0 introduziu a capacidade de usar paralelismo ao criar um backup usando o utilitário gbak, entre outras funções paralelas. Inicialmente, essa função apareceu no HQbird 2.5, depois no HQbird 3.0 e 4.0, e então foi portada para o Firebird 5.0.

Neste artigo, consideraremos as funções usadas ao criar backups paralelos dentro do utilitário gbak. Também mostraremos como elas podem ser usadas em suas aplicações para leitura paralela de dados.

É importante observar que aqui não estamos falando de varredura paralela de tabelas dentro do mecanismo Firebird ao executar consultas SQL, mas sim de leitura de dados dentro de fluxos paralelos da sua aplicação.

Ferramenta de exemplo FBCSVExport

Para demonstrar a leitura paralela de dados do SGBD Firebird, foi escrita uma utilidade de exemplo que exporta dados de uma ou mais tabelas para o formato CSV.

Sua descrição e seu código-fonte aberto estão aqui: https://github.com/IBSurgeon/FBCSVExport.git

Como você pode ver na descrição da utilidade e no artigo abaixo, através do processamento paralelo é possível exportar dados e realizar outras operações paralelas de 2 a 10 vezes mais rápido do que em 1 thread (dependendo do hardware).

Para qualquer dúvida, entre em contato pelo e-mail [email protected].

Leitura paralela

Vamos pensar em como ler dados de várias tabelas em paralelo. Como você sabe, o Firebird permite executar consultas em paralelo somente se cada consulta for executada em uma conexão separada.

Vamos criar um pool de threads de trabalho. A thread principal da aplicação também é uma thread de trabalho, portanto, o número de threads de trabalho adicionais deve ser N - 1, onde N é o número total de trabalhadores paralelos. Cada thread de trabalho executará sua própria conexão e transação.

O primeiro problema: como garantir a consistência dos dados lidos?

Leitura consistente de dados

Como cada thread de trabalho usa sua própria conexão e sua própria transação, surge o problema de leituras inconsistentes - se a tabela for alterada simultaneamente por outros usuários, os dados lidos podem ser inconsistentes. No modo de thread única, o gbak usa uma transação com modo de isolamento SNAPSHOT, o que possibilita ler informações consistentes no início da transação SNAPSHOT. Mas aqui temos múltiplas transações e é necessário que elas vejam o mesmo “snapshot” para que leiam os mesmos dados imutáveis.

O mecanismo para criar um snapshot compartilhado para diferentes transações com o modo de isolamento SNAPSHOT foi introduzido no Firebird 4.0 (originalmente no HQBird 2.5, mas no Firebird 4.0/HQBird 4.0 é mais simples e eficiente). Existem duas maneiras de criar um snapshot compartilhado:

1. Com SQL

• obter o número do snapshot da transação principal (que é iniciada na thread de trabalho principal).

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

• iniciar outras transações com o seguinte SQL:

SET TRANSACTION SNAPSHOT AT NUMBER snapshot_number

onde snapshot_number é o número obtido pela consulta anterior.

2. Com API

• obter o número do snapshot da transação principal (que é iniciada na thread de trabalho principal) com a função isc_transaction_info ou ITransaction.getInfo com a tag fb_info_tra_snapshot_number;

• iniciar outras transações com a tag isc_tpb_at_snapshot_number com o número de snapshot obtido.

A ferramenta de exemplo FBCSVExport, assim como o gbak, usa a segunda abordagem. Essas abordagens podem ser misturadas - por exemplo, obter o número do snapshot com SQL e usar o número obtido para iniciar outras transações com API, ou vice-versa.

No FBCSVExport, obtemos o número do snapshot com o seguinte código:

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

Para iniciar a transação com o número do snapshot, usamos o seguinte código:

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

Agora os dados lidos de diferentes conexões serão consistentes, então podemos distribuir a carga entre as threads de trabalho.

Como exatamente distribuir a carga entre as threads de trabalho? No caso de exportação completa de todas as tabelas ou de uma cópia de backup, a opção mais simples será uma thread de trabalho por tabela. Mas com essa abordagem, temos o seguinte problema: se houver muitas tabelas pequenas em um banco de dados e uma tabela grande, ou até mesmo apenas uma tabela e ela for enorme, não veremos a melhoria. Nesse caso, alguma thread receberá uma tabela grande, e as demais threads ficarão ociosas. Para evitar que isso aconteça, é necessário processar uma tabela grande em partes.

Dividir tabela grande em partes

Digamos que temos apenas uma tabela grande que queremos ler por completo e o mais rápido possível. Propõe-se dividi-la em várias partes e ler cada parte de seu próprio fluxo de forma independente. Cada thread deve ter sua própria conexão com o banco de dados.

Nesse caso, surgem as seguintes perguntas:

• Em quantas partes de processamento a tabela deve ser dividida?

• Qual é a melhor maneira de dividir a tabela em termos de acesso aos dados?

Vamos responder a essas perguntas em ordem.

Em quantas partes de processamento a tabela deve ser dividida?

Vamos supor o cenário ideal - o servidor e o cliente são dedicados ao Firebird, ou seja, todas as CPUs estão completamente à nossa disposição. Então é recomendado:

a) Usar como número máximo de partes paralelas o dobro do número de núcleos de CPU no servidor. Por que 2x núcleos? Sabemos com certeza que haverá atrasos relacionados a E/S, então podemos permitir algum uso extra de CPU. No entanto, esse número deve ser considerado como uma configuração inicial; na prática, depende dos dados.

b) Levar em conta o número de núcleos no cliente: se houver muitos mais no servidor (situação usual), então pode fazer sentido limitar ainda mais o número de partes da partição, para não sobrecarregar o cliente (ele não conseguirá processar mais de qualquer forma, e os custos de alternância de fluxos não desaparecem). Será possível decidir com mais precisão monitorando a carga de CPU do cliente e do servidor - se estiver 100% no cliente, mas visivelmente menor no servidor, então faz sentido reduzir o número de partes.

c) se o cliente e o servidor forem o mesmo host, veja (a).

Se o cliente e/ou servidor estiverem ocupados com outra coisa, pode ser necessário reduzir o número de partes. Isso também pode ser afetado pela capacidade dos discos no servidor de processar muitas solicitações de E/S simultaneamente (monitore o tamanho da fila e o tempo de resposta).

Qual é a melhor maneira de dividir a tabela em termos de acesso aos dados?

Para implementar um processamento paralelo eficaz, é importante garantir uma distribuição uniforme dos trabalhos entre os manipuladores e minimizar sua sincronização mútua. Além disso, você precisa lembrar que a sincronização dos manipuladores pode ocorrer tanto no lado do servidor quanto no lado do cliente. Por exemplo, vários manipuladores não devem usar a mesma conexão com o banco de dados. Um exemplo menos óbvio: é ruim se diferentes manipuladores leem registros das mesmas páginas de banco de dados. Por exemplo, quando dois manipuladores leem registros pares e ímpares - isso não é eficaz. A sincronização no cliente pode ocorrer durante a distribuição de tarefas, durante o processamento dos dados recebidos (alocação de memória para resultados) e assim por diante.

Um dos problemas com a partição “justa” é que o cliente não sabe como os registros estão distribuídos pelas páginas (e pelas chaves de índice), quantos registros ou páginas de dados existem (para tabelas grandes, contar o número de registros antecipadamente levaria muito tempo).

Vamos ver como o gbak resolve esse problema.

Para o gbak, uma unidade de trabalho é um conjunto de registros de páginas de dados (DPs) pertencentes à mesma página de ponteiro (PP). Por um lado, é um número bastante grande de registros para manter o manipulador ocupado sem precisar solicitar frequentemente um novo pedaço de dados (sincronização). Por outro lado, mesmo que esses conjuntos de registros não tenham exatamente o mesmo tamanho, isso permitirá carregar os trabalhadores de forma relativamente uniforme. Ou seja, é bem possível que um trabalhador leia N registros de uma PP, e o outro leia M registros, e M seja bem diferente de N. Essa abordagem não é ideal, mas é bastante simples de implementar e geralmente é bastante eficaz, pelo menos em grandes escalas (com dezenas ou centenas (ou mais) de PPs).

Como obter o número de PPs (Páginas de Ponteiro) para uma determinada tabela? É bem fácil e, o mais importante, rápido calculá-lo a partir da tabela RDB$PAGES:

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

Em seguida, poderíamos simplesmente dividir o número de PPs pelo número de trabalhadores e dar a cada trabalhador seu próprio pedaço. Isso é adequado para o cenário em que o processamento paralelo é feito pelo desenvolvedor que conhece a distribuição dos dados. Mas, para um cenário mais comum, não há garantia de que esses “grandes” pedaços significarão a mesma quantidade de trabalho. Não estamos interessados em ver a situação em que 15 trabalhadores terminaram seu trabalho e ficam ociosos, e o 16º lê seus 10 milhões de registros por um longo tempo.

É por isso que o gbak faz isso de forma diferente. Há um coordenador de trabalho que emite para cada processador 1 PP por vez. O coordenador sabe quantas PPs existem no total e quantas já foram emitidas para trabalho. Quando o trabalhador termina a leitura de seus registros, ele contata o coordenador para obter um novo número de PP. Isso continua até que as PPs se esgotem (ou haja trabalhadores ativos). É claro que essa interação dos trabalhadores com o coordenador requer sincronização. A experiência mostra que a quantidade de trabalho dada por uma PP permite não sincronizar com muita frequência. Essa abordagem permite carregar praticamente todos os trabalhadores (e, portanto, os núcleos de CPU) de forma uniforme com trabalho, independentemente do número real de registros pertencentes a cada PP.

Como o handler lê registros do seu PP? Para isso, a partir do Firebird 4.0 (apareceu pela primeira vez no HQBird 2.5) existe uma função embutida MAKE_DBKEY(). Com a ajuda dela, você pode obter o RDB$DB_KEY (número físico do registro) para o primeiro registro no PP especificado.

E com a ajuda desses RDB$DB_KEY os registros necessários são selecionados:

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)

Por exemplo, se você definir loPP = 0 e hiPP = 1, então todos os registros com PP = 0 serão lidos, e somente dele.

Agora que você tem uma ideia de como o gbak funciona, você pode passar para uma descrição da implementação do utilitário FBCSVExport.

Implementação do utilitário FBCSVExport

O utilitário FBCSVExport é projetado para exportar dados de tabelas de banco de dados Firebird para o formato CSV.

Cada tabela é exportada para um arquivo chamado .csv. No modo normal (single-threaded), os dados das tabelas são exportados sequencialmente em ordem alfabética dos nomes das tabelas.

No modo paralelo, as tabelas são exportadas em paralelo, cada tabela em uma thread separada. Se a tabela for muito grande, ela é dividida em partes, e cada parte é exportada em um fluxo separado. Para cada parte de uma tabela grande, um arquivo separado é criado com o nome .csv.partN, onde N é o número da parte.

Quando todas as partes de uma tabela grande são exportadas, os arquivos das partes são mesclados em um arquivo chamado .csv.

Uma expressão regular é usada para especificar quais tabelas serão exportadas. Apenas tabelas regulares podem ser exportadas (tabelas do sistema, GTT, views, tabelas externas não são suportadas). As expressões regulares devem estar em sintaxe SQL, ou seja, aquelas que são usadas no predicado SIMILAR TO.

Para selecionar uma lista de tabelas exportadas, bem como uma lista de seus PPs no modo multi-threaded, usamos a seguinte consulta:

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

No modo single-threaded, esta consulta pode ser simplificada para

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

No modo single-threaded, os valores dos campos PAGE_SEQUENCE e PP_CNT não são usados; eles são adicionados à consulta para unificar as mensagens de saída.

O resultado desta consulta é formado em um vetor de estruturas:

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

Este vetor é preenchido usando uma função declarada como:

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

O último parâmetro singleWorker alterna o modo de preenchimento de std::vector; se singleWorker = true, então a consulta para o modo single-threaded é usada; se singleWorker = false, então uma consulta mais cara e complexa é usada para o modo multi-threaded. Não vou fornecer a implementação em si, é bastante simples e você pode vê-la no código-fonte do projeto.

Para exportar uma tabela para o formato CSV, a classe CSVExportTable foi desenvolvida, que contém os seguintes métodos:

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

O método prepare é destinado a construir e preparar uma consulta que é usada para exportar uma tabela para o formato CSV. A consulta interna é construída de forma diferente dependendo do parâmetro withDbkeyFilter. Se withDbkeyFilter = true, então a consulta é construída com filtro pelo intervalo RDB$DB_KEY:

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

caso contrário, uma consulta simplificada é usada:

SELECT *
FROM tableName

O valor do parâmetro withDbkeyFilter é definido como true se o modo multi-threaded for usado e a tabela for grande. Consideramos a tabela grande se pp_cnt > 1.

O método printHeader é destinado a imprimir o cabeçalho de um arquivo CSV (nomes das colunas da tabela).

O método printData imprime os dados da tabela em um arquivo CSV a partir do número de página PP ppNum, se a consulta foi preparada usando um filtro pelo intervalo RDB$DB_KEY, e todos os dados da tabela caso contrário.

Agora vamos olhar o código para o modo single-threaded

...

// Abrindo a conexão principal
Firebird::AutoRelease att(
    provider->attachDatabase(
        &status,
        m_database.c_str(),
        dbpLength,
        dpb
    )
);

// Iniciando a transação principal no modo de 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)
    )
);
// Obtendo uma lista de tabelas usando a expressão regular em m_filter.
// m_parallel define o número de threads paralelas; quando é igual a 1,
// então uma consulta simplificada é usada para obter a lista de tabelas,
// caso contrário, uma lista de PPs e seu número é gerada para cada tabela.
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) {
        // não faz sentido usar um filtro de intervalo RDB$DB_KEY aqui
        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);
    }
}

Tudo aqui é bastante simples e não requer explicação adicional, então vamos passar para a parte multi-threaded.

Para que a exportação ocorra no modo multi-threaded, é necessário criar threads de trabalho adicionais m_parallel - 1. Por que o número de threads adicionais é 1 a menos? Sim, porque a thread principal também exportará dados e ela é igual às threads adicionais. Vamos mover a parte comum do fluxo principal e adicional para uma função separada:

void ExportApp::exportByTableDesc(Firebird::ThrowStatusWrapper* status, FBExport::CSVExportTable& csvExport, const TableDesc& tableDesc)
{
    // Se tableDesc tem pp_cnt > 1, então ele descreve apenas parte da tabela, e é necessário construir
    // consulta usando um filtro pelo intervalo 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 esta não é a primeira parte da tabela, então escreva esta parte no arquivo .csv.part, onde
    // N - número do PP. Posteriormente, as partes da tabela serão combinadas em um único arquivo .csv
    if (tableDesc.page_sequence > 0) {
        fileName += ".part_" + std::to_string(tableDesc.page_sequence);
    }
    csv::CSVFile csv(m_outputDir / fileName);
    // O cabeçalho do arquivo CSV deve ser impresso apenas na primeira parte da tabela.
    if (tableDesc.page_sequence == 0 && m_printHeader) {
        csvExport.printHeader(status, csv);
    }
    csvExport.printData(status, csv, tableDesc.page_sequence);
}

As descrições das tabelas ou suas partes estão localizadas em um vetor comum com estruturas TableDesc. Deste vetor, cada thread de trabalho pega uma tabela ou a próxima parte. Para evitar corridas de dados, é necessário sincronizar o acesso ao recurso compartilhado. Mas std::vector em si não muda, então você pode sincronizar apenas a variável compartilhada, que é o índice neste vetor. Isso pode ser facilmente feito usando std::atomic como tal variável.

if (m_parallel == 1) {
    ...
}
else {
    // Determinando o número de threads de trabalho adicionais
    const auto workerCount = m_parallel - 1;

    // Obtendo o número do snapshot da transação principal
    auto snapshotNumber = getSnapshotNumber(&status, tra);
    // variável para armazenar a exceção dentro da thread
    std::exception_ptr exceptionPointer = nullptr;
    std::mutex m;
	// contador atômico
    // é o índice da próxima tabela ou parte dela
    std::atomic<size_t> counter = 0;
    // pool de threads de trabalho
    std::vector<std::thread> thread_pool;
    thread_pool.reserve(workerCount);
    for (int i = 0; i < workerCount; i++) {
		// para cada thread criamos nossa própria conexão
        Firebird::AutoRelease workerAtt(
            provider->attachDatabase(
                &status,
                m_database.c_str(),
                dbpLength,
                dpb
            )
        );
		// e nossa transação para a qual passamos o número do snapshot
        // para criar um snapshot compartilhado
        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)
            )
        );
		// criando uma 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) {
                    // incrementando o contador atômico
                    size_t localCounter = counter++;
					// se as tabelas ou suas partes acabaram, saia
                    // do loop infinito e termine a thread
                    if (localCounter >= tables.size())
                        break;
                    // obtendo uma descrição da tabela ou parte dela
                    const auto& tableDesc = tables[localCounter];
                    // e fazendo a exportação
                    exportByTableDesc(&status, csvExport, tableDesc);
                }
                if (tra) {
                    tra->commit(&status);
                    tra.release();
                }

                if (att) {
                    att->detach(&status);
                    att.release();
                }
            }
            catch (...) {
				// se uma exceção ocorrer, salve-a para
                // liberação subsequente na thread principal
                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);
    }
    ...

Resta apenas combinar os arquivos que foram criados para as partes das tabelas em um único arquivo para cada uma dessas tabelas.

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

Vamos medir o desempenho da ferramenta no modo de thread única e no modo multithread.

Benchmark da ferramenta FBCSVExport

Primeiro, vamos analisar os resultados da comparação entre os modos de exportação multithread e de thread única em um computador doméstico moderado. === Windows

• Sistema operacional: Windows 10 x64.

• Processador: Intel Core i3 8100, 4 núcleos, 4 threads.

• Memória: 16 GB

• Subsistema de disco: NVME SSD (banco de dados), SATA SSD (pasta para armazenar arquivos CSV).

• Firebird 4.0.4 x64

Resultados:

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

Pelo resultado do teste, fica claro que ao usar duas threads, a aceleração foi de 1,8 vezes, o que é um bom resultado. Mas a execução paralela da exportação em 4 threads também mostrou uma melhoria de 1,8 vezes. Por que não 3-4? O fato é que o servidor Firebird e o utilitário de exportação estão rodando no mesmo computador, que possui apenas 4 núcleos. Assim, o próprio servidor Firebird usa 4 threads para ler a tabela, e o utilitário FBCSVExport também usa 4 threads. Obviamente, neste caso, é bastante difícil alcançar uma aceleração de mais de 2 vezes. Portanto, vamos tentar em outro hardware, onde o número de núcleos é significativamente maior.

Linux

• Sistema operacional: CentOS 8.

• Processador: 2 processadores Intel Xeon E5-2603 v4, total de 12 núcleos, 12 threads.

• Memória: 32 GB

• Subsistema de disco: SAS HDD (RAID 10)

• Firebird 4.0.4 x64

Resultados:

[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

Neste caso, o número ideal de threads para exportação é 6 (6 threads para o Firebird e 6 threads para o utilitário FBCSVExport). Ao mesmo tempo, conseguimos alcançar uma aceleração de 5 vezes, o que indica uma escalabilidade bastante boa. No servidor Linux e no computador Windows, usamos bancos de dados idênticos, e você provavelmente notou que a exportação de thread única no Windows foi quase 2 vezes mais rápida: isso se deve a um subsistema de disco mais rápido (o drive NVME é muito mais rápido que os drives SAS combinados em RAID).

Resumo

Neste artigo, consideramos como ler efetivamente dados de tabelas do SGBD Firebird usando paralelismo. Além disso, foi mostrado um exemplo de como você pode usar alguns dos recursos do SGBD Firebird para organizar essa leitura em seu software.

Muitos agradecimentos a Vladislav Khorsun, desenvolvedor principal do Firebird, pela ajuda com este material.

Para quaisquer perguntas ou comentários, envie um e-mail para [email protected].