Потоки... Потоки? Потоки!, FPC Опции

[Archon]

Сел переписывать движок Doomed Game под потоки и... понял, что мои знания о потокобезопасности оставляют желать лучшего. Может быть кто-нибудь (volvo?) проведет несколько лекций на эту тему? Думаю, это не только меня может заинтересовать. Или может я просто задам свои вопросы?

[volvo]

Название темы -

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

[Archon]

Окей. Тогда рассмотрим пример. Мое приложение состоит из нескольких глобальных объектов. Не знаю, насколько это оправдано, но в свое время мне это показалось неплохим решением. Объекты примерно такие:

unit u_test;
{$mode objfpc}

interface

type
	TTest = class
	private
		FData: Integer;
	public
		constructor Create;
		destructor Destroy; override;
		procedure OutputData;
		procedure IncData(Value: Integer);
		procedure RandomData;
		property Data: Integer read FData write FData;
	end;

var
	Test: TTest = nil;

implementation

constructor TTest.Create;
begin
	FData := 0;
end;

destructor TTest.Destroy;
begin
	WriteLn('I''m dying!');
end;

procedure TTest.OutputData;
begin
	WriteLn('Data: ', FData);
end;

procedure TTest.IncData(Value: Integer);
begin
	Inc(FData, Value);
end;

procedure TTest.RandomData;
var
	Temp: Integer;
begin
	Temp := Random(10);
	FData := Temp;
end;

initialization
	Test := TTest.Create;
finalization
	Test.Destroy;
end.

Так как с объектами могут взаимодействовать (вызывать их методы) сразу несколько потоков, их следует сделать потокобезопасными. Предлагаю это сделать . Если такая схема приложения кажется тебе неподходящей, можно ее пересмотреть.

[volvo]

Какие методы планируется вызывать из разных потоков? Все? Как пытался сделать потокобезопасность? (ну, и более глобальный вопрос, как вообще обеспечивается межпоточное взаимодействие, знаешь?) Что именно (по-твоему) может быть опасного в тех методах, которые ты привел, почему нельзя прямо так взять и работать с объектом из разных потоков?

[Archon]

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

В общем случае не все. Но кто его знает, как потом понадобится. Поэтому давай попробуем сделать все.

Как пытался сделать потокобезопасность?

Пока никак не пытался, только доки читал.

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

Знаю на теоретическом уровне критические секции, события, мутексы, семафоры. Использовать не пробовал.

Что именно (по-твоему) может быть опасного в тех методах, которые ты привел, почему нельзя прямо так взять и работать с объектом из разных потоков?

Так... Думаю, проблемы могут возникнуть при обращении к данным (FData). Сам по себе вызов методов, кажется, безопасен (у потоков отдельные стеки). Причем если пересекутся две попытки записи, проблемы очевидны, но я слышал, что и чтение/запись тоже вызывает проблемы (рассинхронизация кэшей процессоров, например). Еще есть подозрение, что при наложении чтение/запись можно прочитать испорченное значение (вряд ли запись переменной происходит в 1 этап).

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

[volvo]

Еще есть подозрение, что при наложении чтение/запись можно прочитать испорченное значение (вряд ли запись переменной происходит в 1 этап).

Вот !!! Все, что нужно, ты уже сказал... Тебе надо просто гарантировать атомарность (неделимость) операции изменения значения переменной. Для этого вместо Inc() используй InterlockedIncrement() или InterlockedExchangeAdd(), обе функции описаны в System... Все Interlocked... функции гарантируют монопольное изменение значения переменной (к тому же, они и выполняются быстрее, Рихтер говорит о примерно 50 тактах против 1000, которые требуются для перехода в Kernel-mode из User-mode)

[Archon]

Вот !!! Все, что нужно, ты уже сказал... Тебе надо просто гарантировать атомарность (неделимость) операции изменения значения переменной. Для этого вместо Inc() используй InterlockedIncrement() или InterlockedExchangeAdd(), обе функции описаны в System... Все Interlocked... функции гарантируют монопольное изменение значения переменной (к тому же, они и выполняются быстрее, Рихтер говорит о примерно 50 тактах против 1000, которые требуются для перехода в Kernel-mode из User-mode)

А неделимость чтения кто гарантировать будет? Вдруг запись произойдет в середине процесса чтения? К тому же, Inc - это только пример. Да и на месте Integer может стоять какой-нибудь TList.
Я набросал примерно вот что в псевдокоде:

unit u_test;
{$mode objfpc}

interface

type
	TTest = class
	private
		FData: Integer;
		{Объявляем Mutex;}
		function GetData: Integer;
		procedure SetData(Value: Integer);
	public
		constructor Create;
		destructor Destroy; override;
		procedure OutputData;
		procedure IncData(Value: Integer);
		procedure RandomData;
		property Data: Integer read GetData write SetData;
	end;

var
	Test: TTest = nil;

implementation

// private

function TTest.GetData: Integer;
begin
	try
		{Пытаемся захватить Mutex в течении N секунд.}
		if {получилось} then
			Result := FData
		else begin
			Result := 0;
			WriteLn('Error reading.');
		end;
	finally
		{Освобождаем Mutex.}
	end;
end;

procedure TTest.SetData(Value: Integer);
begin
	try
		{Пытаемся захватить Mutex в течении N секунд;}
		if {получилось} then
			FData := Value;
		else
			WriteLn('Error writing.');
	finally
		{Освобождаем Mutex;}
	end;
end;

// public

constructor TTest.Create;
begin
	FData := 0;
	{Создаем Mutex;}
end;

destructor TTest.Destroy;
begin
	WriteLn('I''m dying!');
end;

procedure TTest.OutputData;
begin
	WriteLn('Data: ', Data);
end;

procedure TTest.IncData(Value: Integer);
begin
	Data := Data + Value;
end;

procedure TTest.RandomData;
var
	Temp: Integer;
begin
	Temp := Random(10);
	Data := Temp;
end;

initialization
	Test := TTest.Create;
finalization
	Test.Destroy;
end.

Добавлено через 1 мин.
Кстати, еще ведь есть и прямой доступ к полю через property , так что там не только Inc.

Исправил: добавил else в условиях.

Сообщение отредактировано: Archon - 28.06.2009 3:05

[volvo]

Да и на месте Integer может стоять какой-нибудь TList.

Тогда, естественно, придется пользоваться другими средствами синхронизации...

Я набросал примерно вот что в псевдокоде:

А чего в "псевдо"? Делай уже Паскалевский код, что там осталось... Только немного не так:

	try
		{Пытаемся захватить Mutex в течении N секунд;}
		if {получилось} then
			FData := Value;
	finally
		{Освобождаем Mutex;}
	end;

Здесь ты в любом случае освобождаешь Mutex, а этого делать нельзя. Ибо освобождает его только тот, кто установил...

	{Пытаемся __установить__ Mutex в течении N секунд;}
	if {получилось} then
		FData := Value;
		{Освобождаем Mutex;}
	end;

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

[Archon]

А чего в "псевдо"? Делай уже Паскалевский код, что там осталось...

Я не знаю процедур . Можно, конечно, WinApi, но наверно и в rtl должны быть кроссплатформенные аналоги. Попробую поискать их завтра, сейчас уже спать хочу.

Здесь ты в любом случае освобождаешь Mutex, а этого делать нельзя. Ибо освобождает его только тот, кто установил...

Точно, спасибо.

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

Облом будет, если мьютекс не освободится в течение N секунд, а в WinApi, например, есть для этой функции константа INFINITE.

[Archon]

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

Кстати да, CriticalSection тут будет проще .

unit u_test;
{$mode objfpc}

interface

type
	TTest = class
	private
		FData: Integer;
		CS: TRTLCriticalSection;
		function GetData: Integer;
		procedure SetData(Value: Integer);
	public
		constructor Create;
		destructor Destroy; override;
		procedure OutputData;
		procedure IncData(Value: Integer);
		procedure RandomData;
		property Data: Integer read GetData write SetData;
	end;

var
	Test: TTest = nil;

implementation

// private

function TTest.GetData: Integer;
begin
	try
		EnterCriticalSection(CS);
		Result := FData;
	finally
		LeaveCriticalSection(CS);
	end;
end;

procedure TTest.SetData(Value: Integer);
begin
	try
		EnterCriticalSection(CS);
		FData := Value;
	finally
		LeaveCriticalSection(CS);
	end;
end;

// public

constructor TTest.Create;
begin
	FData := 0;
	InitCriticalSection(CS);
end;

destructor TTest.Destroy;
begin
	DoneCriticalSection(CS);
	WriteLn('I''m dying!');
end;

procedure TTest.OutputData;
begin
	WriteLn('Data: ', Data);
end;

procedure TTest.IncData(Value: Integer);
begin
	Data := Data + Value;
end;

procedure TTest.RandomData;
var
	Temp: Integer;
begin
	Temp := Random(10);
	Data := Temp;
end;

initialization
	Test := TTest.Create;
finalization
	Test.Destroy;
end.

Проверяй .

Добавлено через 6 мин.
Почитал про Interlocked-функции. Я так понимаю, что чтение 32-битной переменной в любом случае атомарно?

PS Пока читал, узнал, что Int64 и QWord - оказывается не ordinal .

[Archon]

Обезопасил модуль u_log.pas. Теперь думаю над u_window.pas. Предпологается использовать этот объект только в одном потоке (в модуле u_graphics.pas), но WinProc - это же отдельный поток, верно? Значит вот так делать нельзя:

function TD3DWindow.MessageProc(Msg: UINT; WParam: WPARAM; LParam: LPARAM): LResult;
begin
	case Msg of
		WM_DESTROY: begin
			PostQuitMessage(0);
		end;
		WM_SYSCOMMAND: begin
			if (WParam = SC_SCREENSAVE) or (WParam = SC_MONITORPOWER) then
				Result := 0
			else
				Result := DefWindowProc(WinHandle, Msg, WParam, LParam);
		end;
		WM_SYSKEYUP: begin
			if WParam = VK_RETURN then SetFullscreenMode(D3DPP.Windowed) { <- Тут прозреваю небезопасный вызов метода. }
			else if Char(WParam) = '1' then SetMode('800x600x32') { <- И тут. }
			else if Char(WParam) = '2' then SetMode('1024x768x32') { <- И тут. }
			else if Char(WParam) = '3' then SetMode('1280x1024x32'); { <- И тут. }
		end;
		else
			Result := DefWindowProc(WinHandle, Msg, WParam, LParam);
  end;
end;

Думаю, от этого вобще лучше избавиться и ловить клавиши исключительно в модуле u_controls.pas (его пока нет, но будет )

Сообщение отредактировано: Archon - 28.06.2009 12:37

[volvo]

Проверяй

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

Я так понимаю, что чтение 32-битной переменной в любом случае атомарно?

И чтение и запись 32-битной переменной само по себе - атомарно. Но Interlocked-функции предназначены для того, чтобы сделать атомарными изменения этих переменных:

This feature is useful in a multitasking operating system, in which the system can interrupt one thread's execution to grant a slice of processor time to another thread. Without such synchronization, two threads could read the same value, increment it by 1, and store the new value for a total increase of 1 instead of 2. The interlocked variable-access functions protect against this kind of error.

Пока читал, узнал, что Int64 и QWord - оказывается не ordinal

С этими типами вообще много неясного... Вот, к примеру:

Ordinal types are countable and ordered, i.e. it is, in principle, possible to start counting them
one bye one, in a specified order. This property allows the operation of functions as Inc, Ord,
Dec on ordinal types to be defined.

, однако и Int64 и QWord прекрасно Inc-рементируются Да и Pred/Succ, которые тоже работают только с перечислимыми типами, с ними работают. А вот цикл for с управляющей переменной Int64/QWord невозможен. Как это понимать?

[Archon]

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

А какие существуют методы тестирования на потокобезопасность?

[Archon]

Продолжаю исправлять классы. Работу с текстурами я просто сокрыл внутри спрайтов. Так что теперь главное - сделать безопасную работу со спрайтами.

TSprite = class(TGraphicObject)
	private
		Texture: TTexture;
		CurrentMode: Integer;
		FramesNum: Integer;
		CurrentFrame: Integer;
		FrameInterval: Double;
		RepeatAnimation: Boolean;
		FrameTime: Double;
		SpriteCenter: TD3DXVECTOR3;
		Scale: TD3DMATRIX;
		Rotate: TD3DMATRIX;
		Translate: TD3DMATRIX;
		Transform: TD3DMATRIX;
		FWidth, FHeight: Integer;
	public
		// Width, Height - размеры спрайта (в тайлах, могут быть дробными).
		// Texture - имя файла с текстурой, которая должна быть наложена на спрайт.
		// AWidth, AHeight - размеры спрайта (размеры одного кадра тектсуры в пикселах).
		constructor Create(const TextureName: AnsiString; AWidth, AHeight: Integer);
		// Деструктор автоматически пытается удалить объект из контейнера Sprites.
		destructor Destroy; override;
		// Установка параметров анимации. Параметры:
		// Mode - номер анимации. Frame - номер текущего кадра. Frames - число кадров.
		// CenterX, CenterY - координаты центра спрайта, относительно которого отсчитываются
		// координаты и происходит поворот.
		// Interval -  время показа кадра в мс, 0, если анимация не нужна.
		// RepeatAni - повторять ли анимацию.
		procedure SetMode(Mode, Frame, Frames, CenterX, CenterY, Interval: Integer; RepeatAni: Boolean);
		// Установить параметры: координаты, угол поворота, масштаб.
		procedure SetParams(X, Y, Z, Angle, ScaleX, ScaleY: Single);
		// Нарисаовать спрайт.
		procedure Draw;
		// Обработчик спрайта. Принимает время, прошедшее с предыдущего вызова в мс.
		procedure Update(Delta: Double); override;
		property Width: Integer read FWidth;
		property Height: Integer read FHeight;
	end;

Предпологается, что один поток будет делать SetMode и SetParams, а другой Update и Draw. Критические секции сделать не проблема, но не будет ли это слишком медленным? Все таки каждый кадр предпологается рисовать множество этих спрайтов.

[volvo]

Что-то не так... Смотри:

uses sysutils, classes,
     u_test;

type
  tmythread = class(tthread)
  private
    msg: string;
    num: string;
  protected
    procedure execute; override;
  public
    constructor create(s, n: string);
  end;

constructor tmythread.create(s, n: string);
begin
  msg := s; num := n;
  inherited create(false);
end;

procedure tmythread.execute;
begin
  repeat
    Log.Write(num, msg);
  until terminated;
end;

const n = 50;
var
  thrds: array[1 .. n] of TMyThread;
  i: integer;
begin
  for i := 1 to n do begin
    thrds[i] := TMyThread.Create('message #' + IntToStr(i), IntToStr(i));
  end;
  readln;
  for i := n downto 1 do begin
    thrds[i].Terminate; thrds[i].Destroy;
  end;
end.

- простейший тест, правда? Примерно одинаковое количество раз в файле должно присутствовать каждое сообщение. Теперь смотри на вывод:
 threads.txt ( 91.51 килобайт ) Кол-во скачиваний: 741


Самое интересное начинается с 89 строки...

[Archon]

Странно, я у себя в том же тесте таких ужасов не наблюдаю. Только в конце 1 повторяется, но там их всего ~2-3 потоков осталось. Может менеджер процессов Windows шалит?

[volvo]

Может менеджер процессов Windows шалит?

Угу, тут прямо шалит, а если сделать:

unit u_test;
{$mode objfpc}

interface

uses
	windows, sysutils;

type
	TLog = class
	private
		FLogFile: Text;
		FLock: TMultiReadExclusiveWriteSynchronizer;
		procedure WriteString(const Str: string);
	public
		constructor Create(const FileName: string);
		destructor Destroy; override;
		procedure Write(const SenderName, Str: string);
	end;

var
	Log: TLog;

implementation

const
	TimeFormat = 'yyyy-mm-dd hh:mm:ss.zzz';

// TLog -----

// private

procedure TLog.WriteString(const Str: string);
begin
	FLock.BeginWrite;
	try
		{$I-}
		WriteLn(FLogFile, Str);
		Flush(FLogFile);
		{$I+}
		if IOResult <> 0 then begin
			MessageBox(0, 'Can''t write to log file. Program halted.', 'File error', MB_OK);
			Halt(0);
		end;
	finally
		FLock.EndWrite;
	end;
end;

// public

constructor TLog.Create(const FileName: string);
begin
	FLock := TMultiReadExclusiveWriteSynchronizer.Create;
	{$I-}
	Assign(FLogFile, FileName);
	if FileExists(FileName) then Append(FLogFile) else Rewrite(FLogFile);
	{$I+}
	if (IOResult = 0) and (FileName <> '') then begin
		WriteString('');
		WriteString('*** Start session ***');
	end else begin
		MessageBox(0, 'Can''t create or open log file. Program halted.', 'File error', MB_OK);
		Halt(0);
	end;
end;

destructor TLog.Destroy;
begin
	WriteString('*** End session ***');
	Close(FLogFile);
	FLock.Free;
end;

procedure TLog.Write(const SenderName, Str: string);
begin
	WriteString(FormatDateTime(TimeFormat, Now) + ' Note: "' + SenderName + '" - ' + Str);
end;

initialization
	Log := TLog.Create('threads.txt');
finalization
   Log.Destroy;
end.

(код тестирующего приложения не меняется), то шалить моментально перестает?

[Archon]

А в чем разница? Разве что в TMultiReadExclusiveWriteSynchronizer используется Mutex, а не CriticalSection.

о шалить моментально перестает?

Неа, у меня на старой версии не шалило, а вот этот лог я получил при первой же проверке новой версии. Интересны строчки

2009-06-28 14:35:14.692 Note: "13" - message #13

в конце лога.

Сообщение отредактировано: Archon - 28.06.2009 15:45

[volvo]

Разве что в TMultiReadExclusiveWriteSynchronizer используется Mutex, а не CriticalSection.

Кто сказал?

Description: Create creates a new instance of TMultiReadExclusiveWriteSynchronizer. It initializes
a TRTLCriticalSection.

Насчет строчек "потока №13" - хм... Из 5409 строк лога этот поток завершает запись в 5289 строке. Это не конец лога совсем... Ты файлы не перепутал?

Добавлено через 5 мин.
Или ты о том, что время НЕпоследовательное, а вперемешку? Так это подразумевается вообще-то при использовании критических секций, порядком входа в секцию управлять нельзя. Вот что Рихтер говорит в частности про LeaveCriticalSection:

Эта функция просматривает элементы структуры CRITICAL_SECTION и уменьшает счетчик числа захватов ресурса вызывающим потоком на 1. Если его значение больше 0, LeaveCriticalSection ничего не делает и просто возвращает управление.

Если значение счетчика достигло 0, LeaveCriticalSection сначала выясняет, есть ли в системе другие потоки, ждущие данный ресурс в вызове EnterCriticalSection. Если есть хотя бы один такой поток, функция настраивает значения элементов структуры, что бы они сигнализировали о занятости ресурса, и отдает его одному из ждущих потоков (поток выбирается "по справедливости"). Если же ресурс никому не нужен, LeaveCriticalSection соответственно сбрасывает элементы структуры.

[Archon]

TMultiReadExclusiveWriteSynchronizer is a default implementation of the IReadWriteSync
(1465) interface. It uses a single mutex to protect access to the read/write resource, resulting in a single
thread having access to the resource.

Цитата из того же файла версии 2.1 за август 2006. У меня компилятор 2.0.4 (оказывается ).

Насчет строчек "потока №13" - хм... Из 5409 строк лога этот поток завершает запись в 5289 строке. Это не конец лога совсем... Ты файлы не перепутал?

Конец - это относительно. Глянь чуть выше на строчки 4642-4902. 4903-4926 тоже интересно. Такие последовательности появляются регулярно, могу еще логов предоставить .