Hagar Meir

Dmitry Basin

Edward Bortnikov

Anastasia Braginsky

Idit Keidar

Gali Sheffi

Oak -A Key-Value Map for Big Data Analytics

Keywords: 2012 ACM Subject Classification Theory of computation → Data structures and algorithms for data management, Theory of computation → Concurrent algorithms shared memory, concurrency, memory management Digital Object Identifier 10.4230/LIPIcs.CVIT.2016.23

ou non, émanant des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

Introduction

Real-time analytics applications are on the rise in the big data world. Modern decision support and machine intelligence engines strive to continuously ingest large volumes of data, while providing up-to-date insights with minimum delay. Serving both updates and queries within the same system is key for achieving this goal. This is in stark contrast to the legacy approach, which addressed data acquisition and analysis through different systems, inflating both the time-to-insight and operational costs. One example of such a new-generation system is Oath's Flurry Analytics [START_REF]Flurry analytics[END_REF]. It provides mobile developers with analysis tools to explore user characteristics (age, gender, location, app context, etc.) and behavior, e.g., which code paths they follow and how they churn. As of late 2017, the Flurry SDK is installed on 2.6B devices, and monitors 1M+ mobile apps [20].

The user function provided in a compute operation expects a view of this handle so it can run computational steps on the value with no concern of concurrency control. In addition, all reads are directed by Oak to the handle, so Oak's get operation returns a view of this handle. The handle assures the atomicity of each read performed on the returned view.

Once a value is removed from Oak, the handle assures that no thread will attempt to read this value, since, in the case of off-heap, that memory may be reclaimed. To this end, the handle has a remove method that performs a logical remove by marking the handle as deleted.

A key is deemed present in Oak only if it is associated with a non-deleted handle. All other handle operations check if the handle is deleted before taking any action. In the off-heap implementation, the handle's remove method is also responsible for memory reclamation.

The handle further offers put and compute methods that are used by Oak to replace and update values, respectively. The handle's put method directs the handle to point to the given value. In case Oak handles memory management on behalf of the application, it first allocates space for the new value and writes the value to it, and also reclaims the replaced value. Alternatively, Oak can be implemented to use a user-provided lambda expression that writes the value into a dedicated ByteBuffer provided by Oak.

Keys are also ByteBuffers, therefore, the user may provide a key comparator, since the keys are often serialized into bytes and the regular ByteBuffer comparison might not be appropriate.

In addition, Oak follows the interface suggested by Java NavigableMap and provides methods for creating subMaps to support non-atomic iterations over different ranges, and descendingMap to allow descending iterators.

Off-heap

Oak supports keeping the keys and values in off-heap memory, in the form of Java ByteBuffer, and provides memory allocation and reclamation for these keys and values. Oak implements epoch-based memory reclamation, as described in Section 3.3.

To this end, we need an indication of the end of each operation. Unlike other operations, the end of the iterator operation is entirely in the user's hand. To allow its discovery, Oak's iterator implements the Java AutoCloseable Interface, which adds a close method to the iterator. We refer to this new iterator as a closeableIterator. We describe the close method of the closeableIterator in Section 3.3.

While Oak provides memory reclamation, the users may supply Oak with methods for allocation and deallocation of the memory that best match their intended usage of Oak. In our implementation we also provide simple allocation and deallocation methods.

Oak algorithm

Oak implements a concurrent key-value map supporting various atomic (linearizable) read and update operations, and non-atomic ascending/descending iterators over the map and sub-maps. Oak is designed for standard hardware and uses widely available atomic hardware operations like CAS, F&I, and F&A. Oak keys and values are variable sized, and updates are done in-place which is important for large values.

Data organization

Chunks and index. Oak's structure is chunk-based, it is organized as a linked list of large blocks of contiguous key ranges, as suggested in [START_REF] Braginsky | Locality-conscious lock-free linked lists[END_REF]. The chunk object has a dedicated rebalance procedure, which splits chunks when they are over-utilized, merges chunks when they are under-utilized, and reorganizes the chunks' internals [START_REF] Basin | Kiwi: A key-value map for scalable real-time analytics[END_REF]. To allow fast access, we follow the approach of [START_REF] Basin | Kiwi: A key-value map for scalable real-time analytics[END_REF][START_REF] Braginsky | Cbpq: High performance lockfree priority queue[END_REF][START_REF] Herlihy | A simple optimistic skiplist algorithm[END_REF][START_REF] Herlihy | The Art of Multiprocessor Programming[END_REF]33] and add an index that maps keys to chunks, as illustrated in Figure 1a. Each chunk is indexed according to the minimal key it holds, which is invariant throughout its lifespan. The index is updated in a lazy manner, and so it may be inaccurate, in which case, locating a chunk may involve a partial traversal of the chunk linked list (see [START_REF] Basin | Kiwi: A key-value map for scalable real-time analytics[END_REF] and [33] for details). The index supports the following methods: lookup, lower, insert, and remove. Lookup is used by operations to locate the relevant chunk. Lower is a lookup that returns a chunk with the greatest minimal key that is strictly lower than the given key, which is used by the descending iterator (elaborated in Section 3.2). Insert and remove are exclusively used by rebalance. In addition, Oak provides the locateChunk(k) method, which returns the chunk that may hold key k, by querying the index and possibly a chunks linked list traversal.

Intra-chunk organization. As shown in Figure 1b, chunks hold three types of objects: entries, keys, and handles, as we explain shortly. Entries are organized as an array-based linked list, sorted in ascending key order. Each entry holds a pointer to a key, a pointer to a handle, and the index of the entry that holds the next key in the linked list. Oak makes sure that a key does not appear in more than once entry. Keys are variable size so each entry holds a pointer to the beginning of a key and its size. Keys are stored in a large ByteBuffer within the chunk while values reside outside the chunk.

We follow the approach of KiWi [START_REF] Basin | Kiwi: A key-value map for scalable real-time analytics[END_REF], whereby the rebalancer creates a new chunk with some prefix of the entries array filled with data, and the suffix consists of empty entries for future allocation. This prefix is initialized sorted, that is, the linked list successor of each entry is the ensuing entry in the array. The sorted prefix can be searched efficiently using binary search. As entries are added, the order in the remainder of the chunk is not preserved, i.e., the ordered prefix usually does not grow. When a new entry is inserted, it resides in the first free cell and connected via a bypass in the sorted linked list. We note that in case the insertion order is random, inserted entries are most likely to be distributed evenly in between the ordered prefix entries, thus creating fairly short bypasses. Given that the prefix and the remainder are of similar sizes, the expected search time remains polylogarithmic.

Nevertheless, in the worst-case, the search time is linear in the size of the remainder of the chunk.

Handles and OakBuffers.

There is a single handle per value, and once a key-value pair is removed from Oak, its handle is deleted and never reused (subjected to GC). Handles are stored within the chunk as shown in Figure 1b.

Since values are ByteBuffers, and all read and update operations are done on the handle, the handle must also have a ByteBuffer interface. To this end, we define oakBuffers and writableOakBuffers with the same interface as a read-only ByteBuffer and a ByteBuffer, respectively, which serve as views to a handle, as demonstrated in Figure 1c. OakBuffers are created on demand by operations that need access to the value; they are ephemeral, and cease to exist once the operation is completed. Chunk objects and rebalancing. A chunk object exposes the following API: lookUp, allocateEntryAndKey, allocateHandle, entriesLLputIfAbsent, writeValue, publish, and unpublish. We now describe their operation; their implementation is straightforward and so we omit it from the pseudo-code. LookUp searches for an entry corresponding to the given key. This is done by first running a binary search on the entries array prefix and continuing the search by traversing the entries linked list. There is at most one relevant entry. AllocateEntryAndKey allocates a new entry and also allocates and writes the given key that it points to. AllocateHandle allocates a new handle. Both of these allocations are done by using atomic hardware operations like F&A and F&I, so that the same space will not be allocated twice. After allocating a new entry we try to link this new entry into the entries linked list by calling entriesLLputIfAbsent, which uses CAS for safe insertion to the linked list, so that the invariant of a key not appearing more than once in Oak is preserved. If it encounters a linked entry with the same key (added by a concurrent insertion operation), then it returns the encountered entry. WriteValue has the given handle point to the given value. In case Oak handles memory management on behalf of the application, it first allocates space for the value (outside of the chunk) and writes the value to it.

C V I T

The allocations procedures (allocateEntryAndKey and allocateHandle) are called by Oak's update operations and "under the hood" may trigger a rebalance. The rebalance procedure reorganizes, compacts, and splits chunks, and merges multiple chunks. If a rebalance is triggered then the allocate procedure returns false and the update operation is retried. The rebalancer in Oak is implemented exactly as in KiWi [START_REF] Basin | Kiwi: A key-value map for scalable real-time analytics[END_REF] and similarly to [START_REF] Braginsky | A lock-free b+tree[END_REF]. Since it is not novel and orthogonal to our contributions, we do not explain it here.

Update operations inform the rebalancer of the action they are about to perform on the chunk by calling the publish method. The KiWi rebalancer helps published operations.

Alternatively, a rebalancer may wait as long as there are published operations (in an RCU-like manner), and reorganize the block once they complete. For simplicity, in Section 3.2 we assume the rebalancer does not help published operations, and hence we always retry an operation upon failure. When the update operation has finished its published action, it calls unpublish. We implement the publish and unpublish methods as in KiWi [START_REF] Basin | Kiwi: A key-value map for scalable real-time analytics[END_REF].

Update operations can also help the rebalance procedure; for simplicity, when describing the algorithm in Section 3.2 we omit this helping. That is, in Section 3.

Oak operations

In this section we describe Oak's operations. First we discuss queries, namely get and iterators. Next we describe Oak's update operations. We separate them into two types: insertion operations -put, putIfAbsent, and putIfAbsentComputeIfPresent are given in Algorithm 2, whereas remove and computeIfPresent, which take actions only when the affected key exists in Oak, are given in Algorithm 3.

Queries

Algorithm 1 get 1: procedure get(key)

2:

C, ei, hi, handle ← ⊥

3:

C ← locateChunk(key)

4:

ei ← C.lookUp(key)

5

:

if ei = ⊥ then hi ← C.entries[ei].hi 6: if hi = ⊥ then handle ←C.handles[hi] 7:
if handle = ⊥ ∨ handle.deleted then return null

8:

else return new OakBuffer(handle)

The get operation (Algorithm 1) returns a read-only view (oakBuffer) of the handle that holds the value that is mapped to the given key, in accordance with our zero-copy policy.

Since it is only a view and not a copy of the value, if the value is then updated by a different operation, the view will refer to the updated value. The returned view cannot be used for updating the value since it is a read-only view. Consistency is achieved by making sure that ByteBuffer operations performed on the oakBuffer are thread-safe, for example, we can use a read-write lock in the handle for concurrency control. Furthermore, a concurrent operation can remove the key from Oak, in which case the handle will be marked as deleted;

any operation using the handle checks this deleted flag.

We first locate the relevant chunk (line 3). We then call lookUp (line 4) to search for an entry with the given key. Finally, we check if the handle is deleted (line 7). If an entry holding a valid and non-deleted handle is found, we create a new oakBuffer that points to the handle and return it. Otherwise, get returns null. The ascending iterator begins by locating the first chunk with a relevant key, in the scanned range using the index, and, if needed, the chunks linked list. It then traverses the entries within each relevant chunk using the intra-chunk entries linked list, and continues to the next chunk in the chunks linked list. For every entry the iterator encounters, it returns a key and an oakBuffer only if the handle index is not ⊥. Otherwise, the iterator continues onto the next entry.

C V I T

The descending iterator begins by locating the last relevant chunk. Within each relevant chunk, it first traverses the entries linked list until the last relevant entry. For easy access to the previous entry, the iterator saves a stack of entries traversed since the last relevant entry that belongs to the chunk's ordered prefix. After exhausting the stack and reaching a prefix entry, the iterator simply proceeds to the previous prefix entry (one cell back in the array)

and rebuilds the stack with the linked list entries in the next bypass.

Figure 2a shows an example of an entries linked list; its ordered prefix ends with 9.

Figure 2b shows the stacks constructed during the traversal. In order to traverse all the entries in descending order we start at the last entry in the linked list. The entry with key 9, the last entry in the linked list, is within the prefix and does not have a next entry, so we can return it. Next we move one entry back in the prefix, to entry 6, and traverse the linked list until returning to an already seen entry within the prefix (9 in this case), while creating the stack 8 → 7 → 6. We then pop and return each stack entry. Now, when the stack is empty, we again go one entry back in the prefix and traverse the linked list. Since after 5 we reach 6, which is also in the prefix, we can stop and return 5. Finally, we reach 2 and create the stack with entries 4 → 3 → 2, which we pop and return.

When exhausting a chunk, the descending iterator continues by querying the index again, but now for a the chunk with the greatest minimal key that is strictly less than the current chunk's minimal key. After reaching a new chunk pointed by the index, we again traverse the chunks linked list until finding the last chunk with a minimal key that is less than the current chunk's minimal key, while making sure that we do not overshoot the last returned value.

By RB1-3 and from the iterators algorithm as described above it is easy to see that the iterators guarantee the following:

1. An iterator returns only keys and values that were inserted to Oak before the start of the iteration and not removed until the end of the iteration.

2. An iterator does not return keys and values that were removed from Oak before the start of the iteration and not inserted until the end of the iteration.

3.

Iterators do not return the same key more than once.

Note, that keys inserted or removed concurrently with an iteration may be either included

Insertion operations

Algorithm 2 Oak's insertion operations The three insertion operations: put, putIfAbsent, and putIfAbsentComputeIfPresent, associate the specified key with the given value if it is not already associated with a value; putIfAbsent returns true in this case. In case the key is present, putIfAbsent returns false, put associates the new value with the key, and putIfAbsentComputeIfPresent computes a new value to be associated with the key. All three use the doPut function.

DoPut first locates the relevant chunk and searches for an entry. We then distinguish between two cases: if a non deleted handle is found (case 1: lines 23 -28) then we say that the key is present. In this case, putIfAbsent returns false (line 24), put calls handle.put (line 25) to associate the new value with the key, and compute calls handle.compute (line 26).

These atomic handle operations can be implemented using a read-write lock, and they return false if the handle is deleted (due to a concurrent remove), in which case we retry (line 27). We complete the insertion by using CAS to make the entry point to the new handle index (line 37). Before doing so, we publish the operation (as explained in Section 3.1), which can also lead to a retry (line 36). After the CAS, we unpublish the operation, as it is no longer pending (line 38). If CAS fails, we retry the operation (line 39).

C V I T

To see why we retry, observe that the CAS may fail because of a concurrent non-insertion operation that sets the handle index to ⊥ or because of a concurrent insertion operation that sets the handle index to a different value. In the latter case, we cannot order (linearize) the current operation before the concurrent insertion, because the concurrent insertion operation might have been a putIfAbsent, and would have returned false had the current operation preceded it.

Notice that the value can be given in the form of a lambda expression, and it is used only when we need to write a new value (line 34). This way we can avoid constructing the value when the key is present in putIfAbsent and putIfAbsentComputeIfPresent.

Non-insertion operations

We now discuss the second type of updates -computeIfPresent and remove, which do not insert new entries. Both invoke the doIfPresent function.

The computeIfPresent operation (Algorithm 3) computes a new value to be associated with the given key only if the key-value pair is present. We first locate the handle; if there is no such handle, we return false (line 51). If the handle exists and is not deleted (case 1), we run handle.compute and return true if it is successful (line 56). Otherwise (case 2), compute makes sure that the key is in fact removed by preforming CAS to change handle index to ⊥ (line 63). Since this affects the chunk's entries, we need to synchronize with a possibly ongoing rebalance, and so here too, we publish before the CAS and unpublish when done. If publish or CAS fails then compute retries (lines 62 and 66). Compute returns false in case

(1) it does not find the entry, or finds the entry but with ⊥ as its handle index (line 51), or

(2) CAS to ⊥ is successful (line 67).

Remove also uses doIfPresent. If a non-deleted handle exists (case 1), like computeIf-Present, it updates the handle (in this case, marking it as deleted), and we say that it is a successful remove. However, unlike computeIfPresent, it then performs a second task, namely, marking the appropriate entry's handle index as ⊥. The former task is done by calling handle.remove (line 58), which marks the handle as deleted. Again this operation can be implemented using a read-write lock. The latter is done in the finalizeRemove function.

The first task makes all other threads aware of the fact that this key has been removed, so there will be no further attempts to read its value. This suffices for correctness. The second task, updating the entry, is an optimization that serves two purposes: first, rebalance does C.unpublish(ei, hi, ⊥, ⊥, rm)

81:

return not check whether a handle is deleted, so changing the handle index to ⊥ is needed to allow garbage collection; second, updating the entry saves some time for other operations, which do not need to read the handle in order to see that it is deleted. Therefore, finalizeRemove (Algorithm 3) tries to CAS the handle index to ⊥. We have to take care, however, in case the handle index had already changed, not to change it to ⊥. To this end, finalizeRemove takes a parameter prev -the handle remove marked as deleted. If the entry no longer points to it, we do nothing (line 75). We save in prev the handle itself and not the handle index, to avoid an ABA problem, since after a rebalance, the handle index might remain the same but reference to a different handle. Remove is linearized at the point where it marks the handle as deleted, and therefore it does not have to succeed in performing the CAS in the finalizeRemove function. If CAS fails, this means that some insertion operation reused this entry or another non-insertion operation already set the handle index to ⊥.

If remove finds an already deleted handle (case 2), it cannot simply return, since by the time remove notices that the handle is deleted, the entry might point to another handle. Therefore, similarly to computeIfPresent, remove makes sure that the key is removed by performing a successful CAS of the handle index to ⊥ (line 63). In this case (case 2) remove does not continue on to perform finalizeRemove, but rather retries if the CAS fails (line 66).

Note the difference between the two cases: in case 1, remove sets the handle to deleted, and so changing the entry's handle index to ⊥ is merely an optimization, and should only C V I T 2 0 1 6

23:12

Oak -A Key-Value Map for Big Data Analytics occur if the entry still points to the deleted handle. In the second case, on the other hand, remove does not delete any handle, and so it must make sure that the entry's handle index is indeed ⊥ before returning.

Linearization points

In Appendix B we show that Oak's operations (except for iterators) are linearizable [START_REF] Herlihy | Linearizability: A correctness condition for concurrent objects[END_REF]; that is, every operation appears to take place atomically at some point (the linearization point) between its invocation and response. Here, we only list the linearization points (l.p.): putIfAbsent -if it returns true the l.p. is the successful CAS of handle index (line 37).

Otherwise, putIfAbsent returns false and the l.p. is when it finds a non-deleted handle (reads handle.deleted = false) (line 23). put -if it inserts a new key the l.p. is the successful CAS of handle index (line 37). Otherwise, put replaces the value of a key and the l.p. is when a non-deleted handle is found and there is a successful nested call to handle put (line 25). putIfAbsentComputeIfPresent -if it inserts a new key the l.p. is the successful CAS of handle index (line 37). Otherwise, updates the value of a key and the l.p. is when a non-deleted handle is found and there is a successful nested call to handle compute (line 26). computeIfPresent -if it returns true the l.p. is when a non-deleted handle is found and there is a successful nested call to handle compute (line 56). Otherwise, computeIfPresent returns false and the l.p. is when the entry is not found, or it is found but with ⊥ as its handle index (line 51), or a successful CAS of handle index to ⊥ (line 63) is performed.

remove -if it is successful the l.p. is when a non-deleted handle is found and a successful nested call to handle remove occurs, setting the handle to deleted (line 58). Otherwise, it is not successful and the l.p. is when the entry is not found, or handle index is ⊥ (line 51), or a deleted handle is found and a successful CAS of handle index to ⊥ occurs (line 63). get -if it returns a handle the l.p. is the read of a non-deleted handle (line 7). Otherwise, it returns null and the l.p. is when lookup entry (line 4) returns ⊥, or when get reads that the handle index is ⊥ (line 5).

If get reads a deleted handle (line 7), then it also returns null. However, the l.p. cannot be the read of deleted, since by the time get reads it a new handle may be inserted into Oak. Therefore, if get reads deleted = true, then we say that the l.p. is the later between the read of handle index by get (line 5) and immediately after the set of deleted = true by remove.

Off heap support -epoch-based reclamation

We use epoch-based reclamation (based on [START_REF] Harris | A pragmatic implementation of non-blocking linked-lists[END_REF]) to support off-heap keys and values. We implement a global timestamp that is incremented at the begin of each operation. When a thread performs an operation (query, insertion, and non-insertion), it first sets its active flag, increments the global timestamp, and updates its local timestamp to match the global one.

At the end of an operation, the thread unsets its active flag. When a thread uses the remove operation it attaches the removed value to a released list, with the current global timestamp.

The keys are released during rebalance. In order to reclaim the memory in the released list, the timestamp of an entry in the released list must be less than the minimum current local timestamp of any active thread.

One deviation which we make from the basic protocol is to expand the active flag from one bit to several bits (we use one byte), to allow nesting of operations, since we consider H. Meir, D. Basin, E. Bortnikov, A. Braginsky, I. Keidar, and G. Sheffi

23:13

an iteration as a single continuous operation. Therefore, a thread starts an operation by incrementing its active counter and decrementing it when the operation ends. A thread is considered to be idle when its active counter is zero. When using an iterator we decrement the active counter by calling the close() method of the iterator. Again, this is why we need a closeableIterator.

Evaluation

We implement Oak in two ways: off-heap and on-heap. Both implementations are in Java, the handles are implemented using a Java ReentrantReadWriteLock [START_REF]Java reentrant read write lock[END_REF], and the index is based on Java ConcurrentSkipListMap [START_REF]Java concurrent skip list map[END_REF]. To support off-heap allocation, we use Oak to allocate space for keys and values; for fairness of the comparison, we also copy new values before inserting them into the Java skiplist. The experiments are run on a hardware platform that features four Intel Xeon E5-4650 processors, each with 8 cores. For the experiments we use the Synchrobench framework [START_REF] Gramoli | More than you ever wanted to know about synchronization: Synchrobench, measuring the impact of the synchronization on concurrent algorithms[END_REF]. Each experiment consists of 10 iterations, a few seconds each, and we report the average as the result. It is preceded by a warm-up period where randomly selected keys are inserted into the map. We first compare Oak's on-heap implementation to the (on-heap) Java skiplist, and then compare the two versions of Oak.

Oak vs. Java skiplist

Oak is a highly scalable map that achieves high throughput. We show this by comparing Oak to Java ConcurrentSkipListMap (skiplist) [START_REF]Java concurrent skip list map[END_REF]. Java skiplist holds arbitrary objects as its keys and values, including ByteBuffers which are held by Oak. Similarly to Oak, Java skiplist supports put, remove, get, and ascending and descending iterators. We first run experiments consisting of these operations. The map is initially filled up with 1M randomly selected keys, out of a range of 2M, keys are 4 bytes and values are 100B (in Appendix A we further report on additional experiments with 1KB values, where the results are similar).

Figure 3 depicts the throughput scalability with the number of threads.

The first experiment is a read-only workload (Figure 3a). Oak's chunk-based structure has better locality than a skiplist, which allows fast searches, and indeed we see that Oak's get operation outperforms Java skiplist by 3.3x. The second (Figure 3b) is a write-only workload with 50% puts and 50% removes. Here, Oak outperforms Java by 1.3x, again, thanks to speeding up the search, which is the first part of an update operation. Next we run ascending (Figure 3c) and descending iterators (Figure 3d), each scanning ranges of a 100 keys. The ascending iterator is twice as fast as Java's, again, thanks to fast first key search. Oak has built-in support for descending iterators while Java skiplist does not, and so in this workload, Oak outperforms Java significantly -by 4.8x.

Next we evaluate the compute operation. Java supports an operation called compute, which also assigns a new value computed by a function received as a parameter, however the returned value from the function is used as the new value, therefore this compute is not an atomic update-in-place as in Oak. If the function updates the received value in-place, there are no atomicity guarantees. To allow for a fair comparison, we create a new object LockableByteBuffer consisting of a ByteBuffer and a Java ReentrantReadWriteLock [START_REF]Java reentrant read write lock[END_REF] (as used in the handle implementation), and store LockableByteBuffers instead in the Java skiplist. We implement a function that first locks the given value using the write lock, then runs the computation steps on that value, and finally unlocks the lock. The function returns a pointer to the same value. In our workload, the computation reads a random byte of the given ByteBuffer and writes it to a random byte of the given ByteBuffer. We run two workloads, one executing only computeIfPresent, and one mixed workload with get, compute (putIfAbsentComputeIfPresent), put, and remove, 25% each. In both workloads, Oak outperforms Java skiplist by 70-80%, as shown in Figures 3e and3f. The speedup is due to the fast search in Oak and the extra CAS in Java's compute, which is used to replace the current value with the returned value computed by the user's function.

Oak Off-heap vs. On-heap

We compare two versions of Oak, off-heap and on-heap. We show that there are use-cases where one can benefit from off-heap allocation. We configure JVM to use the same amount of memory for both implementations. Since our off-heap implementation currently supports only 2GB of off-heap memory, we configure the off-heap heap size to 500MB, and the on-heap heap size to 2.5GB (= 2GB + 500MB). In the following experiments we use ~100K keys, 4

bytes each, with ~1KB values; the memory used to hold these keys and values nearly reaches the 2GB capacity of Oak's off-heap memory.

We implement and supply Oak with simple off-heap allocation and deallocation methods.

At the initialization, one continuous 2GB off-heap ByteBuffer is allocated, and the allocation method uses an atomically incremented index over the ByteBuffer to manage Oak's requests.

For the deallocation we hold simple lists of reclaimed memory. Note that in contrast, Oak's on-heap implementation uses the highly optimized Java's GC to manage the memory for keys and values, since they all reside in the Java heap.

The main scenarios in Druid, the platform that Oak is designed to support, are ones with conditional updates on large values. When Druid runs out of memory it usually archives it in persistent storage. Therefore, we run experiments demonstrating these scenarios, and show that in these cases even our simple (unoptimized) allocation/deallocation is sufficient, and Oak off-heap prevails, since Java's GC works hard on trying to free space for future allocations.

In the first experiment, depicted in Figure 4a, we run a mixed workload with equal share of get, put, remove, and compute (putIfAbsentComputeIfPresent) operations with 1KB values. In this experiment Oak is initially filled with 50K randomly selected keys, out of a range of 100K. In this case on-heap outperforms off-heap since the Java GC is better optimized for this case compared to Oak's simple GC implementation.

Next we run the Druid scenarios, with bigger values (5KB) and conditional updates. In these experiments, shown in Figures 4b,4c, and 4d, off-heap outperforms on-heap by 40-80%.

In the first experiment we use putIfAbsent to fill up the data structure (with 300K keys), which is initially empty. In the second experiment we run a mixed workload of putIfAbsent and putIfAbsentComputeIfPresent, after the data structure is initially filled with 150K keys out of a range of 300K. In the last experiment we run putIfAbsentComputeIfPresent with a data structure initially filled with 300K keys (out of a range of 300K). As expected, when monitoring the GC using Java's JConsole [START_REF]JConsole[END_REF], we observe that Java's GC wastes time by trying to free space while Oak's simple deallocation method is almost idle.

Conclusion

We presented Oak, a scalable concurrent KV-map for big data analytics. Two main requirements guided us when designing Oak. The first is supporting large keys and values. To this end, Oak enforces a zero-copy policy, which allows updates and reads to occur concurrently and atomically on the same memory location. It further supports off-heap allocation (and reclamation) of these keys and values, which is a recent trend in systems like HBase [START_REF]Hbase off-heap read-path in production[END_REF] and Druid [START_REF]Druid off-heap[END_REF]. The second requirement is supporting an analytics API. In addition to the standard get, put, and remove, Oak provides compute methods for performing an update of the value in-place. Oak also has built-in support for ascending and descending scans.

Our experiments have shown that Oak is faster by 1.3-4.8x than the currently standard concurrent KV-map, the Java ConcurrentSkipListMap. In addition, our results demonstrated that off-heap allocation is beneficial in scenarios with conditional updates of large values.

C V I T

A Oak vs. Java, 1KB values results

Oak vs. Java, 1KB values (Figure 5).

B Correctness

In this section, we prove Oak's correctness. Since the rebalancer is orthogonal to our contribution, we omit it from the discussion of Oak's correctness. We only assume that RB1-3 hold. We note that a similar rebalance was fully proven in [START_REF] Braginsky | A lock-free b+tree[END_REF].

B.1 Preliminaries

We consider a shared memory system consisting of a collection of shared variables accessed by threads, which also have local variables. An algorithm defines the behaviors of threads as deterministic state machines, where state transitions are associated with either an instance of a shared variable primitive (read, write, CAS, etc.) or a local step affecting the thread's local variables. A configuration describes the current state of all local and shared variables.

An initial configuration is one where all variables hold an initial value. A data structure implementation provides a set of operations, each with possible parameters. We say that operations are invoked and return or respond. The invocation of an operation leads to the execution of an algorithm by a thread. Both the invocation and the return are local steps of a thread. A run of algorithm A is an alternating sequence of configurations and steps, beginning with some initial configuration, such that configuration transitions occur according to A. We say that two operations are concurrent in a run r if both are invoked in r before either returns. We use the notion of time t during a run r to refer to the configuration reached after the t th step in r. An interval of a run r is a sub-sequence that starts with a step and ends with a configuration. The interval of an operation op starts with the invocation step of op and ends with the configuration following the return from op or the end of r, if there is no such return.

An implementation of concurrent data structure is linearizable [START_REF] Herlihy | Linearizability: A correctness condition for concurrent objects[END_REF] (a correctness condition for concurrent objects) if it provides the illusion that each invoked operation takes effect instantaneously at some point, called the linearization point (l.p.), inside its interval. A linearization of a run r (lin(r)) is the sequential run constructed by serially executing each operation at its l.p. Proof. Assume that h is not associated with k at time (t + 1).

B.2 Linearizability Proof

If there is no handle associated with k at time t + 1, then by Definition 1 either h.deleted = true or the entry's handle index (hi) is ⊥. In the first case, the only possible step that marks a handle as deleted is the l.p. of a successful remove(k). In the second case, only non-insertion operations turn hi to ⊥ by using CAS (lines 63 and 79), and according to Algorithm 3 this is only possible when the handle is deleted. However, at time t, h is still associated with k. Therefore, the entry's handle index (hi) is not ⊥.

Otherwise, there is a different handle h = h that is associated with k at time t + 1 (h = ⊥). This change can only be done by an insertion operation using CAS (line 37).

According to Algorithm 2 an insertion operation reaches that CAS only if the handle (h) is C V I T 2 0 1 6

23:20

Oak -A Key-Value Map for Big Data Analytics already deleted (line 23). However, in time t, h is still associated with k, and so there is no different handle that is associated with k.

Therefore, as long as the (t + 1) st step is not the l.p. of a successful remove(k), then h is still associated with k at time t + 1 in r, and there is no handle associated with k if the (t + 1) st step is a l.p. of a successful remove(k), as required.

Claim 4. Assume no handle is associated with key k at time t in a run r. Then, no handle is associated with k at time t + 1 in r if and only if the (t + 1) st step in r is not the l.p. of a successful insertion operation of k.

Proof. If no handle is associated at time t, and at time t + 1 there is an associated handle, then according to Definition 1 either a handle's deleted flag turned from false to true, or the entry's handle index turned from ⊥ to a valid one. The former is not possible because the handles are initialized as not deleted and only become deleted by a remove; no operation turns a deleted handle to a non-deleted one. In the second case, this can only be done by a successful insertion operation, at its l.p. (line 37), as required.

Look at the linearization lin(r) of run r using l.p.s defined in Section 3.2.4. From Claims 3 and 4, by induction on the steps of a run, we get:

Corollary 5. At any point in a concurrent run r, the set of keys associated with handles is exactly the same as the set of inserted keys and not removed keys, associated with the same handles, in lin(r) up to that point.

Claim 6 (Get). In run r, if get(k) returns h then the corresponding get(k) in lin(r) returns h, and if get(k) returns null then the corresponding get(k) in lin(r) returns null.

Proof. There are three cases for get's l.p.:

1. Get(k) finds a non-deleted handle h (line 7), then get(k) returns h and by Claim 2 h is associated with k. By Corollary 5, in lin(r) k is inserted and not removed (the map holds k) and since this is the l.p. of get then the corresponding get(k) in lin(r) returns h as well.

2. LookUp(k) by get(k) (line 4) returns ⊥ or if get(k) reads that the handle index is ⊥ (line 5), then there is no handle associated with key k, and get(k) returns null. By Corollary 5, in lin(r) the map does not hold k, and since this is the l.p. of get then the corresponding get(k) in lin(r) returns null as well.

3. Get(k) finds a deleted handle h at time t 2 (line 7) and returns null. Then its l.p. is the later between the read of handle index hi by get(k) at time t 1 < t 2 (line 5) and immediately after the set of deleted = true by remove(k) at some time t < t 2 . Again there are two cases:

a. If t > t 1 then the l.p. is immediately after the set of deleted = true then there is no handle associated with key k, and by Corollary 5, in lin(r) the map does not hold k, and the corresponding get(k) in lin(r) returns null as well.

b. If t 1 > t then the l.p. is the read of handle index hi by get(k) (line 5) at time t 1 , after the set of deleted = true at time t. We need to show that at no time between t and t 1 the handle index changed to hi = hi and now it does not point to a deleted handle. Notice that only an insertion operation l.p. can change hi to hi . Assume by contradiction that the l.p. of such an operation occurs between t and t 1 . Then when get sees hi at time t 1 , it is already hi and not hi. A contradiction. Hence, at the l.p.

of get(k), there is no handle associated with key k, and by Corollary 5, in lin(r) the map does not hold k, so the corresponding get(k) in lin(r) returns null as required.

In

  Section 3.1 we explain Oak's chuck-based data organization on-and off-heap. Section 3.2 focuses on the different operations offered by Oak and their implementation. Section 3.3 discusses the epoch-based memory reclamation.

Figure 1

 1 Figure 1 Oak structure.

Figure 2

 2 Figure 2 Example of stacks build during a descending iterator traversal in a chunk.

Algorithm 3

 3 Oak's non-insertion operations 41: procedure computeIfPresent(key, func) 42: return doIfPresent(key, func, comp) 43: procedure remove(key) if ei = ⊥ then hi ← C.entries[ei].hi 51: if hi = ⊥ then return false 52: handle ←C.handles[hi] 53: if ¬handle.deleted then Case 1: handle exists and not deleted 54: if op = comp ∧ 55: handle.compute(func) then 56: ¬C.publish(ei, hi, ⊥, func, op) then 62: return doIfPresent(key, func, op) 63: res ← CAS(C.entries[ei].hi, hi, ⊥) 64: C.unpublish(ei, hi, ⊥, func, op) if ei = ⊥ then hi ← C.entries[ei].hi 73: if hi = ⊥ then return 74: handle ←C.handles[hi] 75: if handle = prev then return 76: if ¬C.publish(ei, hi, ⊥, ⊥, rm) .entries[ei].hi, hi, ⊥) 80:

Figure 3

 3 Figure 3 Oak vs. Java, 100B values.

Figure 4

 4 Figure 4Oak Off-heap vs. On-heap, 1-5KB values.

Figure 5

 5 Figure 5 Oak vs. Java, 1KB values.

Meir, D. Basin, E. Bortnikov, A. Braginsky, I. Keidar, and G. Sheffi 23:7

  Note, however, that chunk methods that only read the chunk (lookUp) or modify existing handles (writeValue), and the unpublish method, may proceed and need not abort and retry.Next, we list the rebalancer guarantees. A path is a sequence of chunks C 1 , C 2 , . . . , C k reached by traversing chunks' next pointers in a run (until reaching a null pointer).

	Traversals(C 0 , r) is the set of paths starting from C 0 in extensions of r. We say that
	key k is in the range of chunk C if k ≥ C.minKey and k < C.next.minKey.
	The rebalancer guarantees the following:
	RB1 If locate(k) returns C in run r, then all keys ≥ k inserted before locate(k) occurs are
	reachable in Traversals(C, r), unless they are later removed.
	RB2 If locate(k) returns C in run r, then all keys ≥ k removed before locate(k) are not in
	Traversals(C, r), unless later inserted.
	RB3 If locate(k) returns C in run r, then reachable keys in chunks in Traversals(C, r) are
	monotonically increasing and disjoint.
	2, any chunk's method
	that tries to modify a chunk while it is being rebalanced fails and retries, instead of helping.

H.

2 0 1 6 23:10 Oak -A Key-Value Map for Big Data Analytics In

  the second case, the key is not present. If we discover a removed entry that points to the same key but with hi = ⊥ or a deleted handle, then we reuse this entry. Otherwise, we call allocateEntryAndKey to allocate a new entry, as well as allocate and write the key that it points to (line 30). Either way, we allocate a new handle (line 32). These functions might fail and cause a retry (line 33). If we allocate a new entry, then we try to link this new entry into the entries linked list by calling entriesLLputIfAbsent (line 31). If entriesLLputIfAbsent receives ⊥ as a parameter then it just returns ⊥ as well. Otherwise, if it encounters an already linked entry then it returns it. In this case, our allocated entry remains unlinked in the entries array and other operations never reach it; the rebalancer eventually removes it from the array. Notice that if we do manage to link the new entry, then its handle index remains ⊥ (as initialized). After allocations of the entry, key, and handle, we allocate and write the value (outside of the chunk), and have the new handle point to it (line 34).

2 0 1 6 23:16 Oak -A Key-Value Map for Big Data Analytics 23:18 Oak -A Key-Value Map for Big Data Analytics 32

  Aravind Natarajan and Neeraj Mittal. Fast concurrent lock-free binary search trees. In Proceedings of the 19th ACM SIGPLAN Symposium on Principles and Practice of Parallel Programming, PPoPP '14, pages 317-328, New York, NY, USA, 2014. ACM. URL: http: //doi.acm.org/10.1145/2555243.2555256, doi:10.1145/2555243.2555256.

	33 Alexander Spiegelman, Guy Golan-Gueta, and Idit Keidar. Transactional data structure
	libraries. In Proceedings of the 37th ACM SIGPLAN Conference on Programming Lan-
	guage Design and Implementation, PLDI '16, pages 682-696, New York, NY, USA, 2016.
	ACM. URL: http://doi.acm.org/10.1145/2908080.2908112, doi:10.1145/2908080.
	2908112.

23:21

Claim 7 (PutIfAbsent). In run r, if putIfAbsent(k) returns true then the corresponding putIfAbsent(k) in lin(r) returns true, and if putIfAbsent(k) returns false then in lin(r) the corresponding putIfAbsent(k) returns false. By Claim 2 h is associated with k before and there is no handle associated with k right after (by Definition 1). Since this is the l.p. of remove, and by Corollary 5 in lin(r) the map does hold k before the l.p. and does not after. Therefore, the corresponding remove(k) in lin(r)

removes k and returns as required.

If lookUp(k) by remove(k) returns ⊥, or if remove(k) reads that the handle index is ⊥ (line 51), then there is no handle associated with key k, and remove(k) returns unsuccessfully.

By Corollary 5, in lin(r) the map does not hold k, and since this is the l.p. of remove then the corresponding remove(k) in lin(r) returns unsuccessfully as required.

Otherwise, a successful CAS of handle index to ⊥ is performed by remove(k) (line 63), from a handle index pointing to a deleted handle (line 53). Then remove(k) returns and by Definition 1 there is no handle associated with k just before the CAS and right after it. By Corollary 5, in lin(r) the map does not hold k, and since this is the l.p. of remove then the corresponding remove(k) in lin(r) returns unsuccessfully.

Having shown that all of Oak's operations behave the same way in a run r and its linearization lin(r), we can conclude the following theorem: Theorem 12. Oak is linearizable with the l.p.s defined in Section 3.2.4.