Software & Systems

Распределенная БД – это набор логически связанных БД, объединенных с помощью компьютерной сети [1]. Распределенная СУБД может быть определена как программная система, управляющая таким набором и предоставляющая пользователю прозрачный интерфейс. Прозрачность интерфейса означает отделение высокоуровневой семантики системы от деталей реализации таким образом, чтобы пользователь мог не обращать внимание на эти особенности. При этом, несмотря на наличие компьютерной сети, подразумевается, что БД фактически находится на нескольких узлах. В противном случае такая система не отличается от централизованной, за исключением необходимости учета факторов передачи данных по сети.

Кроме стандартных требований, предъявляемых к СУБД, в распределенной СУБД должен быть выполнен дополнительный набор требо- ваний, связанный с сетевым взаимодействием узлов, а также с эффективностью межпоточного взаимодействия, в частности, обеспечение вы- сокопроизводительного сетевого взаимодействия с клиентскими узлами, эффективное взаимодей- ствие между узлами СУБД, корректность работы с реплицированными и фрагментированными данными и др. [1]. В настоящей работе авторы акцентируют внимание на первых двух свойствах.

Можно сказать, что одна из важнейших задач распределенной СУБД – обеспечение высокоэффективной обработки потоков запросов, исходящих от нескольких клиентов. Важность сетевого и многопоточного аспектов распределенной системы остается острой и до сего момента вследствие того, что характеристики сети продолжают оставаться одним из важнейших факторов для систем подобного рода [2].

В данной статье рассматриваются некоторые аспекты реализации распределенной СУБД на примере исполнителя запросов, разработанного авторами. Этот исполнитель принимал участие в соревновании распределенных СУБД «Distributed query engine programming contest» [3], проводимом на конференции ACM SIGMOD в 2010 г., где занял третье место. Данное соревнование ежегодно проводится среди студентов и аспирантов, целью его является создание полнофункциональной распределенной реляционной СУБД. В рамках соревнования участникам предлагается разработать один из компонентов, который войдет в нее.

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

Постановка задачи

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

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

1) все процессы выполняются в рамках одной операционной системы, при этом взаимодействие по IP осуществляется через интерфейс-петлю (вырожденный случай);

2) каждый процесс выполняется на выделенном узле, при этом узлы находятся в одном сегменте ethernet (основной случай).

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

Репликация (replication) – механизм раскопирования данных, позволяющий повысить производительность и отказоустойчивость системы. Владение многими копиями данных желательно скрывать от пользователя, особенно если система не ориентирована на обновления.

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

Предполагается, что отношения дизъюнктно фрагментированы с помощью техники горизонтального фрагментирования (horizontal partition­ing). Важность дизъюнктности фрагментирования заключается в том, что при такой постановке задачи отпадает необходимость отслеживать дубликаты, происходящие из различных секций. Это, в свою очередь, является важным аргументом при выборе архитектуры исполнителя запросов. Каждая секция может находиться более чем на одном узле, то есть быть реплицированной.

Авторы полагают, что перед началом работы данные уже присутствуют на жестких дисках рабочих узлов, а мастер-узел обладает полной схемой.

Рассматриваемые типы запросов. Разработанная система ориентирована на класс запросов, называемый SPJ Conjunctive Queries [4]. Этот класс состоит из запросов типа SELECT-FROM-WHERE, где отсутствуют подзапросы и агреги- рование. При этом предикаты в блоке WHERE объединены с помощью операции конъюнкции. Каждый предикат может быть либо условием на равенство атрибута какому-либо значению, либо фильтрацией атрибута по условию типа больше. Однако в данной работе рассматриваются только запросы выборки по значению. Данное ограничение является удобным при проведении тестов и позволяет наиболее точно выделить описываемую проблему.

Несмотря на простоту определения, обработка такого класса запросов очень важна и актуальна вследствие широкой распространенности в коммерческих системах. Исследованием, проведенным Amazon.com, установлено, что около 65 % всех запросов системы этой компании составляя- ют обращения к данным только по первичному ключу, второе место занимают простые мультиатрибутные запросы, задействующие несколько таблиц [5].

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

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

Индекс. В рассматриваемой системе использовалась высокопроизводительная библиотека для работы с индексом, разработанным победителем соревнования «SIGMOD programming contest main memory transactional index» в 2009 г. Данный индекс является локальным для каждого узла, хранящего отношение или его секцию. Предполагается, что индекс полностью помещается в оперативную память узла.

Предложенная архитектура

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

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

В нашем случае все узлы располагаются максимально близко друг к другу и соединены весьма надежными каналами связи: ethernet и интерфейс-петля. Однако, если два узла пытаются передать данные на третий, используя максимальную пропускную способность канала, такая ситуация приводит к перегрузке (congestion) и потере пакетов. Значит, выбранные протоколы вышестоящих уровней должны обеспечивать управление перегрузкой (congestion control). Это можно реализовать на уровне приложения, используя ненадежный транспортный протокол, либо предоставить данную задачу стеку протоколов, уже реализованному в операционной системе, то есть использовать протокол TCP.

Приняв во внимание технические трудности, авторы решили использовать протокол TCP.

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

-    обеспечение одновременного доступа к сети для нескольких (~50) рабочих нитей;

-    обеспечение справедливого доступа, то есть никакая рабочая нить не должна ждать доступа к сетевым ресурсам неопределенно долгое время;

-    исключение ситуации, при которой нить пытается писать в сокет, буфер которого переполнен;

-    получение ответа от рабочих узлов.

Для решения этих задач была введена дополнительная нить-мультиплексор, выполняющая следующие функции:

1)   маршрутизация и упорядочение запросов от рабочих нитей к конкретному TCP-сокету, тем самым рабочие нити лишаются исключительного доступа к коммуникационным сокетам;

2)   слежение за состоянием сокета, то есть в сокет не могут быть записаны новые данные, если системный буфер переполнен;

3)   принятие данных из сети (ответа от рабочих узлов) и передача их нужной нити.

Реализация нити-мульти­плексора основана на функции poll(), которая принимает на вход набор дескрипторов ввода/вывода вместе с ожидаемыми состояниями (есть возможность прочесть или записать данные), ожидает момента, когда хотя бы один дескриптор окажется в нужном состоянии, и завершается, возвращая набор дескрипторов, находящихся в нужном состоянии.

В данном случае имеются следующие наборы дескрипторов с состояниями:

-    дескрипторы сетевых соединений с рабочими узлами с ожидаемой возможностью чтения; входят все TCP-сокеты;

-    дескрипторы сетевых соединений с рабочими узлами с ожидаемой возможностью записи; входят все TCP-сокеты, в которые есть потребность отправить данные;

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

Упрощенное поведение нити-мультиплексора при наступлении соответствующих событий показано на рисунке 1.

Аспекты межпотокового взаимодействия. Межпроцессное и межпотоковое взаимодействия играют также важную роль в производительности системы. Авторы столкнулись с классической задачей писателя–читателя. В качестве простейшего решения этой задачи на ранних этапах разработки были использованы UNIX-сокеты как простейшее в реализации решение, вписываемое в архитектуру системы c мультиплексором.

Рассмотрим подробнее возможные методы межпоточного взаимодействия.

·     UNIX-сокеты достаточно медленны, каждая операция записи/чтения переключает процесс в режим ядра, и при этом происходит копирование данных из пространства пользователя в пространство ядра (и обратно). С другой стороны, сокеты очень легко вписываются в предложенную архитектуру.

·     Анонимные каналы (FIFO) также достаточно медленны и подвержены тем же недостаткам, что и сокеты; их, как и сокеты, можно легко применять в использованной архитектуре.

·     Очереди в пространстве пользователя. Для разделения доступа между нитями должны использоваться дополнительные механизмы: мьютексы, спин-блокировки или неблокируемые очереди [6, 7]. Однако в используемой архитектуре этот способ непосредственно неприменим.

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

На основании изложенного можно получить схему, изображенную на рисунке 2.

Эксперименты

В процессе выполнения данной работы была проведена серия экспериментов. В качестве аппаратной платформы использовался кластер рабочих станций, состоящий из 8 узлов следующей конфигурации: ЦПУ – Intel Core 2 Duo 2.00 GHz; сеть – 1 Гбит/с; ОЗУ – 512 Мбайт; ОС – GNU/Linux 2.6.33.

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

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

1.   Данные представлены одним отношением, целиком находящимся на мастер-узле. Серия запросов образована выборкой по первичному ключу с фиксированным значением для этого ключа: SELECT a._id, a.a FROM a AS a WHERE a._id=1. Это позволяет вычитывать данные из дискового кэша, снижая таким образом нагрузку на диск. Результаты показаны на рисунке 3а.

2.   Данные представлены одним отношением, целиком находящимся на мастер-узле. Серия запросов образована выборкой по первичному ключу с последовательно увеличивающимся значением для этого ключа: SELECT a._id, a.a FROM a AS a WHERE a._id=n, где n – порядковый номер запроса в серии (рис. 3б).

3.   Данные представлены пятью отношениями, секции которых распределены по всем узлам кластера. Серия запросов образована выборкой по первичному ключу со случайно подобранным значением для этого ключа: SELECT a._id, a.a FROM a AS a WHERE a._id= k (n), где n – номер запроса в серии, k равномерно распределена на отрезке значений для первичного ключа (рис. 3в).

На основе полученных результатов (рис. 3) можно сделать следующий вывод: неблокируемые очереди даже вкупе с механизмом eventfd, который использует переключение в контекст ядра, демонстрируют прирост в производительности на простых тестах в 5–10 % с учетом погрешностей.

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

Литература

1.   Tamer M. Özsu and Patrick Valduriez. Principles of Distributed Database Systems (2nd Ed.). Prentice-Hall, Inc., Upper Saddle River, NJ, USA. 1999.

2.   Kossmann Donald. The state of the art in distributed query processing. ACM Comput. Surv. 32, 4 (December 2000), URL: http://doi.acm.org/10.1145/371578.371598 (дата обращения: 13.01.2011).

3.   Genzmer C. [et al.]. The SIGMOD 2010 Programming Contest: A Distributed Query Engine. SIGMOD Record, 39(2), pp. 61–64, June 2010.

4.   Amol Deshpande, Zachary Ives, and Vijayshankar Raman. Adaptive query processing. Found. Trends databases 1, 1 (January 2007), 1-140. DOI=10.1561/1900000001 URL: http://dx.doi.org/ 10.1561/1900000001 (дата обращения: 13.01.2011).

5.   Vogels Werner. Data access patterns in the Amazon.com technology platform. In Proceedings of the 33rd international conference on Very large data bases (VLDB '07). VLDB Endow- ment 1-1.

6.   Michael M.M., Scott M.L. Simple, fast, and practical non-blocking and blocking concurrent queue algorithms. In: Proceedings of the 15th Annual ACM Symposium on Principles of Distributed Computing (PODC ’96), ACM, NY, 1996, pp. 267–131.

7.   Ladan-Mozes E., Shavit N. An Optimistic Approach to Lock-Free FIFO Queues. DISC 2004, LNCS 3274, Springer-Verlag Berlin Heidelberg. 2004, pp. 117–131.