Snapshot Isolation (SI) level is extensively used in commercial database systems. We developed a simple SI implementation protocol for distributed DBMS and implemented it in the Apache HBase. The work presents the performance evaluation of the protocol. We have measured the performance of a single-node system and modeled the performance of a distributed HBase cluster.
In the last ten years the special class of data-intensive web-application has emerged. Youtube, Google Maps, social networks, etc. represent the applications of the class. This applications can be characterized by the following features: they work with large volumes of data and typically do not need ACID transactions.
The choice of the concurrency control to employ in the application is of great importance, since it affects the performance of the application as well as the complexity and cost of development. For example, choice of eventual consistency (EC) as an isolation level may result in increased complexity of programming model. Thus, application designer has to carefully look for the most suitable compromise between level of guarantees he gets and performance of isolation level protocols.
The Snapshot Isolation (SI) isolation level was
introduced in [
The transaction executed under SI reads data from a snapshot of the data as of the time the transaction started. Two transactions are called competitive if they have overlapping execution intervals and they have written the same data item. When the transaction is ready to commit, it is assigned the commit timestamp and in case if there is no competing transactions in the system it commits. This feature is called “First-Commiter-Wins”.
We developed SI protocol for distributed DBMS. We
have presented it in [
The remainder of this paper is organized as follows. In Section 2 we discuss literature works on the DBMS performance evaluation with concurrency control performance comparison being in focus. Section 3 contains the description of the protocol and its implementation. Performance model is provided in Section 4. Section 5 presents results of conducted simulation experiments. Section 6 contains discussion of protocol implementation performance along with model validation issues. In section 7 the results are summed up. 2
A lot of concurrency control algorithms was introduced in the last 30 years. Considerable research was devoted to evaluating the performance of concurrency control algorithm in the 80s and the earlier 90s.
In [
Works [
The work [
Recently there was a number of studies related to SI
implementation in distributed systems [
The article [
In the paper [
In this work we present performance models for our protocol and HBase and the comparison of simulation results. To the best of our knowledge the performance model of Key Value store like HBase has not been presented yet. 3 3.1
Let us suppose that current transaction state, its start timestamp and end timestamp (if a transaction has been already committed or aborted) are available at the execution time.
Read operation execution is straightforward. To read a data object x the transaction t performs a selection of all versions of x, committed before start time of t. Then it traverses the obtained list of versions to find the latest committed version. If the list contains version written by t, this version is read instead of the latest committed.
Before the value of x can be updated the transaction has to ensure, that no competitive transaction had already written x. If x was updated by transaction, which was committed after t had started or by active transaction, than t aborts.
We have formally proved the correctness of the
protocol in [
We have implemented our protocol in HBase. HBase is
an open-source distributed data storage written in Java. It
was made after Google Bigtable. In [
Authors has called Bigtable “a sparse distributed persistent multidimensional sorted map”. Bigtable API is very simple. HBase API is also restricted to get, put, delete and scan operations. The data processing necessary for a query can not be pushed as close as possible to data, as it is usually done in distributed RDBMS. If a client wants to update data according to a special rule, it requests the data, updates it locally and puts the updated data to the storage. That is, complex query execution is decomposed into two or more phases. As it was said before, initially only eventual consistency isolation guarantees were provided by HBase. HBase/Bigtable users who needed transactional execution, were suggested to implement it themselves using data timestamps.
HBase cluster consists of servers of two types: master nodes and region nodes. Every region server manages a number of regions. HBase uses horizontal partitioning, each region contains data for an interval of keys.
HBase is column oriented storage, that is, data is stored on a per column basis in main memory and in the file system. Every column store has its own set of files and a per file buffer pool employing the classical LRU discipline to manage pages. HBase use immutable files to store data. When the total space occupied by updates kept in main memory exceeds specified threshold a new file is written to the file system. This is done in a special thread. HBase uses write ahead log (WAL), that is, every update is written to log before it gets to main memory store. 3.3
The “First-Commiter-Wins” feature has an optimistic nature, thus SI definition naturally accepts the optimistic protocol. Our approach is not optimistic, it should be rather called “First-Updater-Wins”.
Although the algorithm is nonlocking, it is more close
to restart oriented locking methods such as
immediaterestart. Immediate-restart algorithm description, its
performance modelling and comparison with blocking and
optimistic methods can be found in [
Before a data object is written the version check is performed. All data object versions made after the transaction had started must be selected and traversed. It may take a considerable amount of time, because some of committed versions of data object may have been already written to the file system. In this case a number of page reads is to be performed.
Some tricks could be used to avoid unnecessary disk accesses or at least reduce their probability. The main memory store could be assigned a timestamp each time its contents are flushed to a new file. The version check performed in main memory is sufficient to ensure “FirstWriter-Wins” feature if the transaction started after the main memory store had been written to disk last time. The second trick which may be used is to write only the oldest versions to file, keeping new ones in memory. Given average transaction execution time tavg, we can keep versions with timestamps greater than now() tavg.
To ensure transactional execution of queries in distributed data storage we used the distributed commit protocol. We have chosen a protocol similar to Two Phase Commit (2PC) for its implementation simplicity. The classical 2PC successfully commit scenario consists of the following steps. The transaction initiator waits for READY message from all the participated servers. When all the participants voted to commit, it sends them COMMIT message and waits for ACK messages. After all ACK messages were received, initiator finishes the transaction.
Since HBase uses WAL, when subtransaction sends READY message to the transaction home node (client machine), it has no additional work to do before it can commit. It can not be aborted by other local subtransaction. When the transaction initiator receives READY message from all the execution region servers, it just sends them COMMIT message and commits the transaction locally. It does not wait for ACK messages from executor servers as in the classical 2PC.
Let us summarize the overheads introduced by the protocol.
Employing distributed commit protocol leads to additional communication overhead.
Version check may lead to additional (local) disk accesses and may consume CPU.
Certain resources are wasted due to additional aborts by protocol reasons. 4
The simulation model was written in Java using Simkit
simulation package [
Our model is rather hardware resource bounded than data contention bounded. We believe that HBase was not intended to be used for the workloads with high data contention. Our model was not aided to solve some fundamental questions, it concentrates on answering the engineering questions only. We have modeled the particular distributed data storage performance, so the model is much less universal than models employed in mentioned related works. However, the model is rather simple. It abstracts from some details. For example, the primary copy replication used in HBase, may affect the performance, but it is not reflected in the model.
The model will be described in terms of queries, we will use the term transaction when we need to outline the differences. 4.1
We use a close system model. There is always the same number of queries in the system. After one query had completed its execution, the new one immediately replaces it. Since the concurrency control performance hardly depends on the active query count in the system, we decided to keep these number constant.
The cluster is modelled as a queueing network, each node is itself modeled as a queueing network.
We use quite common assumption called
“homogeneity assumption” in [
All database sites have the same structure and the same service capacity.
The amounts of data allocated to each unit are equal.
cores per PE processor capacity (MIPS) BOE and EOE steps per object reference for message for IO operation count per PE access time mean (ms) page transmission time (ms) throughput (Mbit/sec) min packet size page size (Kb) count of pages per PE count of pages in buffer pool
Data accesses are spread among nodes in uniform manner.
We do not use a sophisticated model of a communication network as well, because we believe that network should not be a bottleneck. In our model there is a virtual connection between every query initiator (client) and executor node with 100Mbit/sec throughput.
Table 1 contains the system settings used for experiments. 4.2
The workload model is homogeneous, i.e. it consists of queries of one type. As was mentioned above, the complex query is executed in a number of steps. We evaluate the performance for the query which has two execution steps.
1. Several regions are scanned. After the scan results are send to the client (query initiator machine). 2. Some of the objects obtained on the first step are updated. Updates take place in the same regions, and thus are performed by the same region servers.
The number of nodes participating in query execution is distributed according to Poisson distribution with mean M . The number of data objects query reads per region server is exponentially distributed with the mean Nr, writes count is distributed in a similar way with the mean Nw. Reads and writes are distributed among the nodes uniformly.
The 99%-1% data access pattern is used. 99% of transactions references 1% of data objects. In our model the data access pattern influences both the buffer pool hit rate and the data contention.
For the model simplicity we assumed that the data object version has a page size. It is reasonable assumption, because HBase was not designated to work with objects of small size. Thus, the granule for CC algorithm is represented by a page in our model.
Workload parameters can be found in the Table 2. 4.3
Each query has a home node. The home node is the client machine it has arrived to. After a query has arrived, it is split to a number of subqueries, each of which is local to one of execution nodes. Each subquery is modeled as a list of data references (pages) to be processed. For each subquery the message is sent to the execution site to start up the execution. Each query joins the Net queue and consumes CPU time to send a message to execution sites.
When the subquery start message arrives to the execution site, the subquery is created and it joins the Net queue on the execution site and consumes CPU time needed to read a message. Than it consumes CPU to begin execution.
The subquery execution consists of Begin-OfExecution (BOE) step, a number of data reference processing iterations and End-Of-Execution (EOE) step.
For each data object reference it goes to the buffer pool to get the data. If the required page is not already in memory, subquery consumes processor time to start IO and joins the disk queue. Than the subquery goes to CPU queue and consumes time to process data reference (to find appropriate version). In case there are some more references the subquery repeats a similar data reference processing iterations.
After the subquery has successfully finished its execution it sends requested data (if present) and the READY message to the home node. After that, the subquery is finished, so it goes to CPU queue and consumes time for EOE step.
The client receives the data from all the participated servers, process it and sends a message to each participant to perform updates of the second execution stage.
A write subquery is executed in the similar way with a read subquery. Its behaviour differs in data processing iteration. The write subquery firstly consumes CPU for reference processing and after that writes the log page to the file system and enqueues the page write (it does not wait until the page will be written). 4.4
Read subtransactions do not differ from read subqueries in their execution model. Although they may consume more CPU time for a data reference processing to find appropriate version, the performed steps are same.
Write subtransaction performs version check before each write. If the version check failed, the transaction is to be aborted, and subtransaction goes to the Net node to send an ABORT message. The probability of version check success will be discussed later.
The probability that versions needed for version check were not flushed to a file since the transaction had started is quite high for not write intensive workload. Therefore, the version check step is modeled as processing of Nvc pages, each of which is already in buffer pool, so disk IO is not needed.
After the write subtransaction has finished execution it sends READY to the home node and waits for COMMIT or ABORT message in reply. The transaction initiator waits until all READY messages (or one ABORT message) will be received. Than it sends the COMMIT (ABORT) message to all executors and the transaction is finished. After the subquery received the COMMIT (ABORT) message, it writes a log page and goes to the CPU queue to perform the EOE step. 4.5
Let us consider the probability of version check failure. Suppose the transaction t which was started at the moment u0 is going to update the value of data object x at the moment u.
The failure of version check may be caused by two types of transactions that had updated x recently. 1. x may been written by the transaction which was committed after t had started. Given the throughput of the server T (u) for the moment u, the number of transactions committed in the time interval [u0; u] can be approximated by Nw T (u) (u u0). 2. x may been written by the transaction, that has not already committed. Let A(u) denote the number of write subtransaction active at the server as for the moment u. The average number of writes made by active subtransactions may be approximated by Nw=2. The number of writes made by all active transaction as for a moment u may be estimated with Nw A(u)=2.
The count of updates written by transactions which execution interval overlaps with execution interval of t may be approximated by
K(u; u0) = Nw
A(u) + 2 T (u) (u
u0) 2 Let us consider the a-b data access pattern and the case when t belongs to majority of transactions. The number of updates made by transactions of major class is a K(t; u), while the number of data objects t can access is D b. The probability of observing update made by a transaction of the same class is a
K(u; u0)
D b The probability of observing updates made by transactions of the other class is (1 a)
K(u; u0) D b Thus, for a transaction of the first type the probability of version check failure can be estimated by
P1(u; u0) = b (1 a) + a
K(u; u0)
D and for a transaction of the second type by
P2(u; u0) =
K(u; u0)
D
The main goal of our performance study consists in evaluation of performance degradation caused by additional overheads introduced by the protocol. We measured the performance of the protocol in terms of the system throughput and average response time.
We evaluated the throughput degradation for a particular parameter settings as the difference between the EC and SI throughputs divided by EC throughput.
In this section we use the abbreviation EC to denote HBase version, which uses eventual consistency, and SI for HBase, which uses our protocol. For most of the experiments we present two diagrams: one depicts normal workload results, while the other reflects the situation when all the available resources are already saturated. We have payed a special attention to the saturated system performance to examine the influence of transaction aborts on the throughput.
As expected the use of distributed commit protocol does not make considerable contribution to the average response time for the long-running transactions. The disk system appears to be the bottleneck for this kind of workload. The performance of the system is substantially influenced by the buffer pool hit rate.
Fig. 7 gives response time against buffer pool hit rate for the system running 70 queries. Response time is not as sensible to buffer pool hit rate as throughput. The graph is close to linear. However, relative response time increase grows rapidly with buffer pool hit rate getting closer to 1 and reaches the maximum of 6.25% for normal workload and 3.87% for saturated workload. We have implemented the protocol in HBase. Initially we have evaluated the protocol performance using the prototype. We have measured the prototype performance for a workload, which consisted of short-running queries. Each query read and updated 3 values of double type. The prototype has shown relatively good results for such workload: the throughput was 3 times lower than that of initial HBase version.
To validate our model mentioned test set was changed. If each query writes 3 double values, size of log entries to be flushed to disk before query finishes is small. So we replaced double values with 4Kb sized byte arrays. After the changes had been made we discovered HBase throughput being ten times greater than that of the prototype.
We suppose the prototype implementation showed such a poor performance, because it copies data objects for internal purposes. For every single get and put operation at least one data object is copied. We suppose that despite the fact that main memory reads and writes are extremely fast, copying of data objects may affect the performance in case of no disk accesses.
The values of model parameters for EC were selected in such a way, that EC throughput was as close as possible to HBase throughput. The experiment conducted with the same parameters using SI model has shown the throughput 1.5 times lower than for EC (1200 vs. 1800).
To have the SI model performance close to that of the prototype, we adjusted the mean page processing time parameter to be 270000 instruction instead of default 25000. The primary goal of this experiment was to validate the abort probability model. Fig. 8 shows the prototype throughput and SI model throughput, and 9 shows abort rates obtained. The difference between the aborts rates obtained from the model and the prototype has not exceeded 3% being about 2% at average. We conclude that our model captures the system behaviour adequate.
In this work we presented SI protocol and evaluated its
performance. The advantage of the protocol under EC is
obvious: the data storage user is provided with SI
consistency guarantees. However, the protocol introduces
several additional overheads, which may affect the
performance. We have modeled the performance for the
particular kind of workload and validated the model using
the prototype we have presented in [