Разработка инструментария для повышения эффективности использования кэш-памяти процессора

Рисунок 9 - Иерархия каталогов страниц

Как видно из рисунка виртуальный адрес логически делится на части, которые представляют смещение от начало каждого из каталогов. Чтобы упростить работу с виртуальным адресом в ядре были реализованы макросы. Пример SHIFT макросов показан на рисунке 10.

Рисунок 10 - Пример SHIFT макросов

Как уже говорилось, каждая запись в каталогах страниц описывается типом pgd_t, pmd_t и pte_t. Хотя значениями каждой записи являются обычные целые числа без знака, любой из типов представляется в виде структуры по нескольким причинам. Во-первых, необходимо хранить для каждой страницы, права доступа, которые описываются как набор битов. Во-вторых, в режиме PAE возникает необходимость в дополнительных параметрах. Все биты описывающие состояние страницы перечислены в таблице 3.

Таблица 3 - Описание битов состояния страницы

_PAGE_PRESENT	Страница расположена в оперативной памяти
_PAGE_PROTNONE	Страница в памяти, но не доступна
_PAGE_RW	Устанавливается, если страница доступна для чтения/записи
_PAGE_USER	Устанавливается, если страница доступна из пользовательского пространства
_PAGE_DIRTY	Устанавливается, если произошла операции записи
_PAGE_ACCESSED	Устанавливается, если страница доступна

Все необходимые макросы по работе с каталогами страниц определены в файле <asm/pgtable.h>. Ориентироваться по каталогам страниц позволяют три основных макроса. pgd_offset() в качестве первого аргумента принимает структуру типа mm_struct, а в качестве второго аргумента виртуальный адрес и возвращает запись типа pgd_t. В макрос pmd_offset() первым аргументом передается значение типа pgd_t, а вторым аргументом передается виртуальный адрес. Данный макрос возвращает запись типа pmd_t. Макрос pte_offset() принимает первым аргументом значение типа pmd_t и вторым аргументом, как и макросы выше виртуальный адрес. Данный макрос возвращает структуры типа struct page, которая описывает физическую страницу памяти.

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

Листинг 21 - Реализации функции обхода каталогов страниц

pgd = pgd_offset(mm, va);

if (pgd_none(*pgd) || pgd_bad(*pgd)) {

pgd_ERROR(*pgd); return 0;

}

pud = pud_offset(pgd, va);

if (pud_none(*pud) || pud_bad(*pud)) {

pud_ERROR(*pud); return 0;

}

pmd = pmd_offset(pud, va);

if (pmd_none(*pmd) || pmd_bad(*pmd)) {

pmd_ERROR(*pmd); return 0;

}

pte = pte_offset_map(pmd, va);

if (!pte || pte_none(*pte) || !pte_present(*pte)) {

pte_ERROR(*pte); return 0;

}

page = pte_page(*pte);

if (!page) {

printk(KERN_INFO "page bad!\n"); return 0;

}

pa = page_to_phys(page); // physical address

pte_unmap(pte);

2.6 Реализации библиотеки IPC

Чтобы минимальным образом, воздействовать на адресное пространство профилируемой программы, было принято решение реализовать словарь кэш-промахов и ложного разделения данных, как отдельный процесс. При такой реализации профилировщика требуется применение средств межпроцессорного взаимодействия. В операционной системе Linux реализованы различные средства позволяющие процессам, взаимодействовать между собой. Одними из самых часто применяемых средств межпроцессорного взаимодействия считаются: очереди сообщений, разделяемую память и доменные сокеты UNIX. В рамках дипломной работы стояла задача реализовать библиотеку, которая позволяла использовать все существующие высокопроизводительные средства IPC (Inter Process Communications).

Очереди сообщений, реализованные в Linux, представлены в виде связного списка сообщений, который хранится в памяти ядра, и идентифицируются уникальным номером очереди сообщений. [5]

Структура struct msqid_ds определяет состояние очереди. Некоторые реализации включают дополнительные поля в эту структуру, которые не определяются в стандарте.

Создание очереди сообщений происходит на стороне словаря. Профилировщик инициализирует клиентскую часть и с помощью вызовов ipc_read, ipc_write взаимодействует со словарем. Библиотека IPC реализована на основе классов c++. Объявление класса ObjectIPCClientMsgQueue показана в листинге 4.22, а код реализующий клиентскую часть соединения в листинге 23.

Листинг 22 - Пример объявления класса ObjectIPCClientMsgQueue

class ObjectIPCClientMsgQueue {

private:

int msgid;

public:

ObjectIPCClientMsgQueue(): msgid(-1) { }

~ObjectIPCClientMsgQueue() { }

ConnectionObject* connect_to_server(void);

};

Листинг 23 - Реализация клиентской части соединения для очереди сообщений

ConnectionObject* ObjectIPCClientMsgQueue::connect_to_server(void)

{

key_t key;

const char *name = "/tmp";

if ((key = ftok(name, 'a')) == (key_t) -1) {

perror("ftok");

return NULL;

}

if ((msgid = msgget(key, 0666)) < 0) {

perror("msgget");

return NULL;

}

return new ConnectionObjectMsgQueue(msgid, 1);

}

К одним из самых производительных механизмов межпроцессорного взаимодействия относят разделяемую память. С каждой областью разделяемой памяти связывается структура struct shmid_ds. Эта структура определяет текущее состояние разделяемой памяти. Некоторые реализации включают дополнительные поля в эту структуру, которые не определяются в стандарте.

Механизм разделяемой памяти позволяет двум или более процессам совместно использовать одну и ту же область памяти. При её использовании существует сложность в обеспечении синхронизированного доступа к ней. Если сервер размещает данные в области разделяемой памяти, клиент не должен пытаться читать данные до тех пор, пока сервер не завершит работу с разделяемой памятью. [5]

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

Всю основную часть работы связанную с инициализацией семафоров и созданием области разделяемой памяти выполняет серверная часть, которая реализуется в классе ObjectIPCServerShdMemory. При вызове функции connect_to_client() создается объект типа ConnectionObjectShdMemory, который перегружает функции ipc_read и ipc_write. Реализации функции ipc_read показана в листинге 4.24, а реализация функции ipc_write в листинге 4.25.

Листинг 24 - Пример реализации функции ipc_read

int

ConnectionObjectShdMemory::ipc_read(profiler_data_t *data)

{

if (data == NULL)

return -1;

semop(semid, &WaitFull, 1);

semop(semid, &WaitMutex, 1);

profiler_data_t *data_in_buffer = NULL;

data_in_buffer =

&ringbuffer->data[ringbuffer->head++];

ringbuffer->head &= MASK_RING_BUFFER;

memcpy(data, data_in_buffer,

sizeof(profiler_data_t));

semop(semid, &SignalMutex, 1);

semop(semid, &SignalEmpty, 1);

return 0;

}

Семафор WaitFull будет захвачен только в том случае, если была вызвана операция ipc_write и в кольцевом буфере имеются элементы. После того, как будет извлечен элемент, семафор SignalEmpty увеличивается на 1, сигнализируя о свободном слоте в кольцевом буфере.

Листинг 25 - Пример реализации функции ipc_write

int ConnectionObjectShdMemory::ipc_write(profiler_data_t *data)

{

if (data == NULL)

return -1;

semop(semid, &WaitEmpty, 1);

semop(semid, &WaitMutex, 1);

profiler_data_t *ptr_on_spot_in_buffer = NULL;

ptr_on_spot_in_buffer =

&ringbuffer->data[ringbuffer->tail++];

memcpy(ptr_on_spot_in_buffer, data,

sizeof(profiler_data_t));

ringbuffer->tail &= MASK_RING_BUFFER;

semop(semid, &SignalMutex, 1);

semop(semid, &SingalFull, 1);

return 0;

}

Семафор WaitEmpty будет захвачен только в том случае, если была вызвана операция ipc_read и в кольцевом буфере имеются свободные слоты. После того, как будет помещен элемент, семафор SignalFull увеличивается на 1, сигнализируя о помещении нового элемента в кольцевой буфер.

Последним механизмом, применяемым при реализации библиотеки IPC, является доменные сокеты UNIX. Доменные сокеты, в отличие от сокетов интернета не используют сетевой протокол для взаимодействия. При создании сокета указывается имя файла в файловой системе, через который будут взаимодействовать процессы. Установление соединения происходит при вызове функции connect(). [8]

Доменные сокеты UNIX - это самый простой в использовании механизм взаимодействия. Объект класса ObjectIPCServerDomainSocket создается словарем, чтобы взаимодействовать с профилируемой программой. Объявление класса ObjectIPCServerDomainSocket показана в листинге 26.



Рисунок 11 - Пример взаимодействия через разделяемую память

Листинг 26 - Пример объявления класса ObjectIPCServerDomainSocket

2.7 Инструментация кода средствами Clang

Компилятор Clang - это компилятор с открытым исходным кодом для языков программирования семейства Си. Компилятор построен на основе LLVM оптимизатора и кода генератора, которые обеспечивают высококачественную оптимизацию и кода генерацию для различных целей. Помимо базовых языков и диалектов, clang поддерживает широкий спектр расширений, которые описываются в документации. Эти расширения включены, чтобы быть совместимым с GCC, Microsoft и другими популярными компиляторами, а также для улучшения функциональности компилятора Clang. Драйвер Clang и функции языка специально разработаны так, чтобы быть совместимыми с GNU GCC компилятором, для ослабления миграции кода из GCC на Clang. [6]

Если потребуется написать анализатор кода или механизм инструментации кода, то компилятор Clang станет наилучшим выбором для этих целей. Главная цель компилятора Clang построение C/C++, Objective C фронтенда для LLVM компилятора. Фронтенд - это программа, которая использует исходный код и преобразует его в структурированное дерево, сформированное дерево называется абстрактно-синтаксическим деревом или Abstract Syntax Tree (AST). В большинстве случаев, Clang запускает препроцессор, который развернет все макросы и разпарсит исходный код в AST. Если имеется абстрактно-синтаксическое дерево программы, то анализ исходного кода становится проще. Например, если потребуется изменить старое название переменной, необходимо всего лишь найти в AST узел VariableDeclaration и задать новое имя.

Почти каждый компилятор и инструмент статического анализа кода использует AST для представления исходного кода. Абстрактно-синтаксическое дерево имеет довольно сложную структуру, но в действительности такое разнообразие различных узлов в дереве позволяет более качественно анализировать код. В общем случае, Clang AST состоит из двух очень гибких классов: Decl и Stmt. Существует множество подклассов для каждого из классов. Например, FunctionDecl - описывает прототипы функций и их определения, BinaryOperator - описывает бинарные выражения типа (a + b), CallExpr - описывает вызовы функций, такие как sin(a). [6]

Для анализа кода компилятор Clang предлагает несколько различных механизмов: LibTooling, LibClang.

LibClang - это очень удобный механизм, в случае когда необходим стабильный API. Интерфейс классов компилятора Clang периодически изменяется, и если используется LibTooling, возможно, придется обновлять исходный код анализатора, чтобы соответствовать изменениям Clang. Механизм LibClang не дает полноценный доступ к абстрактно-синтаксическому дереву, он предоставляет только высокоуровневый доступ. Для анализа AST часто рекомендуют использовать LibTooling, который обладает большой гибкостью и имеет полный доступ к абстрактно-синтаксическому дереву.

2.8 Библиотека времени выполнения

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

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



Рисунок 12 - Пример взаимодействия основного потока и потока обработчика

Листинг.27 - Создание потока обработчика

Поток обработчик открывает файл /dev/mcva, для того чтобы взаимодействовать с модулем ядра. Взаимодействие потока обработчика с основным потоком осуществляется при помощи потокобезопасной очереди, реализованной на с++. Основной поток заносит информацию о виртуальном адресе и идентификаторе потока, который обращался к памяти в очередь. Потребность в потокобезопасной очереди возникла, чтобы улучшить производительность runtime библиотеки. Поток обработчик в цикле считывает данные из очереди и формирует их для отправки модулю ядра. Как только модуль ядра по виртуальному адресу найдет физический адрес памяти, он передаст эту информацию обратно потоку обработчику, который уже с помощью средств межпроцессорного взаимодействия передаст всю необходимую информацию словарю. В листинге 28 показана реализация основного цикла потока обработчика.

Листинг 28 - Основной цикл потока обработчика

for (;;) {

std::shared_ptr<T> data = queue->wait_and_pop();

T* ptr_data = data.get();

/* Формируем данные перед отправкой модулю ядра */

input_data.pid = ptr_data->pid;

input_data.va = ptr_data->va;

printf("thread handler va = %lu\n", ptr_data->va);

write(fd, &input_data, sizeof(in_data_t));

read(fd, &output_data, sizeof(out_data_t));

printf("thread handler pa = %lu\n", output_data.pa);

/* Формируем данные перед отправкой симулятору кэша */

memcpy(&profiler_data.name, ptr_data->name,

strlen(ptr_data->name));

profiler_data.operation = ptr_data->operation;

profiler_data.pa = output_data.pa;

profiler_data.va = output_data.va;

if (conn_object->ipc_write(&profiler_data) < 0) {

std::cerr << "failed sending data" << std::endl;

exit(-1);

}

if (ptr_data->va == 0)

break;

}

3. Результаты профилирования

Чтобы продемонстрировать работу профилировщика, была выбрана программа, реализующая алгоритм умножения матриц, в неоптимизированной версии, которой выявлена проблема с частыми кэш промахами, из-за нарушения локальности обращения к близлежащим ячейкам памяти. Реализация алгоритма показана в листинге 5.1, а результаты профилирования показаны в таблице 4. Размер матриц задается константно в программе и равен 100x100.

Листинг 3.1 - Неоптимизированный алгоритм умножения матриц

Таблица 4 - Результаты профилирования программы из листинга 5.1

Профилировщик	Число кэш промахов
perf	168726
valgrind	134365
Раз. Инструмен.	135757

Простейший вариант оптимизации алгоритма умножения матриц показан в листинге 3.2. Результаты профилирования показаны в таблице 5.2.

Листинг 3.2 - Оптимизированный алгоритм умножения матриц

Таблица 5 - Результаты профилирования программы из листинга 5.2

Профилировщик2.8 Библиотека времени выполнения	Число кэш промахов
perf	160714
valgrind	127599
Раз. Инструмен.	119651

Заключение

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

память кэш синхронизация код

Библиография

1 Таненбаум Э. Современные операционные системы. - СПб.: Питер, 2010. - 1120 с.

2 Randal E. Bryant, David R. O'Hallaron. Computer Systems: A Programmer's Perspective. - Addison-Wesley, 2014. - 1084 с.

3 Бовет Д., Чезати М. Ядро Linux, 3-е изд. - СПб.: БХВ-Петербург, 2007. - 1104 с.

4 Лав Р. Ядро Linux: описание процесса разработки, 3-е изд. - М.: Вильямс, 2015. - 496 с.

5 Стивенс У., Раго С. UNIX. Профессиональное программирование, 2-е изд. - СПб.: Символ, 2007. - 1035 с.

6 Clang - Официальный сайт поддержки компилятора Clang. - URL: http://clang.llvm.org/ (Дата обращения 09.06.2016)

7 Intel - Официальный сайт компании Intel. - URL: https://software.intel.com (Дата обращения 05.06.2016)

8 Лав Р. Linux. Системное программирование. 2-е изд. - СПб.: Питер, 2014. - 448 с.

Приложение 1

Наиболее употребляемые текстовые сокращения

CPU - central processing unit

SMT - simultaneous multithreading

SMP - symmetric multiprocessing

NUMA - non-uniform memory architecture

AST - abstract syntax tree

VFS - virtual file system

API - application programming interface

UMA - uniform memory architecture

PGD - page global directory

PMD - page middle directory

GCC - gnu compiler collection

Приложение 2

Листинг В.1 - Реализация модуля ядра

/* mcva.h */

#ifndef MCVA_H

#define MCVA_H

#include <linux/wait.h>

#include <linux/semaphore.h>

#include <linux/module.h>

#include <asm/atomic.h>

#include "data.h"

#include "queue.h"

typedef struct information_converting_virtual_addresses {

wait_queue_head_t readq;

struct semaphore semread;

struct semaphore semwrite;

atomic_t readiness_read;

queue_t *data_queue;

} icva_t;

#endif // MCVA_H

/* data.h */

#ifndef DATA_H

#define DATA_H

#define NAME_LENGTH 64

typedef enum {

OPER_READ,

OPER_WRITE

} rw_oper_t;

typedef struct processing_data {

pid_t pid;

char name[NAME_LENGTH];

unsigned long va;

rw_oper_t operation;

} proc_data_t;

/* Структуры обрабатываемые модулем ядра */

typedef struct input_data {

pid_t pid;

unsigned long va;

} in_data_t;

typedef struct output_data {

unsigned long va;

unsigned long pa;

} out_data_t;

typedef struct profiler_data {

int pid;

char name[NAME_LENGTH];

unsigned long pa;

unsigned long va;

rw_oper_t operation;

} profiler_data_t;

endif

/* mcva_dev_main.c */

include <linux/fs.h>

include <linux/pid.h>

include <linux/slab.h>

include <linux/cdev.h>

include <linux/init.h>

include <linux/sched.h>

include <asm/uaccess.h>

include <asm/pgtable.h>

include <linux/device.h>

include <linux/mm_types.h>

include <linux/rcupdate.h>

include "mcva.h"

define DEVICE_NAME "mcva"

define DEVICE_FIRST 0

define DEVICE_COUNT 1

MODULE_LICENSE("GPL");

MODULE_AUTHOR("Kirill Ryzhkov");

MODULE_VERSION("1.0");

static struct class *mcva_class;

static struct cdev mcva_cdev;

static dev_t mcva_dev_number;

static int flag_busy = 0;

static inline void initialization_icva(icva_t *p)

{

init_waitqueue_head(&p->readq);

atomic_set(&p->readiness_read, 0);

sema_init(&p->semread, 1);

sema_init(&p->semwrite, 1);

}

int mcva_open(struct inode *inode, struct file *file)

{

if (flag_busy == 0) {

icva_t *__module;

__module = kmalloc(sizeof(icva_t), GFP_KERNEL);

if (__module == NULL)

return -ENOMEM;

if ((__module->data_queue = queue_create()) == NULL)

return -ENOMEM;

initialization_icva(__module);

file->private_data = __module;

flag_busy = 1;

} else {

printk(KERN_INFO "module is currently busy\n");

return -EBUSY;

}

return 0;

}

int mcva_release(struct inode *inode, struct file *file)

{

icva_t *__module = file->private_data;

kfree(__module->data_queue);

kfree(__module);

file->private_data = NULL;

flag_busy = 0;

return 0;

}

int checking_virtual_address(struct mm_struct *mm, unsigned long va)

{

struct vm_area_struct *vm_area;

for (vm_area = mm->mmap;

vm_area != NULL; vm_area = vm_area->vm_next) {

if (vm_area->vm_start >= va && va <= vm_area->vm_end)

return 1;

}

return 0;

}

unsigned long traversal_page_directory(struct mm_struct *mm, unsigned long va)

{

unsigned long pa;

struct page *page;

pgd_t *pgd;

pud_t *pud;

pmd_t *pmd;

pte_t *pte;

pgd = pgd_offset(mm, va);

if (pgd_none(*pgd) || pgd_bad(*pgd)) {

pgd_ERROR(*pgd);

return 0;

}

pud = pud_offset(pgd, va);

if (pud_none(*pud) || pud_bad(*pud)) {

pud_ERROR(*pud);

return 0;

}

pmd = pmd_offset(pud, va);

if (pmd_none(*pmd) || pmd_bad(*pmd)) {

pmd_ERROR(*pmd);

return 0;

}

pte = pte_offset_map(pmd, va);

if (!pte || pte_none(*pte) || !pte_present(*pte)) {

pte_ERROR(*pte);

return 0;

}

page = pte_page(*pte);

if (!page) {

printk(KERN_INFO "page bad!\n");

return 0;

}

pa = page_to_phys(page);

pte_unmap(pte);

return pa;

}

inline unsigned long get_pa(struct task_struct *task, unsigned long va)

{

return traversal_page_directory(task->mm, va);

}

static ssize_t mcva_bread(struct file *file, char __user *buf,

size_t count, loff_t *f_pos)

{

struct pid *pid;

struct task_struct *task;

out_data_t output_data;

in_data_t *input_data;

unsigned long page_address;

unsigned long offset_address;

icva_t *__module = file->private_data;

if (down_interruptible(&__module->semread))

return -ERESTARTSYS;

if (atomic_read(&__module->readiness_read) == 1) {

if ((input_data = (in_data_t *)

queue_dequeue(__module->data_queue)) == NULL)

atomic_set(&__module->readiness_read, 0);

}

while (atomic_read(&__module->readiness_read) == 0) {

up(&__module->semread);

if (file->f_flags & O_NONBLOCK)

return -EAGAIN;

if (wait_event_interruptible(__module->readq,

atomic_read(&__module->readiness_read) != 0))

return -ERESTARTSYS;

if (down_interruptible(&__module->semread))

return -ERESTARTSYS;

/* wake up due to signals */

if ((input_data = (in_data_t *)

queue_dequeue(__module->data_queue)) == NULL)

atomic_set(&__module->readiness_read, 0);

}

rcu_read_lock();

pid = find_vpid(input_data->pid);

rcu_read_unlock();

if ((task = get_pid_task(pid, PIDTYPE_PID)) == NULL) {

printk(KERN_DEBUG "<%s> get_pid_task failed\n", __func__);

up(&__module->semread);

return -EINVAL;

}

page_address = get_pa(task, input_data->va);

offset_address = page_address +

(input_data->va & (PAGE_SIZE - 1));

output_data.va = input_data->va;

output_data.pa = offset_address;

if (copy_to_user(buf, (void* )&output_data, count)) {

up(&__module->semread);

return -EFAULT;

}

kfree(input_data);

up(&__module->semread);

return count;

}

static ssize_t mcva_bwrite(struct file *file,

const char __user *buf, size_t count, loff_t *f_pos)

{

int ret;

in_data_t *input_data;

icva_t *__module = file->private_data;

if (down_interruptible(&__module->semwrite))

return -ERESTARTSYS;

if ((input_data =

kmalloc(sizeof(in_data_t), GFP_KERNEL)) == NULL) {

printk(KERN_DEBUG "<%s> kmalloc failed\n", __func__);

up(&__module->semwrite);

return -ENOMEM;

}

if (copy_from_user(input_data, buf, count)) {

printk(KERN_DEBUG

"<%s> copy_from_user failed\n", __func__);

up(&__module->semwrite);

return -EFAULT;

}

if ((ret =

queue_enqueue(__module->data_queue, (void *)input_data)) < 0) {

printk(KERN_DEBUG "<%s> queue_enqueue failed\n", __func__);

up(&__module->semwrite);

return ret;

}

if (atomic_read(&__module->readiness_read) == 0) {

atomic_set(&__module->readiness_read, 1);

wake_up_interruptible(&__module->readq);

}

up(&__module->semwrite);

return count;

}

static struct file_operations mcva_fops = {

owner = THIS_MODULE,

open = mcva_open,

read = mcva_bread,

write = mcva_bwrite,

release = mcva_release,

};

static int __init mcva_init(void)

{

int ret;

Request dynamic allocation of a device major number */

if (alloc_chrdev_region(&mcva_dev_number, DEVICE_FIRST,

DEVICE_COUNT, DEVICE_NAME) < 0) {

printk(KERN_DEBUG "=== Can not register device === \n");

return -1;

}

mcva_class = class_create(THIS_MODULE, DEVICE_NAME);

cdev_init(&mcva_cdev, &mcva_fops);

mcva_cdev.owner = THIS_MODULE;

if ((ret = cdev_add(&mcva_cdev, mcva_dev_number, 1)) < 0) {

unregister_chrdev_region(mcva_dev_number, DEVICE_COUNT);

printk(KERN_DEBUG "=== Can not add char device === \n");

goto err;

}

device_create(mcva_class, NULL, mcva_dev_number, "%s", DEVICE_NAME);

printk(KERN_INFO "=== mcva installed %d:%d === \n",

MAJOR(mcva_dev_number), MINOR(mcva_dev_number));

err:

return ret;

}

static void __exit mcva_exit(void)

{

Release the major number */

unregister_chrdev_region(mcva_dev_number, DEVICE_COUNT);

device_destroy(mcva_class, mcva_dev_number);

class_destroy(mcva_class);

cdev_del(&mcva_cdev);

}

module_init(mcva_init);

module_exit(mcva_exit);

Листинг В.2 - Реализация потокобезопасной очереди применяемой в ядре

/* queue.h */

ifndef QUEUE_H

define QUEUE_H

include <linux/list.h>

include <linux/spinlock.h>

typedef struct queue_item {

struct list_head list;

void *item;

} queue_item_t;

typedef struct queue_data {

struct list_head list;

spinlock_t lock;

} queue_t;

queue_t* queue_create(void);

void queue_delete(queue_t *q);

int queue_enqueue(queue_t *q, void *item);

void* queue_dequeue(queue_t *q);

int queue_is_empty(queue_t *q);

int queue_lock(queue_t *q);

int queue_unlock(queue_t *q);

endif // QUEUE_H

/* queue.c */

include <linux/init.h>

include <linux/slab.h>

include <linux/module.h>

include "queue.h"

MODULE_LICENSE("GPL");

MODULE_AUTHOR("Kirill Ryzhkov");

MODULE_VERSION("1.0");

queue_t* queue_create(void)

{

queue_t * q;

q = (queue_t *)kmalloc(sizeof(queue_t), GFP_KERNEL);

if (!q) {

printk(KERN_DEBUG "<%s> kmalloc failed\n", __func__);

return NULL;

}

INIT_LIST_HEAD(&(q->list));

spin_lock_init(&(q->lock));

return q;

}

void queue_delete(queue_t *q)

{

kfree((void *)q);

}

int queue_enqueue(queue_t *q, void *item)

{

queue_item_t *queue_item;

queue_item =

(queue_item_t *)kmalloc(sizeof(queue_item_t), GFP_KERNEL);

if (!queue_item) {

printk(KERN_DEBUG "<%s> kmalloc failed\n", __func__);

return -ENOMEM;

}

queue_item->item = item;

queue_lock(q);

list_add_tail(&(queue_item->list), &(q->list));

queue_unlock(q);

return 0;

}

void* queue_dequeue(queue_t *q)

{

queue_item_t *queue_item;

void *item;

if (queue_is_empty(q))

return NULL;

queue_lock(q);

queue_item = list_first_entry(&(q->list), queue_item_t, list);

item = queue_item->item;

list_del(&(queue_item->list));

kfree(queue_item);

queue_unlock(q);

return item;

}

int queue_is_empty(queue_t *q)

{

int ret;

queue_lock(q);

ret = list_empty(&(q->list));

queue_unlock(q);

return ret;

}

int queue_lock(queue_t *q)

{

spin_lock(&(q->lock));

return 0;

}

int queue_unlock(queue_t *q)

{

spin_unlock(&(q->lock));

return 0;

}

Листинг В.3 - Реализация потокобезопасной очереди применяемой в runtime библиотеке

/* threadsafe_queue.hpp */

ifndef THREADSAFE_QUEUE_H

define THREADSAFE_QUEUE_H

include <mutex>

include <memory>

include <condition_variable>

template <typename T>

class threadsafe_queue {

private:

struct node {

std::shared_ptr<T> data;

std::unique_ptr<node> next;

};

std::mutex head_mutex;

std::unique_ptr<node> head;

std::mutex tail_mutex;

node* tail;

std::condition_variable data_cond;

node* get_tail()

{

std::lock_guard<std::mutex> tail_lock(tail_mutex);

return tail;

}

std::unique_ptr<node> pop_head()

{

std::unique_ptr<node> old_head = std::move(head);

head = std::move(old_head->next);

return old_head;

}

std::unique_lock<std::mutex> wait_for_data()

{

std::unique_lock<std::mutex> head_lock(head_mutex);

data_cond.wait(head_lock,

[&]{return head.get() != get_tail();});

return std::move(head_lock);

}

std::unique_ptr<node> wait_pop_head()

{

std::unique_lock<std::mutex>

head_lock(wait_for_data());

return pop_head();

}

std::unique_ptr<node> wait_pop_head(T& value)

{

std::unique_lock<std::mutex>

head_lock(wait_for_data());

value = std::move(*head->data);

return pop_head();

}

std::unique_ptr<node> try_pop_head()

{

std::lock_guard<std::mutex> head_lock(head_mutex);

if (head.get() == get_tail()) {

return std::unique_ptr<node>();

}

return pop_head();

}

std::unique_ptr<node> try_pop_head(T& value)

{

std::lock_guard<std::mutex> head_lock(head_mutex);

if (head.get() == get_tail()) {

return std::unique_ptr<node>();

}

value = std::move(*head->data);

return pop_head();

}

public:

threadsafe_queue(): head(new node), tail(head.get()) {}

threadsafe_queue(const threadsafe_queue& other) = delete;

threadsafe_queue&

operator=(const threadsafe_queue& other) = delete;

std::shared_ptr<T> try_pop()

{

std::unique_ptr<node> old_head = try_pop_head();

return old_head ? old_head->data:

std::shared_ptr<T>();

}

bool try_pop(T& value)

{

std::unique_ptr<node> const old_head =

try_pop_head(value);

return old_head;

}

std::shared_ptr<T> wait_and_pop()

{

std::unique_ptr<node> const old_head =

wait_pop_head();

return old_head->data;

}

void wait_and_pop(T& value)

{

std::unique_ptr<node> const old_head =

wait_pop_head(value);

}

void push(T new_value)

{

std::shared_ptr<T>

new_data(std::make_shared<T>(std::move(new_value)));

std::unique_ptr<node> p(new node);

{

std::lock_guard<std::mutex>

tail_lock(tail_mutex);

tail->data = new_data;

node* const new_tail = p.get();

tail->next = std::move(p);

tail = new_tail;

}

data_cond.notify_one();

}

bool empty()

{

std::lock_guard<std::mutex> head_lock(head_mutex);

return (head.get() = get_tail());

}

};

endif // THREADSAFE_QUEUE_H

Листинг В.4 - Реализация серверных классов библиотеки IPC

/* ObjectIPCServerSocket.cpp */

include <stdlib.h>

include <errno.h>

include "mutex.h"

include "ringbuffer.h"

include "ConnectionObject.h"

class ObjectIPCServerDomainSocket {

private:

int __socket;

struct sockaddr_un addr;

public:

ObjectIPCServerDomainSocket()

{

__socket = -1;

memset(&addr, 0, sizeof(struct sockaddr_un));

}

ObjectIPCServerDomainSocket();

int create_ipc(void);

int close_ipc(void);

ConnectionObject* connect_to_client(void);

};

ObjectIPCServerDomainSocket::~ObjectIPCServerDomainSocket()

{ close_ipc(); }

int ObjectIPCServerDomainSocket::close_ipc(void)

{

if (__socket != -1) {

close(__socket);

__socket = -1;

}

return 0;

}

int ObjectIPCServerDomainSocket::create_ipc(void)

{

ssize_t len;

const char *path = "/tmp/profiler_socket";

if ((__socket = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {

perror("socket");

return (-1);

}

addr.sun_family = AF_UNIX;

strcpy(addr.sun_path, path);

unlink(path);

len = strlen(addr.sun_path) + sizeof(addr.sun_family);

if (bind(__socket, (struct sockaddr *)&addr, len) < 0) {

perror("bind");

return (-1);

}

if (listen(__socket, 10) < 0) {

perror("listen");

return (-1);

}

return 0;

}

ConnectionObject* ObjectIPCServerDomainSocket::connect_to_client(void)

{

int __client_socket;

socklen_t sock_len;

if ((__client_socket =

accept(__socket, (struct sockaddr *)&addr, &sock_len)) < 0) {

perror("accept");

return NULL;

}

return new ConnectionObjectDomainSocket(__client_socket);

}

class ObjectIPCServerMsgQueue {

private:

int msgid;

public:

ObjectIPCServerMsgQueue(): msgid(-1) { }

~ObjectIPCServerMsgQueue() { close_ipc(); }

int create_ipc(void);

int close_ipc(void);

ConnectionObject* connect_to_client(void);

};

int ObjectIPCServerMsgQueue::create_ipc(void)

{

key_t key;

const char *name = "/tmp";

if ((key = ftok(name, 'a')) == (key_t) -1) {

perror("ftok");

return -1;

}

if ((msgid = msgget(key, IPC_CREAT | 0666)) < 0) {

perror("msgget");

return -1;

}

return 0;

}

int ObjectIPCServerMsgQueue::close_ipc(void)

{

int err;

if (msgid != -1) {

if ((err = msgctl(msgid, IPC_RMID, NULL)) < 0) {

perror("msgctl");

return (-1);

}

msgid = -1;

}

return 0;

}

ConnectionObject* ObjectIPCServerMsgQueue::connect_to_client(void)

{

return new ConnectionObjectMsgQueue(msgid, 0);

}

static union semun

{

int val;

struct semid_ds *buf;

short *array;

} seminfo;

class ObjectIPCServerShdMemory {

int shmid;

int semid;

public:

ObjectIPCServerShdMemory(): shmid(-1), semid(-1) { }

~ObjectIPCServerShdMemory();

int create_ipc(void);

int close_ipc(void);

ConnectionObject* connect_to_client(void);

};

ObjectIPCServerShdMemory::~ObjectIPCServerShdMemory()

{

close_ipc();

}

int ObjectIPCServerShdMemory::close_ipc(void)

{

struct shmid_ds shminfo;

if (semid != -1) {

if (semctl(semid, SEM_COUNT - 1, IPC_RMID, 0) < 0) {

perror("semctl");

return -1;

}

semid = -1;

}

if (shmid != -1) {

if (shmctl(shmid, IPC_RMID, &shminfo) < 0) {

perror("shmid");

return -1;

}

shmid = -1;

}

return 0;

}

int ObjectIPCServerShdMemory::create_ipc(void)

{

key_t key;

const char *name = "/tmp";

if ((key = ftok(name, 'b')) == (key_t) -1) {

perror("ftok");

return -1;

}

if ((semid = semget(key, 3, IPC_CREAT | 0666)) < 0) {

perror("semget");

return -1;

}

seminfo.val = BUF_SIZE;

semctl(semid, SEM_EMPTY, SETVAL, seminfo);

seminfo.val = 1;

semctl(semid, SEM_MUTEX, SETVAL, seminfo);

seminfo.val = 0;

semctl(semid, SEM_FULL, SETVAL, seminfo);

if ((key = ftok(name, 'c')) == (key_t) -1) {

perror("ftok");

return -1;

}

if ((shmid =

shmget(key, sizeof(ring_buffer_t), IPC_CREAT | 0666)) < 0) {

perror("shmget");

return -1;

}

ring_buffer_t *ringbuffer =

(ring_buffer_t *) shmat(shmid, 0, SHM_W);

if (ringbuffer == (void *) (-1)) {

perror("shmat failed");

return -1;

}

/* initialize ring buffer */

ringbuffer->head = 0;

ringbuffer->tail = 0;

if (shmdt(ringbuffer) < 0) {

perror("shmdt failed");

return -1;

}

return 0;

}

ConnectionObject* ObjectIPCServerShdMemory::connect_to_client(void)

{

return new ConnectionObjectShdMemory(shmid, semid);

}

Листинг 5 - Реализация клиентских классов библиотеки IPC

/* ObjectIPCClientSocket.cpp */

include <stdio.h>

include <stdlib.h>

include <errno.h>

include <string.h>

include <unistd.h>

include <stdlib.h>

include <sys/types.h>

include <sys/socket.h>

include <sys/un.h>

include "data.h"

include "ConnectionObject.h"

class ObjectIPCClientDomainSocket {

private:

int __socket;

public:

ObjectIPCClientDomainSocket() { }

~ObjectIPCClientDomainSocket() { close (__socket); }

ConnectionObject* connect_to_server(void);

};

ConnectionObject* ObjectIPCClientDomainSocket::connect_to_server(void)

{

struct sockaddr_un remote;

const char *path = "/tmp/profiler_socket";

if ((__socket = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {

perror("socket");

return NULL;

}

remote.sun_family = AF_UNIX;

strcpy(remote.sun_path, path);

ssize_t len =

strlen(remote.sun_path) + sizeof(remote.sun_family);

if (connect(__socket, (struct sockaddr *)&remote, len) < 0) {

perror("connect");

return NULL;

}

return new ConnectionObjectDomainSocket(__socket);

}

class ObjectIPCClientMsgQueue {

private:

int msgid;

public:

ObjectIPCClientMsgQueue(): msgid(-1) { }

~ObjectIPCClientMsgQueue() { }

ConnectionObject* connect_to_server(void);

};

ConnectionObject* ObjectIPCClientMsgQueue::connect_to_server(void)

{

key_t key;

const char *name = "/tmp";

if ((key = ftok(name, 'a')) == (key_t) -1) {

perror("ftok");

return NULL;

}

if ((msgid = msgget(key, 0666)) < 0) {

perror("msgget");

return NULL;

}

return new ConnectionObjectMsgQueue(msgid, 1);

}

class ObjectIPCClientShdMemory {

private:

int shmid;

int semid;

public:

ObjectIPCClientShdMemory(): shmid(-1), semid(-1) { }

~ObjectIPCClientShdMemory() { }

ConnectionObject* connect_to_server(void);

};

ConnectionObject* ObjectIPCClientShdMemory::connect_to_server(void)

{

key_t key;

const char *name = "/tmp";

if ((key = ftok(name, 'b')) == (key_t) -1) {

perror("ftok");

return NULL;

}

if ((semid = semget(key, 0, 0666)) < 0) {

perror("semget");

return NULL;

}

if ((key = ftok(name, 'c')) == (key_t) -1) {

perror("ftok");

return NULL;

}

if ((shmid = shmget(key, 0, 0666)) < 0) {

perror("shmget");

return NULL;

}

return new ConnectionObjectShdMemory(shmid, semid);

}

Листинг В.6 - Реализация класса ConnectionObject

/* mutex.h */

ifndef MUTEX_H

define MUTEX_H

include <sys/sem.h>

include <sys/ipc.h>

define SEM_MUTEX 0

define SEM_EMPTY 1

define SEM_FULL 2

define SEM_COUNT 3

struct sembuf WaitMutex={SEM_MUTEX, -1, 0};

struct sembuf SignalMutex={SEM_MUTEX, 1, 0};

struct sembuf WaitEmpty={SEM_EMPTY, -1, 0};

struct sembuf SignalEmpty={SEM_EMPTY, 1, 0};

struct sembuf WaitFull={SEM_FULL, -1, 0};

struct sembuf SingalFull={SEM_FULL, 1, 0};

endif // MUTEX_H

/* ringbuffer.h */

ifndef RING_BUFFER_H

define RING_BUFFER_H

include "data.h"

define BUF_SIZE 10

define MASK_RING_BUFFER (BUF_SIZE - 1)

typedef struct ringbuffer {

int head;

int tail;

profiler_data_t data[BUF_SIZE];

} ring_buffer_t;

endif // RING_BUFFER_H

/* ConnectionObject.c */

include <stdio.h>

include <stdlib.h>

include <unistd.h>

include <sys/un.h>

include <sys/types.h>

include <sys/socket.h>

include <sys/msg.h>

include <sys/shm.h>

include <sys/ipc.h>

include "data.h"

include "mutex.h"

include "ringbuffer.h"

typedef struct wrap_profiler_data {

long type;

profiler_data_t data;

} msg_data_t;

class ConnectionObject {

public:

ConnectionObject() { }

~ConnectionObject() { }

virtual int ipc_read(profiler_data_t *data) = 0;

virtual int ipc_write(profiler_data_t *data) = 0;

};

class ConnectionObjectDomainSocket: public ConnectionObject {

private:

int __client_socket;

public:

ConnectionObjectDomainSocket(int __socket):

client_socket(__socket) { }

ConnectionObjectDomainSocket() { }

int ipc_read(profiler_data_t *data)

{

int err;

if ((err = recv(__client_socket,

data, sizeof(profiler_data_t), 0)) < 0) {

perror("recv");

return (-1);

}

return 0;

}

int ipc_write(profiler_data_t *data)

{

int err;

if ((err = send(__client_socket,

data, sizeof(profiler_data_t), 0)) < 0) {

perror("send");

return (-1);

}

return 0;

}

};

class ConnectionObjectMsgQueue: public ConnectionObject {

private:

int mode;

int msgid;

public:

ConnectionObjectMsgQueue(int __socket, int __mode /* 0 - server, 1 - client */)

msgid(__socket), mode(__mode) { }

ConnectionObjectMsgQueue() { }

int ipc_read(profiler_data_t *data)

{

int err = 0;

msg_data_t recvbuffer;

if (mode == 0) {

if ((err = msgrcv(msgid, (void *)&recvbuffer,

sizeof(msg_data_t) - sizeof(long), 2L, 0)) < 0){

perror("msgrcv");

return -1;

}

} else {

if ((err = msgrcv(msgid, (void *)&recvbuffer,

sizeof(msg_data_t) - sizeof(long), 1L, 0)) < 0){

perror("msgrcv");

return -1;

}

}

memcpy(data, &recvbuffer.data,

sizeof(profiler_data_t));

return err;

}

int ipc_write(profiler_data_t *data)

{

int err = 0;

msg_data_t sendbuffer;

if (mode == 0) {

sendbuffer.type = 1L; // client data

memcpy(&(sendbuffer.data), data,

sizeof(profiler_data_t));

if ((err = msgsnd(msgid, (void *)&sendbuffer,

sizeof(msg_data_t) - sizeof(long), 0)) < 0) {

perror("msgsnd");

return (-1);

}

} else {

sendbuffer.type = 2L; // server data

memcpy(&(sendbuffer.data), data,

sizeof(profiler_data_t));

if ((err = msgsnd(msgid, (void *)&sendbuffer,

sizeof(msg_data_t) - sizeof(long), 0)) < 0) {

perror("msgsnd");

return (-1);

}

}

return err;

}

};

class ConnectionObjectShdMemory: public ConnectionObject {

private:

ring_buffer_t *ringbuffer;

int shmid;

int semid;

public:

ConnectionObjectShdMemory(int __shmid, int __semid):

shmid(__shmid), semid(__semid)

{

ringbuffer =

(ring_buffer_t *)shmat(shmid, 0, SHM_R | SHM_W);

if (ringbuffer == (void *) -1) {

ringbuffer = NULL;

}

}

ConnectionObjectShdMemory()

{

if (ringbuffer != NULL) {

shmdt(ringbuffer);

}

}

int ipc_read(profiler_data_t *data)

{

if (data == NULL)

return -1;

semop(semid, &WaitFull, 1);

semop(semid, &WaitMutex, 1);

profiler_data_t *data_in_buffer = NULL;

data_in_buffer =

&ringbuffer->data[ringbuffer->head++];

ringbuffer->head &= MASK_RING_BUFFER;

memcpy(data, data_in_buffer, sizeof(profiler_data_t));

semop(semid, &SignalMutex, 1);

semop(semid, &SignalEmpty, 1);

return 0;

}

int ipc_write(profiler_data_t *data)

{

if (data == NULL)

return -1;

semop(semid, &WaitEmpty, 1);

semop(semid, &WaitMutex, 1);

profiler_data_t *ptr_on_spot_in_buffer = NULL;

ptr_on_spot_in_buffer =

&ringbuffer->data[ringbuffer->tail++];

memcpy(ptr_on_spot_in_buffer, data,

sizeof(profiler_data_t));

ringbuffer->tail &= MASK_RING_BUFFER;

semop(semid, &SignalMutex, 1);

semop(semid, &SingalFull, 1);

return 0;

}

};

Листинг В.7 - Реализация runtime библиотеки

include <mutex>

include <iostream>

include <thread>

include <condition_variable>

include <fcntl.h>

include <unistd.h>

include <stdlib.h>

include <sys/un.h>

include <sys/stat.h>

include <sys/types.h>

include <sys/socket.h>

include <pthread.h>

include "data.h"

include "threadsafe_queue.h"

include "ObjectIPCDomainSocket.h"

include "ObjectIPCShdMemory.h"

include "ConnectionObject.h"

static int initialization = 0;

threadsafe_queue<struct processing_data> * queue;

static int termination_condition = 0;

pthread_mutex_t __mutex = PTHREAD_MUTEX_INITIALIZER;

pthread_cond_t __condition = PTHREAD_COND_INITIALIZER;

template <typename T>

void Handler(void)

{

ObjectIPCClientShdMemory ipc;

ConnectionObject * conn_object;

conn_object = ipc.connect_to_server();

if (conn_object == NULL) {

std::cerr <<

"connection on the server failed" << std::endl;

exit(-1);

}

/* Открываем файл устройства для взаимодействия

с модулем ядра */

int fd = open("/dev/mcva", O_RDWR);

if (fd < 0) {

std::cerr << "failed open /dev/mcva" << std::endl;

exit(-1);

}

in_data_t input_data;

out_data_t output_data;

profiler_data_t profiler_data;

for (;;) {

std::shared_ptr<T> data = queue->wait_and_pop();

T* ptr_data = data.get();

/* Формируем данные перед отправкой модулю ядра */

input_data.pid = ptr_data->pid;

input_data.va = ptr_data->va;

printf("thread handler va = %lu\n", ptr_data->va);

write(fd, &input_data, sizeof(in_data_t));

read(fd, &output_data, sizeof(out_data_t));

printf("thread handler pa = %lu\n", output_data.pa);

/* Формируем данные перед отправкой симулятору кэша */

memcpy(&profiler_data.name, ptr_data->name,

strlen(ptr_data->name));

profiler_data.operation = ptr_data->operation;

profiler_data.pa = output_data.pa;

profiler_data.va = output_data.va;

profiler_data.pid = ptr_data->pid;

if (conn_object->ipc_write(&profiler_data) < 0) {

std::cerr << "failed sending data" << std::endl;

exit(-1);

}

if (ptr_data->va == 0)

break;

}

pthread_mutex_lock(&__mutex);

termination_condition = 1;

pthread_mutex_unlock(&__mutex);

pthread_cond_signal(&__condition);

}

void mem_ref(const char *name, unsigned long va, int operation)

{

struct processing_data data;

memset(&data, 0, sizeof(struct processing_data));

if (initialization == 0) {

queue = new threadsafe_queue<struct processing_data>;

std::thread thread(Handler<struct processing_data>);

initialization = 1;

thread.detach();

}

memcpy(&data.name, name, strlen(name));

data.operation = (operation == 1) ? OPER_WRITE: OPER_READ;

data.pid = getpid();

data.va = va;

queue->push(data);

/* Дождаться завершения потока */

if (va == 0) {

std::cout << "waiting the end a thread" << std::endl;

pthread_mutex_lock(&__mutex);

while (termination_condition != 1)

pthread_cond_wait(&__condition, &__mutex);

pthread_mutex_unlock(&__mutex);

}

}

Размещено на Allbest.ru