Версия для печати темы
Нажмите сюда для просмотра этой темы в оригинальном формате
Форум программистов > Delphi: Общие вопросы > Ограничение количества соединений с базой


Автор: drkot 7.6.2014, 02:17
Общие положения:
- есть N нитей (N постоянно изменяется от 0 то тысяч)
- каждая нить пишет в базу
- для данной задачи выделено 10 соединений с базой

Проблема:
- идут отказы в обслуживании от базы

Вопрос: как 1000 нитей заставить работать через 10 соединений с базой?

Работа видится примерно так:
1) есть пул из 10 соединений
2) запрос свободного соединения из пула или ожидание пока освободится
3) работа с базой
4) возврат соединения в пул

Собственно затруднение вызывает 2й пункт. 


PS: заставить работать через ОДНО соединение с базой проблем не вызывает, но это не эффективно.

Автор: PointerToNil 7.6.2014, 07:46
WaitForMultipleObjects?

Автор: drkot 7.6.2014, 15:08
и кого ждать?

Автор: PointerToNil 7.6.2014, 16:09
CreateEvent / SetEvent / ResetEvent

Автор: drkot 7.6.2014, 17:12
PointerToNil, Вы прямо таки "капитан очевидность" smile
Вы говорите "ЧЕМ", а вопрос стоит "КАК".... Велосипед склепать не проблема, только нет гарантии что он поедет...
Наверняка есть типовой алгоритм решения задачи, он и интересует.

Автор: PointerToNil 7.6.2014, 17:30
drkot, вы из каких вариантов-то выбираете? огласите, плиз
я просто не вижу тут ни вариантов, ни каких-то проблем или непонятностей

кстати, насчет велосипедов - конечно, он возможен: массив из 10 булеанов - признаков "занято" по соединениям, попытка занять свободный нитью делается через InterlockedExchange (если вернул то же, что и положили - значит, кто-то успел раньше), а если перебор в цикле от 1 до 10 не обнаруживает свободного - небольшой sleep() и снова перебор
вот это был бы велосипед, но я предложил специальные функции, которые долго изобретали, реализовывали и тестировали программисты компании microsoft, а другие старались и описывали их в msdn и толстых книгах 

Автор: drkot 7.6.2014, 18:26
PointerToNil, конкретный набор функций предполагает конкретное решение на их основе.
Ваш набор: 
Код

WaitForMultipleObjects
CreateEvent / SetEvent / ResetEvent

предполагает что?... Ставить Event на каждое соединение и ждать их? Как минимум Eventы должны быть с автосбросом и есть ограничение на 64 соединения. Из явных проблем, о которых стоит упомянуть, это именование Eventов, так как при совпадении имен можем получить не предсказуемый эффект. Из не явных затруднена обработка и отслеживание упавших соединений и изменение их общего количества.

Как по  мне, 
- либо надо использовать один Event (который срабатывает при освобождение любого соединения) и оборачивать счетчик и прочее в критическую секцию 
- либо Semaphore (но как его прикрутить к данной задаче пока не понятно), возможно он и не применим тут...

Главное чтобы не возникла проблема "узкого горлышка", а это уже конкретный алгоритм...
В том же интерпретаторе PHP есть подобная реализация, да и не только в нем...

Автор: PointerToNil 7.6.2014, 19:07
> Ставить Event на каждое соединение и ждать их?
ну как бы ничего другого даже в голову не приходит

> Как минимум Eventы должны быть с автосбросом 
какой "автосброс", зачем?

> есть ограничение на 64 соединения. 
вы же сказали, что их всего 10 - к чему тут 64??

> Из явных проблем, о которых стоит упомянуть, это именование Eventов
а зачем их именовать-то??
завели массив из 10 THandle/интегеров и сложили туда

> затруднена обработка и отслеживание упавших соединений и изменение их общего количества
как затруднена и по сравнению с чем? надеюсь, вы ведь сами обрабатываете эти падения?
если оно упало во время занятости, так и восстанавливаться будет, сохраняя занятость
если упало, будучи свободным, то займется, обнаружится факт падения и тоже будет восстанавливаться, будучи занятым

> либо надо использовать один Event (который срабатывает при освобождение любого соединения)
тогда вам все равно понадобится что-то типа массива из 10 булеанов для признаков занятости каждого конкретного соединения
я предлагаю вместо них массив из 10 ивентов

> либо Semaphore
в семафоре есть счетчик, но вам-то не достаточно одного счетчика свободных или занятых соединений - вам нужны 10 отдельных признаков занятости/свободы для каждого конкретного из 10 соединений 

> Главное чтобы не возникла проблема "узкого горлышка"
это обсуждайте с тем, кто вам ограничение в 10 соединений предписал - этого вполне может и мало оказаться
но сделать транзакции нитей с БД как можно короче - возможно, в вашей компетенции 

вообще говоря, это уже работающую систему нужно профилировать на тему того, что насколько загружено и что является горлышком (сеть, процессор или диски и на каком из серверов - БД или веб и т.д.) и что нужно расширять (или будет видно, что все работает вполсилы из-за искусственных ограничений в софте)

Автор: drkot 7.6.2014, 20:01
Цитата(PointerToNil @  7.6.2014,  19:07 Найти цитируемый пост)
какой "автосброс", зачем?

а как Вы видите работу WaitForMultipleObjects на которой ждет 10 потоков без автосброса?

Цитата(PointerToNil @  7.6.2014,  19:07 Найти цитируемый пост)
что их всего 10 - к чему тут 64??

нужно всегда думать о будущем

хотел дальше отвечать, но передумал... нет желания комментировать глупости... 

Автор: PointerToNil 7.6.2014, 20:15
> а как Вы видите работу WaitForMultipleObjects на которой ждет 10 потоков без автосброса?
как я представляю себе внутреннюю реализацию WaitForMultipleObjects? это точно стоит описывать?
но, если что - таймаут в ней задается четвертым параметром - если вдруг он вам понадобится
и, если что, можете еще завести специальный 11-й ивент для "аварийной сигнализации" (типа "кончай работу, пора обедать")

> нужно всегда думать о будущем
проблем с увеличением числа соединений не вижу никаких - ну, расширили динамический массив, завели новый ивент, аналогично и с уменьшением

> хотел дальше отвечать, но передумал... нет желания комментировать глупости...  
правильно, за работу приниматься пора - там и разберетесь что к чему, даже метод тыка быстрее форумных дискуссий, я уж не говорю о rtfm

Автор: Sajtran 19.8.2014, 19:01
Добрый день, коллеги
Поделитесь наработками, что сделали
самому пришлось решать похожую проблему
мой код ниже, есть большие сомнения ...

Код

unit ResPools;
{$mode objfpc}{$H+}
interface
uses
  Classes, SysUtils,SyncObjs;

type

  { ResourcePool }

  generic ResourcePool<TObj>=class(TObject)
   private
    Event:TEventObject;
    Objs:array of TObj;
    FUseCount:integer;
    function GetSize: integer;inline;
    function GetWaitCount: integer;
   public
    constructor Create(ACount:integer);
    destructor Destroy;override;
    function Lock(Timeout : Cardinal=0):TObj;
    procedure Unlock(const O:TObj);
    property WaitCount:integer read GetWaitCount;
    property UseCount:integer read FUseCount;
    property Size:integer read GetSize;
  end;


implementation

{ ResourcePool }

function ResourcePool.GetWaitCount: integer;
begin
  Result:=FUseCount-Size;
  if Result<0 then Result:=0;
end;

function ResourcePool.GetSize: integer;
begin
  Result:=Length(Objs);
end;

constructor ResourcePool.Create(ACount: integer);
var i:integer;
begin
  inherited Create;
  Event:=TEventObject.Create(nil,false,false,'');
  SetLength(Objs,ACount);
  for i:=0 to ACount-1 do begin
    Objs[i]:=TObj.Create;
  end;

end;

destructor ResourcePool.Destroy;
var i:integer;
begin
  for i:=0 to Size-1 do begin
    FreeAndNil(Objs[i]);
  end;
  FreeAndNil(Event);
  inherited Destroy;
end;

function ResourcePool.Lock(Timeout: Cardinal): TObj;
var i:integer;
  b:boolean;
begin
  b:=false;
  repeat
    if InterLockedIncrement(FUseCount)>Size then begin
      b:=true;
      Event.WaitFor(Timeout);
    end;
    for i:=Size-1 downto 0 do begin
      Result:=Objs[i];
      if Result=nil then continue;
      if InterlockedCompareExchange(Pointer(Objs[i]),nil,Pointer(Result))=Pointer(Result) then exit;
    end;

    InterLockedDecrement(FUseCount);
  until (Timeout>0)and b;
  Result:=nil;
end;

procedure ResourcePool.Unlock(const O: TObj);
var i:integer;
begin
  for i:=Size-1 downto 0 do begin
    if InterlockedCompareExchange(Pointer(Objs[i]),Pointer(O),nil)=nil then break;
  end;

  if InterLockedDecrement(FUseCount)>Size then
    Event.SetEvent;
end;

end.

Автор: drkot 21.8.2014, 07:50
Во первых событие должно быть с автосбросом.
Во вторых не понятно какие объекты заняты, а какие свободны.
Про странный цикл и смысл условия в нем я вообще молчу... все таки не смолчал...

Но ход мыслей правильный.

Автор: Sajtran 23.8.2014, 19:50
Цитата(drkot @ 21.8.2014,  09:50)
Во первых событие должно быть с автосбросом.
Во вторых не понятно какие объекты заняты, а какие свободны.
Про странный цикл и смысл условия в нем я вообще молчу... все таки не смолчал...

Но ход мыслей правильный.

Код

      constructor Create(EventAttributes : PSecurityAttributes;
        AManualReset,InitialState : Boolean;const Name : string);

а разве у меня не автосброс?
Код

Event:=TEventObject.Create(nil,false,false,'');

и что в цикле не нравится, просто объект забирается, а потом возвращается
    function Lock(Timeout : Cardinal=0):TObj; - забирает
    procedure Unlock(const O:TObj);                - отдаёт

Добавлено через 3 минуты и 44 секунды
Unlock правда ещё подправил немного
Код

procedure ResourcePool.Unlock(const O: TObj);
var i:integer;
begin
  for i:=Size-1 downto 0 do begin
    if InterlockedCompareExchange(Pointer(Objs[i]),Pointer(O),nil)=nil then break;
  end;

  if InterLockedDecrement(FUseCount)>=Size then
    Event.SetEvent;
end;   

Автор: PointerToNil 24.8.2014, 20:41
а у меня вот такой минимализм получился:
Код

type
  TPoolItemState = (isFree, isBusy);
  ItemPool = class
  private
    FFlags: array of TItemState;
  public
    constructor Create(size: integer);
    destructor Destroy; override;
    function AcquirePoolItem(try_count: integer = -1): integer;
    procedure FreePoolItem(index: integer);
  end;

implementation

function ItemPool.Create(size: integer);
begin
  SetLength(FFlags, size); // default = TPoolItemState(0) = isFree
end;

destructor ItemPool.Destroy;
begin
  SetLength(FFlags, 0);
end;

function ItemPool.AcquirePoolItem(try_count: integer = -1): integer;
var i, try_n: integer;
begin
  try_n := 0;
  repeat
    for i:=0 to SizeOf(FFlags)-1 do
      if (FFlags[i] = isFree) and (InterLockedExchange(FFlags[i], isBusy) = isFree) then 
        begin Result := i; exit end;
    sleep(10); inc(try_n); 
  until  (try_count > 0) and (try_n >= try_count); 
  Result := -1;
end;

procedure FreePoolItem(index: integer);
var i:integer;
begin
  FFlags[index] = isFree;
end;

(чёта я подумал, зачем какие-то еще ивенты, если есть массив признаков занятости и все равно цикл по нему крутить)
и "объектов пула" внутри нет - SRP!
try_count может быть менее удобным, чем Timeout - это можно переделать

Автор: drkot 24.8.2014, 22:50
Цитата(Sajtran @  23.8.2014,  20:50 Найти цитируемый пост)
и что в цикле не нравится, просто объект забирается, а потом возвращается

в целом нет гарантии (на мой взгляд), получения экземпляра объекта. Возможно это ошибочное мнение, но думается код получения и возврата желательно делать в критической секции.

Сам реализовывал примерно так. В качестве пула использовался список.
Код

  if FEvent.WaitFor(TimeOut) = wrSignaled then
  begin
    FListSection.Enter;
    try
      try
        { TODO : Захват объекта }
    finally
      FListSection.Leave;
    end;
    if FCountCurrent < FCountMaximum then
      FEvent.SetEvent;
  end
  else
    Result:= nil;


Код

  FListSection.Enter;
  try
      { TODO : Возврат объекта }
  finally
    FListSection.Leave;
  end;
  FEvent.SetEvent;

возможно это избыточно... Но для меня важна стабильность при активной конкуренции потоков за ресурс.
Вообще простейший пул можно сделать на базе TThreadList.

Цитата(Sajtran @  23.8.2014,  20:50 Найти цитируемый пост)
а разве у меня не автосброс?

Извините, не заметил.

Цитата(PointerToNil @  24.8.2014,  21:41 Найти цитируемый пост)
зачем какие-то еще ивенты

ивенты для того, чтоб 100500 потоков не крутили один и тот же цикл в ожидании освободившегося объекта. Как то так.

Цитата(PointerToNil @  24.8.2014,  21:41 Найти цитируемый пост)
а у меня вот такой минимализм получился:

наверно надо провести нагрузочное тестирование и посмотреть насколько этот минимализм стабилен.

Автор: PointerToNil 24.8.2014, 23:49
Цитата(drkot @  24.8.2014,  22:50 Найти цитируемый пост)
ивенты для того, чтоб 100500 потоков не крутили один и тот же цикл в ожидании освободившегося объекта. Как то так.
если 100500 потоков вызовут Event.WaitFor, то все они как бы "приостановятся". но ведь системный код (WaitForSingleObject), исполняющийся при этом в контексте каждого из потоков отдельно (т.е. в числе 100500 экземпляров), будет при этом что-то делать! есть веские подозрения, что таки крутить цикл (скорее всего, тоже со sleep-ом или подобным) - иных вариантов просто нет

Автор: drkot 25.8.2014, 14:11
Цитата(PointerToNil @  25.8.2014,  00:49 Найти цитируемый пост)
иных вариантов просто нет

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

Автор: PointerToNil 25.8.2014, 14:59
Цитата(drkot @  25.8.2014,  14:11 Найти цитируемый пост)
сознается некая очередь потоков, и как минимум крутится один цикл, а скорее всего все висит на аппаратном прерывании таймера. Следовательно нулевое потребление ресурсов. 
думается, всё это - часть функциональности sleep() (т.е. по крайней мере должна и в нем быть, а у меня он есть)
(halt + аппаратное прерывание таймера - применимо только для того редчайшего случая, когда ВСЕ потоки вызовут sleep)

Автор: drkot 25.8.2014, 15:15
PointerToNil,  предлагаю сделать тест и посмотреть.
Пул из 15 объектов и 50 потоков. После захвата объекта поток спит 100мс.
в качестве оценочных параметров можно считать: среднее время ожидания захвата, количество захватов (каждым потоком), количество отказов в захвате.

Автор: Sajtran 28.8.2014, 07:47
спс, PointerToNil, ваша 
Код

(try_count: integer = -1)

заметил у себя фигню, ивенты просто не работали (не ждали :-() 

drkot, я не думаю что сильно стоит бояться FreeLock-код. Вот, например, зачем состояния хранить, если можно объект просто изъять

вот подправка
Код

unit ResPools;
{$mode objfpc}{$H+}
interface
uses
  Classes, SysUtils,SyncObjs;

type

  { ResourcePool }

  generic ResourcePool<TObj>=class(TObject)
   private
    Event:TEventObject;
    Objs:array of TObj;
    FUseCount:integer;
    function GetSize: integer;inline;
    function GetWaitCount: integer;
   public
    constructor Create(ACount:integer);
    destructor Destroy;override;
    function Lock(Timeout : Cardinal=-1):TObj;
    procedure Unlock(const O:TObj);
    property WaitCount:integer read GetWaitCount;
    property UseCount:integer read FUseCount;
    property Size:integer read GetSize;
  end;


implementation

{ ResourcePool }

function ResourcePool.GetWaitCount: integer;
begin
  Result:=FUseCount-Size;
  if Result<0 then Result:=0;
end;

function ResourcePool.GetSize: integer;
begin
  Result:=Length(Objs);
end;

constructor ResourcePool.Create(ACount: integer);
var i:integer;
begin
  inherited Create;
  Event:=TEventObject.Create(nil,false,false,'');
  SetLength(Objs,ACount);
  for i:=0 to ACount-1 do begin
    Objs[i]:=TObj.Create;
  end;

end;

destructor ResourcePool.Destroy;
var i:integer;
begin
  for i:=0 to Size-1 do begin
    FreeAndNil(Objs[i]);
  end;
  FreeAndNil(Event);
  inherited Destroy;
end;

function ResourcePool.Lock(Timeout: Cardinal): TObj;
var i:integer;
  b:boolean;
begin
  b:=false;
  repeat
    if InterLockedIncrement(FUseCount)>Size then begin
      //если ожидающих больше, просто ждём нашего события и может нам повезёт
      b:=true;
      Event.WaitFor(Timeout);
    end;
    for i:=Size-1 downto 0 do begin
      Result:=Objs[i];
      if Result=nil then continue;// если удалось поменять на nil значит можно выйти с ним
      if InterlockedCompareExchange(Pointer(Objs[i]),nil,Pointer(Result))=Pointer(Result) then exit;
    end;

    InterLockedDecrement(FUseCount);
  until (Timeout>=0)and b;
  Result:=nil;
end;

procedure ResourcePool.Unlock(const O: TObj);
var i:integer;
begin
  for i:=Size-1 downto 0 do begin
    //если удалось заменить (т.е. там было пусто), то мы его вернули
    if InterlockedCompareExchange(Pointer(Objs[i]),Pointer(O),nil)=nil then break;
  end;
  InterLockedDecrement(FUseCount); //здесь подправил так надёжнее всё таки
  //if InterLockedDecrement(FUseCount)>=Size then
    Event.SetEvent;
end;

end.


вот так проверял


Код

program TestPool;

{$mode objfpc}{$H+}

uses
  {$IFDEF UNIX}{$IFDEF UseCThreads}
  cthreads,
  {$ENDIF}{$ENDIF}
  Classes, SysUtils, ResPools;
type
  TPool=specialize ResourcePool<TObject>;

  { TTestThread }

  TTestThread=class(TThread)
   protected
     procedure Execute;override;
     constructor Create(ASuspended:boolean);
     destructor Destroy;override;

  end;
var
  Pool:TPool;
  ThreadCount:integer;
  { TTestThread }

procedure TTestThread.Execute;
var O:TObject;
  i:integer;
begin
  FreeOnTerminate:=true;
  for i:=1 to 1 do begin
    O:=Pool.Lock();
    try

      Sleep(1000);
    finally
      Pool.Unlock(O);
    end;

  end;

end;

constructor TTestThread.Create(ASuspended: boolean);
begin
  InterLockedIncrement(ThreadCount);
  inherited Create(ASuspended);
end;

destructor TTestThread.Destroy;
begin
  InterLockedDecrement(ThreadCount);
  inherited Destroy;
end;

var i:integer;
begin
  Pool:=TPool.Create(10);
  try
    for i:=1 to 150 do begin
      TTestThread.Create(false);
    end;
    repeat
      sleep(500);
      Writeln('Use=',Pool.UseCount,'  wait=',Pool.WaitCount);
    until ThreadCount=0;
    Writeln('Ok');
  finally
    Pool.Free;
  end;
end.

Powered by Invision Power Board (http://www.invisionboard.com)
© Invision Power Services (http://www.invisionpower.com)