Код, написанный в предыдущей статье этой серии, следовал следующему базовому алгоритму:

1. SPU ставит на выполнение передачу DMA GET для помещения части набора данных из основной памяти в буфер.
2. SPU ожидает заполнения буфера.
3. SPU обрабатывает буфер.
4. SPU ставит на выполнение передачу DMA PUT для передачи буфера обратно в основную память.
5. SPU ожидает окончания передачи буфера.
6. Если остаются необработанные данные, процедура повторяется.

Проблема вышеуказанной процедуры заключается в том, что очень много процессорного времени тратится впустую. Шаги, на которых происходит передача данных, не задействуют SPU вообще, а задействуют только контроллер MFC (который является частью элемента SPE). В коде, написанном до этого момента, SPU просто ожидал, пока контроллер MFC закончит работу, и только после этого выполнял какие-то другие операции. Определенно, мы сможем чем-нибудь занять его в то время, пока он простаивает в ожидании.

Двойная буферизация

Это работает почти как кабинет врача – я знаю, что когда я попаду туда, мне придется провести много времени в комнате ожидания. Поэтому я всегда беру с собой что-нибудь, чем бы я мог заняться в это время. Тот же самый принцип применим и к программированию. Вместо того чтобы тратить полезные такты процессора, ожидая завершения передачи данных, ваш код мог бы иметь второй буфер для обработки. Таким образом, пока вы ожидаете завершения передачи одной части данных, вы можете обрабатывать другую их часть. С учетом этого новый алгоритм обработки выглядит следующим образом:

1. SPU ставит на выполнение передачу DMA GET для помещения части набора данных из основной памяти в буфер #1.
2. SPU ставит на выполнение передачу DMA GET для помещения части набора данных из основной памяти в буфер #2.
3. SPU ожидает заполнения буфера #1.
4. SPU обрабатывает буфер #1.
5. SPU (а) ставит на выполнение передачу DMA PUT для передачи содержимого буфера #1, а затем (б) ставит на выполнение передачу DMA GETB после передачи PUT для заполнения буфера следующей частью данных из основной памяти.
6. SPU ожидает заполнения буфера #2.
7. SPU обрабатывает буфер #2.
8. SPU (а) ставит на выполнение передачу DMA PUT для передачи содержимого буфера #2, а затем (б) ставит на выполнение передачу DMA GETB после передачи PUT для заполнения буфера следующей частью данных из основной памяти.
9. Процедура повторяется с шага 3 до тех пор, пока не будут обработаны все данные.
10. Ожидание завершения работы со всеми буферами.

Конечно, это алгоритм приводит к дополнительным вопросам. Прежде всего, заметьте, что вы потенциально выполняете большое количество ненужной работы, когда буфер заканчивается, поскольку вы обрабатываете два буфера на каждой итерации цикла. Вы могли бы вставить несколько операторов if для выхода и остановки процесса заполнения буфера в тот момент, когда у вас заканчиваются данные. Тем не менее, в этой программе я предпочел этот метод, поскольку он вносит большой объем дополнительной обработки для каждой итерации. Если в коде обрабатываются большие массивы данных, то затраты на каждую итерацию гораздо более важны, нежели затраты на сборку и освобождение. Поэтому, чтобы избежать ветвлений, я, насколько это возможно, освобождаю SPE от работы с условиями. В случаях обработки буфера контроллер MFC рассматривает запрос данных с нулевым размером в качестве команды холостого хода, так что я могу создавать запросы даже в тех случаях, когда данных для чтения нет. В случаях фактической обработки буфера функция вполне способна обрабатывать буферы с нулевым размером путем простого возвращения. Итак, все случаи уже обработаны, и любые ветвления для удаления дополнительных шагов по освобождению будут служить только для замедления случая по умолчанию.

Другой вопрос – как запланировать передачи PUT и GET для одного и того же буфера, избежав конфликтов. После каждого шага обработки данных вы вызываете передачу PUT для передачи данных в основную память и передачу GET – для получения следующего пакета данных. Поскольку по умолчанию контроллер MFC обрабатывает запросы в произвольном порядке, который он сам определяет, как вы можете запланировать определенный порядок выполнения? Как обсуждалось в последней статье, решением являются барьеры (barriers) и ограничители (fences). Помещение ограничителя в запрос заставляет все предшествующие запросы MFC, входящие в одну и ту же группу тэгов, обрабатываться перед текущим запросом. Тем не менее, этот способ не определяет порядок выполнения последующих передач. Барьер похож на ограничитель за исключением того, что он выполняет упорядочивание как предыдущих, так и последующих запросов. Таким образом, послав второй запрос с ограничителем или барьером, вы можете заставить контроллер MFC обрабатывать запросы в нужном порядке, и, поскольку они находятся в одной и той же группе тэгов, то к моменту начала использования буфера вы можете просто подождать завершения обработки целой группы тэгов. GETB, PUTB, GETF и PUTF являются основными командами DMA для работы с ограничителями и барьерами одиночных буферов.

Теперь давайте подумаем, как вы могли бы применить этот алгоритм к текущему коду преобразования в верхний регистр. Для справки приведем исходный код файла convert_driver_c.c:

				
#include <spu_intrinsics.h>
#include <spu_mfcio.h> /* объявление констант для MFC */
typedef unsigned long long uint64;
typedef unsigned int uint32;

void convert_buffer_to_upper(char *conversion_buffer, int current_transfer_size);

#define MAX_TRANSFER_SIZE 16384
char conversion_buffer[MAX_TRANSFER_SIZE];

typedef struct {
	uint32 length __attribute__((aligned(16)));
	uint64 data __attribute__((aligned(16)));
} conversion_structure;

int main(uint64 spe_id, uint64 conversion_info_ea) {
	conversion_structure conversion_info; /* Информация о данных из PPE */

	/* Новые переменные для отслеживания нашего местоположения в данных */
	uint32 remaining_data; /* Сколько данных осталось в целой строке */
	uint64 current_ea_pointer; /* Наше местоположение в системной памяти */
	uint32 current_transfer_size; /* Насколько велик размер текущей передачи (может
	                               * быть меньше, чем MAX_TRANSFER_SIZE) */

	/* В этой программе мы используем только один тэг */
	mfc_write_tag_mask(1<<0);

	/* Соберем информацию о преобразовании */
	mfc_get(&conversion_info, conversion_info_ea, sizeof(conversion_info), 0, 0, 0);
	spu_mfcstat(MFC_TAG_UPDATE_ALL); /* Wait for Completion */

	/* Настройка цикла */
	remaining_data = conversion_info.length;
	current_ea_pointer = conversion_info.data;

	while(remaining_data > 0) {
		/* Определим, сколько данных осталось передать */
		if(remaining_data < MAX_TRANSFER_SIZE)
			current_transfer_size = remaining_data;
		else
			current_transfer_size = MAX_TRANSFER_SIZE;

		/* Получим реальные данные */
		mfc_getb(conversion_buffer, current_ea_pointer, 
		current_transfer_size, 0, 0, 0);
		spu_mfcstat(MFC_TAG_UPDATE_ALL);

		/* Выполним преобразование */
		convert_buffer_to_upper(conversion_buffer,
		 current_transfer_size);

		/* Поместим данные обратно в системную память */
		mfc_putb(conversion_buffer, current_ea_pointer, 
		current_transfer_size, 0, 0, 0);

		/* Перейдем к следующей части данных */
		remaining_data -= current_transfer_size;
		current_ea_pointer += current_transfer_size;
	}
	spu_mfcstat(MFC_TAG_UPDATE_ALL); /* Wait for Completion */
}

Для этой программы требуются следующие дополнительные файлы из предыдущих статей: файл convert_buffer_c.c из части 5 и файл ppu_dma_main.c из части 3 (позже в этой статье появится еще одна версия). Скомпилируйте и запустите программу точно так же, как и в предыдущей статье (эти команды будут работать для всех примеров из данной статьи):

spu-gcc convert_buffer_c.c convert_driver_c.c -o spe_convert
embedspu -m64 convert_to_upper_handle spe_convert spe_convert_csf.o
gcc -m64 spe_convert_csf.o ppu_dma_main.c -lspe -o dma_convert
./dma_convert

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

• адрес самого буфера
• адрес основной памяти, из которого был заполнен буфер
• размер обрабатываемых данных

С учетом этого, создайте следующую структуру, содержащую всю информацию, относящуюся к буферу:

struct {
	uint64 effective_address __attribute__((aligned(16)));
	uint32 size __attribute__((aligned(16)));
	char data[MAX_TRANSFER_SIZE] __attribute__((aligned(16)));
} buffer;

Затем вам необходимо лишь объявить глобальный массив, состоящий из этих двух буферов:

buffer buffers[2];

Теперь разделим процесс преобразования на два вызова процедур:

1. начало загрузки данных в буфер
2. ожидание, обработка и сохранение данных обратно в буфер

Процесс разделяется данным образом по той причине, что здесь присутствуют независимые устройства, которые должны быть перегруппированы. Начало загрузки данных необходимо вызвать в начале программы, таким образом, этот этап должен быть выделен в отдельную функцию. Итак, ниже представлена версия кода процессора MFC с двойной буферизацией (снова это файл convert_driver_c.c):

				
#include <spu_intrinsics.h>
#include <spu_mfcio.h>

/* Константы */
#define MAX_TRANSFER_SIZE 16384

/* Структуры данных */
typedef unsigned long long uint64;
typedef unsigned int uint32;
typedef struct {
	uint32 length __attribute__((aligned(16)));
	uint64 data __attribute__((aligned(16)));
} conversion_structure;

typedef struct {
	uint32 size __attribute__((aligned(16)));
	uint64 effective_address __attribute__((aligned(16)));
	char data[MAX_TRANSFER_SIZE] __attribute__((aligned(16)));
} buffer;

/* Глобальные переменные */
buffer buffers[2];

/* Вспомогательные функции */
inline uint32 MIN(uint32 a, uint32 b) {
	return a < b ? a : b;
}

inline void wait_for_completion(uint32 mask) {
	mfc_write_tag_mask(mask);
	spu_mfcstat(MFC_TAG_UPDATE_ALL);
}

inline void load_conversion_info(uint64 cinfo_ea, uint64 *data_ea, uint32 *data_size) {
	conversion_structure cinfo;
	mfc_get(&cinfo, cinfo_ea, sizeof(cinfo), 0, 0, 0);
	wait_for_completion(1<<0);
	*data_size = cinfo.length;
	*data_ea = cinfo.data;
}

/* Функции обработки данных */
inline void initiate_transfer(uint32 buf_idx, uint64 *current_ea_pointer,
uint32 *remaining_data) {
	/* Настройка информации буфера */
	buffers[buf_idx].size = MIN(*remaining_data, MAX_TRANSFER_SIZE);
	buffers[buf_idx].effective_address = *current_ea_pointer;
	/* Начало передачи с использованием индекса буфера в качестве тэга DMA */
	mfc_getb(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);
	/* Перемещение указателей */
	*remaining_data -= buffers[buf_idx].size;
	*current_ea_pointer += buffers[buf_idx].size;
}

inline void process_and_put_back(uint32 buf_idx) {
	wait_for_completion(1<<buf_idx);
	/* Выполнение преобразования */
	convert_buffer_to_upper(buffers[buf_idx].data, buffers[buf_idx].size);
	/* Начало обратной передачи DMA 
	с использованием индекса буфера в качестве тэга DMA */
	mfc_putb(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);
}

/* Основной код */
int main(uint64 spe_id, uint64 conversion_info_ea) {
	uint32 remaining_data;
	uint64 current_ea_pointer;

	load_conversion_info(conversion_info_ea, &current_ea_pointer, &remaining_data);

	/* Начинаем заполнять буферы для подготовки к циклу (цикл предполагает, что оба
	 * буфера содержат поступающие данные) */
	initiate_transfer(0, &current_ea_pointer, &remaining_data);
	initiate_transfer(1, &current_ea_pointer, &remaining_data);

	do {
		/* Обработка буфера 0 */
		process_and_put_back(0);
		initiate_transfer(0, &current_ea_pointer, &remaining_data);

		/* Обработка буфера 1 */
		process_and_put_back(1);
		initiate_transfer(1, &current_ea_pointer, &remaining_data);
	} while(buffers[0].size != 0);

	wait_for_completion(1<<0|1<<1);
}

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

В начало

Мультибуферизация

В предыдущем разделе использовалась распространенная идея, называющаяся конвейерной обработкой программного обеспечения. То есть вы делите обработку на этапы, которые могут перекрываться во время выполнения, обеспечивая максимальную пропускную способность. В данном случае ваша конвейерная обработка имеет только два этапа – загрузка/сохранение и обработка. Тем не менее, обобщая эту идею для других случаев, может быть установлено любое количество этапов конвейерной обработки. Основная идея заключается в том, чтобы предоставить каждому конвейеру свой собственный буфер для обработки, и затем обрабатывать каждый буфер одновременно за этап. Когда при конвейерной обработке используется более двух буферов, это называется мультибуферизацией. В случае с SPU двухэтапная конвейерная обработка (такая, как эта) работает для большинства приложений лучше всего, поскольку перемещением данных занимается контроллер MFC, а не процессор, благодаря чему этапы конвейерной обработки могут работать одновременно. Особенность параллельной обработки и передачи данных обеспечивает преимущества двухэтапной конвейерной обработки в программировании SPE.

В дополнение к добавлению этапов конвейерной обработки, существует другой способ получить преимущества от использования дополнительных буферов. Главное преимущество заключается в том, что можно начать множество передач данных в контроллере MFC, а затем позволить ему самостоятельно определять порядок обработки. Например, предположим, что одна область памяти в данный момент находится в пространстве свопинга, в то время как другая область находится в памяти. Имея множество передач, ожидающих выполнения в MFC, контроллер может определить наилучший порядок, в котором необходимо выполнить передачи. Кроме того, это помогает устранить конфликтные ситуации при обращении к шине – когда шина заполнена, программа может обрабатывать дополнительные буферы вместо того, чтобы ждать освобождения шины. Когда шина свободна, она может заполнить дополнительные буферы. В нашей конкретной программе обработка буферов таким способом не влияет существенно на время выполнения, а для некоторых наборов данных сказывается даже отрицательно. Тем не менее, не смотря ни на что, это полезный пример, чтобы показать еще один способ работы с буферами и, в особенности, как использовать MFC_TAG_UPDATE_ANY.

Новый процесс будет выглядеть следующим образом:

1. Постановка на выполнение передач DMA GET для всех буферов. Отметка каждого буфера статусом "filling", если они передают более нуля байтов. Каждый буфер получает уникальный идентификатор тэга DMA.
2. Если нет ни одного буфера со статусом "filling," ожидается завершение всех передач DMA PUT и выход.
3. Ожидание заполнения одного буфера, отмеченного статусом "filling".
4. Обработка буфера.
5. Постановка на выполнение DMA PUT для передачи буфера назад в основную память.
6. Постановка на выполнение передачи DMA GETB для заполнения буфера дополнительными данными после того, как существующие данные сохранились назад.
7. Если передача DMA на предыдущем шаге предназначена, как минимум, для передачи одного байта, (другими словами, действительно остались данные для передачи), буфер отмечается статусом "filling".
8. Возврат к шагу 2.

В этом алгоритме порядок, в котором обрабатываются буферы, существенно менее детерминирован. Здесь ключевая проблема заключается в сведении к минимуму количества ветвлений. Возможные источники ветвлений возникают тогда, когда нужно определить, должен ли буфер быть отмечен статусом "filling", а также опросить буферы с целью выяснения их доступности. Оба этих источника могут быть легко устранены при точном выборе встроенных функций SPU и хорошей разработке структуры данных.

Организовать ожидание готовности буферов, на самом деле, довольно просто. Имея маску нужных вам буферов, вы можете вызвать инструкцию spu_mfcstat(MFC_STAT_UPDATE_ANY), которая возвратит маску всех тех буферов, которые не содержат незаконченных операций (другими словами, все операции завершены), а также будет ожидать до тех пор, пока хотя бы один из буферов не станет доступен. Думайте об этом, как о специализированной версии функции select библиотеки языка C, но только для передач DMA. Итак, она возвратит все доступные буферы, хотя вам нужен только один. Следовательно, вам необходимо преобразовать маску в один индекс, который затем можно использовать для указания обрабатываемого вами буфера, и вам нужно сделать это, не используя ветвлений. Для этого идеально подходит инструкция SPU clz (count leading zeroes – сосчитать ведущие нули, называющаяся в языке C spu_cntlz). Вы можете преобразовать результирующую маску в один индекс, сосчитав ведущие нули и вычтя их из 31. Последовательность команд на языке ассемблера, выполняющая эти действия, может выглядеть следующим образом:

	#предположим, что маска находится в $10
	#сосчитаем ведущие нули
	clz $11, $10
	#вычтем это из 31
	sfi $12, $11, 31
	#$12 теперь содержит индекс буфера, который мы хотим использовать.

На языке C этот код можно написать следующим образом:

	/* buffers_completed содержит маску  */
	spu_extract(
		spu_sub(
			(int32)31,
			spu_cntlz(
				spu_promote(
					(uint32)buffers_completed, 0
				)
			)
		),
		0
	);

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

Теперь вам нужно определить, как запомнить буферы с текущим статусом "filling", а также суметь установить для них соответствующие флаги, не прибегая к ветвлениям. Лучший способ сохранить информацию о статусе – это использовать маску тэгов так, чтобы ее можно было применить непосредственно в качестве маски для инструкции spu_mfcstat. Однако установить эти биты на основании условий, не прибегая к ветвлениям, будет немного сложнее. Последовательность команд на языке ассемблера, выполняющая эти действия, выглядит следующим образом:

	#$10 содержит нашу маску буфера
	#$11 содержит размер последней передачи
	#$12 содержит индекс текущего буфера

	#Преобразуем индекс текущего буфера в бит для битовой маски (хранится в $14)
	il $13, 1
	shl $14, $13, $12

	#Сбросим бит в исходной маске
	xor $10, $10, $14

	#последняя передача больше нуля? (ответ хранится в $15)
	cgti $15, $11, 0

	#Установим или сбросим бит, 
	основываясь на предыдущем результате (ответ хранится в $14)
	and $14, $14, $15

	#Установим бит, основываясь на наших существующих результатах
	or $10, $10, $14

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

	/* сбросим бит */
	*buffers_with_data &= ~(1<<buf_idx);
	/* установим бит в зависимости от условий*/
	*buffers_with_data |= (buffers[buf_idx].size > 0 ? (1<<buf_idx) : 0);

Поскольку задача тривиально распараллеливается, в этой программе вы можете фактически иметь столько буферов, сколько поддерживает локальная память SPU. Поскольку в этой программе каждый буфер может (теоретически) иметь две активных передачи DMA (сохранение и загрузка), программа может использовать максимум восемь буферов, так как контроллер MFC может одновременно обрабатывать только 16 ожидающих операций DMA. Если теперь вы превысите этот лимит, это не затронет логику программы. В этом случае, если вы добавите семнадцатую операцию DMA, SPU просто будет находиться в режиме останова до тех пор, пока одна из операций, ожидающих выполнения, не завершится, после чего программа продолжит свою работу со следующей запланированной операции.

Ниже приведен код новой версии программы (это снова файл convert_driver_c.c):

				
#include <spu_intrinsics.h>
#include <spu_mfcio.h>
typedef unsigned long long uint64;
typedef unsigned int uint32;
typedef int int32;

/* Константы */
#define MAX_TRANSFER_SIZE 16384
#define NUM_BUFFERS 8 /* Контроллер MFC поддерживает только передачи из 16 очередей,
                       * и мы имеем до двух активных на каждый буфер */

/* Структуры данных */
typedef struct {
	uint32 length __attribute__((aligned(16)));
	uint64 data __attribute__((aligned(16)));
} conversion_structure;

typedef struct {
	uint32 size __attribute__((aligned(16)));
	uint64 effective_address __attribute__((aligned(16)));
	char data[MAX_TRANSFER_SIZE] __attribute__((aligned(16)));
} buffer;

buffer buffers[NUM_BUFFERS];

/* Вспомогательные функции */
inline uint32 MIN(uint32 a, uint32 b) {
	return a < b ? a : b;
}

/* Обработка буфера, постановка в очередь передачи DMA 
для возвращения данных и очистка 
 * бита "waiting for data" в buffers_with_data */
inline void process_and_put_back(uint32 buf_idx, uint32 *buffers_with_data) {
	convert_buffer_to_upper(buffers[buf_idx].data, buffers[buf_idx].size);
	mfc_putb(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);
	*buffers_with_data &= ~(1<<buf_idx); /* Clear out bit for this buffer */
}

/* Постановка в очередь передачи DMA GET, 
если есть данные для передачи, установка
 * соответствующего бита в buffers_with_data для индикации того, что мы ожидаем
 * данные в этом буфере */
inline void initiate_transfer(uint32 buf_idx, uint32 *buffers_with_data,
uint64 *current_ea_pointer, uint32 *remaining_data) {
	/* Настройка буфера*/
	buffers[buf_idx].size = MIN(*remaining_data, MAX_TRANSFER_SIZE);
	buffers[buf_idx].effective_address = *current_ea_pointer;

	/* Перемещение указателей */
	*remaining_data -= buffers[buf_idx].size;
	*current_ea_pointer += buffers[buf_idx].size;
	/* Начало передачи (если данных нет, ничего не происходит) */
	mfc_get(buffers[buf_idx].data, buffers[buf_idx].effective_address,
		buffers[buf_idx].size, buf_idx, 0, 0);

	/* Установка бита "Buffer Waiting for Data" только если есть данные для чтения */
	*buffers_with_data |= (buffers[buf_idx].size > 0 ? (1<<buf_idx) : 0);
}

/* Ожидание завершения работы со всеми заданными буферами */
inline void wait_for_completion(uint32 mask) {
	mfc_write_tag_mask(mask);
	spu_mfcstat(MFC_TAG_UPDATE_ALL);
}

/* Загрузка информации о процессе преобразования в целом */
inline void load_conversion_info(uint64 conversion_info_ea, uint64 *current_ea_pointer,
uint32 *remaining_data) {
	conversion_structure conversion_info;

	mfc_get(&conversion_info, conversion_info_ea, sizeof(conversion_info), 0, 0, 0);
	wait_for_completion(1<<0);

	*remaining_data = conversion_info.length;
	*current_ea_pointer = conversion_info.data;
}

/* Возвращение индекса первого доступного буфера с данными */
inline uint32 get_next_buffer(uint32 buffers_with_data) {
	uint32 buffers_completed; /* Будет содержать маску буферов, чьи
	                           * передачи завершены */

	/* Буферы, которые нужно искать */
	mfc_write_tag_mask(buffers_with_data);

	/* Ожидаем, пока хотя бы один буфер не станет доступен */
	buffers_completed = spu_mfcstat(MFC_TAG_UPDATE_ANY);

	/* Используйте "count leading zeros" для определения индекса буфера из
	 * маски buffers_completed */
	return spu_extract(
		spu_sub(
			(int32)31,
			spu_cntlz(
				spu_promote((uint32)buffers_completed, 0)
			)
		),
		0
	);
}

/* Шаги пронумерованы в соответствии с описанием в этом разделе */
int main(uint64 spe_id, uint64 conversion_info_ea) {
	uint32 remaining_data;
	uint64 current_ea_pointer;
	uint32 buffers_with_data = 0; /* 
	Битовая маска для каждого буфера, ожидающего данные,
	                               * используемая для spu_mfcstat в основном цикле */
	uint32 all_buffers = 0; /* Используется для ожидания всех оставшихся передач в
	                         * конце программы */
	uint32 current_buffer_idx;

	load_conversion_info(conversion_info_ea, &current_ea_pointer, &remaining_data);

	/* Шаг 1: Загрузка всех буферов 
	(поскольку NUM_BUFFERS является константой, компилятор
	 *         должен выполнить раскрутку цикла в любом случае) */
	for(current_buffer_idx = 0; 
	current_buffer_idx < NUM_BUFFERS; current_buffer_idx++) {
		initiate_transfer(current_buffer_idx, &buffers_with_data,
			&current_ea_pointer, &remaining_data);
		all_buffers |= 1<<current_buffer_idx;
	}

	/* Шаг 2: Продолжаем, пока еще есть ожидающие буферы*/
	while(buffers_with_data != 0) {
		/* Шаг 3: Получим следующий заполняемый буфер */
		current_buffer_idx = get_next_buffer(buffers_with_data);
		/* Шаги 4 и 5: Обработка буфера и постановка в очередь передачи DMA, 
		 * передающей данные назад в основную память */
		process_and_put_back(current_buffer_idx, &buffers_with_data);
		/* Шаги 6 и 7: Постановим в очередь перезагрузку буфера
		 и пометим буфер как 
                   * "filling" (установив соответствующий бит в remaining_data) */
		initiate_transfer(current_buffer_idx, &buffers_with_data,
			&current_ea_pointer, &remaining_data);
	}

	/* Ожидание завершения всех передач PUT */
	wait_for_completion(all_buffers);
}

Этот код построен специальным образом и обеспечивает то, что основной цикл и вызов функции convert_buffer_to_upper являются единственными обязательными ветвлениями. Другие возможные ветвления в этом коде – это либо встраиваемые функции (которые, очевидно, могут быть встроены компилятором), либо функции, в которых ветвления с легкостью устранены компилятором. Почти так же, компилятор (как GCC, так и XLC) может устранить любое ветвление, которое можно привести к тройному оператору ? : с использованием кода, не содержащего каких-либо побочных эффектов или вызовов невстраиваемых функций.

Сейчас программа PPE, которую мы до сих пор использовали для тестирования программы SPE, работает только с одним буфером, поэтому мы не получим какой-либо выгоды от нашей оптимизации, и проверить разницу в производительности будет довольно трудно. Для того чтобы посмотреть, как эти программы работают с большими наборами данных, ниже приведена версия программы ppu_dma_main.c, использующая большие объемы данных и определяющая время работы SPU.

				
#include <stdio.h>
#include <libspe.h>
#include <errno.h>
#include <string.h>
#include <sys/time.h>
#include <malloc.h>

/* Размер буфера – ДОЛЖЕН быть кратным 16 */
#define BUF_SIZE (16 * 200000)

/* embedspu фактически определяет это в генерируемом объектном файле, 
здесь нам нужна только внешняя ссылка */
extern spe_program_handle_t convert_to_upper_handle;

/* Это структура параметров, которую ожидает наш код SPE */
/* Заметьте, что границы выравнивания всех данных, которые
 * будут переданы SPE, составляют 16 байтов */
typedef struct {
	int length __attribute__((aligned(16)));
	unsigned long long data __attribute__((aligned(16)));
} conversion_structure;

int main() {
	int status = 0;
	int i;
	struct timeval initial_time, final_time;

	/* Создание строки в выровненной границе */
	char *str = memalign(16, BUF_SIZE);

	/* Заполнение строки данными */
	for(i = 0; i < BUF_SIZE - 1 ; i++) {
		str[i] = 'a' + i % 26;
	}

	/* Строка с последним элементом, равным нулю */
	str[BUF_SIZE - 1] = '\0';
	/* Создание структуры преобразования в выровненной границе */
	conversion_structure conversion_info __attribute__((aligned(16)));

	/* Установка элементов данных в структуре параметров */
	conversion_info.length = BUF_SIZE; /* добавим один для нулевого байта */
	conversion_info.data = (unsigned long long)str;

	/* Отметим время начала */
	gettimeofday(&initial_time, NULL);

	/* Создание потока и проверка на предмет ошибок */
	speid_t spe_id = spe_create_thread(0, &convert_to_upper_handle, 
	&conversion_info, NULL, -1, 0);
	if(spe_id == 0) {
		fprintf(stderr, "Unable to create SPE thread: errno=%d\n", errno);
		return 1;
	}

	/* Ожидание завершения работы потока SPE */
	spe_wait(spe_id, &status, 0);

	/* Отметим окончательное время */
	gettimeofday(&final_time, NULL);

	/* Распечатка времени выполнения, затраченного SPU */
	fprintf(stderr, "%llu microseconds\n",
		((long long)final_time.tv_sec * 1000000 + final_time.tv_usec) -
		((long long) initial_time.tv_sec * 1000000 + initial_time.tv_usec));

	/* Вывод результата – раскомментируйте, если вы хотите увидеть его */
	//printf("The converted string is: %s\n", str);

	return 0;
}

В начало

Заключение

В этой статье были рассмотрены два метода управления буферами элемента SPE – двойная буферизация и мультибуферизация. Вы узнали, как можно модифицировать существующий код, чтобы обеспечить одновременную работу с несколькими активными буферами и позволить контроллеру MFC определять порядок, в котором они заполняются, что обеспечивает соответствующее структурирование кода, при котором устраняются все ненужные ветвления.

Ресурсы

• Примите участие в обсуждении материала на форуме.
  
  
• Оригинал статьи Programming high-performance applications on the Cell BE processor, Part 6 (EN).
  
  
• Прочитайте другие статьи серии Программирование высокопроизводительных приложений на процессоре Cell BE.
  
  
• MIT провела курс по программированию PS3. Майкл Перроне (Michael Perrone) из IBM выступил с отличным докладом (EN) (в формате PDF), в котором среди прочего раскрывается, как работает шина EIB (Element Interconnect Bus - шина, соединяющая элементы SPE с основной памятью и друг с другом), и какие она имеет ограничения.
  
  
• В этом же курсе Родрик Рабба (Rodric Rabbah) из IBM выступил с двумя докладами по шаблонам разработки параллельного программирования, включающими в себя некоторые концепции конвейерного программирования, о которых мы говорили (часть 1 (EN), часть 2 (EN) - обе в формате PDF).
  
  
• Другой пример двойной буферизации для SPU (а также масса другой полезной информации о передачах DMA) содержится в этом слайд-шоу (EN) (в формате PDF) из библиотеки IBM Cell/B.E..
  
  
• Вы можете найти полный набор встроенных функций языка C/C++ в спецификации PPU & SPU C/C++ Language Extension Specification (EN).
  
  
• Следите за новостями в мире Cell BE - подпишитесь на новостную ленту IBM microNews (EN).
  
  

Об авторе

Оценить эту статью

Комментарии

В начало

Содержание

• Двойная буферизация
• Мультибуферизация
• Заключение
• Ресурсы
• Об авторе
• Комментарии