Добавление данных в FB Embedded в несколько потоков.

[Bill Humble]

На Windows 7 x64 написанная на Delphi 6 утилита читает много больших текстовых таблиц и помещает данные из них в таблицу FB Embedded 2.5 для последующего анализа. Процесс не очень быстрый, появилась идея попробовать его распараллелить. Но первый же тест привел в уныние:

unit Unit1;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, DBXpress, DB, SqlExpr, FMTBcd, DBClient, Provider, Grids,
  DBGrids, StdCtrls, Buttons, IBDatabase, IBQuery, IBCustomDataSet, IBTable,
  ComCtrls;

type
  TForm1 = class(TForm)
    IBDatabase1: TIBDatabase;
    IBTransaction1: TIBTransaction;
    IBQuery1: TIBQuery;
    Button1: TButton;
    Button2: TButton;
    procedure FormCreate(Sender: TObject);
    procedure Button1Click(Sender: TObject);
    procedure Button2Click(Sender: TObject);
  private
    { Private declarations }
  public
    { Public declarations }
  end;

TThread1 = class(TThread)
protected
  procedure Execute; override;
end;

TThread2 = class(TThread)
protected
  procedure Execute; override;
end;

TThread3 = class(TThread)
protected
  procedure Execute; override;
end;

var
  Form1: TForm1;

implementation

uses Md5;
{$R *.dfm}


procedure TForm1.FormCreate(Sender: TObject);
begin
  DeleteFile('.\gdb\data.gdb');
  IBDatabase1.Params.Clear;
  IBDatabase1.DatabaseName:='.\gdb\data.gdb';
  IBDatabase1.Params.Add('user ''SYSDBA'' password ''masterkey'' ');
  IBDatabase1.Params.Add('page_size 4096');
  IBDatabase1.Params.Add('default character set win1251');
  IBDatabase1.CreateDatabase;

  IBQuery1.SQL.Clear;
  IBQuery1.SQL.Add('create table TEST(id int not null,');
  IBQuery1.SQL.Add('hash varchar(32))');
  IBQuery1.ExecSQL;
  IBQuery1.Close;

  IBTransaction1.Commit;
end;

procedure TForm1.Button1Click(Sender: TObject);
var
  I : Integer;
  Hash : String;
  S : String;
begin
  for I := 1 to 1000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    IBQuery1.SQL.Clear;
    IBQuery1.SQL.Add(S);
    IBQuery1.ExecSQL;
  end;
end;

procedure TForm1.Button2Click(Sender: TObject);
var
  Thread1, Thread2, Thread3 : TThread;
begin
  Thread1 := TThread1.Create(True);
  Thread1.Priority := tpIdle;
  Thread1.Resume;
  Thread2 := TThread2.Create(True);
  Thread2.Priority := tpIdle;
  Thread2.Resume;
  Thread3 := TThread2.Create(True);
  Thread3.Priority := tpIdle;
  Thread3.Resume;
end;

procedure TThread1.Execute;
var
  Q : TIBQuery;
  T : TIBTransaction;
  I : Integer;
  Hash : String;
  S : String;
begin
  Q := TIBQuery.Create(nil);
  Q.Database := Form1.IBDatabase1;
  T := TIBTransaction.Create(nil);
  T.AddDatabase(Form1.IBDatabase1);
  T.StartTransaction;
  for I := 1000001 to 2000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    Q.SQL.Clear;
    Q.SQL.Add(S);
    Q.ExecSQL;
  end;
  T.Commit;
  T.Free;
  Q.Free;
  Terminate;
end;

procedure TThread2.Execute;
var
  Q : TIBQuery;
  T : TIBTransaction;
  I : Integer;
  Hash : String;
  S : String;
begin
  Q := TIBQuery.Create(nil);
  Q.Database := Form1.IBDatabase1;
  T := TIBTransaction.Create(nil);
  T.AddDatabase(Form1.IBDatabase1);
  T.StartTransaction;
  for I := 2000001 to 3000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    Q.SQL.Clear;
    Q.SQL.Add(S);
    Q.ExecSQL;
  end;
  T.Commit;
  T.Free;
  Q.Free;
  Terminate;
end;

procedure TThread3.Execute;
var
  Q : TIBQuery;
  T : TIBTransaction;
  I : Integer;
  Hash : String;
  S : String;
begin
  Q := TIBQuery.Create(nil);
  Q.Database := Form1.IBDatabase1;
  T := TIBTransaction.Create(nil);
  T.AddDatabase(Form1.IBDatabase1);
  T.StartTransaction;
  for I := 3000001 to 4000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    Q.SQL.Clear;
    Q.SQL.Add(S);
    Q.ExecSQL;
  end;
  T.Commit;
  T.Free;
  Q.Free;
  Terminate;
end;

end.

Button1Click для миллиона записей выполняется за 55 сек, загрузка CPU 12.5%
Button2Click три потока по миллиону записей выполняются 2 мин 35 сек, загрузка CPU 21.5%

Складывается ощущение, что все три TIBQuery трудятся в одном потоке, загружая единственное ядро. Есть какая-либо возможность заставить FB Embedded исполнять запросы разных потоков на разных физических ядрах?

[Arigato]

У вас во всех потоках используется одно и тоже подключение IBDatabase1. Попробуйте в каждом потоке создать свое подключение и работать с ним.

[Bill Humble]

Стало только хуже. Код

procedure TThread1.Execute;
var
  D : TIBDatabase;
  Q : TIBQuery;
  T : TIBTransaction;
  I : Integer;
  Hash : String;
  S : String;
begin
  D := TIBDatabase.Create(nil);
  D.Params.Clear;
  D.DatabaseName:='.\gdb\data.gdb';
  D.Params.Add('user SYSDBA');
  D.Params.Add('password masterkey');
  D.Params.Add('page_size 4096');
  D.Params.Add('default character set win1251');
  D.LoginPrompt := False;
  D.Open;

  Q := TIBQuery.Create(nil);
  Q.Database := D;
  T := TIBTransaction.Create(nil);
  T.AddDatabase(D);
  Q.Transaction := T;
  T.StartTransaction;
  for I := 1000001 to 2000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    Q.SQL.Clear;
    Q.SQL.Add(S);
    Q.ExecSQL;
  end;
  T.Commit;
  T.Free;
  Q.Free;
  Terminate;
end;

procedure TThread2.Execute;
var
  D : TIBDatabase;
  Q : TIBQuery;
  T : TIBTransaction;
  I : Integer;
  Hash : String;
  S : String;
begin
  D := TIBDatabase.Create(nil);
  D.Params.Clear;
  D.DatabaseName:='.\gdb\data.gdb';
  D.Params.Add('user SYSDBA');
  D.Params.Add('password masterkey');
  D.Params.Add('page_size 4096');
  D.Params.Add('default character set win1251');
  D.LoginPrompt := False;
  D.Open;

  Q := TIBQuery.Create(nil);
  Q.Database := D;
  T := TIBTransaction.Create(nil);
  T.AddDatabase(D);
  Q.Transaction := T;
  T.StartTransaction;
  for I := 2000001 to 3000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    Q.SQL.Clear;
    Q.SQL.Add(S);
    Q.ExecSQL;
  end;
  T.Commit;
  T.Free;
  Q.Free;
  Terminate;
end;

procedure TThread3.Execute;
var
  D : TIBDatabase;
  Q : TIBQuery;
  T : TIBTransaction;
  I : Integer;
  Hash : String;
  S : String;
begin
  D := TIBDatabase.Create(nil);
  D.Params.Clear;
  D.DatabaseName:='.\gdb\data.gdb';
  D.Params.Add('user SYSDBA');
  D.Params.Add('password masterkey');
  D.Params.Add('page_size 4096');
  D.Params.Add('default character set win1251');
  D.LoginPrompt := False;
  D.Open;

  Q := TIBQuery.Create(nil);
  Q.Database := D;
  T := TIBTransaction.Create(nil);
  T.AddDatabase(D);
  Q.Transaction := T;
  T.StartTransaction;
  for I := 3000001 to 4000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));
    S := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')';
    Q.SQL.Clear;
    Q.SQL.Add(S);
    Q.ExecSQL;
  end;
  T.Commit;
  T.Free;
  Q.Free;
  Terminate;
end;

отработал за 3 минуты 55 секунд с утилизацией процессом CPU менее 21%

[Arigato]

В качестве эксперимента попробуйте переписать код потока примерно так:

T.StartTransaction;
Q.SQL.Text := 'INSERT INTO TEST (ID, HASH) VALUES (:ID, :HASH)';  // Параметрический запрос

for I := 1000001 to 2000000 do
begin
  Hash := MD5DigestToStr(MD5String(IntToStr(I)));
  
  Q.ParamByName('ID').AsInteger := I;       // Передаём значение ID
  Q.ParamByName('HASH').AsString := Hash;   // Передаём значение HASH
  Q.ExecSQL;

  if (I mod 5000) = 0 then  // Коммит каждые 5000 записей
  begin
    T.Commit;
    T.StartTransaction;  // Начинаем новую транзакцию
  end;
end;

T.Commit;  // Финальный коммит после завершения всех операций

[Bill Humble]

4 минуты 08 секунд и 11% процессорного времени.
Есть подозрения, что теперь уже прохлаждается даже одно ядро, а процессы стоят в очереди за доступом к файлу.

[Arigato]

Проблема кроется скорее всего в том, что потоки банально мешают друг другу. Ведь данные хранятся в файле. Нагрузка ЦП тут вообще не показатель, так как узким местом будет именно операции с файлами. Чуть меньше процессор стал нагружаться из-за параметрического запроса.

Можно попробовать отключить Forced Writes командой в CMD:

gfix -write async .\gdb\data.gdb

Вполне вероятно, что FB Embedded вообще не приспособлен для интенсивной многопоточной работы.

Эффект прироста производительности может дать пакетная вставка, то есть за один запрос сразу по несколько записей:

const
  BatchSize = 100;  // Размер пакета
var
  BatchCounter: Integer;
  SQLBatch: String;
...
  SQLBatch := '';  // Инициализируем пустую строку для SQL

  BatchCounter := 0;  // Счётчик для пакетов
  for I := 1000001 to 2000000 do
  begin
    Hash := MD5DigestToStr(MD5String(IntToStr(I)));

    // Формируем пакетный SQL-запрос
    if SQLBatch = '' then
      SQLBatch := 'INSERT INTO TEST (ID, HASH) VALUES (' + IntToStr(I) + ', ''' + Hash + ''')'
    else
      SQLBatch := SQLBatch + ', (' + IntToStr(I) + ', ''' + Hash + ''')';  // Добавляем новые значения

    Inc(BatchCounter);

    // Выполняем пакетный запрос, когда собрали 100 записей
    if BatchCounter >= BatchSize then
    begin
      Q.SQL.Text := SQLBatch;
      Q.ExecSQL;  // Выполняем вставку пакетом

      SQLBatch := '';  // Очищаем SQLBatch для следующего пакета
      BatchCounter := 0;  // Сбрасываем счётчик
    end;

    // Периодический коммит через каждые 1000 записей (или другой разумный лимит)
    if (I mod 1000) = 0 then
    begin
      T.Commit;
      T.StartTransaction;
    end;
  end;

  // Вставка оставшихся данных, если пакет не полностью заполнен
  if SQLBatch <> '' then
  begin
    Q.SQL.Text := SQLBatch;
    Q.ExecSQL;
  end;

  T.Commit;  // Финальный коммит

[Bill Humble]

Всю однопоточную оптимизацию я давно прошел. Вплоть до EXECUTE BLOCK.
Сейсас специально сделал неоптимальный код, чтоб он максимально загружал процессор и хорошо было видно, разлеглись потоки по ядрам или нет.

Вряд-ли диск - узкое место. На SSD, пусть и SATA, образовавшийся в результате добавления 3 миллионов записей 250-мегабайтный .GDB копируется в соседнюю папку за 2 секунды.

Исходя из того, что однопоточный код грузит одно ядро на 100% все-таки склонен думать, что узкое место - какие-то вычисления самого FB, которые не умеют распараллеливаться.

Склонен оставить задачу, как бесперспективную. Все-таки в документации написано, что FBE "Thread safe", но никто не обещал, что он "Thread optimized". Спасибо за помощь.

[Arigato]

Сообщение от Bill Humble

Вряд-ли диск - узкое место.

Дело не в скорости диска, а в разделении ресурсов. Проще говоря, пока один из потоков производит запись данных, другие потоки вынуждены ждать. Почему я и предложил пакетный SQL, по идеи он должен ускорить работу, даже в случае однопотока за счет уменьшения количества обращений к СУБД.

[Bill Humble]

Пакетный SQL действительно повышает производительность. Может раза в полтора максимум. Это не тот выигрыш, который я хотел получить, заставив исполнять каждый IBQuery в отдельном потоке.