Distributed key-value stores are used extensively on modern platforms, particularly in cloud applications [1,8,11,17,18], but also in large-scale databases [12,20,24,36,41] and distributed file systems [15,22,39]. In these applications, there is a large corpus of data items (e.g., key-value pairs, objects, files) stored across many servers. Clients of the system send requests, which access some particular data. Each request is routed to the server that has the items, and the server is responsible for processing this request and responding to the client. If the server is busy, the request may be queued.
For such systems, an important optimization criterion is throughput -the number of requests that the system can process on average per time period. If most requests end up accessing a small number of servers, most system capacity might be idle while a few servers' queues grow unboundedly. The client may care about latency -the amount of time they wait between sending a request and receiving a reply. Since long queues impact latency, distributed storage systems implement bounded queues for servers such that a server with a full queue rejects future requests.
The goal of the system is to accept or consume as many requests as possible, or inversely to reject as few requests as possible while keeping the queue size small -this is equivalent to maximizing throughput while keeping latency small. To avoid rejecting many requests (thereby reducing the system throughput), distributed storage systems try to balance load across servers. However, they have no direct control over the load, since requests are generated by clients and the request access pattern may change over time. Therefore, if all the "hot items" -data items that are being accessed frequently -are on the same server, that server will experience a high load. Avoiding overload may require load balancing via moving these items to other servers dynamically.
Load balancing algorithms for distributed storage systems generally have the following steps: (1) initially distribute data across servers in some manner; (2) each server has a queue with bounded size (decided by the algorithm) and rejects incoming requests if its queue is full; (3) do dynamic data distribution by choosing some data to move from an "overloaded" servers to other servers. The question considered in this work is how one should do these steps. This problem has been studied empirically using various heuristics based on past system behaviors or theoretically using queuing theoretic assumptions on request arrivals [9,30].
In this paper, we consider this problem from a theoretical perspective. More formally, assume that the data is divided into 𝑛 chunks that must be housed on 𝑚 servers. At every time step, each server can process one request. We assume that the requests are generated by an adversarial process that generates 𝑚 requests per time step and maximizes the number of rejected requests. Our goal is to understand what strategies with small queues may work against various adversarial assumptions. We have the following results: Simple deterministic strategies work against random clients, but no deterministic strategy is good against an adversarial client: If the client is not adversarial and picks a random chunk to request in every time step, then any simple strategy that divides the chunks evenly across servers works well; however, if the client is adversarial, then, for any deterministic algorithm, there exists a request sequence such that the algorithm rejects most requests. Simple randomized strategy works reasonably well against oblivious adversary: A simple randomized strategy that places chunks on servers uniformly at random and then never moves them performs does ok -that is, it accepts at least a constant fraction of the requests in expectation. In addition, this strategy provides a strong bound of rejecting only 𝑂 (1/𝑚) fraction of the requests if either of the following is true: (1) if the adversary is weakened so that there is a bound on how frequently it can access the same chunk; or (2) servers have speed augmentation and each server consumes 𝑂 (log 𝑚) requests per time step instead of 1. Data transfer is useful: We show that no strategy without data transfer and constant speed can consume 1 -1/𝑚 fraction of requests with an adversarial client. We also design an algorithm that starts with a randomized allocation and then moves chunks out of heavily loaded servers to balance the load. Given constant speedups of processing and data transfer, we show that this strategy consumes 1 -1/𝑚 fraction of the requests even against a fully adversarial client. Empirical evaluations are promising: To demonstrate the practicality of the system model considered in this work, we construct a case study using FoundationDB [41] to run realistic benchmarks on a real cluster. We conduct simulation experiments using adversarial and realistic workloads to evaluate the empirical performance of our proposed algorithms. Results show that randomized algorithms with and without data transfers only need a small speedup to achieve the same performance as an offline optimal algorithm.
In distributed data stores, we have 𝑚 servers that store a large collection of items, such as key-value pairs, objects, or files. We can partition the data into 𝑛 (𝑛 > 𝑚 2 ) data chunks, where a chunk is a collection of keys or a contiguous range of keys. The system distributes chunks to servers according to its load balancing policies. In this paper, we are going to abstract away the details of exactly how the data partition is done and assume that 𝑛 and 𝑚 are given and do not change over time. We also leave other problems such as data replication (and other database mechanisms, such as write-ahead log and multi-version control) as future work.
Clients send requests to the cluster online and we assume that each request accesses a single data chunk. When a request arrives, it is delivered to the server that hosts the appropriate data chunk. For analysis purposes, we divide the timeline into identical time slots, which have a length equal to the time to process a request. For simplicity, we do not distinguish the read, write, and read-range operations. Each server has a FIFO queue (first-in-first-out) with the maximum length of 𝑞 to store the unserved client requests. When requests arrive, they are stored in the respective server's queue unless there is no space in which case the request is rejected. Since a server can consume one request in a slot, the cluster can consume at most 𝑚 requests in the ideal case where the 𝑚 requests access chunks at 𝑚 different servers. Therefore, to make the workload feasible, we assume that at most 𝑚 requests arrive at the cluster in a time slot and each of these requests accesses a different chunk.
During system initialization, the system allocates chunks to servers according to some load balancing algorithms, after which client requests are served. For the first few results in this paper, we assume that the allocation is fixed after initialization. However, some servers may be overloaded, and a load-balancing algorithm may want to move data from one server to another; therefore, later sections of the paper analyze algorithms that move data. We formally model the process of the data movement, named data transfer, as follows. At any time, a server can be involved in one data transfer. In each transfer, at most 𝑠 chunks to one other server, and this takes 𝑠 time slots. In other words, transferring 𝑠 or fewer chunks from one server to another takes 𝑠 time; therefore, the transfer time is a stepped function, not linear. This models the fact that the latency of moving data from one server to another is often large, but the bandwidth is also large -therefore, sending or receiving 1 chunk vs. several chunks up to the bandwidth takes the same amount of time. On the other hand, moving two sets of chunks from or to two different servers takes 2𝑠 time slots, since the server must prepare the transfer and initiate the transfer. We use the single parameter 𝑠 for modeling both the maximum number of chunks in a data transfer (with a total size below the bandwidth) and the latency of a data transfer. Our case study in Section 6 indicates that this model is reasonable on a real platform.
As noted in Section 1, the goal is to minimize the number of rejected requests, which is a measure of throughput.
Definition 1 (Throughput). Given a sequence 𝜎 of requests arriving over time, 𝐴(𝜎) denotes the number of requests that are accepted by algorithm 𝐴 on sequence 𝜎.
We assume the input sequence of client requests is generated by an oblivious adversary defined as follows.
Definition 2 (Oblivious adversary). The oblivious adversary knows how the online algorithm works, but it does not know the random choices made by the algorithm.
For deterministic data allocation, the oblivious adversary always knows the exact locations of chunks at any point in time; therefore, unsurprisingly, we were able to show that no deterministic algorithm performs well. Most of this paper analyzes the performance of various randomized algorithms.
The theoretical performance of an algorithm can be analyzed by comparing its throughput with optimal throughput. Definition 3 (Constant competitive). An online algorithm 𝐴 is (constant) 𝑐-competitive, if 𝐴(𝜎) ≥ 𝑐 • 𝑂𝑃𝑇 (𝜎) for any finite input sequence 𝜎, where 𝑂𝑃𝑇 is the offline optimal. An algorithm's performance is better when 𝑐 gets closer to 1. While constant competitiveness is nice, we would prefer not to reject a constant fraction of the requests. Thus, we define the following, stronger performance criterion.
Definition 4 (Almost optimal). An online algorithm 𝐴 is almost optimal, if 𝐴(𝜎) ≥ (1 -𝑂 (1/𝑚))𝑂𝑃𝑇 (𝜎) for any finite input sequence 𝜎, where 𝑂𝑃𝑇 is the optimal offline scheduler.
For randomized algorithms, similar definitions apply except that we compare the expected throughput of the algorithm with the throughput of 𝑂𝑃𝑇 . Note that in the best case for an offline optimal where all requests arriving in a time slot access different servers and are processed in this slot, all the |𝜎 | requests in a finite input sequence 𝜎 can be accepted by the optimal. Hence, an online algorithm is near optimal if it can accept (1 -1/𝑚)|𝜎 | requests for all input sequences.
Finally, some of our algorithms will require resource augmentation or speedup. Resource augmentation implies that we allow the algorithm to perform certain operations faster than the optimal algorithm can. We consider two types of resource augmentation: (1) speed of processing requests on servers and (2) speed of data transfer. Often resource augmentation is necessary to achieve nontrivial results. However, the smaller the resource augmentation required, the better the algorithm. We generally want the resource augmentation factor to be no more than a constant.
This section analyzes deterministic policies for allocating data chunks to servers. For warmup, we will first show that if the client requests access to random chunks, then any policy that evenly balances the number of chunks on each server performs well. On the other hand, no deterministic policy performs well against an oblivious adversary -for any deterministic policy, there exists an access pattern that causes the system to accept only a small number of requests.
Simple policies work well against random clients. We now do a simple analysis showing that all reasonable allocations work for random clients where each request picks a chunk to access uniformly at random and independently. Without loss of generality, we relax the assumption and allow multiple requests to access the same chunk even in the same time slot. For this random client, we consider a simple even allocation, where 𝑛/𝑚 chunks are allocated to each server. The allocation is generated once and need not change. The following lemma is easy to prove using Chernoff bounds.
Lemma 1. For a random client and an even allocation, in any log 𝑚 consecutive time slots, the probability that more than 3 log 𝑚 requests arrive at a particular server is less than 1/𝑚 2 .
Lemma 1 leads to the following theorem. Proof. Divide the execution into rounds, where each round has 𝑚 log 𝑚 requests -therefore, each round has at least log 𝑚 time steps, so a server can process 3 log 𝑚 requests with speed 3 from its queue. From Lemma 1, the probability that a server receives more than 3 log 𝑚 requests in a round is at most 1/𝑚 2 . By union bound, in a particular round 𝑟 , the probability that at least one server receives more than 3 log 𝑚 requests is at most 1/𝑚.
For a particular round 𝑟 , we will prove via induction that the number of leftover requests from the previous round in any server's queue is at most 3 log 𝑚 for all servers at the start of every round. We have two cases. First, In round 𝑟 , no server gets more than 3 log 𝑚 requests. In this case, the queue size of server 𝑎 at the end of the round does not increase, since server 𝑎 can process 3 log 𝑚 requests during a round if available. During this round, there can be at most 6 log 𝑚 requests in any queue and no requests are rejected. Second, in round 𝑟 , some server get more than 3 log 𝑚 requests. We can pessimistically assume that all the 3 log 𝑚 requests that arrived in 𝑟 are rejected maintaining the inductive hypothesis. Since requests are rejected only in the second type of rounds, which occur with a probability at most 1/𝑚, we reject at most 1/𝑚 fraction of requests in expectation.
No deterministic policy works against an adversarial client. We now consider an oblivious adversary. Intuitively, since the algorithm is deterministic and the adversary knows the allocation, it can always overload some server. Here we argue that no deterministic policy, even if it moves data to load balance, can work well with small queues.
Theorem 2. Consider a deterministic algorithm 𝐷 running on a system with 𝑚 servers, 𝑛 chunks, max queue length 𝑞, data transfer time 𝑠, and constant speedup for both request processing and data transfer, where 𝑛 > 𝑚 2 . There exists a request sequence 𝜎 such that 𝑂𝑃𝑇 (𝜎)/𝐷 (𝜎) = Ω(𝑚𝑠/(𝑞 +𝑐𝑠)) where 𝑂𝑃𝑇 is the offline optimal.
Proof. Since 𝑛 > 𝑚 2 , at any time instant, there exists a server with more than 𝑚 chunks. Since the algorithm is deterministic, the adversary always knows which server has more than 𝑚 chunks and can send all 𝑚 requests to these chunks.
Since the server can process only 𝑐 chunks per time step, the queue will soon fill up causing most requests to be rejected. Now, say that the 𝐷 moves chunks. We divide the timeline into rounds. Each round has 𝑚𝑠 time slots, where 𝑠 is the number of slots to perform a data transfer. Here is the adversary's strategy. At the start of round 𝑖, pick 𝑚 chunks that are all in a particular server for 𝐷, say 𝑎, and send requests to these chunks only for the next 𝑠 time steps. By the end of 𝑠 steps, 𝐷 can move 𝑠 of these chunks away from 𝑎 with 𝑞 requests. With speed 𝑐 for request processing, 𝐷 can process at most 𝑐𝑠 requests in the 𝑠 time steps and store at most 𝑞 requests in the queue. Therefore, it accepts at most 𝑐𝑠 + 2𝑞 of these 𝑚𝑠 requests. After this, for the rest of the round (the remaining 𝑚𝑠 -𝑠 time steps), the adversary does not send any requests.
Since OPT knows the sequence, in the previous round 𝑖 -1, it will set up to ensure that all these 𝑚 chunks were on different servers so it can accept all requests. In the remaining (𝑚 -1)𝑠 time steps of round 𝑖, OPT (in collaboration with the adversary) sets up for the next round. It knows which server will have at least 𝑚 chunks at the start of the next round for 𝐷, so it distributes these 𝑚 chunks across 𝑚 servers for OPT so that OPT can answer all 𝑚𝑠 requests in the next round. The adversary can then repeat for the next round. □
In this section, we consider a simple randomized strategy that randomly allocates chunks to servers, which is similar to the game of throwing randomized balls into bins. We will show that it is constant competitive. We will also prove that no strategy that doesn't move data can be almost optimal against an oblivious adversary. In the next section, we will see a strategy that is almost optimal.
Definition 5 (M 𝑏𝑎𝑙𝑙-𝑏𝑖𝑛 ). For 𝑛 chunks and 𝑚 servers, the algorithm picks a server for each chunk uniformly at random and independently. A M 𝑏𝑎𝑙𝑙-𝑏𝑖𝑛 mapping denotes the mapping between chunks and servers.
Upper Bounds for Balls into Bins. We will now prove a theorem against an adversarial client.
Theorem 3. Given a system with 𝑚 servers and 𝑛 requests. For any sequence 𝜎, say 𝐸 [𝐵 𝜎 ] is the expected number of requests accepted by balls into bins allocation 𝐵. Then, 𝐸 [𝐵 𝜎 ]/|𝜎 | ≥ 1-1/𝑒 even with queue size 𝑞 = 1 and no resource augmentation.
Proof. Consider the balls into bins allocation and consider a particular time step 𝑡 when the client sends requests to 𝑘 ≤ 𝑚 distinct chunks. These 𝑘 chunks were randomly thrown on 𝑚 servers. Now consider a particular server 𝑎. The probability that exactly 𝑖 of these 𝑘 requests hit 𝑎 is:
Therefore, the expected number of servers that get at least 1 request is at least
Any server that gets at least one request processes at least one request -therefore, at least 𝑥 requests out of 𝑘 total requests were consumed at this time step. Therefore, by adding over all time steps, we get the result. □
We see that the balls into bins allocation is constant competitive with no speed augmentation and with very small queues. What if we give it speed augmentation? We now show that with sufficient (Θ(log 𝑚)) speed augmentation, it is almost optimal. Proof. Consider a particular time step 𝑡 when the client sends requests to 𝑘 ≤ 𝑚 distinct chunks. These 𝑘 chunks were randomly thrown on 𝑚 servers when the allocation was done. Now consider a particular server 𝑎. The probability that at least 𝑥 requests access this server at this time step is at most 2 for 𝑥 = 𝑐 log 𝑚 with large enough 𝑐. At any time step, with Θ(log 𝑚) speed, 𝑥 = Θ(log 𝑚) requests can be processed by a server. Therefore, the balls into bins allocation can process all the 𝑘 requests that arrive on that time step with a probability at least 1-1/𝑚. Even assuming pessimistically that all requests at other times are rejected, summing over time still gives us the result. □ Lower bound on algorithms with no data transfer. We saw that the balls into bins algorithm is almost optimal with Θ(log 𝑚) speed; however, what about constant speed? We now show a lower bound, saying that any algorithm without moving data cannot be almost optimal with constant speed. Theorem 5. Given any policy 𝐷 for allocating chunks, for any queue with length 𝑞, a constant speedup of processing, if 𝑛 > 𝑚 2 , then there exists an input sequence, such that
Here is the adversary's strategy: It simply selects 𝑚 different chunks at random and repeatedly requests these chunks in every time slot. The challenge is to show that there is no randomized strategy that can provide an almost optimal acceptance ratio in expectation for this sequence. To do this, we will define a concept of a group, which will allow us to extract a common property of all distributions. Definition 6 (group). Let a 𝑇 -group consists of exactly 𝑇 chunks that are in the same server, and different groups do not have any chunks in common.
We want all groups to have the same number of chunks in order to compute probabilities. If a server has at least 𝑇 chunks, we create groups with 𝑇 chunks each and leave the leftover chunks ungrouped. Repeatedly put 𝑇 ungrouped chunks into groups until we have fewer than 𝑇 ungrouped chunks. The following observation just says that there are a large number of groups.
Observation 1. For any position of 𝑛 chunks in 𝑚 servers, the number of (𝑛/2𝑚)-groups is at least 𝑚.
We now only consider grouped chunks and we will show that a large number of requests sent to grouped chunks will be rejected. In particular, we will first show that the probability that a group gets a large number of requests is not too small. Lemma 2. Given a specific distribution of the chunks, and suppose the adversary randomly selects 𝑚 chunks. Then for 𝑘 = 𝑜 (𝑚) and 𝑛 > 𝑚 2 , the probability that an (𝑛/2𝑚)-group has at least 𝑘 requests is greater than
We can say that these groups are overloaded.
Proof. Since the adversary randomly chooses 𝑚 distinct chunks, the number of chunks in each group follows a Hypergeometric Distribution. In particular, consider 𝑘 = 𝑜 (𝑚) and group size 𝑇 = 𝑛/2𝑚. We can assume that 𝑚 -𝑘 > 𝑚/2 and 𝑇 -𝑘 > 𝑇 /2 and 𝑚 -𝑘 < 𝑛/𝑚. We can then bound the probability that a particular group has 𝑘 chunks among the 𝑚 chunks by the following:
This proves the lemma. □
Now we are able to give a proof for Theorem 5. Since the adversary repeatedly requests the same 𝑚 chunks, overloaded groups remain overloaded. Therefore, regardless of the queue size, servers with overloaded groups will eventually reject many of their requests.
Proof. According to lemma 2, each (𝑛/2𝑚)-group is overloaded with probability 𝑝 -they get at least 𝑘 requests on every time step. Therefore, with any speedup smaller than 𝑘, these groups remain overloaded forever since their total capacity is smaller than their average load. Therefore, these chunks will reject at least one request per time step. Since the expected number of overloaded groups is 𝑚𝑝 = 𝑚 8 𝑘 𝑘! •𝑒 -3 2 , with speed < 𝑘, at least 𝑚𝑝 requests are rejected on every time step within expectation while the optimal algorithm can handle all 𝑚 requests. Therefore, the fraction of the requests that any algorithm accepts is 𝑝, and 𝑝 is a constant for any constant 𝑘. □
Constraining the request sequence. The argument in the previous section indicates that the worst-case workload for any randomized strategy that does not move data is an adversary which requests the same 𝑚 chucks repeatedly. However, this is often not what happens in the real world. In this section, we show that we can do better if the adversary can not request the same chunks repeatedly -we assume that the client issues at most one request to a particular chunk in any consecutive log 𝑚 time steps and show that balls into bins strategy is almost optimal against this adversary.
We divide time into phases of size log 𝑚. The following Lemma is similar to Lemma 1, except that the randomness comes from the randomized allocation rather than the client. Lemma 3. Against a constraint adversary which can not send more than one request to the same server in one phase, the probability that any server gets more than 3 log 𝑚 requests in a phase is at most 1/𝑚 2 .
With Lemma 3, we can prove Theorem 6 in a manner very similar to the proof of Theorem 1. Theorem 6. Given a queue with size 𝑞 = 𝑂 (log 𝑚) and a constrained adversarial input 𝜎, the expected number of requests consumed by balls into bins policy is
The result indicates that a workload is great if requests are not successively and repeatedly access to the same chunk.
In this section, our goal is to design a policy that is almost optimal (consumes 1 -𝑂 (1/𝑚) fraction of the requests) with constant speedup and without placing any restrictions on the input sequence. In particular, we propose an algorithm, say 𝑌 which satisfies the following strong guarantee.
Theorem 7. Say we have a constant speedup of data movement and a constant speedup of processing on the system 𝑌 with the queue length 𝑞 = Θ(𝑠 log 2 𝑚), where 𝑠 is the number of time slots to complete a data transfer. For any input sequence 𝜎, the expected number of requests consumed by 𝑌 is at least
An overview of the system 𝑌 : In prior sections, we have seen that while deterministic policies can not be shown to be constant competitive (even allowing data movement) (Theorem 2), simple randomized policies such as allocating chunks uniformly at random can achieve constant competitiveness (Theorem 3). In particular, with balls into bins randomization, a constant fraction of the requests can be consumed on each time step. However, since some servers get more than a constant number of requests (some servers get Ω(log 𝑚) requests) on each time step, these servers are overloaded and the adversary can keep them overloaded -therefore, with at most constant speed and with no restrictions on the input sequence, no randomized policy without data movement can be almost optimal (Theorem 5).
To address this issue, we will design a system, we call it, 𝑌 which moves data to get a good load balance. In particular, we start with a randomized balls into bins allocation just like the previous sections. Therefore, all chunks have a home server where they were allocated through balls into bins. However, when the queue of the server contains a request that has been in the system for Θ(𝑠 log 𝑚) time, this indicates that this server is overloaded and can not handle all the requests being sent to it. At this point, a batch movement process is triggered on this server and the chunks in the server that have pending requests are distributed to other servers so that these old pending requests can be handled efficiently by other servers. Once these pending requests are handled, these chunks are moved back to their home server so that the original randomized allocation is restored.
Modeling assumptions: For simplicity in analysis, we will make some modeling assumptions that do not impact the overall result. In particular, we will assume that each server has two processors: a primary processor which consumes the requests that arrive at this server from the client and a secondary processor which consumes the requests that sent to it from other servers for load balancing purposes from other processors. Each server has its own queues, where the primary and secondary queues are both of size Θ(𝑠 log 2 𝑚). Note that this does not impact the theorem statement since we allow for constant speedup -if we allow for processor speed of 𝜌 𝑝 , then each of the processors can run at half the speed. Similarly, the queue length can be split among the two queues while only impacting constant factors. In this analysis, we will not try to optimize constant factors; therefore, the constant factors computed will be large. In the evaluation section, we will see that the algorithm requires a quite small constant in practice to perform very well.
To restate the algorithm using these terms: when a request arrives at a server from a client, it is put in the primary queue. If the primary queue is full, then the request is rejected. If the age of the oldest request in the primary queue is 6𝑠 log 𝑚, the server triggers a batch movement process, and any chunk that has a request in the primary queue is moved to other servers and their corresponding requests are moved to those servers' secondary queues (move out phase). These moved requests are processed by the target processor's secondary server. Once a server's secondary queue is empty, the chunks that have been moved there are sent back to their home servers (move back phase). We will define the precise policy of movement momentarily, but let us first consider what the challenges are in designing this policy.
Challenges: Intuitively, the system 𝑌 should work since (1) it moves the requests out of the primary queue in a timely manner to avoid filling the primary queue; (2) it spreads the requests among the secondary queues, which efficiently consumes the requests and avoids filling secondary queue. However, there are a few challenges. First, a batch movement process might take a long time if the batches are large. When a server triggers a batch movement, many chunks in that server may have pending requests in the primary queue and all these chunks must be moved. Recall, from Section 2, that it takes 𝑠 time to move at most 𝑠 chunks between two particular servers -we call this one operation a transfer. Therefore, depending on how many transfers are required, a full batch movement may take time. This can cause the queue of the server to get longer while the batch process is executing as well as blocking other batch processes from starting. Second, we may get unlucky and a server may get a very large number of requests from the client at the same time step. When this happens, the primary queue can become overloaded very quickly, potentially causing downstream effects. Third, the batch process is deterministic -therefore, in principle, the adversary can guess the location of chunks when they are being processed away from their home server and can overload the servers where these chunks are located by sending too many requests there.
Algorithm Description. We can now describe 𝑌 and how it handles these challenges. The queue size of each server is Θ(𝑠 log 2 𝑚) for a sufficiently large constant hidden in the Θ-notation.
(I) Batch trigger: A batch process at a particular server 𝑎 is triggered when the oldest request at this server that is not already a part of an older batch process is 6𝑠 log 𝑚 old.
(II) Batch start: In Y, batch processes may not start as soon as they are triggered; they are executed in order. Say, the batch process 𝑖 is triggered before the batch process 𝑗 is triggered. If the processes are triggered on the same server, then 𝑗 executes after 𝑖 completes since the same chunks may be involved in both. If they are on different servers, they can execute in parallel. Our distribution algorithm ensures that a server is involved in at most transfer at a time.
(III) Data distribution during moveout: Once a batch is triggered, we know which chunks are part of the batchthat is which chunks have pending requests which are part of this batch. These chunks are divided into log 𝑚 packages for transfer so that each bucket has 𝑂 (𝑠) chunks and at most 𝑂 (𝑠 log 𝑚) requests. We later show that this is possible with high probability. Therefore, each of these packages can be moved to destination servers using one transfer each. When the batch starts, these buckets are moved to log 𝑚 different servers and the corresponding requests are moved to those server's secondary queues.
(IV) Process the chunks and move back: The secondary processor of the target server processes the requests from the secondary queue in order. Once all requests of a particular transfer have been processed, the chunks are moved back to their home server.
(V) Request handling during batch: Even when a chunk has been moved to a different server for processing its pending requests, any new requests to this chunk are still sent to its home server's primary queue and these are processed once the chunk has moved back to its home server.
(VI) Flow control and batch cut-off: In order to avoid some boundary conditions, we perform some controls: (1) If a single server receives more than 2 log 𝑚 requests in a single time step, then some of the requests are immediately rejected to ensure that only 2 log 𝑚 requests are added any primary queue on any single time step. ( 2) If the server has more than 24𝑠 log 𝑚 chunks with requests in the primary queue, the server only moves 24𝑠 log 𝑚 chunks with the most requests, and the server rejects the requests of unmoved chunks.
Causes of rejection. We want to show that this process rejects at most 𝑂 (1/𝑚 2 ) fraction of the requests in expectation. Note that requests can be rejected due to the following reasons: (1) flow control and batch cut; (2) rejection from the primary queue if it becomes full; and (3) rejection from the secondary queue if it becomes full. We will show that the first two reasons cause few rejects in expectation and the last reason causes no rejections for a good load balancing policy. However, in order to show this, we must first bound the time it needs to complete a batch process.
Bounding the execution time of a batch process. We first bound the execution time of a batch process -the time between the triggering and completion of a batch process. We will divide time into phases of size 3𝑠 log 𝑚. We say that 𝑋 𝑖 is the set of batch processes that were triggered during phase 𝑖. We will prove the following key Lemma. Lemma 4. Any batch process that was triggered during phase 𝑖 will be completed by the end of phase 𝑖 + 1 with constant speedup on both processing speed and transfer speed.
In order to prove this lemma, we first prove some supporting lemmas. First, we recall that a batch process is triggered when the oldest request in the queue which is not already part of the batch process has been in the queue for 6𝑠 log 𝑚 time. Therefore, a batch process can contain requests from at most 6𝑠 log 𝑚 time steps. We can bound the number of requests that are part of a batch process as follows: Lemma 5. Given the balls into bins allocation, constant 𝑐 ≥ 1 and 𝑠 < 𝑚/(𝑐 log 𝑚). In 6𝑠 log 𝑚 consecutive time slots, the probability that a particular server receives 12𝑠 log 2 𝑚 or more requests is less than 1/𝑚 2 .
Proof. Note that the client can send requests to the same chunk in different time steps, but in one time slot it must send 𝑚 different requests. In each time step, for a particular server 𝑎, say 𝑋 𝑖 𝑎 is the random variable representing the number of requests that hit this server at time step 𝑖. For 𝑥 = 2 log 𝑚, we have
The last equation works when 𝑚 > 4. Therefore, we can sum up 𝑋 𝑖 𝑎 through all 𝑐𝑠 log 𝑚 steps and use Bernoulli's inequality to get the bound we need. □
We can also bound the number of chunks involved in a batch process by a method very similar to Lemma 1. Lemma 6. Given a balls into bins allocation, the total number of chunks from a particular server that can be requested within 6𝑠 log 𝑚 time is 24𝑠 log 𝑚 with probability of at least 1 -1/𝑚 2 . Therefore, the total number of chunks involved in a batch process is 24𝑠 log 𝑚 with probability of 1 -1/𝑚 2 . Hence, each batch process causes log 𝑚 transfers to different targets.
Given these lemmas, we can define a transfer packaging policy. Recall that each data transfer can transfer up to 𝑠 chunks to a particular server. Once we have defined a batch with at most 𝑂 (𝑠 log 2 𝑚) requests and 𝑂 (𝑠 log 𝑚) chunks, we package these into log 𝑚 transfers greedily. We keep adding chunks to a transfer until either the number of requests in the transfer is greater than 𝑐 1 𝑠 log 𝑚 for some constant 𝑐 1 or the number of chunks in the transfer is greater than 𝑐 2 𝑠. At this point, this transfer is complete and we start a new transfer. Due to the previous two lemmas, the total number of transfers for a particular batch process is at most log 𝑚 for suitable choices of 𝑐 1 and 𝑐 2 . Therefore, we have the following corollary:
Corollary 7.1. The amount of time it takes to move all the chunks that are part of the same batch process to log 𝑚 target servers is 𝑠 log 𝑚 assuming that we have data movement speedup of 24 -that, we can move 24𝑠 chunks in 𝑠 time.
We can also bound the number of requests in the secondary queue under certain conditions. Consider batch processes in 𝑋 𝑖 ∪ 𝑋 𝑖+1 -the batch processes triggered during two consecutive phases. We want to bound the number of requests that end up in the same server's secondary queue due to these batch processes. Lemma 7. Considering only requests which are part of batches from 𝑋 𝑖 ∪ 𝑋 𝑖+1 , the number of these requests that end up in a particular secondary server's queue is at most 𝑐 3 𝑠 log 𝑚 for some constant 𝑐 3 .
Proof. All the requests that are part of batch processes in 𝑋 𝑖 ∪ 𝑋 𝑖+1 arrive within a time interval of 12𝑠 log 𝑚 since any request which is in batch from 𝑋 𝑖 must have arrived at most 6𝑠 log 𝑚 time before the beginning of 𝑋 𝑖 . Therefore, these batches can contain a maximum of 12𝑠𝑚 log 𝑚 requests. Since requests are evenly balanced across servers during the batch movements and we have 𝑚 servers, no server has more than 𝑐 3 𝑠 log 𝑚 requests for a suitably large 𝑐 3 .
□
We now observe the following invariant about consecutive phases since a batch is triggered when the oldest request which is not part of a batch arrived 6𝑠 log 𝑚 time ago. Observation 2. Consider any two consecutive phases 𝑖 and 𝑖 + 1, and consider the set 𝑋 𝑖 ∪ 𝑋 𝑖+1 -the set of batch processes triggered during these phases. No two batch processes in this set can be triggered by the same server.
Proof. The base case is trivial since the first phase doesn't trigger any batch processes. For Inductive Hypothesis, assume that all batch processes triggered during phase 𝑖 pleted by the end of phase 𝑖 +1. Therefore, all batch processes triggered during phase 𝑖 + 1 are ready to start at the beginning of phase 𝑖 + 2 since all prior batch processes at their respective servers have completed.
None of the batch processes triggered during phase 𝑖 + 1 have the same source (Observation 2). In addition, they trigger log 𝑚 transfers each (Corollary 6). Therefore, there are a total of at most 𝑚 sources and at most 𝑚 log 𝑚 targets of the transfers -these can be scheduled in 𝑠 log 𝑚 time without source or destination conflicts with sufficient large constant speedup in data transfer speed. Now consider the processing of these requests by the secondary server. The secondary server can only contain requests due to batch processes triggered during phase 𝑖 + 1 and phase 𝑖 +2 (since phase 𝑖 +2 has started, some of the batch processes in this phase may have also started). By Lemma 7, the total number of requests in any secondary queue due to requests from these two phases is at most 𝑐 3 𝑠 log 𝑚 for some constant 𝑐 3 . Therefore, again, with sufficient constant speedup on processing, all these requests can be processed by the secondary server in the next 𝑠 log 𝑚 steps. Finally, all the chunks from batch processes in phase 𝑖 + 1 must move back and again, which can be done in 𝑠 log 𝑚 time by using a reverse schedule from the move-out schedule. Therefore, in at most 3𝑠 log 𝑚 time after the start of phase 𝑖 + 1 -that is, at the end of phase 𝑖 + 2 -all these batches have completed. □ Bounding the number of rejected requests. We can now show that the number of rejected requests is small. Below is the proof of Theorem 7.
Proof. Recall that requests can be rejected due to the two control strategies. First, Lemma 6 implies that the probability that a batch has more than 24𝑠 log 𝑚 chunks involved is less than 1/𝑚 2 . Since rejecting a request due to the second policy happens only on these batches, this probability is also less than 1/𝑚 2 . Second, consider the case where 2 log 𝑚 requests arrive at the same server. For a particular server 𝑎, say 𝑋 𝑎 is the random variable representing the number of requests that hit this server at this time step. Denote 𝑥 = 2 log 𝑚, we have
For 𝑚 > 4 we have 1 𝑥! < 1/𝑚 3 . According to the union bound through all the servers, the probability that it is rejected due to the first policy is less than 1/𝑚 2 .
In addition to the control strategies, a request may be rejected if the primary queue is full. However, recall that each batch process completes in time at most 𝑂 (𝑠 log 𝑚) and due to Lemma 5 the total number of requests that arrive in this time period is at most 𝑂 (𝑠 log 2 𝑚) with probability 1 -1/𝑚 2 . Since the primary queue has this capacity, the probability of a request being rejected due to this is small. Finally, Lemma 7 indicates that requests can never be rejected from the secondary queue since the number of items in it are at most 𝑂 (𝑠 log 𝑚). □
In this section, we evaluate the practicality of our theoretical study on load balancing in distributed key-value stores via a case study real-world database cluster and extensive simulations based on statistics measured from realistic workloads.
Our case study platform is an Amazon Kubernetes Service cluster running FoundationDB [41], which is an open-source distributed database supporting transactional key-value stores. We develop experimental components in FoundationDB to measure the runtime statistics and customize workloads to mimic real-world applications. The case study aims to verify the applicability of the theoretical model for the load balancing problem in distributed key-value stores and collect statistics to be used in simulation experiments.
Experimental design. We set up a FoundationDB cluster on the Amazon Kubernetes Service with 70 storage servers and one data distributor, managed by the FoundationDB Operator. Each storage server runs in a dedicated Kubernetes pod with Amazon EBS storage and contains more than 100 data chunks. The RocksDB [5] is used as the storage engine.
The storage layer of the FoundationDB implementation that supports snapshots, including RocksDB, works as follows. The keys (values) are persistent in RocksDB checkpoints. Each checkpoint, denoted as a data chunk in previous sections, is stored as a physical file on the disk. Each key is persistent in one checkpoint file. A data transfer from server 𝑖 to server 𝑗 is implemented as fetching and transmitting at most 𝑠 checkpoint files from server 𝑖 to 𝑗.
In FoundationDB, a client uses transactions to interact with the storage. A transaction contains a series of requests (operations) on keys, such as read, write, and scan (read range) requests. Since this work focuses on the storage layer, we regard transactions as the proxy of clients.
Our case study runs the realistic workload using the Yahoo! Cloud Serving Benchmark (YCSB) [10], with an extension that client requests have a 90/10 Read/Write ratio, which mimics the pattern in real-world applications. We run the workload for one hour containing more than one million requests, from which we randomly sample and measure 100K requests with specific properties under consideration.
Since the data distributor issues a data transfer and monitors it until the data chunks complete moving, we measure the data transfer latency inside the data distributor. Hence, the latency measurement is accurate without needing clock synchronization between storage servers. When recording a data transfer, we collect the number of bytes in the moved data chunks, its end-to-end latency, the source storage server, and the destination storage server.
Data transfer latency. Recall that our theoretical model assumes that the latency of a data transfer does not increase with the number of moved chunks, as long as the total size is below the network bandwidth. To validate this assumption, we use our FoundationDB cluster to measure the data transfer latencies under different settings: (1) moving chunks with different total data sizes from one server to another server, (2) receiving chunks in multiple data transfers from different servers at the same destination server, and (3) sending chunks in multiple data transfers from the same source server to different servers.
Figure 1(a) presents the latency of standalone data transfers with different total sizes. We can see that the latency does not increase with larger data sizes. This is mainly because the network bandwidth between storage servers is approximately above 1GB per second, which is larger than the total size. Thus, the latency is dominated by the CPU overhead of the storage server to prepare for the data transfer, which is independent of the chunk size.
For multiple data transfers to the same destination server or from the same source server in Figures 1(b) and 1(c), we can see that the latency increases roughly linearly with an increasing number of data transfers. This is because the storage server can only process the data transfers sequentially, which is consistent with our theoretical model. Data transfer latency vs. request processing latency. For analyzing theoretical bounds, the time to process a client request is considered as 1 time slot, while the time to perform data transfer is denoted as 𝑠 time slots. Our empirical measurements show that the request processing latency and data transfer latency are at the magnitude of 0.001 and 0.1 seconds, respectively. Besides, the read and scan requests have smaller latencies than that of write. Hence, the ratio 𝑠 is around 100 to 1000 in practice. Thus, our simulation experiments below use 𝑠 = 100.
We conduct simulation experiments with different workloads to evaluate our proposed data distribution policies.
Experimental design. We implement a simulator for a FoundationDB cluster with 𝑚 storage servers, where 𝑛 data chunks are distributed among the servers according to the chosen policy. During a simulation, the simulator generates requests that access some servers following some workload pattern. When adding a generated request to the server's queue, the simulator rejects (drops) the request if the number of requests waiting in the queue is already equal to the queue length 𝑞. Note that the policy in Section 5 utilizes two queues. In this case, a request is rejected if both queues with a total size 𝑞 are full.
We experimented with two request workloads: Adversary and Zipfian. The adversarial workload always generates requests that access the same set of 𝑚 chunks during the simulation, which is especially difficult for policies without knowing this access pattern. In comparison, the Zipfian workload generates requests that access chunks following the Zipfian distribution [30] with the parameter 𝑎 = 2, which represents more realistic workloads. For each setting, we run 10 simulations, each with 4.5 million generated requests, measure the request rejection ratios and report the median.
Policies and baselines. We implemented our proposed randomized policy without data transfers (labeled Random) and with data transfers (labeled DataMove). To examine how speedup of request processing and data transfers improves the performance of our policies given the same workload, we also enable the simulator to have different speeds. In particular, at speed 1, a server processes one request in a time slot. At speed 2, for instance, a server processes two requests in a time slot. In the simulation, there is no speedup of data transfer.
As a baseline comparison, we implemented and evaluated a policy that deterministically distributes data chunks to servers. Additionally, if requests evenly access all servers, the requests generated at the beginning of a time slot can all be fully processed in this time slot at speed 1. Therefore, the lower bound of an optimal policy can achieve zero rejection.
Evaluation results. Figure 2 shows the rejection ratios under the adversarial and ZipFian workloads, where the queue lengths are set to 𝑞 = 200𝑠 log 𝑚 and 𝑞 = 20𝑠 log 𝑚, respectively. The queue length is chosen so that we can clearly observe the performance differences under different policies. Simulation results show that randomized policy significantly outperforms deterministic policy. Adding data transfers to the randomized policy further reduces the rejection ratios for both workloads. Moreover, we can see that our proposed Random and DataMove policies can already achieve zero rejection with a speedup of 2 and 3, respectively. This result reveals that the required speedup of our policies in practice can be significantly smaller than that in the theoretical results. We have also run experiments with larger 𝑠, 𝑚, and 𝑛 and observed similar performance trends.
Both the problem and the solution are related to many topics studied by researchers. Here we provide a brief overview of some of the related work.
Distrbuted key-value stores have been designed both for academia and commercial use [12,15,23,24,36,41]. Most of these systems are evaluated empirically and most of this work finds that handling online requests that might be skewed towards certain keys is often an important challenge [2,10]. Most of the theoretical analysis of these systems is done by making stochastic assumptions on the arrival pattern of requests [28,29,32]. In this paper, we take a different tack and analyze these systems under adversarial inputs.
Another related area of research is distributed hash tables, which have received extensive theoretical and empirical investigation. For instance, consistent hashing [21,27] is widely used to partition the data in distributed hash maps [12,24,34]. Randomization is widely used to build and maintain the distributed hash tables [13,25,38]. In addition, some distributed hash tables consider network topology in the design of the hash table itself [3,33]. The problem we consider in this paper is also related to general load balancing problems where tasks can be dispatched to various servers.
The techniques used in this paper such as balls-into-bins analysis were originally developed in the load balancing context. These games have been extensively analyzed for various metrics [4,16,31]. In addition, variations of these games such as power of two choices [14,26,29] have also been analyzed extensively and sometimes used in the design of distributed hash tables. The problem we considered in this paper is different but related, to the distributed hash table problem as well as the load balancing problem. In particular, the model for queuing and the model for moving data from one location to another is generally different in distributed hash tables which leads to different algorithmic and analytical challenges.
Since most online systems are dynamic where the system load changes and the keys which are accessed frequently change over time, data migration is an important tool in this system design. There has been extensive empirical work for designing data migration schemes and evaluating them in a diverse range of systems [6, 7, 19, 23, 35-37, 40, 41] for various distributed workloads in key-value stores and distributed databases. Most of this work is empirical or uses stochastic assumptions on data arrivals for analysis.
We have explored the problem of load balancing in distributed storage systems, assuming an oblivious adversary sends requests. With reasonable queue sizes and good algorithm designs, it is possible to consume almost all requests in expectation. Several open problems remain. First, we assume that each chunk is stored in at most one server. Real systems often store each chunk on multiple servers via replication to aid the load balancing and recovery from faults. We will design algorithms to support replication. Second, we assume an oblivious adversary that gets no information from the system once it runs. One can imagine an adaptive adversary that can glean the requests' distribution from the rejection pattern or reply latencies. Finally, we would like to explore if there are algorithms that can provide results similar to ours but using smaller queue sizes and thus smaller latencies.
PPoPP'23, Feb 25-Mar 01, 2023, Montreal, Canada Zhe Wang, Jinhao Zhao, Kunal Agrawal, He Liu, Meng Xu, and Jing Li
PPoPP'23, Feb 25-Mar 01, 2023, Montreal, Canada