Оптимизация работы с множествами set of byte

[alcaedo]

Глобальные переменные:

Type
  hodS = record p,n:byte end;
Var
  wayS:array[1..255] of hodS;
  wayScnt:byte=0;
  SminNumCicle,SmaxNumCicle:byte;
  isRunning:boolean;
  poleSwork,poleSg:set of byte;
  MainCounter:int64;

Рекурсивная функция:

FUNCTION WORKS:boolean;
var
  Wo,fpos:byte;
  fnp,fep:integer;
begin
  inc(MainCounter);
  if (MainCounter mod 10000)=0 then Application.ProcessMessages;
 
  if not isRunning then
    begin
      Result:=false;
      exit;
    end;
  Wo:=0;
  for fpos := SminNumCicle to SmaxNumCicle do if (fpos in poleSwork) then
    begin
      inc(Wo);
 
      fnp:=fpos-2;{ UP }
      fep:=fpos-1;
      if (fep in poleSwork) and not (fnp in poleSwork) and not (fnp in poleSg) and not (fep in poleSg) then
        begin
          exclude(poleSwork,fpos);
          exclude(poleSwork,fep);
          include(poleSwork,fnp);
          inc(wayScnt);
          wayS[wayScnt].p:=fpos;
          wayS[wayScnt].n:=0;
          if WorkS then
            begin
              Result:=true;
              exit
            end else
            begin
              include(poleSwork,fpos);
              include(poleSwork,fep);
              exclude(poleSwork,fnp);
              dec(wayScnt);
            end;
        end;
 
      fnp:=fpos+2;{ DOWN }
      fep:=fpos+1;
      if (fep in poleSwork) and not (fnp in poleSwork) and not (fnp in poleSg) and not (fep in poleSg) then
        begin
          exclude(poleSwork,fpos);
          exclude(poleSwork,fep);
          include(poleSwork,fnp);
          inc(wayScnt);
          wayS[wayScnt].p:=fpos;
          wayS[wayScnt].n:=1;
          if WorkS then
            begin
              Result:=true;
              exit
            end else
            begin
              include(poleSwork,fpos);
              include(poleSwork,fep);
              exclude(poleSwork,fnp);
              dec(wayScnt);
            end;
        end;
 
      fnp:=fpos-32;{ LEFT }
      fep:=fpos-16;
      if (fep in poleSwork) and not (fnp in poleSwork) and not (fnp in poleSg) and not (fep in poleSg) then
        begin
          exclude(poleSwork,fpos);
          exclude(poleSwork,fep);
          include(poleSwork,fnp);
          inc(wayScnt);
          wayS[wayScnt].p:=fpos;
          wayS[wayScnt].n:=2;
          if WorkS then
            begin
              Result:=true;
              exit
            end else
            begin
              include(poleSwork,fpos);
              include(poleSwork,fep);
              exclude(poleSwork,fnp);
              dec(wayScnt);
            end;
        end;
 
      fnp:=fpos+32;{ RIGHT }
      fep:=fpos+16;
      if (fep in poleSwork) and not (fnp in poleSwork) and not (fnp in poleSg) and not (fep in poleSg) then
        begin
          exclude(poleSwork,fpos);
          exclude(poleSwork,fep);
          include(poleSwork,fnp);
          inc(wayScnt);
          wayS[wayScnt].p:=fpos;
          wayS[wayScnt].n:=3;
          if WorkS then
            begin
              Result:=true;
              exit
            end else
            begin
              include(poleSwork,fpos);
              include(poleSwork,fep);
              exclude(poleSwork,fnp);
              dec(wayScnt);
            end;
        end;
 
    end;
  Result:=(Wo<2);
end;

Как можно ускорить выполнение функции WorkS? Может заменить как-то процедуры include и exclude на что-то (ассемблер?), или условия if (fep in poleSwork) and not (fnp in poleSwork) and not (fnp in poleSg) and not (fep in poleSg) then можно оптимизировать?

В первую очередь интересует именно ускорение этого алгоритма, а не вариант другого. Но если у кого-то появятся варианты, то вот описание:
Функция двигает фишки на квадратном поле 16х16 (poleSwork) пока не останется одна фишка.
Граница, через которую фишки двигать нельзя - poleSg.
Все сделанные ходы от начальной расстановки до текущей хранятся в массиве wayS. В нём wayS[XXX].p - позиция подвинутой фишки, а wayS[XXX].n - направление сдвига.
Фишки двигаются только перескоком через другую фишку, при этом перепрыгнутая фишка удаляется.
Счётчик MainCounter нужен чисто на время отладки.
Прерывание работы функции по флагу isRunning.

[BDA]

Во-первых, сократим-ка длину кода:

const
  dirs: array [0 .. 3, 0 .. 1] of ShortInt = (
    (-2, -1), {UP}
    (2, 1), {DOWN}
    (-32, -16), {LEFT}
    (32, 16) {RIGHT}
  );
//...
FUNCTION WORKS: boolean;
var
  Wo, fpos, n: byte;
  fnp, fep: integer;
begin
  inc(MainCounter);
  if (MainCounter mod 10000) = 0 then
    Application.ProcessMessages;

  if not isRunning then
  begin
    Result := false;
    exit;
  end;
  Wo := 0;
  for fpos := SminNumCicle to SmaxNumCicle do
    if (fpos in poleSwork) then
    begin
      inc(Wo);
      for n := 0 to 3 do
      begin
        fnp := fpos + dirs[n, 0];
        fep := fpos + dirs[n, 1];
        if (fep in poleSwork) and not(fnp in poleSwork) and not(fnp in poleSg) and not(fep in poleSg) then
        begin
          exclude(poleSwork, fpos);
          exclude(poleSwork, fep);
          include(poleSwork, fnp);
          inc(wayScnt);
          wayS[wayScnt].p := fpos;
          wayS[wayScnt].n := n;
          if WORKS then
          begin
            Result := true;
            exit
          end
          else
          begin
            include(poleSwork, fpos);
            include(poleSwork, fep);
            exclude(poleSwork, fnp);
            dec(wayScnt);
          end;
        end;
      end;
    end;
  Result := (Wo < 2);
end;

Во-вторых, include и exclude достаточно хорошо записаны в ассемблерном коде, хотя лучше бы отказаться от типа integer для fep и fnp (при типе byte будет 3 ассемблерных команды вместо 4).
В-третьих, из-за ленивых вычислений лучше выстроить условия в самом длинном ифе так, чтобы проверка при отрицательном исходе прекращалась как можно раньше (но я не смог решить, какое из условий "отрицательно заряжено" ).
В-четвертых, не совсем понял, что хранится в poleSg.
В-пятых, можно произвести профилирование кода, для поиска узких мест (http://www.gunsmoker.ru/2013/01/opti...on-basics.html).
Предложить другой алгоритм не могу - спать хочется

[alcaedo]

1) Оно примерно так и было изначально, специально развернул цикл и избавился от лишних процедур

2) я бы отказаться от типа integer в пользу byte, но там диапазон возможных значений от -32 до +287. И ещё вот по поводу типа этих переменных такая непонятка:
fnp,fep:smallint; {скорость 2760 KOps/s }
fnp,fep:integer; {скорость 3085 KOps/s }
fnp,fep:int64; {скорость 3075 KOps/s }
Скорости указаны на момент, когда функция ещё работала в главном потоке программы.

3) Это я подумаю. Так сходу сложно сказать какое из них вначале выставить. Но смысл понятен.

4) В poleSg - граница, через которую фишки двигать нельзя и на которую двигать нельзя. Работает как "исключающая маска" на poleSwork. На картинке красные квадратики - это poleSg, синие - poleS.



Протестировал. Сейчас, работая в отдельном потоке, вариант с циклом выдаёт скорость 2630 килоопераций в секунду, без цикла (как в первом посте) 3270. Как-то бы ещё распараллелить по четырём направлениям, а то четырёхядерный проц загружен на 25%.

[evg_m]

Как-то бы ещё распараллелить по четырём направлениям

Ну например, выполнить ПЕРВЫЕ четыре варианта хода

fpos:=SminCicle;
      for n := 0 to 3 do //в варианте BDA со свернутым циклом
      begin

ПОЛУЧИТЬ ЧЕТЫРЕ НОВЫХ НАЧАЛЬНЫХ СОСТОЯНИЙ.
Далее каждое их ЭТИХ четырех состояний (четыре КОПИИ!!!) анализировать независимо, в потоках. (той же самой WORKS но все глобальные переменные сделать ЛИЧНЫМИ данными потока (те самые КОПИИ!!).
недостаток. любой поток может закончится "слишком рано" если ему попадутся "плохие" данные.
Хорошо бы в таком случае потоку иметь возможность "отобрать"(попросить) себе новое исходное состояние у других потоков.

[alcaedo]

Сообщение от evg_m

Ну например, выполнить ПЕРВЫЕ четыре варианта хода

Проблема в том, что чаще всего первые 8-9 вариантов хода тупиковые

Сообщение от evg_m

Хорошо бы в таком случае потоку иметь возможность "отобрать"(попросить) себе новое исходное состояние у других потоков.

Вот это и хотелось бы, но пока не знаю, как

[evg_m]

как вариант с бесконечным числом потоков
каждый поток создает себе четыре новых потока для проверки СВОИХ под-деревьев
и спокойно засыпает ДО получения результатов от своих субпотоков.

бесконечное число потоков переделываем в пул потоков.
при старте потока заносим в очередь четыре новых узла для проверки и ждем засыпаем до результатов проверки.

Рекурисию вызовов Works превращаем в очередь проверок ТЕКУЩЕГО состояния.
замена обхода в глубину(проверка сразу на всю глубину ветви)
на обход в ширину (проверяем "сразу все ветки" но только на один шаг в глубину)
Но очередь получится длинная (при этом надо будет хранить много-много текущих состояний) хотя и будет идти быстро.
Хотя текущее состояние легко вычисляется по исходному при наличии "записи" ходов.
вариант совместить обход в ширину с обходом глубину.
1. пока очередь маленькая делаем обход в ширину (наращиваем очередь)
2. при достижения "критической" длины начинаем проводить анализ до конца. переходим на анализ в глубину)
3. как только какой либо "глубокий" анализ заканчивается (т.е. уменьшается очередь) снова пробуем наращивать очередь (п.1)

[alcaedo]

Распараллелил. Программа с формой целиком: BG.zip

Тут всплыла одна проблема. Несколько ходов в разной последовательности могут приводить из одного начального состоя в одинаковое конечное. Программа посчитает такие последовательности ходов разными и будет обрабатывать все варианты (либо одновременно в разных потоках, либо по очереди в одном потоке). То есть, программа выполняет бесполезную работу. Ведь, если мы приходим к состоянию, которое уже обрабатывалось или которое сейчас обрабатывается в соседнем потоке, то можно его откинуть. Как вы думаете, какой способ отсева будет самым не затратным?



ps: не нашёл, как вставить гифку в спойлер

[evg_m]

Если мы не хотим повторно обрабатывать, то надо помнить ВСЕ промежуточные состояния.

1) работать только с очередью.(только анализ в ширину)
2) при добавлении проверять наличие "дубликата". (для ускорения поиска дубликата нужно иметь упорядоченный каким-либо образом список, но упорядочивание снижает скорость добавления).
3) после анализа НЕ удалять из очереди а помечать как обработанный.

В конце получаем очень большую очередь включающую ВСЕ допустимые состояния. ( а хватит ли памяти!)

возврат к "нормальной"очереди проверок.
можно ПОМНИТЬ все проверенные маршруты (последовательности ходов, не состояния).

Хотя текущее состояние легко вычисляется по исходному при наличии "записи" ходов.

А состояния для проверок генерировать по всем и каждому из сохраненных маршрутов. Увеличиваем время, но может быть(!) сократим память.
после добавления новых сгенерированных маршрутов, исходный маршрут исключается из очереди.
Мы его в любом случае получим (и даже не один раз) при восстановлении из дочерних маршрутов.

НУ и комбинация приемов.(восстанавливаем анализ в глубину)
При каждом СПУСКЕ (проверяем нашу очередь на дубликаты) и либо возвращаемся либо регистрируем новый обработанный путь и ...
Хотя ЗДЕСЬ надо внимательно "смотреть за синхронизацией".
Один начал но не успел добавить, и тут же второй пришел к этому.

посмотрел код

begin { следующий поток без работы: отдать свою (сделав ход на поле другого потока) и продолжить, как будто ход тупиковый }

Работа с чужими данными даже если поток и "стоит" лучше делать с синхронизацией.

по хорошему поток НИЧЕГО не должен знать о других потоках.
Максимум с кем он должен общаться это головной поток, поддерживающий очередь.
отдать / не отдать должен решать не сам поток, проверяя данные других потоков, а головной поток ЗНАЮЩИЙ сколько у него обрабатывающих(4, 8 вообще один), и чем они в данный момент заняты.
Представьте себе у вас появился 8-ядерный процессор, вы захотели увеличить число потоков обработки, и ... полная (капитальная) переделка ВСЕХ потоковых процедур.
Или же внесения изменений в список обрабатывающих (увеличение списка запускаемых потоков и всЁ)

Все обрабатывающие потоки должны быть ОДИНАКОВЫ.
т.е. не четыре класса с РАЗНЫМИ процедурами обработки и разными ГЛОБАЛЬНЫМИ данными
а ОДИН класс в котором КАЖДЫЙ экземпляр имеет СВОИ ЛОКАЛЬНЫЕ данные.

ГЛОБАЛЬНЫМ по хорошему счету должен быть МАССИВ ЭКЗЕМПЛЯРОВ рабочего потока.
и ОДИН экземпляр управляющего потока (он может и не быть, если управляющим потоком будет выступать основной).

ВСЕ остальные данные (если это не обще программные константы) должны быть локальными ПОЛЯМИ потока.

[alcaedo]

Сообщение от evg_m

Работа с чужими данными даже если поток и "стоит" лучше делать с синхронизацией.

по хорошему поток НИЧЕГО не должен знать о других потоках.
Максимум с кем он должен общаться это головной поток, поддерживающий очередь.
отдать / не отдать должен решать не сам поток, проверяя данные других потоков, а головной поток ЗНАЮЩИЙ сколько у него обрабатывающих(4, 8 вообще один), и чем они в данный момент заняты.

Конкретно в этой программе каждый поток имеет ровно 1 управляющий поток. Первый управляется Нулевым, Второй управляется Первым и т.д. Каждый поток сейчас знает, сколько потоков находится под его управлением - ровно один. И каждому управляющему известно, в каком состоянии находится управляемый.

Сейчас это выглядит так:



Главное, чтобы поток получал задания только из одного места. И не важно, из какого. Ниже - ещё несколько безопасных вариантов:
1)

2)

3)

[alcaedo]

Сообщение от evg_m

Представьте себе у вас появился 8-ядерный процессор, вы захотели увеличить число потоков обработки, и ... полная (капитальная) переделка ВСЕХ потоковых процедур.
Или же внесения изменений в список обрабатывающих (увеличение списка запускаемых потоков и всЁ)

Для добавления ещё 1 рабочего потока достаточно:
- внутри функций NWSx3 заменить по 4 упоминания "z0" на "z4"
- создать пару новых функций NWS24 и NWS34, в которых в тех же 4-х местах будет стоять "z0", а первой строкой стоит "begin with z4 do"
- ещё, конечно, надо добавлять переменную z4 и "procedure TMyWorkerThread4.Execute;", но это не имеет отношения к капитальной переделке

Сообщение от evg_m

Все обрабатывающие потоки должны быть ОДИНАКОВЫ.

Ну, они почти одинаковы. У каждого есть управляющий и подчинённый.

Сообщение от evg_m

а ОДИН класс в котором КАЖДЫЙ экземпляр имеет СВОИ ЛОКАЛЬНЫЕ данные.

Не очень понимаю, как VCL-поток будет брать текущее состояние из локальных данных рабочих потоков

Вообще, я не могу придумать ситуацию, в которой глобальные данные могли бы быть повреждены или прочитаны неверно в этой программе (не считая момента, когда VCL читает, например, переменную счётчика, а в этот же момент какой-то поток делает запись в эту переменную, что не критично). В каждой ситуации, где к одной переменной могут обратиться несколько потоков, во-первых, есть флаг JNR, который управляющий поток может только менять на FALSE, а управляемый только менять на TRUE. Общающиеся потоки меняют флаг только тогда, когда в них закончилась обработка глобальных переменных zx. Во-вторых, у каждого управляемого потока есть только 1 управляющий. Это значит, что если управляющий поток видит JNR=TRUE и начинает работать с переменной z, то никакой другой поток в это время с переменной z не работает. Просто нет другого управляющего потока. А управляемый не работает с z, т.к. ждёт JNR=FALSE. Сами переменные JNR занимают в памяти 1 байт, а значит запись и чтение в них происходит одной операцией. Ситуации, когда управляемый поток пытается записать JNR=TRUE, но приостанавливается на середине, а потом управляющий пытается прочесть недозаписанные данные, не может возникнуть.
С VCL-потоком сложнее. Тут может возникнуть ситуация чтения недозаписанных данных (счётчики, переменные zx.w), но это не может привести к ошибке, а только в временному (до следующего тика таймера) искажению информации на форме. Переменные zx.isStartTh и zx.isFinishTh, которые читает VCL-поток по таймеру, а выставить может какой-то рабочий поток, тоже однобайтовые. Они прочитаются правильно. Переменная isRunning может быть изменена и рабочим и VCL-потоком. Возможна ситуация, когда одновременно её пытаются изменить из нескольких мест, но тут фишка в том, что эта однобайтовая переменная может меняться только с TRUE на FALSE. Даже, если все пять потоков программы будут записывать isRunning:=FALSE, всё-равно в результате получим FALSE.

UPD: всё-таки, чтение VCL-потоком переменной zx.w может вызвать ошибку в процедуре pole_upd вот тут:

puW[0]:=z0.w;puW[1]:=z1.w;puW[2]:=z2.w;puW[3]:=z3.w;
  for h := 0 to 3 do with puW[h] do
    begin
      puc:=_puc;if puc>wc then puc:=wc;pl:=poleS;
      for i:=1 to puc do
        begin
          exclude(pl,wp[i]);
          case wn[i] of
            0:begin exclude(pl,wp[i]-1); include(pl,wp[i]-2)  end;
            1:begin exclude(pl,wp[i]+1); include(pl,wp[i]+2)  end;
            2:begin exclude(pl,wp[i]-16);include(pl,wp[i]-32) end;
            3:begin exclude(pl,wp[i]+16);include(pl,wp[i]+32) end;
            else showmessage('ERROR 77');
          end;
        end;
    end;

Если на момент переноса puW[0]:=z0 в переменной z0 части z0.pl, z0.wp, z0.wn и z0.wc не будут соответствовать друг другу, это может спровоцировать попытку включения или исключения из множества чисел меньше 0 или больше 255. Ну, или может сработать "else showmessage('ERROR 77');", хотя у меня такого не происходило ни разу. Надо будет просто добавить несколько корректирующих проверок в pole_upd.