Sonia Ben Mokhtar
email: sonia.benmokhtar@insa-lyon.fr

Louis-Claude Canon
email: louis-claude.canon@femto-st.fr

Anthony Dugois
email: anthony.dugois@ens-lyon.fr

Loris Marchal
email: loris.marchal@ens-lyon.fr

Etienne Rivière
email: etienne.riviere@uclouvain.be

Taming Tail Latency in Key-Value Stores: a Scheduling Perspective (extended version)

Keywords: Online Scheduling, Key-Value Store, Replica Selection, Tail Latency, Lower Bound Hash ring M1

Distributed key-value stores employ replication for high availability. Yet, they do not always efficiently take advantage of the availability of multiple replicas for each value, and read operations often exhibit high tail latencies. Various replica selection strategies have been proposed to address this problem, together with local request scheduling policies. It is difficult, however, to determine what is the absolute performance gain each of these strategies can achieve. We present a formal framework allowing the systematic study of request scheduling strategies in key-value stores. We contribute a definition of the optimization problem related to reducing tail latency in a replicated key-value store as a minimization problem with respect to the maximum weighted flow criterion. By using scheduling theory, we show the difficulty of this problem, and therefore the need to develop performance guarantees. We also study the behavior of heuristic methods using simulations, which highlight which properties are useful for limiting tail latency: for instance, the EFT strategy-which uses the earliest available time of servers-exhibits a tail latency that is less than half that of state-of-the-art strategies, often matching the lower bound. Our study also emphasizes the importance of metrics such as the stretch to properly evaluate replica selection and local execution policies.

Introduction

Online services are used by a large number of users accessing ever-increasing amounts of data. One major constraint is the high expectation of these users in terms of service responsiveness. Studies have shown that an increase in the average latency has direct effects on the use frequency of an online service, e.g., experiments at Google have shown that an additional latency of 400 ms per request for 6 weeks reduced the number of daily searches by 0.6 % [START_REF] Brutlag | Speed matters for google web search[END_REF].

In modern cloud applications, data storage systems are important actors in the evolution of overall userperceived latency. Considerable attention has been given, therefore, to the performance predictability of such systems. Serving a single user request usually requires fetching multiple data items from the storage system. The overall latency is often that of the slowest request. As a result, a very small fraction of slow requests may result in overall service latency degradation for many users. This problem is known as the tail latency problem. In large-scale deployments of cloud applications, it has been observed that the 95 th and 99 th percentiles in the query distribution show latency values that can be several orders of magnitude higher than the median [START_REF] Atikoglu | Workload analysis of a large-scale key-value store[END_REF][START_REF] Dean | The tail at scale[END_REF].

In this study, we focus on the popular class of storage systems that are key-value stores, where each value is simply bound to a specific key [START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF][START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF]. These systems scale horizontally by distributing responsibility for fractions of the key space across a large number of storage servers. They ensure disjoint-access parallelism, high availability and durability by relying on data replication over several servers. As such, read requests may be served by any of these replica.

Replica selection strategies [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF][START_REF] Jiang | Haste makes waste: The on-off algorithm for replica selection in key-value stores[END_REF][START_REF] Suresh | C3: Cutting tail latency in cloud data stores via adaptive replica selection[END_REF] dynamically schedule requests to different replicas in order to reduce tail latency. When the request reaches the selected replica, it is inserted into a queue and a local queue scheduling strategy may decide to prioritize certain requests over others. These combinations of global and local strategies are well adapted to the distributed nature of key-value stores, as they assume no omniscient or real-time knowledge of the status of each replica, or of concurrently-scheduled requests. It remains difficult, however, to systematically assess their potential. On the one hand, there is no clear upper bound on the performance that a global, omniscient strategy could theoretically achieve. On the other hand, it is difficult to determine what is the impact of using only local or partial information on achievable performance. Our goal in this paper is to bridge this gap, and equip designers of replica selection and local scheduling strategies with tools enabling their formal evaluation. By modeling a corresponding scheduling problem, we develop a number of guarantees that apply to a variety of designs.

Outline. We make the following contributions:

a formal model to describe replicated key-value stores and the scheduling problem associated to the minimization of tail latency (Section 3); a polynomial-time offline algorithm, a (2 -1 m )-approximation guarantee and a NP-completeness result for related scheduling problems (Section 4); online heuristics to solve the online optimization of maximum weighted flow, representing compromises in locally available information at the different servers of the key-value store (Section 6); the comparison of these heuristics in extensive simulations (Section 7).

The algorithms, the related code, data and analysis are available online 5 .

Related Work

We review related work on key-value stores and system contributions for reducing latency, and on latency minimization work in scheduling theory.

Dealing with Tail-Latency in Key-Value Stores

The principles of key-value stores were first documented by Amazon with Dynamo [START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF]. Cassandra [START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF] is a widely-used open-source key-value store, following principles similar to that of Dynamo; other popular key-value stores include Redis [START_REF] Carlson | Redis in action[END_REF], memcached [START_REF] Jose | Memcached design on high performance rdma capable interconnects[END_REF], and document stores such as MongoDB [START_REF] Chodorow | MongoDB: the definitive guide: powerful and scalable data storage[END_REF].

Key-value stores implement data partitioning for horizontal scalability. Typically, data is spread over a cluster of servers using consistent hashing. This consists of treating the output of a hashing function as a ring; each server is then assigned a position on this circular space and becomes responsible of all data between it and its predecessor (the position of a data item is decided by hashing the corresponding key) [START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF][START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF]. Replication is implemented on top of data partitioning, by duplicating each data item on the successors of its assigned server. Fig. 1 shows an example of a cluster with a replication factor of 3: value whose keys' hash values fall in the range of M 2 can also be requested from servers M 3 and M 4 .

Design and implementation improvements have been proposed to improve response time and in particular reduce tail-latency in key-value stores. They include the use of redundant requests [START_REF] Vulimiri | Low latency via redundancy[END_REF][START_REF] Wu | Costlo: Cost-effective redundancy for lower latency variance on cloud storage services[END_REF], performing smart resource allocation [START_REF] Didona | Size-aware sharding for improving tail latencies in in-memory key-value stores[END_REF], or employing hybrid scheduling [START_REF] Delgado | Job-aware scheduling in eagle: Divide and stick to your probes[END_REF][START_REF] Delgado | Hawk: Hybrid datacenter scheduling[END_REF]. We are interested in this paper in replica selection strategies [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF][START_REF] Jiang | Haste makes waste: The on-off algorithm for replica selection in key-value stores[END_REF][START_REF] Suresh | C3: Cutting tail latency in cloud data stores via adaptive replica selection[END_REF]. They seek to avoid that a request be sent to a busy server when a more available one would have answered faster. The server receiving a request (the coordinator) is generally not the one in charge of the corresponding key. All servers know, however, the partitioning and replication plans. Coordinators can, therefore, associate a key with a list of replicas and select the most appropriate server to query. Cassandra uses Dynamic Snitching [START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF], which selects the replica with the lowest average load over a time window. This strategy is prone to instabilities, as lowly-loaded servers tend to receive swarm of requests. C3 [START_REF] Suresh | C3: Cutting tail latency in cloud data stores via adaptive replica selection[END_REF] uses an adaptive replica selection strategy that can quickly react to heterogeneous service times and mediate this instability. Dynamic snitching and C3 both assume that values are served with the same latency. This does not reflect the reality of storage workloads, where values may be highly heterogeneous in size [START_REF] Atikoglu | Workload analysis of a large-scale key-value store[END_REF]. Requests for small values may be scheduled behind requests for large values, creating a head-of-line blocking problem and increasing tail latency. Coordinator servers do not have, however, directly access to the size of data items, and only know the requested key. Héron [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF] addresses this problem by propagating across the cluster the identity of values whose size is over a threshold, together with load information and pending requests to such large values. This information is stored at each coordinator using a Bloom filter, and Héron avoids scheduling requests for small values behind pending requests for large values. Size-aware sharding [START_REF] Didona | Size-aware sharding for improving tail latencies in in-memory key-value stores[END_REF] avoids head-of-line blocking on a specific server, by specializing some of its cores to serve only large values. Local scheduling can take into account the specificities of the data structure used for storing updates to the local values under write-heavy workloads, such as with Log-Structured Merge Key-Value Stores [START_REF] Balmau | Silk+ preventing latency spikes in log-structured merge key-value stores running heterogeneous workloads[END_REF]. Other systems, such as REIN [START_REF] Reda | Rein: Taming tail latency in key-value stores via multiget scheduling[END_REF] or TailX [START_REF] Jaiman | TailX: Scheduling heterogeneous multiget queries to improve tail latencies in key-value stores[END_REF], focus on the specific case of multi-get operations, whereby multiple keys are read in a single operation. We intend to consider multi-get queries in our future work, as an extension of our formal models.

All the solutions mentioned above empirically improve tail latency under the considered test workloads. There is, however, no strong evidence that no better solution exists as the proposed heuristics are not compared to any formal ground. In contrast, and similarly to our objective, Li et al. [START_REF] Li | Tales of the tail: Hardware, OS, and application-level sources of tail latency[END_REF] propose a single-node model of a complete hardware, application and operating system stack using queuing theory. This allows determining expected tail latencies in the modeled system. The comparison of the model and an actual hardware and software stack shows important discrepancies. The authors were able to identify performanceimpacting factors (e.g. request re-ordering, limited concurrency, non-uniform memory accesses, etc.) and address them, matching close to optimal performance under the knowledge of predictions from the model. Our goal is to be an enabler for such informed optimization and development for the case of distributed (multi-node) storage services.

Latency Minimization in Scheduling Theory

Minimization of latency-the time a request spends in the system-is commonly approached as the optimization of flow time in theoretical works, and a great diversity of scheduling problems deal with this criterion. The functions that usually constitute the objective to minimize are the max-flow (F max ) and the sum-flow ( F i ).

Minimizing maximum (weighted) flow. It is well-known that the maximum flow criterion is minimized by the FIFO (First In First Out) strategy on a single-machine [START_REF] Bender | Flow and stretch metrics for scheduling continuous job streams[END_REF]. This scheme is also (3 -2 m )-competitive on m machines when preemption is allowed or not [START_REF] Bender | Flow and stretch metrics for scheduling continuous job streams[END_REF][START_REF] Mastrolilli | Scheduling to minimize max flow time: Off-line and on-line algorithms[END_REF], and Ambühl et al. gave a (2 -1 m )-competitive algorithm for the preemptive case, which is optimal [START_REF] Ambühl | On-line scheduling to minimize max flow time: an optimal preemptive algorithm[END_REF]. Sometimes the maximum weighted flow max w i F i is considered in order to give more importance to some requests. For example Bender et al. introduced the stretch, where the weight is the inverse of the request serving time (w i = 1/p i ), to express and study the notion of fairness for scheduling HTTP requests in web servers [START_REF] Bender | Flow and stretch metrics for scheduling continuous job streams[END_REF]. For single-machine problems, they proved that no polynomial-time algorithm can approximate the offline non-preemptive problem of optimizing the max-stretch criterion within a factor Ω(n 1-ε ), for any ε > 0, unless P = NP. They also exhibit an FPTAS for the preemptive case, and they derive an O( √ ∆)-competitive algorithm from the EDF (Earliest Deadline First) strategy, with ∆ being the ratio between the largest processing time to the smallest one (∆ = max pi min pi ). Later, Legrand et al. presented a polynomial-time algorithm to solve the offline minimization of max weighted flow time on unrelated machines when preemption is allowed [START_REF] Legrand | Minimizing the stretch when scheduling flows of divisible requests[END_REF]. The FIFO strategy is also shown to be ∆-competitive for the online problem of minimizing the max-stretch on one machine, and a lower bound of (1 + ∆) on a single server and 1 2 (1 + ∆ m+1 ) on m servers [START_REF] Saule | Optimizing the stretch of independent tasks on a cluster: From sequential tasks to moldable tasks[END_REF]. Finally, Dutot et al. closed the online problem of minimizing the stretch on a single machine by proving a lower bound of 1 2 ∆( √ 5 -1), which is tight [START_REF] Dutot | Online non-preemptive scheduling to optimize max stretch on a single machine[END_REF]. Some additional works deal with the minimization of maximum weighted flow time under different assumptions and models [START_REF] Anand | Minimizing maximum (weighted) flow-time on related and unrelated machines[END_REF][START_REF] Bansal | Minimizing maximum flow-time on related machines[END_REF][START_REF] Lucarelli | Online non-preemptive scheduling to minimize maximum weighted flow-time on related machines[END_REF].

Table 4 provides a more exhaustive summary of results on max-flow minimization.

Minimizing sum (weighted) flow. Optimizing the average performance is obtained through minimizing the sum-flow criterion. It is well known that this problem is hard without preemption, while the preemptive version can be polynomially solved by the SRPT (Shortest Remaining Processing Time) strategy [START_REF] Baker | Introduction to sequencing and scheduling[END_REF]. Kellerer et al. gave an O( √ n)-approximation algorithm for the non-preemptive case, and proved that no polynomialtime algorithm can approximate the offline non-preemptive problem of minimizing the sum-flow within a factor Ω(n 1 2 -ε ), for any ε > 0, unless P = NP [START_REF] Kellerer | Approximability and nonapproximability results for minimizing total flow time on a single machine[END_REF]. On m servers, Leonardi et al. analyzed SRPT and showed a competitive ratio of O(min(log ∆, log n m )) [START_REF] Leonardi | Approximating total flow time on parallel machines[END_REF]. They provide a lower bound of Ω(n 1 3 -ε ) on the approximability of this problem, unless P = NP. For the weighted sum-flow, SRPT has been proven 2competitive on a single server, and 14-competitive on parallel machines [START_REF] Muthukrishnan | Online scheduling to minimize average stretch[END_REF]. Chekuri et al. improved this last competitiveness result by providing a 9.82-competitive algorithm on m machines, and a 17.32-competitive algorithm for non-migratory version of the problem [START_REF] Chekuri | Algorithms for minimizing weighted flow time[END_REF]. They also present a (2 + ε)-approximation for the offline optimization of weighted sum-flow on one machine. Note that one must be careful when using the (weighted) sum-flow criterion, as it may lead to starvation: some requests may be infinitely delayed in an optimal solution [START_REF] Bender | Flow and stretch metrics for scheduling continuous job streams[END_REF]. This is one reason why we discard this criterion in our study.

For a more detailed survey about sum-flow optimization, see Table 5.

Replication in scheduling. An important consequence of replication is that a given request cannot be executed by any server; it must be processed by a server in the subset of replicas able to handle it. In scheduling literature, this constraint is known as "multipurpose machines", "processing set restrictions" or even "eligibility constraints". Brucker et al. proposed a formalization and analyzed the complexity of some of these problems [START_REF] Brucker | Complexity of scheduling problems with multi-purpose machines[END_REF], by introducing the M i term in the common α|β|γ notation. They also used a routine based on solving the minimum cost matching problem to polynomially solve Q|M i , p i = 1| w i U i and

P |M i , r i , p i = 1| w i U i .
Another notable result is that P |M i , pmtn| C i can be solved in polynomial time by transforming any preemptive schedule in a non-preemptive one without worsening the objective. A more recent survey is also proposed by Leung et al. [START_REF] Leung | Scheduling with processing set restrictions: A survey[END_REF], and Shabtay et al. treated a similar problem where there are two job types that can be processed by a specific subset of machines [START_REF] Shabtay | Online scheduling of two job types on a set of multipurpose machines with unit processing times[END_REF].

To the best of our knowledge, there exists no work considering replication for the minimization of the maximum (weighted) flow.

Formal Model

We propose a formal model of a distributed and replicated key-value store. This section describes the theoretical framework and states the optimization problem related to the minimization of tail latency.

Application and Platform Models

We start by defining a key-value map (K, V ) as a set of associations between keys and values. We associate c keys K = {K 1 , . . . , K c } to c values V = {V 1 , . . . , V c }: each unique key K l refers to a unique value V l whose size is z l > 0.

The considered problem is to schedule a set of n requests T = {T 1 , . . . , T n } on m parallel servers M = {M 1 , . . . , M m } in a replicated key-value store. The set K is spread over these servers. For one server M j , the function Ψ gives the subset of keys Ψ (M j ) ⊆ K that is owned by M j . Each request T i carries a key that will be used to retrieve a specific value in the store. For one request T i , the function ϕ gives this key ϕ(T i ) = K l . The same key can be carried by different requests. Fig. 2 shows the relationship between requests, keys and values. A server M j may execute a request T i if and only if ϕ(T i ) ∈ Ψ (M j ), i.e., M j holds the value for the carried key of T i .

All requests are independent: no request has to wait for the completion of another request, and no communication occurs between requests. We limit ourselves to the non-preemptive problem, as real implementations of key-value stores generally do not interrupt requests.

In addition, each request T i has a processing time p i = αz l + β, where α, β > 0 (with ϕ(T i ) = K l ). Processing time is equal to the average network latency β plus data sending time, which is proportional to the size of the value this request is looking for (factor α represents the inverse of the bandwidth). A request is also unavailable before time r i ≥ 0 and its properties are unknown as well.

As a server M j may execute a request T i only if it holds the key ϕ(T i ), we treat the multipurpose machines scheduling problem where the set M i ⊆ M represents the set of machines able to execute the request T i , i.e., M i = {M j | ϕ(T i ) ∈ Ψ (M j )}. In the Graham α|β|γ notation of scheduling problems, this constraint is commonly denoted by M i in the β-part. This aspect of the problem models data replication on the cluster. Key-value stores tend to express the replication factor, i.e., the number of times the same data is duplicated, as a parameter k of the system. Therefore, we have

|M i | = k.

Problem Statement

There is no objective criterion that can straightforwardly represent the formal optimization of tail latency, as there is no formal definition of this system concept. Different works consider the 95 th percentile, the 99 th percentile, or the maximum, and it should be highlighted that we do not want to degrade average performance too much. We propose to approach the tail latency optimization by minimizing a well-known criterion in online scheduling theory: the maximum time spent by requests in the system, also known as the maximum flow time max F i , where F i = C i -r i expresses the difference between the completion time C i and the release time r i of a request T i .

However, it seems unfair to wait longer for a request for a small value to complete than for a large one: for example, we know that a user's tolerance for the response time of a real system is greater when a process considered to be heavy is in progress. Hence, the latency should be weighted to emphasize the relative importance of a given request; we are looking for a "fairness" property. To formalize this idea, we associate a weight w i to each T i . The definition of this weight is flexible, in order to allow the key-value store system designer to consider different kinds of metrics. We focus on three weighting strategies in our simulations. First, the flow time (w i = 1) gives an importance to each request that is proportional to its cost, which favors requests for large values. Second, the stretch (w i = 1 pi ) gives the same importance to each request, but this favors requests for small values because they are more sensitive to scheduling decisions. Third, the inverse of the square root of the processing time (w i = 1

√ pi ) constitutes a compromise between the two previous cases. This last weighting strategy is denoted as the "weak stretch" in this paper.

In summary, our optimization problem consists in finding a schedule minimizing the maximum weighted flow time max w i F i under the following constraints:

-P : there are m parallel identical servers.

-M i : each request T i is executable by a subset of servers.

online-r i : each request T i has a release time r i ≥ 0 and request characteristics (r i , p i and w i ) are not known before r i .

A solution to this problem (P |M i , online-r i | max w i F i ) is to find a schedule Π, which, for a request T i , gives its executing server and starting time. Then we define a pair Π(T i ) = (M j , σ i ), where the server M j executes T i at time σ i ≥ r i . Server M j must hold the required value (M j ∈ M i ), and there are no simultaneous executions: two different requests cannot be executed at the same time on the same server.

As mentioned earlier, we are also interested in minimizing the average latency ( w i F i ) as a secondary objective; even if the main goal is to reduce tail latency, it would not be reasonable to degrade average performance doing so. This bi-objective problem could be approached by the optimization of the more general p -norm function of flow times [START_REF] Bansal | Server scheduling in the lp norm: a rising tide lifts all boat[END_REF], but it is left for future work.

Maximum Weighted Flow Optimization

In order to evaluate the performance of replica selection heuristics, it would be interesting to derive optimal or guaranteed algorithms for the offline version of our problem, namely the minimization of the maximum weighted flow time of requests, or even for restricted variants. We show here that we can derive optimal or approximation algorithms when tasks are all released at time 0, but as soon as we introduce release dates, the problem gets harder to tackle. Nevertheless, a lower bound can be computed.

Without Release Times

We first focus on the non-preemptive problem of minimizing the maximum weighted flow time on a single server when there is no release times, i.e., all requests are available at time 0. Remark that in this case, the more common completion times equivalently replace flow times (i.e., C i = F i ). We consider a simple algorithm named Single-Simple (Algorithm 1), which schedules requests by non-increasing order of weights w i , to solve this scheduling problem.

Theorem 1. Single-Simple (Algorithm 1) solves 1|| max w i C i in polynomial time.
Proof. Let OPT be an optimal schedule. If all requests are ordered by non-increasing weight, then Single-Simple is optimal. If they are not, we can find two consecutive requests T j and T k in OPT such that w j ≤ w k .

Algorithm 1 Single-Simple

Input: wi Output: a schedule 1: schedule requests by non-increasing order of wi 2: return schedule Then, their contribution to the objective is

C = max(w j C j , w k (C j + p k )) = w k (C j + p k ) because w j C j ≤ w j (C j + p k ) ≤ w k (C j + p k ).
If we swap the requests, then the contribution becomes

C = max(w k C k , w j (C k + p j )) where C k is the completion time of request T k in this new schedule. By construction, C j + p k = C k + p j . We have w k C k ≤ w k (C k + p j ), w j (C k + p j ) ≤ w k (C k + p j ) and w k (C k + p j ) = w k (C j + p k ) = C. Therefore, max(w k C k , w j (C k + p j )) = C ≤ C.
It follows that if two consecutive requests are not ordered by non-increasing weight in OPT , we can switch them without increasing the objective. By repeating the operation request by request, we transform OPT in another optimal schedule where requests are sorted by non-increasing w i . Hence, Single-Simple is optimal.

Single-Simple does not extend to m parallel machines in the general case. However, it solves the case where all requests have homogeneous size p.

Theorem 2. Single-Simple (Algorithm 1) solves

P |p i = p| max w i C i in polynomial time.
Proof. We use the same kind of argument than for the single-machine case. Let OPT be an optimal schedule. If all requests are ordered by non-increasing weight, then Single-Simple is optimal. If they are not, let T j and T k be two requests in OPT such that w j ≤ w k . By definition, p j = p k = p. Without loss of generality, we consider that there is no idle time intervals in OPT , and T j starts at time t, whereas T k starts at time t + cp with c ≥ 0 (on any server).

Their contribution is

C = max(w j C j , w k (C j + cp)) = w k (C j + cp) because w j C j ≤ w k C j ≤ w k (C j + cp). If we switch T j and T k , then the contribution is C = max(w k C k , w j (C k + cp)). By construction, C k + cp = C j + cp, i.e., C k = C j . We have w k C k = w k C j ≤ w k (C j + cp) and w j (C k + cp) = w j (C j + cp) ≤ w k (C k + cp). Hence, C ≤ C.
It follows that we can transform OPT in another optimal schedule by switching repeatedly non-sorted requests. Then, Single-Simple is optimal.

For m machines when processing times are not identical, the problem is trivially NP-hard even with unit weights because P ||C max is NP-hard [START_REF] Lenstra | Complexity of machine scheduling problems[END_REF]. We prove that Single-Simple is an approximation algorithm.

Theorem 3. Single-Simple (Algorithm 1) computes a (2-1

m )-approximation for the problem P || max w i C i , and this ratio is tight.

Proof. Let us consider a schedule S built by Single-Simple and an optimal schedule OPT . We denote by T j the request for which w j C j = max w i C S i , i.e., the request that reaches the objective in S. Then we remove from S and OPT all T i such that w i < w j (it does not change the objective max w i C S i in S and can only decrease max w i C OPT i in OPT ). Let C * max denote the optimal makespan when scheduling only the remaining requests. As S is a list-scheduling (in the sense of Graham), we have [START_REF] Graham | Bounds for certain multiprocessing anomalies[END_REF], where C S max is the completion time of the last request in S (i.e., C j = C S max ). Let T k be the last completed request in OPT , such that C k = C OPT max . This makespan is bounded by the optimal one (i.e.,

C S max ≤ 2 -1 m • C * max
C * max ≤ C OPT k ). Therefore, max w i C S i = w j C S j = w j C S max ≤ 2 - 1 m • w j C * max ≤ 2 - 1 m • w j C OPT k ≤ 2 - 1 m • w j w k • w k C OPT k ≤ 2 - 1 m • w j w k • max w i C OPT i .
As we removed all requests weighted by a smaller value than w j , we have

wj w k ≤ 1 and it follows that max w i C S i ≤ 2 -1 m • max w i C OPT i
. We now prove that this bound is asymptotically tight, by considering the instance with m machines and n = m(m -1) + 1 requests T 1≤i≤n with the following weights and processing times:

-w i = W + 1, p i = 1 for all 1 ≤ i < n; -w n = W , p n = m.
Request T n will be scheduled last in S, which gives an objective of max w i C S i = (2m -1)W , whereas an optimal schedule starts this request at time 0 and has an objective of max w i C OPT i = m(W + 1). On this instance, the approximation ratio 2

-1 m • W W +1 tends to 2 -1 m as W → ∞.

Offline Problem With Release Times

Legrand et al. solved the scheduling problem R|r i , pmtn| max w i F i in polynomial time using a linear formulation of the model [START_REF] Legrand | Minimizing the stretch when scheduling flows of divisible requests[END_REF]. This offline problem is very similar to the one we are interested in, as the platform relies on unrelated machines, which generalizes our parallel multipurpose machines environment (P |M i | max w i F i is a special case of R|| max w i F i [START_REF] Leung | Scheduling with processing set restrictions: A survey[END_REF]). In fact, it only differs on one specific aspect: it allows preempting and migrating jobs, which we do not permit in our model. We establish below the complexity of the problem P |r i , pmtn * | max w i F i , where non-migratory6 preemption is allowed. Interestingly, preventing migration makes the problem NP-complete. The proof of this result consists in a reduction from the NP-complete problem P ||C max [START_REF] Lenstra | Complexity of machine scheduling problems[END_REF].

Definition 1 (NonMigratory-Dec(T, M, B)). Given a set of requests T , a set of machines M and a bound B, if we define the deadline

d i = r i + B
wi for all T i , is it possible to build a non-migratory preemptive schedule where each request meets its deadline?

Theorem 4. The problem NonMigratory-Dec(T, M, B) is NP-complete.
Proof. We prove the NP-completeness of this problem by reduction from P ||C max , which is NP-complete [START_REF] Lenstra | Complexity of machine scheduling problems[END_REF]. Obviously, NonMigratory-Dec(T, M, B) belongs to NP.

Building instance. We consider an instance I 1 of the decision problem associated to P ||C max : given a set of requests T , a set of servers M and a bound B , is it possible to schedule each request T i non-preemptively on M such that C max = B ? We construct the following instance I 2 of NonMigratory-Dec(T, M, B) from I 1 . We first set M = M and B = B . For each request T i we define a request T i to which we associate a release time r i = 0 and a weight w i = 1; hence, d i = B. I 2 can clearly be constructed in polynomial time in the size of I 1 .

Equivalence of problems. A solution to I 1 trivially constitutes a non-preemptive (thus non-migratory) solution to I 2 .

Assume now that I 2 has a solution S. It means that for each server M j , we know a set A j ⊆ T of requests that are preemptively scheduled on M j exclusively (recall migration is not allowed), and max Ti∈Aj C S i is the makespan of M j in S. For each request T i , we denote the associated set of n i processing intervals by

Λ i = {(σ i,k , δ i,k ) | 1 ≤ k ≤ n i }, where σ i,k is the start time of the k-th processing interval of T i and δ i,k is its duration. For all i, k, σ i,k + δ i,k ≤ σ i,k+1
, and for all j,

max Ti∈Aj C S i = Ti∈Aj ni k=1 δ i,k ≤ C S max = B.
We can build a solution S to I 1 by removing preemptions from S, i.e., for each request T i , we rearrange its n i intervals (without migrating them) such that for all k, σ i,k +δ i,k = σ i,k+1 . As we only switch processing intervals of requests, it does not change the makespan values of servers, therefore, max

Ti∈Aj C S i = Ti∈Aj ni k=1 δ i,k = max Ti∈Aj C S i ≤ C S max = B = B .

Online Problem

We now study problems in an online context, where properties of requests are not known before their respective release time. We prove that there exists no optimal online algorithm for the minimization of maximum weighted flow even on the very simple case of a single machine and unit request sizes, as outlined in the following theorem.

Theorem 5. No online algorithm can be optimal for the scheduling problem 1|online-r i , p i = 1| max w i F i .

Proof. Let us prove this by building an example instance. We consider 10 requests T 1≤i≤10 with the following weights and release times:

-

w i = 3, r i = i -1 for all 1 ≤ i ≤ 8; -w 9 = 1, r 9 = 0; -w 10 = 1, r 10 = 0.
We can construct an optimal schedule OPT by starting T 9 and T 10 before T 8 , which gives an objective of max w i F OPT i = 9. Now let T 11 be a request released at r 11 = 9 with a weight w 11 = 11. The new optimal schedule OPT has to schedule T 8 before T 11 , i.e., T 9 or T 10 will start last, to obtain an objective of max w i F OPT i = 11. OPT cannot be built from OPT ; adding a new request may lead to reconsider our previous scheduling choices. Hence, the competitive ratio is always strictly greater than 1.

We now present an adaptation of Single-Simple to the online case, restricted to unit tasks, Single-Unit (Algorithm 2): at each time step, we consider all submitted requests at this time and schedule the one whose flow (if processed now) is the largest. This gives priority to the currently most impacting requests. Unfortunately, even on unit size tasks, this strategy does not lead to an approximation algorithm, as outlined by the following theorem. Theorem 6. The competitive ratio of Single-Unit (Algorithm 2) is arbitrarily large for the scheduling problem 1|online-r i , p i = 1| max w i F i .

Proof. First, we build an instance designed to reach an arbitrary large ratio. Then, we determine a lower bound on the objective achieved with Single-Unit and finally, an upper bound on the optimal one. Instance characteristics. For an arbitrary competitive ratio k ≥ 1, we build the following instance with n requests. The first k requests have a processing cost of w i = k and release time r i = 0. Then, a new request arrives at each new time step with a processing cost that is the highest integer lower than or equal to 1 + 1 k times the previous submitted request (i.e.,

w i = (1 + 1 k )w i-1 and r i = i -k for k < i ≤ n).
In total, n = k 2 + 11 requests are submitted.

Single-Unit lower bound. At time t = 0, Single-Unit starts one of the first k requests because they are the only ones that are ready. We now prove that at any time t such that 1 ≤ t < k, Single-Unit starts one of the remaining first k requests, which delays all arriving requests (any request T i such that k < i < 2k). On one hand, w i (t + 1 -r i ) = k(t + 1) for any of the first k requests (1 ≤ i ≤ k). On the other hand, for

k < i ≤ n, w i ≤ (1 + 1 k )w i-1 and thus, w i ≤ (1 + 1 k ) i-k k. Therefore, w i (t + 1 -r i ) ≤ (1 + 1 k ) i-k k(t + 1 -i + k).
It remains to prove that at any time t such that 1 ≤ t < k, any of the first requests has the highest value, that is k(t + 1)

≥ (1 + 1 k ) i-k k(t + 1 -i + k) for all k < i ≤ t + k.
By changing variables (j = i -k and t = t + 1), this corresponds to proving (1 + 1 k ) j (t -j) ≤ t for all 1 ≤ j < t ≤ k. We show by induction that (1 + 1 k ) j (t -j) ≤ t for all 0 ≤ j and for a given t ≥ 2 and k ≥ t . The induction basis with j = 0 is direct (t ≤ t ). The induction step assumes (1 + 1 k ) j (t -j) ≤ t to be true for a given j ≥ 0.

(

+ 1 k ) t -j -1 t -j = (1 + 1 k )(1 - 1 t -j ) = 1 + 1 k - 1 t -j - 1 k(t -j) ≤ 1
The last line is obtained by remarking that t ≤ k and j ≥ 0. Thus,

1 k ≤ 1 t -j . Therefore, (1 + 1 k ) j+1 (t -(j + 1)) = (1 + 1 k ) j (t -j)(1 + 1 k ) t -j-1
t -j ≤ t , which concludes the induction proof. At time t = k, all of the first k requests have been completed. We now prove that at any time t such that k ≤ t < n, Single-Unit starts request T t+1 . This would mean that at time t, only requests T i such that t < i ≤ t + k are ready and not completed. We prove by induction that at time k ≤ t < n, all requests T i with i ≤ t are completed. The induction basis with t = k is already proved above. Assume the hypothesis is true for a given k ≤ t < n, it remains to prove that at time

t = t + 1, T t+2 is started among requests T i such that t < i ≤ t + k. On one hand, w i (t + 1 -r i ) = w t+2 k for request T t+2 . On the other hand, for t + 1 < i ≤ t + k, w i (t + 1 -r i ) ≤ (1 + 1 k ) i-t -1 w t+2 (t + 1 -i + k). It remains to prove that (1 + 1 k ) i-t -1 w t+2 (t + 1 -i + k) < w t+2 k for t + 1 < i ≤ t
+ k and for a given k ≤ t < n. By changing variables (j = i -t -1), this corresponds to proving (1 + 1 k ) j (k -j) < k for all 0 < j < k. We show this again by induction on j for a given k ≥ 1. For the induction basis, (1

+ 1 k )(k -1) = k + 1 -1 -1 k < k.
For the induction step, we can show that (1 + 1 k ) k-j+1 k-j ≤ 1 by remarking that k > k -j, which concludes the induction proof.

To conclude on the performance of Single-Unit, request T i is started at time t = i -1 and therefore, the objective value is at least

w n F n = w n (C i -r i ) = kw n .
Optimal upper bound. A better objective value can be obtained by starting all requests as soon as they arrive except for the first k ones: request T 1 is started at time t = 0; then, request T i is started at time t = i -k for k < i ≤ n; finally, the remaining requests among the first k ones are started (T i is started at time t = n -k + i -1 for 1 < i ≤ k). We analyse the objective value for request T k because it the last one to be executed among the first k requests, and T n because it is the one with the highest weight among the last n -k requests. For T k ,

w k F k = k(C k -r k ) = kn. For T n , w n F n = w n .
We prove that w n ≥ kn by deriving a lower bound on w n . The weights increase in multiple stages. At first, each increment is unitary w i+1 = w i + 1 for k ≤ i < 2k. Then, the increment increases at the second stage and w i+1 = w i + 2 for 2k ≤ i < 2k + k/2 . At the kth stage, w i+1 = w i + k for a single request. At a given stage j, the increment of the weight is j for at most k/j requests. Let n 1 = k j=1 k/j be the number of such requests (assuming n -k ≥ n 1 ). Finally, the remaining n 2 = n -k -n 1 requests are incremented by a value that increases by at least 1 for each new request:

w i+1 ≥ w i + (k + i -n + n 2 ) for n -n 2 < i ≤ n.
The last weight w n is at least the sum of the increments of all these stages:

w n ≥ k + k j=1 j k/j + n2 j=1 (k + j). Thus, w n ≥ k(k + 1) + kn 2 + n 2 2 /2.
Our hypothesis is that w n ≥ kn, which would be verified if

k(k + 1) + kn 2 + n 2 2 /2 ≥ kn k(k + 1) + k(n -k -n 1 ) + (n -k -n 1 ) 2 /2 ≥ kn 2k(1 -n 1 ) + (n -k -n 1 ) 2 ≥ 0 n -k -n 1 ≥ 2k(n 1 -1) n ≥ k + n 1 + 2k(n 1 -1).
We bound n 1 using the asymptotic expansion of the harmonic number H k :

n 1 = k j=1 k/j < k j=1 k/j + k < k(H k + 1) < k(log(k) + γ + 1 2k + 1)
where γ ≈ 0.577 is the Euler-Mascheroni constant. What remains to be proved is

n ≥ k(log(k) + γ + 1 2k + 2 + 2(log(k) + γ + 1 2k + 1).
Numerical analysis shows this occurs when n = k 2 + 11, which proves that w n ≥ kn. Thus, the optimal objective is at most w n and the one achieved with Single-Unit is at least kw n , which concludes the proof.

Lower Bound

We have seen that our initial scheduling problem, with heterogeneous processing times, no preemption and in an online setting is far from being solvable or even approximable. This motivates the search of lower bounds to constitute a formal baseline and derive performance guarantees. One way to do this is to compute an optimal solution to the offline problem (or a relaxed version) and compare it to a solution generated by an online heuristic.

Maximum Weighted Flow with Machines Restrictions

We propose an ILP (Integer Linear Program) to solve the scheduling problem P |M i , r i , p i = 1| max w i F i , which we adapt to provide a lower bound to the case with arbitrary processing times p i . We denote allocations of requests to servers by x i,j,t : x i,j,t = 1 if request T i is allocated to server M j at time t, and 0 otherwise. x i,j,t is defined for each request T i ∈ T , for each server M j ∈ M i , for all times r i ≤ t ≤ τ where τ = max i r i + n, which yields at most nkτ variables. Then solving the Mixed Integer Linear Program (1) provides a solution to the scheduling problem:

we want to minimize the maximum weighted flow time (Equation (1a)); each request must be scheduled (Equation (1b), n constraints); each request starts after its release time (Equation (1c), n constraints); there is no simultaneous execution on the same machine (Equation (1d), kτ constraints).

minimize max i w i     j,t t • x i,j,t   + p i -r i   (1a) subject to ∀i, j,t x i,j,t = 1 (1b) ∀i, j,t t • x i,j,t ≥ r i (1c) ∀j, ∀t, i x i,j,t ≤ 1 (1d) x i,j,t ∈ {0, 1} (1e)
In order to constitute a lower bound for the problem with arbitrary processing times, we transform each non-unitary request and we adapt the objective. Each request T i arriving at time r i with a processing time p i is replaced with p i requests arriving at times r i , r i + 1, . . . , r i + p i -1. The new objective function to minimize is max w i (C i -r i ) where i is the index of the original non-unitary request.

Even if this ILP theoretically constitutes a lower bound to the considered problem, its resolution is unfortunately too difficult in practice for non-trivial instances that would be representative (n > 1000 requests). A more practical solution would require to reduce the number of variables and constraints, or relax the model to a pure Linear Program. Alternatively, we tackle these performance issues by focusing on another related problem that is polynomially solvable and provides a lower bound to our scheduling problem.

Maximum Weighted Flow with Preemption

The solution to R|r i , pmtn| max w i F i provides such a lower bound, which is found by performing a binary search on a Linear Program [START_REF] Legrand | Minimizing the stretch when scheduling flows of divisible requests[END_REF], followed by the reconstruction scheme from Lawler et al. [START_REF] Lawler | On preemptive scheduling of unrelated parallel processors by linear programming[END_REF]. In order to facilitate the work for the reader, we explain the complete procedure here (refer to the mentioned articles for details about the correctness).

The algorithm of Legrand et al. relies on the fact that this problem is equivalent to the deadline scheduling problem: we want to find the minimum objective f such that, when we fix a deadline d i (f ) = r i + f wi to each request T i , it is possible to find a feasible schedule where all requests are executed during their feasible interval [r i , d i (f )].

We define the ordered set of epochal times E f = {r 1 , . . . , r n , d 1 (f ), . . . , d n (f )}, for any value f . Each epochal time E f t has a position t in the set, and let T r (i) (resp. T d (i)) give the position of the value r i (resp.

d i (f )) in E f . Adjacent epochal times constitute time intervals [E f t , E f t+1 ] (of course, for t = 2n, the considered interval is [E f t , +∞[).
It is critical to observe that the relative ordering of epochal times only changes for specific values f , i.e., there is an ordered set {F k } ∈ 2 Q such that, for all k and for any f, g ∈]F k , F k+1 [, f < g, the relative ordering of E f is the same as the relative ordering of E g . Each F k is called a milestone, and corresponds to a value f for which one deadline is equal to the release time or the deadline of another request.

Computing milestones. We first need to get the set of milestones, i.e., all values f for which the relative ordering of epochal times E f changes. This happens when the deadline of a request T i coincides with the release time or the deadline of a request T j , with i = j. There are two cases to consider:

-d i (f ) = r j =⇒ r i + f wi = r j =⇒ f = w i (r j -r i ); -d i (f ) = d j (f ) =⇒ r i + f wi = r j + f wj =⇒ f = wiwj wj -wi (r j -r i ), where w i = w j 7 .
Solving in a milestone interval. Let F 1 and F 2 be two consecutive milestones. We want to know if there exists f ∈ [F 1 , F 2 ] such that a preemptive deadline scheduling is feasible, and if so, get the minimal one. Let x i,j,t be the fraction of T i processed by M j during the time interval [E f t , E f t+1 ]. Then the Linear System (2) provides a solution:

we want to minimize the objective value (Equation (2a)); each request must be completely processed (Equation (2b)); the total processing time during a time interval cannot exceed its capacity (Equation (2c)); the processing time of a request during a time interval cannot exceed its capacity, i.e., a given request cannot be simultaneously executed by several servers (Equation (2d)); a request cannot be executed before its release time (Equation (2e)); a request cannot be executed after its deadline (Equation (2f)).

minimize f (2a)

subject to ∀i,

j,t x i,j,t = 1 (2b) ∀j, ∀t, i x i,j,t • p i,j ≤ E f t+1 -E f t (2c)
∀i, ∀t,

j x i,j,t • p i,j ≤ E f t+1 -E f t (2d) ∀i, ∀j, ∀t, T r (i) ≥ t + 1 =⇒ x i,j,t = 0 (2e) ∀i, ∀j, ∀t, T d (i) ≤ t =⇒ x i,j,t = 0 (2f) F 1 ≤ f ≤ F 2 (2g)
Optimal solution. We have the set of milestones and a way to obtain the optimal solution in a milestone interval if there is one, hence we are able to find the globally optimal objective value by performing a binary search on the set of milestones. Let us now build the schedule from the provided solution f .

Schedule reconstruction.

Let us assume that we are considering the t-th time interval [E f t , E f t+1 ] (we will repeat the same procedure for all intervals, and concatenate the partial schedules). First, we build the m × n cost matrix A such that A j,i = x i,j,t • p i,j , which represents the fraction of request that should be executed during the current time interval. The procedure is to build iteratively the schedule by choosing a subset D of elements in A, called the decrementing set (at most one element per row and per column), and a time length δ at each step, until all elements of A are equal to zero. Let us construct the (m + n) × (m + n) bistochastic matrix

B = A D m D n A T ,
where A T is the transpose of A, and D m (resp. D n ) is an m × m (resp. n × n) diagonal matrix whose elements are such that each row sum and column sum of B is equal to

E f t+1 -E f t .
As stated by the Birkhoffvon Neumann theorem, each bistochastic matrix is a convex combination of permutation matrices, i.e., B = k c k P k , where each c k is a coefficient and P k is a (m + n) × (m + n) permutation matrix. The top-left m × n block of any P k gives a decrementing set D to schedule in the current iteration: if P kj,i = 1, then A j,i ∈ D, which means the request T i may be executed by the server M j .

We now compute the maximum processing time δ allowed during the current iteration. We denote a row j in A as tight if its sum is equal to E f t+1 -E f t , and slack otherwise. The same terminology is used for a column i. δ is chosen to be maximum subject to the following constraints:

-if A j,i ∈ D and is in a tight row or column, then δ ≤ A j,i ; -if A j,i ∈ D and is in a slack row, then δ ≤ A j,i + (E f t+1 -E f t ) -k A j,k ; -if A j,i ∈ D and is in a slack column, then δ ≤ A j,i + (E f t+1 -E f t ) -k A k,i ; -if row j contains no element of D, then δ ≤ (E f t+1 -E f t ) -k A j,k ; -if column i contains no element of D, then δ ≤ (E f t+1 -E f t ) -k A k,i .
Finally, for each A j,i ∈ D, T i is scheduled on M j as soon as possible for min(δ, p i,j ) time units, and A j,i is replaced by max(0, A j,i -δ) in A.

Online Heuristics

We recall that a solution to our problem consists, for each request, in choosing a server among the ones holding a replica of the requested data as well as a starting time for each request. These two decisions appear at different places in a real key-value store: the selection strategy R used by the coordinator gives a replica R(T i ) = M r , whose execution policy E r defines the request starting time E r (T i ) = σ i . This section describes several online replica selection heuristics and execution policies that we then compare by simulation.

Replica Selection

We consider several replica selection heuristics with different levels of knowledge about the cluster state. Some of these levels are hard to achieve in a real system; for instance, the information about the load of a given server will often be slightly out of date. Similarly, the information about the processing time can only be partial, as the size of the requested value cannot be known by the coordinator for large scale data sets, and practical systems generally employ an approximation of this metric, e.g., by keeping track of size categories of values using Bloom filters [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF]. However, we exploit this exact knowledge in our simulations to estimate the maximal performance gain that a given type of information allows. We now describe selection heuristics.

Random. The replica is chosen uniformly at random among compatible servers: M r = rand M i . This strategy has no particular knowledge.

LeastOutstandingRequests ( LOR). Let R(M j ) be the number of outstanding requests sent to M j , i.e., the number of sent requests that received no response yet. The chosen replica minimizes R(M j ):

M r = argmin Mj ∈Mi R(M j ).
It is easy to implement, as it only requires local information; in fact, it is one of the most commonly used in load-balancing applications [START_REF] Suresh | C3: Cutting tail latency in cloud data stores via adaptive replica selection[END_REF].

Héron. We also consider an omniscient version of the replica selection heuristic used by Héron [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF]. It identifies requests for values with size larger than a threshold, and avoids scheduling other requests behind such a request for a large value by marking the chosen replica as busy. When the request for a large value completes, the replica is marked available again. The replica is chosen among compatible servers that are available according to the scoring method of C3 [START_REF] Suresh | C3: Cutting tail latency in cloud data stores via adaptive replica selection[END_REF]. The threshold is chosen according to the wanted proportion of large requests in the workload.

EarliestFinishTime ( EFT). Let FinishTime(M j ) denote the earliest time when the server M j becomes available, i.e., the time at which it will have emptied its execution queue. The chosen replica is the one with minimum FinishTime(M j ) among compatible servers:

M r = argmin Mj ∈Mi FinishTime(M j ).
Knowing FinishTime is hard in practice, because it assumes the existence of a mechanism to obtain the exact current load of a server. A real system would use a degraded version of this heuristic.

EarliestFinishTime-Sharded ( EFT-S). For this heuristic, servers are specialized: we define small servers, which execute only requests for small values, and large servers, which execute all requests for large values and some requests for small values when possible (similarly to size-aware sharding [START_REF] Didona | Size-aware sharding for improving tail latencies in in-memory key-value stores[END_REF]). Each request for a large value is scheduled on large servers using the EFT strategy, while each request for a small value is scheduled on any server (small or large), also using EFT.

For the following experiments, we define large servers as the set of servers {M b } 1≤b≤m such that b mod k = 0 (recall k is the replication factor). This makes sure that one server in each interval M i is capable of treating requests for large values. We define a threshold parameter ω to distinguish between requests for small and large values: requests with duration larger than ω are treated by large servers only, while others can be processed by all available servers.

We derive the threshold ω from the size distribution. In the best case, when all servers in each interval of replicas are perfectly balanced, requests for small values are scheduled on small servers only and requests for large values on large servers only. It means that the total work is k times larger than the work on large servers on average. Let X be the random variable that models the size distribution, and f X denote its probability density function. We denote by p(X) = αX + β the duration of the corresponding request (where α is the inverse of the bandwidth and β the latency), and by p ω (X) the duration if it is a large value (and zero otherwise), that is:

p ω (X) = p(X) if p(X) ≥ ω 0 otherwise.
Then, the expected work on large servers when any request is submitted is

E [p ω (X)] = ω-β α x=0 0f X (x)dx + ∞ x= ω-β α (αx + β)f X (x)dx.
It should be equal to the expected work when any request is submitted, E[p(X)], divided by k. This leads to finding ω such that:

E [p ω (X)] = 1 k • E[p(X)]. (3)
This heuristic has to be able to distinguish requests for small and large values with respect to ω; it could be achieved in practice with combined Bloom filters, in a similar fashion than Héron [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF].

StaticWindow ( SW). The requests are no longer scheduled on reception, but every q time units, where q is a parameter of the heuristic. The set Q denotes the requests received during this window of q time units. Let t be the time at which requests from Q must be scheduled (requests with t < r i ≤ t + q form the next batch and must be scheduled at time t + q). We assume here a centralized system, where a unique scheduler receives and schedules all requests. The underlying idea is to be able to make choices based on more exhaustive workload information than previous greedy heuristics. This heuristic must therefore also decide the order in which the requests of Q are scheduled. We derive two versions.

Sufferage-SW (SSW). This strategy is inspired from the Sufferage heuristic [START_REF] Maheswaran | Dynamic mapping of a class of independent tasks onto heterogeneous computing systems[END_REF]. Let F be the function giving the estimated weighted flow F(T i , M j ) = w i (max(r i , FinishTime(M j ))+p i -r i ) of T i when scheduled on M j as soon as possible. Let ρ(T i ) = argmin Mj ∈Mi F(T i , M j ) be the "best" server for T i , i.e., the one minimizing its weighted flow, and ρ (T i ) = argmin Mj ∈Mi\ρ(Ti) F(T i , M j ) be the "second best" server for T i . Then, we define the sufferage value Suff(T i ) = F(T i , ρ (T i )) -F(T i , ρ(T i )) > 0 as the difference of weighted flow values on ρ (T i ) and ρ(T i ). The request we choose to schedule is the one which suffers the most if we schedule it on its second best server:

T s = argmax Ti∈Q Suff(T i ).
The chosen replica is ρ(T s ):

M r = ρ(T s ) = argmin Mj ∈Mi F(T s , M j ).
Algorithm 3 Sufferage-SW 1: loop {every q time units} 2: for all Ti ∈ Q do 3:

ρ(Ti) ← argmin M j ∈M i F(Ti, Mj) 4: ρ (Ti) ← argmin M j ∈M i \ρ(T i ) F(Ti, Mj) 5: Suff(Ti) ← F (Ti, ρ (Ti)) -F (Ti, ρ(Ti)) 6:
end for 7:

while Q is not empty do 8:

Ts ← argmax T i ∈Q Suff(Ti) 9:

schedule Ts on ρ(Ts) 10:

Q ← Q \ {Ts} 11:
update ρ, ρ and Suff 12:

end while 13: end loop Request T s is then removed from Q, and we update sufferage values of remaining requests. Algorithm 3 describes this procedure. This strategy runs in time O(n 2 • m) and uses a space O(n) per time window.

MaxMin-SW (MSW)

. This strategy is inspired from the Max-Min heuristic [START_REF] Maheswaran | Dynamic mapping of a class of independent tasks onto heterogeneous computing systems[END_REF]. We build a matrix Mat whose rows are requests of set Q and columns are servers, where

Mat[T i , M j ] = F(T i , M j ) if M j ∈ M i +∞ otherwise.
The best weighted flow of request

T i is F best (T i ) = F(T i , ρ(T i )) = min Mj ∈M Mat[T i , M j ].
Then, we schedule the request T s whose "best" objective value is the highest:

T s = argmax Ti∈Q F best (T i ).
The chosen replica minimizes the objective value of T s :

M r = argmin Mj ∈M Mat[T s , M j ].
The request T s is then removed from the set Q, as well as the related row in the matrix Mat, and the column M r is updated with new values. These operations are repeated until Q is empty (see Algorithm 4). This strategy runs in time O(n 2 • m) and uses a space O(n • m) per time window.

Table 1 summarizes the properties of our heuristics.

Local Queue Scheduling Policies

We now present scheduling policies locally enforced by replicas. Each replica handles an execution queue Q in which coordinators send requests, and then decides of the order of executions. In a real key-value store, these policies should be able to extract exact information on the local values, and in particular their sizes, as a single server is responsible for a limited number of keys. We consider the following local policies.

FirstInFirstOut ( FIFO). This is a classic strategy, which is commonly used as a local scheduling policy in key-value stores (e.g., Cassandra [START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF]). The requests in Q are ordered by non-increasing insertion time, i.e., the first request that entered the queue (the one with the minimum r i ) is the first to be executed.

MaxWeightedFlow ( MWF). We propose another strategy, which locally reorders requests. When the server becomes available at time t, the next request T s to be executed is the one whose weighted flow is the highest:

T s = argmax Ti∈Q w i (t + p i -r i ).
This is a general execution policy that considers the request weights as defined by the system designer. In any case, starvation is not a concern: focusing on the maximum weighted flow ensures that all requests will eventually be processed. Note that when coupled with the stretch metric (w i = 1 pi ), MWF is equivalent to the strategy that selects the request with maximal stretch (MaxStretch). This favors requests for small values in front of requests for large ones, and thus is a way to mitigate the problem of head-of-line blocking.

Table 2 summarizes the properties of our heuristics.

Simulations

We analyse the behavior of previously described strategies and compare them with each other in simulations.

We built a discrete-event simulator based on Python 3.8 and salabim 21.0.1 for this purpose, which mimics a real key-value store: coordinators receive user requests and send them to replicas in the cluster, which execute these requests. Each request is first headed to the queue of a server holding a replica of the requested data by the selection heuristic. Then, the queue is reordered by the local execution policy and requests are processed in this order.

Workload and Settings

We designed a synthetic heterogeneous workload to evaluate our strategies: value sizes follow a Weibull distribution with scale η = 32 000 and shape θ = 0.5, which gives a mean size equal to η • Γ 1 + 1 θ = 64 kB, where Γ is the gamma function; these parameters yield a long-tailed distribution that is consistent with existing file sizes characterizations [START_REF] Feitelson | Workload modeling for computer systems performance evaluation[END_REF]. User requests arrive at coordinators according to a Poisson process with arrival rate λ = mL/p, where m is the number of servers, L is the wanted average server load (defined as the average fraction of time spent by servers on serving requests), and p = α • η • Γ 1 + 1 θ + β is the mean processing time of requests. Each key has the same probability of being requested, i.e., we do not model skewed popularity. The cluster consists in m = 15 servers and we set the replication factor to k = 3, which is a common configuration in real implementations [START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF][START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF]. The network bandwidth is set to 1/α = 100 Mbps and the average latency is set to β = 1 ms. For the threshold between requests for small and large values, we plug the density function of our Weibull distribution in Equation ( 3) and solve it numerically for ω:

∞ x= ω-β α (αx + β) • θ η • x η θ-1 • e -( x η ) θ dx = 1 k • ∞ x=0 (αx + β) • θ η • x η θ-1 • e -( x η ) θ
This yields a threshold ω ≈ 26.4 ms (for a size of 318 kB), resulting in a proportion of 5 % of requests for large values in the workload. Each experiment is repeated on 10 different scenarios; a given scenario defines the processing times p i , the release times r i , and the replication groups M i according to described settings.

Weight Values

We recall that each request in our model is associated to a weight value w i . Thus far, we considered these weights to be completely arbitrary. We now describe and explain the values we used in our simulations:

w i = 1 for all T i ∈ T . This is the classic flow time (or latency) metric.

w i = 1 pi . Pure latency tends to favor large requests over the small ones. One way to work around this behavior is to consider the stretch (weighting the latency with the processing time): it measures the slowdown of a request, i.e., the cost for sharing resources with other requests.

-w i = 1 √ pi .
Although the stretch metric is more fair than pure latency, we noted in some experiments that it tends to be inappropriate under heterogeneous workloads where the majority of requests are small. Small requests are too favored. For instance, if a small request of 1 ms and a large request of 100 ms have a stretch value of 2, then the large request can tolerate a 100 ms delay (F i = 200), whereas the small one can only tolerate a 1 ms delay (F i = 2). Yet it seems reasonable to delay small requests a little more to avoid impacting the large ones too much. This weighting seems to be a tradeoff between latency and stretch metrics, and we denote it as the "weak stretch".

Results

Fig. 3 shows Empirical Cumulative Distribution Functions (ECDF) of the flow, the stretch and the weak stretch, for each combination of distributed selection heuristic and execution strategy. The dashed horizontal lines respectively represent median, 95 th and 99 th percentile. Files are requested with a load L = 0.9, and the simulations run for 120 s. We show in Fig. 4 ECDF of window-based strategies when servers are subject to a burst, i.e., the arrival rate is very high and the average load is greater than 1. We measure the metrics with average load values ranging from 1.0 to 9.0, combined to a FIFO execution. For SSW and MSW, we consider the stretch weighting (w i = 1 pi ), to favor small requests that are in majority in the workload. We recall that these heuristics are centralized, i.e., all requests are scheduled by one coordinator, and the time window is set to q = 100 ms. The simulations run over 3 s in order to simulate a short burst of requests.

Fig. 5 shows the 99 th quantiles of each metric as a function of average server load for each combination of selection and execution heuristics and for load values ranging from 0.5 to 0.9. In this context, the maximum of the distribution is impacted by rare events of varying amplitude, which makes this criterion unstable. The stability of the 99 th quantile allows comparing more confidently the performance between scenarios with identical settings. For the local execution policy MWF, we discard the case w i = 1/ √ p i , as it exhibits performances always worst than the case w i = 1/p i . The simulations run for 120 s.

The comparison of online heuristics with the lower bound introduced in Section 5 is shown in Fig. 6. We normalize the maximum objective max w i F i generated by a given heuristic with the lower bound. Each boxplot8 represents the distribution of these normalized maximums among 10 different scenarios, for each combination of strategies. Horizontal red bars help to locate the lower bound. Files are requested with a load L = 0.9, and the 10 scenarios are solved over 1200 requests.

The first thing to note in Fig. 3-6 is that the choice on replica selection heuristic is indeed critical for read latency, as the 99 th quantile can often be improved by a factor 2 compared to state-of-the-art strategies LOR and Héron, without increasing median performance as confirmed in Fig. 3. This highlights the fact that some properties of the cluster and the workload are more suitable to taming tail latency; in particular, knowing the current load of a server, and thus its earliest available time, allows implementing the EFT strategy and getting very close to the lower bound (Fig. 6).

Fig. 6 also shows that EFT yields the most stable maximums between scenarios, as more than 50 % of normalized max-flow range from 1.0 to 1.15, in particular when coupled with FIFO. This improves the confidence that this strategy will perform close to optimal in a majority of cases, and cannot be significantly improved. On the opposite, when considering the stretch, the gap between the best achieved performance and the lower bound increases significantly. It is yet unclear whether this is because the lower bound is far from the optimal as it exploits migration, or whether the proposed heuristics are not the best suited to the stretch metric, even if EFT-S shows the best results. On a side note, the effect of switching from FIFO to MWF and the relative performance between the heuristics are consistent with Fig. 5.

For the stretch metric, where latencies are weighted by processing times, EFT-S performs even better than EFT (Fig. 3,5), yielding a 99 th quantile of 30 (resp. 18) when coupled with FIFO (resp. MWF (w i = 1/p i )). This is due to the nature of EFT-S that favors requests for small values, which are in majority in the workload. However, EFT-S does not perform well for the last quantiles in the latency distribution; this corresponds to the 5 % of requests for large values that are delayed in order to avoid head-of-line blocking situations. Fig. 3, 5-6 also illustrate the significant impact of local execution policies on the stretch metric: local reordering according to MWF (w i = 1/p i ) favors requests for small values, which results in an improvement for all selection strategies, even on the median values. Note that this does not necessarily improve latency, as FIFO is well-known to be the optimal strategy for max-flow on a single machine [START_REF] Bender | Flow and stretch metrics for scheduling continuous job streams[END_REF]. It is confirmed by our observations, as MWF worsen the tail-latency.

When a burst occurs, Fig. 4 shows the value of our window-based heuristics. Interestingly, these replica selection strategies do not benefit a lot from centralized and global information about the workload, and are not even effective for realistic load values. When the average load exceeds 300 % (L ≥ 3) we see that ECDF of EFT and SSW or MSW are similar, but the window-based heuristics never outperforms EFT. This seems to confirm that EFT is a close-to-optimal strategy in average, as additional information do not allow to increase performance.

Conclusion

This study defines a formal model of a key-value store in order to derive maximal performance achievable by a real online system, and states the associated optimization problem. We also provide theoretical results on various problems related to our main scheduling problem. After showing the difficulty of this problem, we describe some investigations on a lower bound. We develop online heuristics and compare them with stateof-the-art strategies such as LOR, Héron [START_REF] Jaiman | Héron: Taming tail latencies in key-value stores under heterogeneous workloads[END_REF] or size-aware sharding [START_REF] Didona | Size-aware sharding for improving tail latencies in in-memory key-value stores[END_REF] using simulations. This allows understanding more finely the impact of replica selection and local execution on performance metrics. We hope that our work will help practitioners draw new scheduling strategies. We plan to continue to improve on a lower bound, for example by using resource augmentation models [START_REF] Choudhury | Rejecting jobs to minimize load and maximum flow-time[END_REF][START_REF] Kalyanasundaram | Speed is as powerful as clairvoyance[END_REF], and we propose to formally analyze EFT with various techniques such as competitive analysis. We wish to study the effect of various assumptions on scheduling, e.g., the impact of skewed key popularity, and to extend the model with multi-get operations [START_REF] Jaiman | TailX: Scheduling heterogeneous multiget queries to improve tail latencies in key-value stores[END_REF][START_REF] Reda | Rein: Taming tail latency in key-value stores via multiget scheduling[END_REF].

A Notations

Table 3 provides a list of the most used notations in this document. -ε ) [START_REF] Kellerer | Approximability and nonapproximability results for minimizing total flow time on a single machine[END_REF] no no -FIFO is P -comp [START_REF] Legrand | Minimizing the stretch when scheduling flows of divisible requests[END_REF] 1|ri, pmtn| Fi SRPT [START_REF] Baker | Introduction to sequencing and scheduling[END_REF] yes no -SRPT is 1-comp 1|ri, pmtn| Fi SRPT no 1 + ε -SETF is (1 +

Fig. 1 .

 1 Fig. 1. Data partitioning and replication on a cluster of 6 servers.

2 - 1

 21 is derived. Saule et al. improved this result by showing a lower bound of 1 2

Fig. 3 .

 3 Fig.3. ECDF of each metric generated by each combination of selection and execution heuristics in steady-state over 120 s, for an average load L = 0.9.

Fig. 4 .Fig. 5 .Fig. 6 .

 456 Fig. 4. ECDF of each metric generated by SSW and MSW in steady-state over 3 s, for load values ranging from 1.0 to 9.0.

Ti ∈ TTable 5 .

 5 request i Mj ∈ M machine j Mi ⊆ M machines able to process Ti K l ∈ K key l V l ∈ Vvalue l wi weight of Ti ri release time of Ti pi processing time of Ti σi start time of Ti Ci completion time of Ti C S i completion time of Ti in a schedule S Fi flow time of Ti Si stretch of Ti z l size of V l ϕ : T → K gives the key carried by a request Ψ : M → 2 K gives the key subset owned by a machine Π : T → M × N gives the machine and start time of a request α bandwidth inverse β latency λ submission rate η Weibull scale parameter θ Weibull shape parameter L load Competitive analysis results related to sum (weighted) flow minimization.

Table 1 .

 1 Properties of replica selection heuristics. ACK-Done denotes the need to acknowledge the completion of sent requests. FinishTime is the knowledge of available times of each server. pi denotes the processing times of local requests and ri their release times. n is the number of requests in Q and m is the total number of servers.

	Heuristic Knowledge	Type	Complexity
	Random None	Distributed O(1)
	LOR ACK-Done	Distributed O(m)
	Héron ACK-Done, pi ≥ ω Distributed O(m)
	EFT FinishTime	Distributed O(m)
	EFT-S FinishTime, pi ≥ ω Distributed O(m)
	SSW FinishTime, pi, ri Centralized O(n 2 • m)
	MSW FinishTime, pi, ri Centralized O(n 2 • m)

Table 2 .

 2 Properties of local queue scheduling heuristics. pi denotes the processing times of local requests and ri their release times. N is the number of local requests in Q and m is the total number of servers.

	Heuristic Knowledge Complexity
	FIFO None	O(1)
	MWF pi, ri	O(N )

Table 3 .

 3 List of notations.

	Symbol Definition
	i index of requests
	j index of machines
	l index of keys/values
	n number of requests
	m number of machines
	c number of keys/values
	k replication factor

Table 6 .

 6 Competitive analysis results related to p-norm minimization.

	Offline		Online
	Problem Opt Approx C S	Competitive Ratio
	1|ri, pmtn| Fi p	yes no	-no n o(1) -comp algo [11]
	1|ri, pmtn| Fi p 1|ri, pmtn| Fi p	yes 1 + ε -SJF is O( 1 ε )-comp -SRPT is O( 1 ε )-comp [11] no 1 + ε -SETF is O( 1 ε 2+2/p )-comp [11]
	1|ri, pmtn| Si p	yes no	-no n o(1) -comp algo [11]
	1|ri, pmtn| Si p	yes 1 + ε -SJF is O( 1 ε )-comp -SRPT is O( 1 ε )-comp [11]
	1|ri, pmtn| Si p	no 1 + ε -LB: Ω(min(n, log P ))
			-SETF is O(	1 ε 3+1/p log 1+1/p P )-comp [11]

https://doi.org/10.6084/m9.figshare.13114196

We express non-migratory preemption as pmtn * in the β-part, not to be confused with the classic pmtn constraint.

With different release times, two deadlines will never coincide if wi = wj.

Each boxplot consists of a bold line for the median, a box for the quartiles, whiskers that extend at most to 1.5 times the interquartile range from the box and additional points for outliers.

B Competitive Analysis

We provide a comprehensive summary of results related to competitive analysis of online scheduling problems which address the minimization of flow time. Tables are organized in the following way: the results are classified into two general columns, where the first one deals with offline optimization and the second one deals with online aspects, and more particularly with competitiveness.

We specify the modalities for each competitive analysis, where the symbol C indicates a clairvoyant problem (i.e., all request properties are known) and S stands for a speed augmentation characteristic (i.e., the online solution is allowed to use an increased fraction of resources compared to offline). Table 4 (resp. Table 5) presents competitive analysis results which are related to maximum (resp. sum) weighted flow. The more general objective p -norm is described in Table 6.

Table 4. Competitive analysis results related to maximum (weighted) flow minimization.

Offline Online

Problem Opt Approx C S Competitive Ratio 1|ri|Fmax FIFO [START_REF] Bender | Flow and stretch metrics for scheduling continuous job streams[END_REF] no no -FIFO is 1-competitive

C Problems Complexity

To complete our survey on scheduling problems related to our study, we present a summary on complexity results in Table 7.