{'fraction_non_alphanumeric': 0.041941923401452946, 'fraction_numerical': 0.011854047045749213, 'mean_word_length': 4.349185380889476, 'pattern_counts': {'":': 0, '<': 2, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
High-Performance Distributed RMA Locks

We propose a topology-aware distributed Reader-Writer lock that accelerates irregular workloads for supercomputers and data centers. The core idea behind the lock is a modular design that is an interplay of three distributed data structures: a counter of readers/writers in the critical section, a set of queues for ordering writers waiting for the lock, and a tree that binds all the queues and synchronizes writers with readers. Each structure is associated with a parameter for favoring either readers or writers, enabling adjustable performance that can be viewed as a point in a three dimensional parameter space. We also develop a distributed topology-aware MCS lock that is a building block of the above design and improves state-of-the-art MPI implementations. Both schemes use non-blocking Remote Memory Access (RMA) techniques for highest performance and scalability. We evaluate our schemes on a Cray XC30 and illustrate that they outperform state-of-the-art MPI-3 RMA locking protocols by 81% and 73%, respectively. Finally, we use them to accelerate a distributed hashtable that represents irregular workloads such as key-value stores or graph processing.


INTRODUCTION
The scale of today's data processing is growing steadily. For example, the size of Facebook's social graph is many petabytes [10,50] and graphs processed by the well-known HPC benchmark Graph500 [40] can have trillions of vertices. Efficient analyses of such datasets require distributed-memory (DM) machines with deep Non-Uniform Memory Access (NUMA) hierarchies.
Locks are among the most effective synchronization mechanisms used in codes for such machines [9]. On one hand, if used improperly, they may cause deadlocks. Yet, they have intuitive semantics and they often outperform other schemes such as atomic operations [43] or transactions [6].
Designing efficient locks for machines with deep hierarchical memory systems is challenging. Consider four processes competing for the same lock. Assume that two of them (A and B) run on one socket and the remaining two (C and D) execute on the other one. Now, in a naive lock design oblivious to the memory hierarchy, the lock may be passed between different sockets up to three times, degrading performance (e.g., if the order of the processes entering the critical section (CS) is A, C, B, and D). Recent advances [12,18] tackle this problem by reordering processes acquiring the lock to reduce inter-socket communication. Here, the order of A, B, C, and D entails only one inter-socket lock transfer, trading fairness for higher throughput. Extending such schemes to DM machines with weak memory models increases complexity. Moreover, expensive inter-node data transfers require more aggressive communicationavoidance strategies than those in intra-node communication [20]. To our best knowledge, no previous lock scheme addresses these challenges. Another property of many large-scale workloads is that they are dominated by reads (e.g., they constitute 99.8% of requests to the Facebook graph [50]). Here, simple locks would entail unnecessary overheads. Instead, the Reader-Writer (RW) lock [38] can be used to reduce the overhead among processes that only perform reads in the critical section (CS). Initial RW NUMA-aware designs have recently been introduced [11], but they do not address DM machines.
In this work, we develop a lock that addresses the above challenges. Its core concept is a modular design for adjusting performance to various types of workloads. The lock consists of three key data structures. First, the distributed counter (DC) indicates the number of readers or the presence of a writer in the CS. Second, the distributed queue (DQ) synchronizes writers belonging to a given element of the memory hierarchy (e.g., a rack). Finally, the distributed tree (DT) binds together all queues at different levels of the memory hierarchy and synchronizes writers with readers. Each of these three structures offers an adjustable performance tradeoff, enabling high performance in various settings. DC can lower the latency of lock acquire/release performed by either readers or writers, DQ can be biased towards improving either locality or fairness, and DT can increase the throughput of either readers or writers. The values of these parameters constitute a three dimensional space that is illustrated in Figure 1. Each point is a specific lock design with selected performance properties.
Most DM machines offer Remote Direct Memory Access (RDMA) [42], a hardware scheme that removes the OS and the CPU from the inter-node communication path. RDMA is the basis of many Remote Memory Access (RMA) [20] programming models. Among others, they offer a Partitioned Global Address Space (PGAS) abstraction to the programmer and enable low-overhead direct access to remote memories with put/get communication primitives. RMA principles are used in various HPC languages and libraries: Unified Parallel C (UPC) [49], Fortran 2008 [30], MPI-3 [39], or SHMEM [4]. We will illustrate how to utilize RMA in the proposed locks for DM machines, addressing the above-mentioned challenges. In the following, we use MPI-3 RMA but we keep our protocols generic and we discuss ( § 6) how other RMA languages and libraries can also be used.
In summary, our key contributions are as follows: • We develop a topology-aware distributed Reader-Writer lock that enables various tradeoffs between fairness, throughput, latency, and locality. • We offer a topology-aware distributed MCS lock that accelerates the state-of-the-art MPI-3 RMA codes [20]. • We illustrate that our designs outperform the state-of-theart in throughput/latency (7.2x/6.8x on average) and that they accelerate distributed hashtables used in key-value (KV) stores or graph processing.

RMA AND LOCKS
We start by discussing RMA ( § 2.1), our tool to develop the proposed locks. Next, we present traditional ( § 2.2) and state-of-the-art ( § 2.3, § 2.4) locks that we use and extend. Notation/Naming: We denote the number of processes as ; we use the notion of a process as it occurs frequently in DM codes such as MPI [39]. Still, our schemes are independent of whether heavyweight processes or lightweight threads are incorporated. Each process has a unique ID called the rank ∈ {1, ..., }. A process in the CS is called active. A null pointer is denoted as ∅. Then, is the number of levels of the memory hierarchy of the used machine. Here, the selection of the considered levels depends on the user. For example, one can only focus on the nodes connected with a network and racks that contain nodes and thus = 3 (three levels: the nodes, the racks, and the whole machine). We refer to a single considered machine part (e.g., a node) as an element. We refer to a node that is a shared-memory cache-coherent domain connected to other such domains with a non-coherent network as a compute node (or just node). One compute node may contain smaller elements that are cache-coherent and together offer non-uniform memory access (NUMA). We refer to such elements as NUMA nodes; an example NUMA node is a socket with a local DRAM. We present symbols used in the paper in Table 1.

Number of processes.
Rank of a process that attempts to acquire/release a lock. Number of levels of the considered machine. Number of machine elements at level ; 1 ≤ ≤ . Index used to refer to the th machine level. Index used to refer to the th element at a given machine level.
RMA Windows: In RMA, each process explicitly exposes an area of its local memory as shared. In MPI, this region is called a window. Once shared, a window can be accessed with puts/gets/atomics and synchronized with flushes. We will refer to such an exposed memory in any RMA library/language as a window.
RMA Functions: We describe the syntax/semantics of the used RMA calls in Listing 1. All ints are 64-bit. For clarity, we also use the bool type and assume it to be an int that can take the 0 (false) or 1 (true) values, respectively. Values returned by Get/FAO/CAS are only valid after the subsequent Flush. The syntax is simplified for clarity: we omit a pointer to the accessed window (we use a single window). We use an origin/a target to refer to a process that issues or is targeted by an RMA call.

Traditional Hardware-Oblivious Locks
We now present hardware-oblivious locks used in this work.

2.2.1
Reader-Writer (RW) Locks. Reader-Writer (RW) locks [15] distinguish between processes that only perform reads when in the CS (readers) and those that issue writes (writers). Here, multiple readers may simultaneously enter a given CS, but only one writer can be granted access at a time, with no other concurrent readers 1 /* Common parameters: : target 's rank ; : an offset 2 * into 's window that determines the location of the 3 * targeted data ; : an operation applied to a remote piece of 4 * data ( either an atomic replace ( REPLACE ) or a sum ( SUM )); 5 * : the operand of an atomic operation . */ 6 7 /* Place atomically _ in 's window . */ 8 void Put ( int src_data , int target , int offset ); 9 10 /* Fetch and return atomically data from 's window . */ 11 int Get ( int target , int offset ); 12 13 [37,44,46] does not distinguish between readers or writers. Instead, it only allows one process at a time to enter the CS, regardless of the type of memory accesses issued by . Here, processes waiting for the lock form a queue, with a process at the head holding the lock. The queue contains a single global pointer to its tail. Moreover, each process in the queue maintains: (1) a local flag that signals if it can enter the CS and (2) a pointer to its successor. To enter the queue, a process updates both the global pointer to the tail and the pointer at its predecessor so that they both point to . A releasing process notifies its successor by changing the successor's local flag. The MCS lock reduces the amount of coherence traffic that limits the performance of spinlocks [2]. Here, each process in the queue spin waits on its local flag that is modified once by its predecessor.

State-of-the-Art NUMA-Aware Locks
We now discuss lock schemes that use the knowledge of the NUMA structure of the underlying machine for more performance. We will combine and extend them to DM domains, and enrich them with a family of adjustable parameters for high performance with various workloads.
2.3.1 NUMA-Aware RW Locks. Many traditional RW locks ( § 2.2.1) entail performance penalties in NUMA systems as they usually rely on a centralized structure that becomes a bottleneck and entails high latency when accessed by processes from remote NUMA elements. Calciu et al. [11] tackle this issue with a flag on each NUMA node that indicates if there is an active reader on that node. This reduces contention due to readers (each reader only marks a local flag) but may entail additional overheads for writers that check for active readers.

Hierarchical MCS Locks.
Hierarchical locks tackle expensive lock passing described in § 1. They trade fairness for higher throughput by ordering processes that enter the CS to reduce the number of such passings. Most of the proposed schemes address two-level NUMA machines [12,17,34,41]. Chabbi et al. consider a multi-level NUMA system [12]. Here, each NUMA hierarchy element (e.g., a socket) entails a separate MCS lock. To acquire the global lock, a process acquires an MCS lock at each machine level. This increases locality [47] but reduces fairness: processes on the same NUMA node acquire the lock consecutively even if processes on other nodes are waiting.

Distributed RMA MCS Locks
Finally, we present a distributed MCS (D-MCS) lock based on an MPI-3 MCS lock [22]. We will use it to accelerate state-of-the-art MPI RMA library foMPI [20] and as a building block of the proposed distributed topology-aware RW and MCS locks ( § 3).

Summary and Key Data Structures.
Here, processes that wait for the D-MCS lock form a queue that may span multiple nodes. Each process maintains several globally visible variables. A naive approach would use one window per variable. However, this would entail additional memory overheads (one window requires Ω( ) storage in the worst case [20]). Thus, we use one window with different offsets determining different variables: a pointer to the next process in the MCS queue (offset NEXT, initially ∅) and a flag indicating if a given process has to spin wait (offset WAIT, initially false). A selected process (rank tail_rank) also maintains a pointer to a process with the queue tail (offset TAIL, initially ∅).

Lock Protocols.
We now describe the protocols for acquire/release. We refer to respective variables using their offsets in the window.
Lock Acquire (Listing 2) First, atomically modifies TAIL with its own rank and fetches the predecessor rank (Line 6). If there is no predecessor, it proceeds to the CS. Otherwise, it enqueues itself (Line 10) and waits until its local WAIT is set to false. Flushes ensure the data consistency. Lock Release (Listing 3) First, checks if it has a successor in the queue (Line 3). If there is none, it atomically verifies if it is still the queue tail (Line 5); if yes, it sets TAIL to ∅. Otherwise, waits for a process that has modified TAIL to update its NEXT field (Lines 9-11). If there is a successor, the lock is passed with a single Put (Line 14).

DISTRIBUTED RMA RW LOCKS
We now present a distributed topology-aware RW lock (RMA-RW) for scalable synchronization and full utilization of parallelism in workloads dominated by reads. We focus on the RW semantics as the key part of the introduced lock. Symbols specific to RMA-RW are presented in Table 2.
Lock Abbreviations We always refer to the proposed topologyaware distributed RW and MCS lock as RMA-RW and RMA-MCS, respectively. Both RMA-RW and RMA-MCS use as their building block a simple distributed topology-oblivious MCS lock ( § 2.4) denoted as D-MCS.
Example In the whole section, we will use the example shown in Figure 2. Here, = 3 and the considered levels are: compute nodes, racks, and the whole machine.

Design Summary and Intuition
As explained in § 1, RMA-RW consists of three types of core data structures: distributed queues (DQs), a distributed tree (DT), and a distributed counter (DC). They are illustrated in Figure 2. First, every machine element (at each considered level) has an associated DQ and thus a D-MCS lock local to this element (as opposed to  the global RMA-RW lock). In our example, every node, rack, and the whole machine have their own DQ (and thus a local MCS lock). Note that some DQs that are associated with elements such as nodes are not necessarily distributed, but we use the same name for clarity. Second, all the DQs form a DT that corresponds to the underlying memory hierarchy, with one DQ related to one tree vertex. For example, DQs associated with nodes that belong to a given rack constitute vertices that are children of a vertex associated with a DQ running on rack . Third, DC counts active readers and writers and consists of several physical counters located on selected processes. DT on its own (without DC and any readers) constitutes RMA-MCS. Writers A writer that wants to acquire a lock starts at a leaf of DT located at the lowest level (a node in our example). At any level (2 ≤ ≤ ), it acquires a local D-MCS lock that corresponds to a subtree of D-MCS locks (and thus DQs) rooted at the given element. Here, it may compete with other writers. When it reaches level 1, it executes a different protocol for acquiring the whole RMA-RW lock. Here, it may also compete with readers. RMA-RW's locality-aware design enables a shortcut: some writers stop before reaching level 1 and directly proceed to the CS. This happens if a lock is passed within a given machine element.
Readers Readers do not enter DQs and DT and thus have a single acquire protocol. This design reduces synchronization overhead among readers.

Key Data Structures
We now present the key structures in more detail.

Distributed Counter (DC)
. DC maintains the number of active readers or writers. It enables an adjustable performance tradeoff that accelerates readers or writers. For this, one DC consists of multiple physical counters, each maintained by every th process; is a parameter selected by the user. To enter the CS, a reader increments only one associated physical counter while a writer must check each one of them. Thus, selecting more physical counters (smaller ) entails lower reader latency (as each reader can access a counter located on a closer machine element) and contention (as each counter is accessed by fewer readers). Yet, higher entails lower latency for a writer that accesses fewer physical counters.
A physical counter associated with a reader is located at a rank ( ); (·) ∈ {1, ..., } can be determined at compile-or run-time. In a simple hardware-oblivious scheme, one can fix ( ) = ⌈ / ⌉. For more performance, the user can locate physical counters in a topology-aware way. For example, if the user allocates processes/node and a node hosts processes with successive ranks starting from ( − 1) + 1, then setting = in the above formula results in storing one physical counter every th node. This can be generalized to any other machine element.
To increase performance, we implement each physical counter as two 64-bit fields that count the readers (assigned to this counter) that arrived and departed from the CS, respectively. This facilitates obtaining the number of readers that acquired the lock since the last writer and reduces contention between processes that acquire and release the lock. We dedicate one bit of the field that counts arriving readers to indicate whether the CS of RMA-RW is in the READ mode (it contains readers) or the WRITE mode (it contains a writer).
RMA Design of DC: Each physical counter occupies two words with offsets ARRIVE (for counting arriving readers) and DEPART (for counting departing readers); physical counters together constitute an RMA window.

Distributed Queue (DQ)
. DQ orders writers from a single element of the machine that attempt to enter the CS. DQs from level have an associated threshold , that determines the maximum number of lock passings between writers running on a machine element from this level before the lock is passed to a process from a different element. We use a separate threshold , for each because some levels (e.g., racks) may need more locality (a higher threshold) than others (e.g., nodes) due to expensive data transfers. This design enables an adjustable tradeoff between fairness and throughput at each level.
DQ extends D-MCS in that the local flag that originally signals whether a process can enter the CS now becomes an integer that carries (in the same RMA operation) the number of past lock acquires within a given machine element. We use this value to decide whether to pass the lock to a different element at a given level (if the value reaches , ) or not (if the value is below , ).
RMA Design of DQ: All DQs at a given level constitute an RMA window. Respective offsets in the window are as follows: NEXT (a rank of the next process in the queue), STATUS (an integer that both signals whether to spin wait and carries the number of past lock acquires in the associated machine element), and TAIL (a rank of the process that constitutes the current tail of the queue). TAIL in DQ at level associated with th element is stored on a process tail_rank[ , ].

Distributed Tree of Queues (DT).
DT combines DQs at different memory hierarchy levels into a single structure. This enables to make progress in acquiring/releasing RMA-RW by moving from level to level 1. Then, at the tree root, writers synchronize with readers. Specifically, the lock is passed from writers to readers (if there are some waiting) when the total number of lock passings between writers reaches a threshold . In our design, = =1 , . To avoid starvation of writers, we also introduce a threshold that is the maximum number of readers that can enter the CS consecutively before the lock is passed to a writer (if there is one waiting). Increasing or improves the throughput of readers or writers because more processes of a given type can enter the CS consecutively.
While climbing up DT, a writer must determine the next DQ (and thus D-MCS) to enter. This information is encoded in a mapping (·, ·) and structure tail_rank [ , ]. ( , ) ∈ {1, ..., } returns the ID of a machine element associated with a process at level . An expression tail_rank[ , ( , )] returns the rank of a process that points to the tail of a DQ at level within a machine element assigned to . This enables to enter D-MCS at the next level on the way to the CS. Similarly to ( ), ( , ) can be determined statically or dynamically.
Depending on , , some writers do not have to climb all DT levels and can proceed directly to the CS. Thus, we further extend the STATUS field used in DQ with one more special value ACQUIRE_PARENT. This indicates that cannot directly enter the CS and should continue up DT.

Discussion on the Status Field. A central part of DQ and
DT is the STATUS field that enables processes to exchange various additional types of information in a single RMA communication action, including: (1) if a lock mode changed (e.g., from READ to WRITE), (2) if a given process should acquire a lock at a higher DT level, (3) if a given process can enter the CS, and (4) the number of past consecutive lock acquires. Two selected integer values are dedicated to indicate (1) and (2). All the remaining possible values indicate that the given process can enter the CS (3); at the same time the value communicates (4).

Distributed Reader-Writer Protocol
We now illustrate how the above data structures play together in the acquire and release protocols. A writer starts at the leaf of DT (level ) both for acquiring and releasing. At any level (2 ≤ ≤ ), it proceeds up the tree executing a protocol for a partial acquire/release of the respective part of the tree ( § 3.3.1, § 3.3.2). At level 1, it executes a different protocol for locking or releasing the whole lock ( § 3.3.3, § 3.3.4). Readers do not follow such a hierarchy and thus have single acquire ( § 3.3.5) and release ( § 3.3.6) protocols.
3.3.1 Writer Lock Acquire: Level to 2 (Listing 4). Intuition: enters the DQ associated with a given level and its home element ( , ); it then waits for the update from its predecessor. If the predecessor does not have to hand over the lock to a process from another element (i.e., has not reached the threshold , ), the lock is passed to that immediately enters the CS. Otherwise, moves to level − 1. Details: first modifies its NEXT and STATUS to reflect it spin waits at the DQ tail (Lines 2-3). Then, it enqueues itself (Line 5). If there is a predecessor at this level, makes itself visible to it with a Put (Line 8) and then waits until it obtains the lock. While waiting, uses Gets and Flushes to check for any updates from the predecessor. If the predecessor reached , and released the lock to the parent level, must itself acquire the lock from level − 1 (Line 23). Otherwise, it can directly enter the CS as the lock is simply passed to it (Line 18). If there is no predecessor at level , also proceeds to acquire the lock for level − 1 (Line 23). 3.3.2 Writer Lock Release: Level to 2 (Listing 5). Intuition: passes the lock within ( , ) if there is a successor and , is not yet reached. Otherwise, it releases the lock to the parent level − 1, leaves the DQ, and informs any new successor that it must acquire the lock at level − 1. Details: first finds out whether it has a successor. If there is one and , is not yet reached, the lock is passed to it with a Put (Line 8). If , is reached, releases the lock for this level and informs its successor (if any) that it has to acquire the lock at level − 1. If there is no known successor, it checks atomically if some process has already entered the DQ at level (Line 15). If so, the releaser waits for the successor to make himself visible before it is notified to acquire the lock at level − 1.

Writer
Lock Acquire: Level 1 (Listing 7). Intuition: This scheme is similar to acquiring the lock at lower levels ( § 3.3.1). However, the predecessor may notify of the lock mode change that enabled readers to enter the CS, forcing to acquire the lock from the readers. Details: first tries to obtain the lock from a predecessor (Lines 2-18). If there is one, waits until the lock is passed. Still, it can happen that the predecessor hands the lock over to the readers

Reader
Lock Acquire (Listing 9). Intuition: Here, first spin waits if there is an active writer or if 's arrival made its associated counter ( ) exceed . Then, it can enter the CS. If ( ) = , then resets DC. Details: In the first part, may spin wait on a boolean barrier variable (Line 5), waiting to get the lock from a writer. Then, atomically increments its associated counter and checks whether the count is below . If yes, the lock mode is READ and enters the CS. Otherwise, either the lock mode is WRITE or is reached. In case of the latter, checks if there are any waiting writers (Line 17). If there are none, resets the DC (Line 20) and re-attempts to acquire the lock. If there is a writer, sets the local barrier and waits for DC to be reset by the writer.

Example
Consider the scenario from Figure 2. Here, there are three machine levels, 12 readers, and 12 writers ( = 0.5). Writer Acquire Assume a new writer running on a node related to DQ 3.1 attempts to acquire RMA-RW (Figure 2, Part 5). First, it enters DQ 3.1 (Listing 4). As this queue is not empty, spins locally (Lines 10-12) until its predecessor 9 modifies 's STATUS. Now, if 9 has not yet reached ,3 , gets the lock and immediately proceeds to the CS (Lines 15-19). Otherwise, it attempts to move to level 2 by updating its STATUS (Line 22) and calling writer-acquire< −1>(). Thus, it enters DQ 2.1 and takes the same steps as in DQ 3.1 : it spins locally until 4 changes its STATUS and it either directly enters the CS or it proceeds up to level 1. Assuming the latter, enters DQ 1.1 and waits for 1 to change its STATUS (Listing 7, Lines 10-12). If STATUS is different from MODE_CHANGE (Line 17), can enter the CS. Otherwise, the lock was handed over to the readers and calls set_counters_to_WRITE() to change all physical counters to the WRITE mode (Line 15), which blocks new incoming readers. At some point, the readers reach the threshold and hand the lock over to . Writer Release Assume writer occupies the CS and starts to release RMA-RW (Figure 2, Part 6). It begins with level 3 (Listing 5).
Here, it first checks if it has a successor in DQ 3.1 and if , 3 is not yet reached (Line 5). Its successor is 10 and assume that the latter condition is true. Then, passes the lock to 10 by updating its STATUS so that it contains the number of lock acquires within the given element. If ,3 is reached, releases the lock at level 2 (Line 12). Here, it repeats all the above steps (its successor is 6 ) and then starts to release the lock at level 1 (Listing 8). Here it hands the lock over to the readers if is reached (Lines 5-8). Finally, it notifies its successors at each level ( to 2) to acquire the lock at the parent level (Listing 5, Line 23).
Reader Acquire A reader that attempts to acquire RMA-RW first increments ( ) (Listing 9, Line 12) and checks if is reached (in the first attempt Lines 6-8 are skipped). If yes, it sets barrier (Line 14), backs off (Line 24), and reattempts to acquire the lock. In addition, if is the first process to reach , it also checks if there are any waiting writers (Lines 15-21). If not, it resets ( ) and sets barrier to false so that it can enter the CS even if was reached. Then, it reexecutes the main loop (Line 3); this time it may enter the loop in Lines 6-8 as the lock was handed over to a writer (if was reached). In that case, waits until its ( ) is reset (Listing 9, Lines 6-8).
Reader Release This is a straightforward scenario in which only increments DEPART at ( ).

RMA-RW vs. RMA-MCS
We also outline the design of RMA-MCS. RMA-MCS consists of DQs and DT but not DC. and are excluded as the are no readers. Similarly, ,1 is not applicable because there is no need to hand the lock to readers. The acquire/release protocols are similar to the ones in Listings 4 and 5 for any ∈ {1, ..., }.

CORRECTNESS ANALYSIS
We now discuss how RMA-RW ensures three fundamental correctness properties: mutual exclusion (ME), deadlock freedom (DF), and starvation freedom (SF) [24]. At the end of this section, we show how we use model checking to verify the design.

Mutual Exclusion
ME is violated if two writers or a reader and a writer enter the CS concurrently. We now discuss both cases.
Writer & Writer: We distinguish between writers that are in the same DQ (case A) or in different ones (case B). In case A, they operate on the same TAIL. Thus, they could only violate ME if both writers do not see any predecessor. This is prevented by using FAO for atomically modifying TAIL. In case B, two writers competing in different DQs have a common DQ in DT where they or their predecessor compete for the lock. Similarly as above, the MCS lock must be acquired at each DT level. If a predecessor has to compete for the lock, a writer waits until he gets notified by its predecessor and thus does not interfere in the lock acquiring process.
Reader & Writer: A reader and a writer can be active at the same time if the lock mode is READ and about to change to WRITE. This is because the reader on its own cannot change the mode and as a consequence cannot acquire a lock while a writer is active. However, a writer can alter the mode to WRITE while a reader is active. This is prevented by a writer that checks each counter again for active readers after changing all of them.

Deadlock Freedom
Here, we also differentiate two base cases: two writers deadlock or a reader and a writer deadlock.
Writer & Writer The only way how writers deadlock is if there is a cycle in a queue. For two writers it means that one becomes the predecessor of the other. Therefore, both wait on the other to get notified. This cannot happen as the processes use an atomic FAO to obtain their predecessor. As explained, this function is atomic and thus we can order the uses of FAO in a timeline. This contradicts that the writers have a cycle in their waiting queue.
Reader & Writer A reader may deadlock after is reached (case A) or the mode goes into WRITE (case B). In case A, either there is no writer active and the reader resets the DC or a writer is waiting and a reader backs off. Thus, the writer changes the mode to WRITE after all readers back off which is done in a finite time. As writers do not deadlock and the last writer changes the mode back to READ, no reader will deadlock in case B either.

Starvation Freedom
Finally, we show that no writer or reader can starve.
Writers A writer may starve while other writers or readers are active. We prevent it with different thresholds. First, there is , at each DT level . After reaching , , writers in one of the associated DQs at release the lock to the next DQ at the same level. Thus, we only need to show that one DQ is starvation-free which is already provided by the underlying MCS queue lock design. Yet, there is the threshold that regulates the number of lock acquires by readers for one counter before the readers associated to the counter back off. We already showed that the readers make progress. Thus, at some point, all counters have reached and a writer changes the mode to WRITE.
Readers There are two ways how readers could starve. First, other readers are active while processes associated with a certain counter back off to let writers acquire the lock. However, there is the threshold for each counter after which the readers associated with this counter back off. Thus, eventually, all readers wait on the writers to take over. This leads us to the second case where the writers have the lock and do not pass it to the waiting readers. This is not possible since there is the , threshold at each level of the writer hierarchy and at most after = =1 , lock passings between writers the lock goes to readers; we have also already illustrated that the writers will make progress until this threshold is reached.

Model Checking
To confirm that RMA-RW provides the desired correctness properties, we also conduct model checking with SPIN [27] (v6.4.5), a software tool for the formal verification of multi-threaded codes. The input to SPIN is constructed in PROMELA, a verification modeling language that allows for the dynamic creation of concurrent processes to model, for example, distributed systems. We evaluate RMA-RW for up to ∈ {1, ..., 4} and a maximum of 256 processes. The machine elements on each level of the simulated system have the same number of children. Thus, for = 3 and four subelements per machine element, the system would consist of 4 3 processes. Each process is defined randomly either as a reader or a writer at the beginning and after that, it tries to acquire the lock 20 times. We choose this value as it generates a feasible number of cases that SPIN has to check even for a high count of processes. During the execution of a test, we use a designated process that verifies that either only one writer or multiple readers hold a lock. All the tests confirm mutual exclusion and deadlock freedom.

EVALUATION
We now illustrate performance advantages of RMA-MCS and RMA-RW over state-of-the-art distributed locks from the foMPI implementation of MPI-3 RMA [20].
Comparison Targets We consider D-MCS and both foMPI locking schemes: a simple spin-lock (foMPI-Spin) that enables  mutual exclusion, and an RW lock (foMPI-RW) that provides both shared and exclusive accesses to the CS. Selection of Benchmarks We conduct six series of experiments. The latency benchmark (LB) measures the latency of both acquiring and releasing a lock; an important performance metric in workloads such as real-time queries. Four other analyses obtain throughput under varying conditions and parameters. The emptycritical-section benchmark (ECSB) derives the throughput of acquiring an empty lock with no workload in the CS. The single-operation benchmark (SOB) measures the throughput of acquiring a lock with only one single operation (one memory access) in the CS; it represents irregular parallel workloads such as graph processing with vertices protected by fine locks. Next, the workload-critical-section benchmark (WCSB) covers variable workloads in the CS: each process increments a shared counter and then spins for a random time (1-4 s) to simulate local computation. The wait-after-release benchmark (WARB) varies lock contention: after release, processes wait for a random time (1-4 s) before the next acquire. The throughput experiments represent data-and communication-intensive workloads. Finally, we integrate and evaluate the proposed locks with a distributed hashtable (DHT) to cover real codes such as key-value stores.
Varied Parameters To evaluate various scenarios, we vary: , , , and . Unless stated otherwise, we set the fraction of writers = 0.2% as it reflects Facebook workloads [50]; however, we also evaluate other values.
Experimentation Methodology To calculate the latency, we derive the arithmetic mean of 100,000 operations per process (for each latency benchmark). Throughput is the aggregate count of lock acquires or releases divided by the total time to run a given benchmark. 10% of the first measurements are discarded (warmup). All time measurements are taken using a high precision rdtsc timer [26].
Experimental Setup We conduct experiments on CSCS Piz Daint (Cray XC30). Each node has an 8-core HT-enabled Intel Xeon E5-2670 CPU with 32 GiB DDR3-1600 RAM. The interconnection is based on Cray's Aries and it implements the Dragonfly topology [19,31]. The batch system is slurm 14.03.7. We use C++ and the GNU 5.2.40 g++ compiler with -O3 optimizations. The utilized Cray DMAPP is 7.0.1-1.0501.8315.8.4.ari. Unless stated otherwise, we use all the compute resources and run one MPI process per one HT resource (16 processes per one compute node).

Machine Model
We consider two levels of the hierarchy: the whole machine and compute nodes, thus = 2.
Implementation Details We use the libtopodisc [23] library for discovering the structure of the underlying compute nodes and for obtaining MPI communicators that enable communication within each node. We group all the locking structures in MPI allocated windows to reduce the memory footprint [20].

Performance Analysis of RMA-MCS
We present the results in Figure 3. The latency of RMA-MCS is lower than any other target. For example, for = 1, 024, it is ≈10x and ≈4x lower than foMPI-Spin and D-MCS, respectively. This is because foMPI-Spin entails lock contention that limits performance. In addition, both foMPI-Spin and D-MCS are topology-oblivious. Then, the throughput analysis confirms the advantages of RMA-MCS across all the considered benchmarks. The interesting spike in ECSB and SOB is because moving from = 8 to = 16 does not entail inter-node communication, initially increasing RMA-MCS's and D-MCS's throughput. We conclude that RMA-MCS consistently outperforms the original foMPI design and D-MCS.

Performance Analysis of RMA-RW
We now proceed to evaluate RMA-RW. First, we analyze the impact of various design parameters ( Figure 4) and then compare it to the state-of-the-art ( Figure 5). Due to space constraints, we only present a subset of the results, all remaining plots follow similar performance patterns.

Influence of
. We first discuss how different values impact performance. We consider ∈ {1, 2, 4} (one physical counter on each compute node and every 2nd and 4th compute node, respectively). We also vary the number of counters on one node (1,2,4,8). The results are presented in Figure 4a. First, lower entails more work for writers that must access more counters while changing the lock mode. This limits performance, especially for high , because of the higher total number of counters. Larger increases throughput (less work for writers), but at some point (e.g., = 512 a counter on every 2nd node) the overhead due to readers (contention and higher latency) begins to dominate. We conclude that selecting the proper is important for high performance of RMA-RW, but the best value depends on many factors and should be tuned for a specific machine. For example, higher might  entail unpredictable performance penalties on Cray XE because the job scheduler does not enforce contiguous job allocations [8].
5.2.2 Influence of , . Next, we analyze the performance impact of , in the considered system ∈ {1, 2}. We fix = 25% to ensure that there are multiple writers per machine element on each level. We start with various =1 , : the maximal number of writer acquires before the lock is passed to the readers; see Figure 4b. As expected, smaller product increases throughput because more readers can enter the CS, but reduces fairness as writers wait longer. In the second step, we analyze how varying each , impacts performance. We first fix =1 , = 1000. As = 2, we use ,2 ∈ (10, 25, 50) and ,1 ∈ (100, 40,20). The outcome is shown in Figure 4c. When more writers consecutively acquire the lock within one node (higher ,2 ), the throughput increases. Still, the differences between the considered options are small (up to 25% of the relative difference), especially for lower . This is because of smaller amounts of inter-node communication. Interestingly, options that increase throughput (e.g.,  also increase latency, see Figure 4d. We conjecture this is due to improved fairness caused by smaller ,2 (more processes from different nodes can acquire the lock). However, the average latency increases because other writers have to wait for a longer time. = 0.2%. The throughput for ∈ {1,000 ; 2,000} drops significantly for > 512 due to the higher overhead of writers. Contrarily, increasing improves the throughput significantly. This is because the latency of readers is lower than that of writers and a higher entails a preference of readers. However, the larger the longer the waiting time for writers is. Finally, we analyze the relationship between and in more detail; see Figure 4f. Here, we vary ∈ {2%, 5%}. The results indicate no consistent significant advantage (<1% of relative difference for most ) of one threshold over others within a fixed .

5.2.4
Comparison to the State-of-the-Art. We now present the advantages of RMA-RW over the state-of-the-art foMPI RMA library [20]; see Figure 5. Here, we consider different rates. As expected, any RW distributed lock provides the highest throughput for = 0.2%. This is because readers have a lower latency for acquiring a lock than writers and they can enter the CS in parallel. The maximum difference between the rates = 0.2% and = 2% is 1.8x and between = 0.2% and = 5% is 4.4x. We then tested other values of up to 100% to find out that for > 30% the throughput remains approximately the same. At such rates, the throughput is dominated by the overhead of writers that enter the CS consecutively.
In each case, RMA-RW always outperforms foMPI by >6x for ≥ 64. One reason for this advantage is the topology-aware design. Another one is the presence of , and that prevent one type of processes to dominate the other one resulting in performance penalties.

Case Study: A Distributed Hashtable
We now illustrate how RMA-RW accelerates a distributed hashtable (DHT) that represents irregular codes. Our DHT stores 64-bit integers and it consists of parts called local volumes. Each local volume consists of a table of elements and an overflow heap for elements after hash collisions. The table and the heap are constructed with fixed-size arrays. Every local volume is managed by a different process. Inserts are based on atomic CASes. If a collision happens, the losing thread places the element in the overflow list by atomically incrementing the next free pointer. In addition, a pointer to the last element is also updated with a second CAS. Flushes are used to ensure memory consistency.
We illustrate a performance analysis in Figure 6. In the benchmark, − 1 processes access a local volume of a selected process with a specified number of inserts and reads targeted at random hashtable elements. We compare the total execution time of foMPI-A (a variant that only synchronizes accesses with CAS/FAO), foMPI-RW, and RMA-RW. For ∈ {2%, 5%, 20%} RMA-RW outperforms both the remaining variants. For = 0%, foMPI-RW and RMA-RW offer comparable performance.

DISCUSSION
Using Different RMA Libraries/Languages In our implementation, we use MPI RMA. Still, the proposed schemes are generic and can be implemented using several other existing RMA/PGAS libraries/languages that support the required operations described in Listing 1. We illustrate this in Table 3 (we omit the distinction between blocking and non-blocking operations as any type can be used in the proposed locks). The analysis indicates that RMA-MCS and RMA-RW can be used in not only traditional HPC domains (by utilizing UPC, SHMEM, or RDMA/IB), but also in TCP/IP-based settings (by using iWARP).
Selecting RMA-RW Parameters To set the parameters, we first find an appropriate value for . This is because our performance analysis indicates that has on average the highest impact on performance of both readers and writers. Here, our evaluation indicates that placing one counter per compute node results in a reasonable balance between reader throughput and writer latency. In the second step, we further influence the reader/writer performance tradeoff by manipulating with and , . To reduce the parameter space, we fix as indicated in Table 2. Selecting , depends on the hardware hierarchy and would ideally incorporate several performance tests before fixing final numbers. One rule of the thumb is to reserve larger values for , associated with components with higher inter-component communication costs, such as racks; this may reduce fairness, but increases throughput.

RELATED WORK
Queue-Based Locks The well-known traditional examples of this family are CLH [16,35] and MCS [37]. Yet, they are oblivious to the memory hierarchy and cannot use this knowledge to gain performance. More recently, Radovic and Hagersten [41] proposed a hierarchical backoff lock that exploits memory locality: a thread reduces its backoff delay if another thread from the same cluster owns the lock. This increases the chance to keep the lock within the cluster, but introduces the risk of starvation. Luchangco et al. [34] improved this scheme by introducing a NUMA-aware CLH queue that ensures no starvation. Yet, it considers only two levels of the memory hierarchy. Chabbi et al. [12] generalized it to any number of memory hierarchy levels. Similarly to our scheme, they introduce an MCS lock for each level. Yet, they do not target DM machines. None of these protocols can utilize the parallelism of miscellaneous workloads where the majority of processes only read the data. RW Locks There exist various traditional RW proposals [28,32]. Recently, Courtois et al. [15] introduced different preference schemes that favor either readers (a reader can enter the CS even if there is a writer waiting) or writers (a writer can enter the CS before waiting readers). Yet, this protocol neither prevents starvation nor scales well. Mellor-Crummey and Scott [38] extended their MCS lock to distinguish between readers and writers. This algorithm however does not scale well under heavy read contention. Next, Krieger et al. [32] use a double-linked list for more flexibility in how processes traverse the queue. Yet, there is still a single point of contention. Hsieh and Weihl [28] overcome this by trading writer throughput for reader throughput. In their design, each thread has a private mutex; the readers acquire the lock by acquiring their private mutex but the writers need to obtain all mutex objects. This introduces a massive overhead for the writers for large thread counts. Other approaches incorporate elaborate data structures like the Scalable Non-Zero Indicator (SNZI) tree [33] that traces readers in the underlying NUMA hierarchy for more locality. Yet, writers remain NUMA-oblivious. Calciu et al. [11] extend this approach with an RW lock in which both readers and writers are NUMAaware. This design improves memory locality but it only considers two levels in a NUMA hierarchy. None of these schemes address DM environments. Distributed Locks To the best of our knowledge, little research has been performed into locks for DM systems. Simple spin-lock protocols for implementing MPI-3 RMA synchronization were proposed by Gerstenberger et al. [20]. Some other RMA languages and libraries (e.g., UPC) also offer locks, but they are not RW, their performance is similar to that of foMPI, and they are hardwareoblivious.
We conclude that our work offers the first lock for DM systems that exploits the underlying inter-node structure and utilizes the RW parallelism present in various data-and communication-intensive workloads.

CONCLUSION
Large amounts of data in domains such as graph computations require distributed-memory machines for efficient processing. Such machines are characterized by weak memory models and expensive inter-node communication. These features impact the performance of topology-oblivious locks or completely prevent a straightforward adoption of existing locking schemes for shared-memory systems.
In this work, we propose a distributed topology-aware Reader-Writer (RMA-RW) and MCS lock that outperform the state-of-theart. RMA-RW offers a modular design with three parameters that offer performance tradeoffs in selected parts of the lock. These are: higher lock fairness or better locality, larger throughput of readers or writers, and lower latency of readers or writers. This facilitates performance tuning for a specific workload or environment. RMA-RW could also be extended with adaptive schemes for a runtime selection and tuning of the values of the parameters. This might be used in accelerating dynamic workloads.
Microbenchmark results indicate that the proposed locks outperform the state-of-the-art in both latency and throughput. Finally, RMA-RW accelerates a distributed hashtable that represents irregular workloads such as key-value stores.