http://muratbuffalo.blogspot.com/2023/04/the-end-of-myth-distributed.html Skip to main content Search This Blog [ ] [Search] Metadata On distributed systems broadly defined and other curiosities. The opinions on this site are my own. The end of a myth: Distributed transactions can scale * Get link * Facebook * Twitter * Pinterest * Email * Other Apps - April 10, 2023 This paper appeared in VLDB'17. The paper presents NAM-DB, a scalable distributed database system that uses RDMA (mostly 1-way RDMA) and a novel timestamp oracle to support snapshot isolation (SI) transactions. NAM stands for network-attached-memory architecture, which leverages RDMA to enable compute nodes talk directly to a pool of memory nodes. [AVvXsEj4KpyQzrQ-bP9ndycEXXi-kyPCvYmmUd7n4z76] Remote direct memory access (RDMA) allows bypassing the CPU when transferring data from one machine to another. This helps relieve a major factor in scalability of distributed transactions: the CPU overhead of the TCP/IP stack. With so many messages to process, CPU may spend most of the time serializing/deserializing network messages, leaving little room for the actual work. We had seen this phenomena first hand when we were researching the performance bottlenecks of Paxos protocols. This paper reminds me of the "Is Scalable OLTP in the Cloud a Solved Problem? (CIDR 2023)" which we reviewed recently. The two papers share one author. I think this earlier NAM paper is more mature in terms of implementation details and evaluation. In contrast the CIDR'23 paper concentrated on explaining the benefits of using multi-writer shared memory architecture and reads more like a position paper, with some discussion of the ScaleStore idea. Both papers use RDMA as a key enabler for scalable distributed transactions, but they differ in how they leverage it. NAM-DB uses RDMA mainly for one-sided operations to access remote memory servers, while shared-writer with coherent cache uses RDMA mainly for two-sided operations to implement cache coherence protocols. Both papers aim to support snapshot isolation (SI) transactions, but they differ in how they handle concurrency control and version management: NAM-DB uses a compare-and-swap operation to lock and install new versions of records on memory servers, while shared-writer with coherent cache uses a cache invalidation protocol to ensure consistency of cached records on compute servers. System design RDMA-enabled networks transforms the system to a hybrid shared-memory and message-passing architecture: it is neither a distributed shared-memory system (as several address spaces exist and there is no cache-coherence protocol), nor is it a pure message-passing system since memory of a remote machine can be directly accessed via RDMA reads and writes. [AVvXsEjdkJbWdxRj2pX2krQRTJqIDKmcIXDqTl6u62qM] Figure 1 shows the network-attached-memory (NAM) architecture. Note the two distinct type of servers: compute servers and memory servers. The NAM architecture logically decouples compute and storage nodes and uses RDMA for communication between all nodes. Memory servers provide a shared distributed memory pool that holds all the data, which can be accessed via RDMA from compute servers that execute transactions. Memory servers hold all data of a database system such as tables, indexes as well as all other state for transaction execution (e.g., logs and metadata). The main task of compute servers is to execute transactions over the data items stored in the memory servers. This is the first design principle in NAM: Separation of Compute and Memory. The data is assumed to be randomly distributed to memory nodes in the shared memory pool. Any memory node is equi-distant to any compute node, and a compute node needs to reach multiple memory nodes for transaction execution. All transactions are by default distributed transactions. Where the data is located is not a primary point of optimization in the NAM architecture. This is the second design principle in NAM: Data Location Independence. I like this data location independence idea for its built-in tolerance to workload skew. Locality is only a tuning parameter for optional/additional optimization, and may be achieved by running a specific compute server and memory server on the same physical machine. Snapshot Isolation (SI) transaction execution [AVvXsEiwrSFmZIVFYd-CF1MLdrl82rlbIzdeXnqtX2WyguxUoeZHBN5D] [AVvXsEitt1_cLHM5654aKGalNrb_V0NlhLOdcNrWiI2H] This part below is mostly verbatim from the paper. The CAS steps and RDMA use are the interesting bits in this protocol. For executing a transaction, the compute server first fetches the read-timestamp rts using an RDMA read (step 1 in Figure 2, line 3 in Listing 1). The rts defines a valid snapshot for the transaction. Then, the compute server executes the transaction, which means that the required records are read remotely from the memory servers using RDMA read operations (e.g., the record with ckey = 3 in the example) and updates are applied locally to these records; i.e., the transaction builds its read- and write-set (step 2 in Figure 2, line 5 in Listing 1). Once the transaction has built its read- and write-set, the compute server starts the commit phase. For committing, a compute server fetches a unique commit timestamp (cts) from the memory server (step 3 in Figure 2, line 7 in Listing 1). Then, the compute server verifies and locks all records in its write-set on the memory servers using one RDMA compare-and-swap operation (line 10-15 in Listing 1). The main idea is that each record stores a header that contains a version number and a lock bit in an 8-Byte memory region. For example, in Figure 2, (3, 0) stands for version 3 and lock-bit 0 (0 means not locked). The idea of the compare-and-swap operation is that the compute server compares the version in its read-set to the version installed on the memory-server for equality and checks that the lock-bit is set to 0. If the compare succeeds, the atomic operation swaps the lock bit to 1 (step 4 in Figure 2, line 13 in Listing 1). If compare-and-swap succeeds for all records in the write-set, the compute server installs its write-set using RDMA writes (line 19-20 in Listing 1). These RDMA writes update the entire record including updating the header, installing the new version and setting the lock-bit back to 0. For example, (6, 0) is remotely written on the header in our example (step 5 in Figure 2). If the transactions fails, the locks are simply reset again using RDMA writes (line 24-28 in Listing 1). Finally, the compute server appends the outcome of the transaction (commit or abort) as well as the commit timestamp cts to a list (ctsList) in the memory server (step 6 in Figure 2, line 32 in Listing 1). Timestamp oracle In the above protocol, a timestamp oracle is responsible for advancing the read timestamp by scanning the queue of completed transactions. It scans ctsList and tries to find the highest commit timestamp where every transaction before that timestamp is also committed. The bottleneck-free implementation of this timestamp oracle is of interest as an example of how a global synchronization/contention register can be partitioned into a one-way-RDMA amenable data structure. This is the third design principle in NAM: Partitionable Data Structures. The timestamp oracle protocol uses a timestamp vector (as in vector clocks) to store the latest commit timestamp for each compute server transaction execution thread. * When a transaction starts, the compute server reads the entire timestamp vector from a memory server and uses it as its read timestamp. * When a transaction commits, it creates a new commit timestamp by incrementing its own counter in the timestamp vector. It then verifies and locks all records in its write-set on the memory servers using RDMA compare-and-swap operations. If the verification succeeds, it installs its write-set using RDMA writes and updates its counter in the timestamp vector using RDMA write. If the verification fails, it aborts and releases the locks using RDMA writes. The timestamp oracle is responsible for advancing the read timestamp by scanning the timestamp vector and finding the highest commit timestamp that is preceded by all committed transactions. This allows other transactions to read more recent snapshots of the database. I think synchronized clocks would be a more practical solution to the timestamping problem. If you are using special hardware for RDMA, why not get hardware support (which is very feasible these days) for synchronized clocks as well. When using synchronized clocks, how would the read timestamp be advanced? Simple, by the current clock at the compute node at transaction's start, T-start. The transaction will then read anything that has committed before its transaction start, T-start, as a consistent snapshot. Implementation The table and index structures are designed to be partitionable and scalable. The table structure uses a hash-based partitioning scheme that maps each record to a fixed-size slot in a memory region. The index structure uses a B+tree that is partitioned by key ranges and stored in separate memory regions. Both structures use RDMA operations to access and update data on memory servers. A database catalog is implemented such that transactions can find the storage location of tables and indexes. The catalog data is hash-partitioned and stored in memory servers. All accesses from compute servers are implemented using two-sided RDMA operations since query compilation does not result in a high load on memory servers when compared to the actual transaction execution. Since the catalog does not change too often, the catalog data is cached by compute servers. [AVvXsEjVEH53afULBBcyir8ZNUPPSnS7LEdgy3egUUBT] The multi-versioning scheme allows compute servers to access and update different versions of records using RDMA operations. The scheme stores the most recent version of each record in a dedicated memory region, and moves older versions to an old-version buffer and an overflow region. It also uses a header section for each record that contains metadata such as version information and lock bits. In order to tolerate memory server failures, each transaction execution thread of a compute server writes a private log journal to memory servers using RDMA writes. The log entries for all transaction statements are written to the database log before installing the write-set on the memory servers. Once a memory server fails, NAM-DB halts the complete system and recover all memory servers to a consistent state from the last persisted checkpoint. The recovery procedure is executed by one dedicated compute server that replays the merged log for all memory servers. This global stall is problematic for production, but NAM is a research proof-of-concept system. Maybe using redundancy/quorums would be a way to solve this, but then that could be introducing challenges for consistency. Not straightforward. Evaluation The experimental setup involves using a cluster of 56 machines connected by an InfiniBand FDR network, and running the TPC-C benchmark with different configurations and workloads. [AVvXsEg08uiKvIN2j7iFwqraPuPZNdpw100P5yQM5B3S] [AVvXsEhJOppe2NLARE_Ue6U86WFUgdDzvBJNZCBAnYa2zQBaW-e4YY92] The scalability results show that NAM-DB achieves linear scale-out for throughput while keeping a low latency, and outperforms other RDMA-based systems such as FaRM. For the standard configuration of TPC-C benchmark, they show that NAM-DB scales linearly to over 3.6 million new-order transactions per second on 56 machines, and 6.5 million new-order transactions with locality optimizations, which is 2 million more transactions per second than what FARM achieves on 90 machines. Discussion 1. Why is SI important for NAM? Why would this not work for serializable isolation? I am not complaining about the use of SI, it makes sense and it is the overwhelming use in real world. I think this question is interesting to get insights about distributed transactions design and implementation. 2. It beats me why we are not leveraging on new hardware more for distributed systems and databases. I don't think the issue is the cost of the hardware. Is RDMA mature (robust/reliable) enough to use in distributed transactions? What are the handicaps? What are the reasons for slow uptake on this? databases distributed transactions RDMA * Get link * Facebook * Twitter * Pinterest * Email * Other Apps Comments Post a Comment Popular posts from this blog Strict-serializability, but at what cost, for what purpose? - August 03, 2022 Image Strict-serializability guarantees that transactions appear to occur in an order consistent with the "real-time" ordering of those transactions: If transaction T1 commits before transaction T2 is invoked, then the commit timestamp of T1 precedes the commit timestamp of T2. This is, in fact, the real-time constraint from linearizability , but applied across transactions not just per-key. A strict-serializability system satisfies both serializability (transactions appear to occur as if they are executed one at a time in isolation) and linearizability per key (after all single-key reads /writes are transactions over one item). Below figure is from https:/ /jepsen.io/consistency . However, this is a one-way implication, the other direction does not hold. You can satisfy both serializability per transactions and linearizability per key, but fail to satisfy strict-serializability. (Below I give an example accompanied with a TLA+ specification to check it.) This is because, in strict- Read more >> Foundational distributed systems papers - February 27, 2021 I talked about the importance of reading foundational papers last week. To followup, here is my compilation of foundational papers in the distributed systems area. (I focused on the core distributed systems area, and did not cover networking, security, distributed ledgers, verification work etc. I even left out distributed transactions, I hope to cover them at a later date.) I classified the papers by subject, and listed them in chronological order. I also listed expository papers and blog posts at the end of each section. Time and State in Distributed Systems Time, Clocks, and the Ordering of Events in a Distributed System. Leslie Lamport, Commn. of the ACM, 1978. Distributed Snapshots: Determining Global States of a Distributed System. K. Mani Chandy Leslie Lamport, ACM Transactions on Computer Systems, 1985. Virtual Time and Global States of Distributed Systems. Mattern, F. 1988. Practical uses of synchronized clocks in distributed systems. B. Liskov, 1991. Expository papers Read more >> Learning about distributed systems: where to start? - June 10, 2020 This is definitely not a "learn distributed systems in 21 days" post. I recommend a principled, from the foundations-up, studying of distributed systems, which will take a good three months in the first pass, and many more months to build competence after that. If you are practical and coding oriented you may not like my advice much. You may object saying, "Shouldn't I learn distributed systems with coding and hands on? Why can I not get started by deploying a Hadoop cluster, or studying the Raft code." I think that is the wrong way to go about learning distributed systems, because seeing similar code and programming language constructs will make you think this is familiar territory, and will give you a false sense of security. But, nothing can be further from the truth. Distributed systems need radically different software than centralized systems do. --A. Tannenbaum This quotation is literally the first sentence in my distributed systems syllabus. Inst Read more >> Speedy Transactions in Multicore In-Memory Databases - February 24, 2023 Image This paper is from SOSP'13. The Polyjuice paper, which we studied last week, built on the Silo codebase and commit protocol, which led me to read the Silo paper. Silo is a single-node multi-core in-memory database. It avoids all centralized contention points, including centralized transaction ID (TID) assignment. Silo's key contribution is a commit protocol based on optimistic concurrency control (OCC) that provides serializability while avoiding all shared-memory writes for records that were only read. Logging and recovery is provided by linking periodically-updated epochs with the commit protocol. Silo achieves almost 700,000 transactions per second on a standard TPC-C workload mix on a 32-core machine, as well as near-linear scalability. That is pretty impressive over 2013 hardware. The Silo paper got 445 citations in 10 years. That is also impressive. So let's dive in. Architecture Silo's organization is typical of databases. Tables are implemented as collections Read more >> The Seattle Report on Database Research (2022) - August 08, 2022 Every 5 years, researchers from academia and industry gather to write a state-of-the-union (SOTU) report on database research. This one was released recently . It is a very readable report, and my summary consists of important paragraphs clipped from the report. Emphasis mine in bolded sentences. I use square brackets for when I paraphrase a long text with a more direct statement. TL;DR: The SOTU is strong (the relational database market alone has revenue upwards of $50B) and growing stronger thanks to boom in cloud computing and machine learning (ML). For the next 5 years, research should scale up to address emerging challenges in data science, data governance, cloud services, and database engines. What has changed in the last 5 years Over the last decade, our research community pioneered the use of columnar storage, which is used in all commercial data analytic platforms. Database systems offered as cloud services have witnessed explosive growth. Hybrid transactional/analytical pro Read more >> Anna: A Key-Value Store For Any Scale - April 29, 2022 Image This paper (ICDE'18) introduces Anna, a CALM / CRDT implementation of a distributed key-value system both at the data structure level as well as system architecture and transaction protocol levels. Anna is a partitioned, multi-mastered key-value system that achieves high performance and elasticity via wait-free execution and coordination-free consistency. Anna employs coordination-free actors that perform state update via merge of lattice-based composite data structures. I love the strongly opinionated introduction of this paper. This is what papers should be about: opinionated, challenging conventions, making bets, and doing hypothesis testing in the small. Conventional wisdom says that software designed for one scale point needs to be rewritten when scaling up by 10x. Anna sets out to disprove this by showing how a key-value storage (KVS) system can be architected to scale across many orders of magnitude. (Spoiler Anna can give you only upto causal consistency, but cannot pro Read more >> Warp: Lightweight Multi-Key Transactions for Key-Value Stores - May 04, 2022 Image This paper introduces a simple yet powerful idea to provide efficient multi-key transactions with ACID semantics on top of a sharded NoSQL data store. The Warp protocol prevents serializability cycles forming between concurrent transactions by forcing them to serialize via a chain communication pattern rather than using a parallel 2PC fan-out/ fan-in communication. This avoids hotspots associated with fan-out/ fan-in communication and prevents wasted parallel work from contacting multiple other servers when traversing them in serial would surface an invalidation/abortion early on in the serialization. I love the elegance of this idea. As far as I can see, this paper did not get published in any conference. The authors published a followup paper to this in NSDI 16, called "The Design and Implementation of the Warp Transactional Filesystem." But that paper does not talk about the internals of Warp protocol like this archive report, rather talks about the Warp Transactional File Read more >> Checking statistical properties of protocols using TLA+ - October 03, 2022 Image In my previous post, I mentioned how excited I was about Jack Vanlightly and Markus Kuppe's talk at TLA+ Conference. They showed a new mode (called "generate") added to the TLA+ checker to enable obtaining statistical properties from the models. Here is how this works in a nutshell. Your TLA+ model/protocol produces a tree of runs, and, in the generate mode the TLA+ checker chooses these runs in a uniform manner. You also specify some state constraints, and when these are satisfied the TLA+ checker executes/evaluates your state constraint formula: you add the statistics logging as the side effect of this formula, and record information about the current run to a CSV file. This way, you can collect statistical hyper properties (abort rates, completion rounds, etc.) about the runs, in addition to checking it for correctness invariants. Later you can use Excel or Rscript to visualize the statistics collected and analyze the model behavior. Checking statistics is very usef Read more >> Amazon Aurora: Design Considerations + On Avoiding Distributed Consensus for I/Os, Commits, and Membership Changes - March 20, 2022 Image Amazon Aurora is a high-throughput cloud-native relational database. I will summarize its design as covered by the Sigmod 17 and Sigmod 18 papers from the Aurora team. Aurora uses MySQL or PostgreSQL for the database instance at top, and decouples the storage to a multi-tenant scale-out storage service. In this decoupled architecture, each database instance acts as a SQL endpoint and supports query processing, access methods, transactions, locking, buffer caching, and undo management. Some database functions, including redo logging, materialization of data blocks, garbage collection, and backup/ restore, are offloaded to the storage nodes. A big innovation in Aurora is to do the replication among the storage nodes by pushing the redo log; this reduces networking traffic and enables fault-tolerant storage that heals without database involvement. In contrast to CockroachDB and FoundationDB , Aurora manages not to use consensus at all. It uses a primary secondary failover at the comp Read more >> Powered by Blogger Theme images by Michael Elkan Murat Demirbas My photo Murat I am a principal applied scientist at AWS. On leave as a computer science and engineering professor at SUNY Buffalo. I work on distributed systems, distributed consensus, and cloud computing. You can follow me on Mastodon or Twitter. Visit profile Pageviews Recent Posts * April2 * March3 * February5 * January3 * December4 * November4 * October1 * September4 * August5 * July6 * June3 * May3 * April4 * March3 * February3 * January3 * December5 * November3 * October6 * September1 * August4 * July2 * June12 * May1 * April1 * March4 * February4 * January4 * December3 * November7 * October4 * September1 * August3 * July6 * June11 * May9 * April8 * March8 * February7 * January9 * December10 * November14 * October6 * September13 * July3 * June3 * May4 * April6 * March2 * February1 * January3 * December4 * November7 * October2 * September2 * August8 * July2 * June4 * May9 * April6 * March9 * February5 * January13 * December15 * November15 * October5 * September8 * August10 * July3 * June3 * May3 * April4 * February4 * January7 * December7 * November9 * October3 * September1 * July4 * June5 * May1 * April4 * March2 * February2 * January4 * December3 * November2 * October3 * September2 * August3 * June1 * May1 * April6 * March6 * February4 * January3 * November4 * October4 * September6 * August2 * July2 * June3 * March3 * February4 * January1 * December1 * November2 * August2 * July4 * June2 * May5 * April8 * January1 * December1 * November7 * October1 * September2 * August1 * May2 * March1 * February1 * January2 * December3 * September5 * June1 * May5 * April5 * March5 * February9 * January5 * December6 * November9 * October9 * September7 * August1 Show more Show less Topics AI1 analytics1 atomic storage1 auditability5 automated reasoning5 aws 1 Azure10 benchmarks3 bestof6 big-data23 Blockchain39 book-review50 chaos2 cloud computing7 consistency24 Cosmos DB10 CosmosDB11 data warehouse1 databases35 dataflow8 disaggregation1 distributed consensus42 distributed transactions23 distSQL5 facebook15 failures16 fault-tolerance38 formal methods11 graph-processing1 htap1 humans10 indexing3 isolation levels1 links2 mad-questions42 main-memory1 misc103 mlbegin7 mldl26 mobile2 my advice16 my-paper10 newsql2 NoSQL1 OLAP1 paper-review148 paxos47 presenting4 programming5 RDMA1 reading-group23 reconfiguration1 research-advice47 research-question43 Rust3 scheduling3 seminar9 serializability1 serverless1 smartphones2 snapshot isolation1 sonification1 SQL3 stabilization6 statistics1 stream-processing10 teaching31 tensorflow11 time10 time synchronization2 tla41 transactions11 trip-report23 wpaxos5 writing26 Show more Show less