Mahmoud Ismail

Gautier Berthou
email: gautier@sics.se

Jim Dowling
email: jdowling@kth.se

Salman Niazi

Leader Election Using NewSQL Database Systems

Introduction

One of the main difficulties when designing a replicated distributed system is to ensure that the nodes will reach agreement on the actions to take. Agreement protocols are complex to design and inefficient in terms of throughput and latency, for example, classical Paxos [START_REF] Lamport | Paxos made simple[END_REF] in a failure-recovery model. As a result, most distributed systems rely on a unique leader node to coordinate the tasks running in the system. For this leader pattern to work correctly the nodes need to be able to solve the general agreement problem [START_REF] Chandra | The weakest failure detector for solving consensus[END_REF] in order to agree on which one of them is the leader. Solving this problem is the purpose of the leader election protocol.

Leader election protocols are a fundamental building block that play a central role in many scalable distributed systems such as stateful middleware [START_REF] Junqueira | The life and times of a zookeeper[END_REF], distributed filesystems [START_REF] Shvachko | The hadoop distributed file system[END_REF], and distributed databases [START_REF] Ronström | Recovery Principles of MySQL Cluster 5.1[END_REF]. The typical role of a leader is to propose global state updates and to disseminate them atomically among the nodes. Having a unique leader is imperative to avoid multiple leaders proposing conflicting updates that would compromise the integrity of the system. Additionally, the failure of the leader should not affect the system availability. Moreover, the detection of the leader failure and the election of a new leader should be low latency events (at most, in the order of seconds).

Implementing an algorithm that provides both uniqueness of the leader and low latency is very challenging. In order to avoid errors and to curtail development time, developers often rely on third-party, standalone coordination services such as Chubby [START_REF] Burrows | The chubby lock service for loosely-coupled distributed systems[END_REF] and ZooKeeper [START_REF] Hunt | Zookeeper: Wait-free coordination for internet-scale systems[END_REF]. These services have the advantages of being widely used and well tested but they introduce additional operational costs and complexity as they must be kept running on at least three machines if the leader-election service itself is to be highly available.

Many existing distributed systems use highly available relational databases or key-value stores to manage their persistent data. Why not build the leader election service using the database as a shared memory? This would allow developers to exploit the leader pattern without paying the extra operational cost of a dedicated coordination service. Implementing the leader election using shared memory is not a new problem; Guerraoui [START_REF] Guerraoui | A Leader Election Protocol for Eventually Synchronous Shared Memory Systems[END_REF] and Fernandez [START_REF] Fernandez | Electing an eventual leader in an asynchronous shared memory system[END_REF][START_REF] Fernandez | A timing assumption and a t-resilient protocol for implementing an eventual leader service in asynchronous shared memory systems[END_REF] have shown that a leader election service can be implemented using shared memory. However, in partially synchronous systems, these solutions do not guarantee that there will be a unique leader while the system is unstable. They only guarantee that nodes will eventually agree on a leader once the system has stabilized. As a result, these solutions are not widely used in production systems.

In contrast, we propose an algorithm based on locking and transaction primitives provided by the database to guarantee that there is at most one leader in the system at any given time. Traditional highly available relational database management systems are not suitable for building our leader election service, since it can take long time for transactions to complete if a database node failure occurs, which would slow down the leader election process. However, NewSQL systems have emerged as a new class of distributed, in-memory databases that are optimized for on-line transaction processing (OLTP) and have low timeouts for transactions, thus, making them a viable platform for building our leader election service.

In this paper, we present a practical leader election service based on shared memory in a NewSQL database. Our implementation uses the Network Database (NDB) storage engine for MySQL Cluster [START_REF] Ronström | Recovery Principles of MySQL Cluster 5.1[END_REF], but our approach is generalizable to all NewSQL databases. Our main contribution is to prove that two-phase commit can be used to implement practical leader election algorithm. We validate our algorithm and show that its performance is comparable to a leader election algorithm implemented using ZooKeeper [START_REF] Hunt | Zookeeper: Wait-free coordination for internet-scale systems[END_REF] for cluster sizes of up to 800 processes.

NewSQL Database Systems

Although a database can be used as shared memory to implement leader election, some databases do not provide sufficient primitives to implement a reliable leader election service. Our leader election service requires a highly available database with support for transactions and locking primitives. Additionally, the database must ensure that database node failures and slow clients do not cause transactions to take too long to complete (commit or abort). We will now discuss different types of database systems and their suitability for leader election service.

Highly available relational databases typically provide high-availability using either an active-standby replication protocol that provides eventually consistent guarantees for data (as used in SQLServer [START_REF]Sqlserver @ONLINE[END_REF] and MySQL [START_REF]Mysql :: The world's most popular open source database @ONLINE[END_REF]) or a shared-state replication protocol, as used in Oracle RAC [START_REF]White paper: Xa and oracle controlled distributed transactions @ON-LINE[END_REF]. For the activestandby model, a crash of the active node will result in the leader election service being unavailable until the standby node takes over. There are no guarantees on how long this failover will take, and, in practice, it can take from seconds up to minutes to complete depending on the degree of lag at the standby node. Moreover, until the failover completes, the system remains vulnerable to failures as the standby node now becomes a single point of failure. For shared-state databases, it can take up to a minute for transactions to complete if a database node failure occurs (the default distributed lock timeout in Oracle RAC is 60 seconds [START_REF]White paper: Xa and oracle controlled distributed transactions @ON-LINE[END_REF]). For these reasons, traditional highly available relational databases are not suitable for building leader election services.

NoSQL systems are highly available, but they only provide eventually consistent guarantees for data [START_REF] Lakshman | Cassandra: A decentralized structured storage system[END_REF][START_REF]Riak, basho technologies @ONLINE[END_REF]. This make them unsuitable as the basis for a leader election service, as eventually consistent data may lead to multiple leaders in the system.

NewSQL systems are a new class of highly available databases that can scale in performance to levels reached by NoSQL systems, but still provide ACID guarantees and a SQL-based declarative query interface [START_REF] Stonebraker | The end of an architectural era: (it's time for a complete rewrite)[END_REF][START_REF] Özcan | Are we experiencing a big data bubble?[END_REF]. NewSQL systems achieve high performance and scalability by redesigning the internal architecture of traditional databases, often to a shared-nothing architecture, that take better advantage of modern multi-core hardware along with increasingly cheap in-memory storage. NewSQL systems can be scaled-out by adding additional nodes. What makes NewSQL systems a viable platform for building a leader election service is that they typically have low timeouts for locks and transactions. Some notable NewSQL systems are the NDB storage engine for MySQL Cluster [START_REF] Ronström | Recovery Principles of MySQL Cluster 5.1[END_REF], FoundationDB [START_REF]Multi-model database -foundationdb @ONLINE[END_REF], and VoltDB [START_REF]In-memory database, newsql and real-time analytics -voltdb @ONLINE[END_REF].

System Model and Eventual Leader Election

Processes The system consists of a time varying finite set of processes p 1 , p 2 , p 3 ... p n . Each process has a unique id assigned by a function that returns monotonically increasing ids. All the processes are assumed to behave according to their protocol specification, that is, the processes are not Byzantine. A process can fail by crashing, but until a process crashes it will execute the protocol and it will not halt for an indefinite amount of time. When a process fails it stops executing all operations. A failed process can recover after the failure, but it is assigned a new id by the monotonically increasing function. There is no restriction on the number of processes that can fail or join during the execution of leader election protocol.

The underlying system is partially asynchronous, as it is impossible to develop a leader election service for purely asynchronous systems [START_REF] Fischer | Impossibility of distributed consensus with one faulty process[END_REF]. In partially synchronous systems there are positive upper and lower bounds on the communication and processing latency. These synchrony primitives of the system are eventually determined by the application. Before these time bounds are determined, a distributed application may not function as expected. The time after which the lower and upper time bounds hold is called global stabilization time (GST). The protocol proceeds in rounds. The duration of these rounds expand until the GST is reached. Moreover, each process' local clock drift is significantly smaller than the round time of protocol.

Shared Memory All processes communicate through reliable atomic registers (shared memory) implemented using rows in a table in the database. A reliable atomic register is always available, moreover, if two read operations r 1 and r 2 return w 1 and w 2 respectively and r 1 precedes r 2 then w 1 precedes w 2 . Atomic registers can easily be implemented in a relational database using a strong enough transaction isolation level. To be considered correct, a process must successfully read and write the register values within a heartbeat period, that can expand during the execution of the protocol.

Leader Election Service Eventually a correct process with the lowest id in the system will be elected as the leader. The service ensures that a correct leader is elected again in the subsequent rounds. Our service provides stronger guarantees than Ω [START_REF] Chandra | The weakest failure detector for solving consensus[END_REF]. With Ω there could be multiple leaders if the GST has not been reached. With the help of transactions, our leader election guarantees at most one leader at any given time and guarantees following properties:

-Integrity: there should never be more than one leader in the system.

-Termination: a correct process eventually becomes a leader.

-Termination: all invocations of the primitive getLeader() invoked by a correct process should return the leader's id.

Leader Election in a NewSQL Database

Logically, all processes communicate through shared registers (implemented as rows in a table ). Each process has its own counter that it updates periodically (in a transaction) to indicate that it is still alive. Each process maintains a local history of the all processes descriptors. Process descriptor contains id, counter, ip and port information. Using the local history a process is declared dead if it fails to update its counter in multiple consecutive rounds. A process declares itself to be the leader when it detects that it has the smallest id among all the alive processes in the system. The leader evicts failed processes, and it is also responsible for increasing the heartbeat round time to accommodate slow processes.

All processes run in parallel, concurrency control could be handled with a transaction isolation level set to serializable, ensuring that conflicting transactions will execute one after another. For example, if two processes, P a and P b , want to become leader simultaneously then the transactions will automatically be ordered such that if P a manages to execute first then P b is put on hold. The transaction P b waits until transaction P a has finished.

However, due to poor performance [START_REF] Thomson | Calvin: Fast Distributed Transactions for Partitioned Database Systems[END_REF], NewSQL systems typically do not provide serializable as the default transaction isolation level, if they even support it at all. The strongest isolation level supported by NDB is the read committed isolation level, guaranteeing that any data read is committed at the moment it is read. However, it is not sufficient for implementing a reliable leader election service. We use row-level locking to implement stronger isolation levels for transactions. Row-level locking complicates the design, but allows for more finegrained concurrency control and thus, higher throughput. Every processes is an element of one of two disjoint sets. The first set contains the majority of processes. These are non-leader processes that only update their counter in each round. The second set of processes contains the leader process, processes contending to become the leader, and processes that have not yet obtained a unique id. Usually this group is very small, and it depends upon the amount of churn in the system.

All the processes in the first set can run concurrently as they only update their own counters. However, the processes in the second set may take decisions or change the state of the global variables which can effect other processes. Therefore, all the transactions of the processes in the second set are serialized. For example, assume the leader wants to evict a slow process. By taking exclusive locks, the leader process prevents the slow process from committing any updates to shared state. When the slow process' transaction is scheduled, it will notice its id is missing and it will have to rejoin the system. Similarly, if two processes are contending to become the leader then their operations should be serialized to prevent a system state where there are multiple leaders. Moreover, the first round of new processes are also serialized to generate monotonically increasing ids.

Leader Election Rounds

Each round encapsulates its operations in a transaction that starts by taking a lock on the VARS register which acts as a synchronization point. Processes belonging to the first group acquire shared locks while the processes in the second group acquire exclusive locks on the VARS register, lines 5 -9.

After acquiring locks on the VARS register all the processes descriptors are read without any locks (read committed). The processes update their counters and check if they can become the new leader. Each process maintains a history of process descriptors to identify dead processes, lines 12 -14. Now, we explain these operations in more detail from the perspective of both groups of processes.

A new process starts by taking exclusive locks in the first round. It obtains a new monotonically increasing id and stores its descriptor, lines 32 -33. An exclusive lock is required to update the maximum process id in the VARS register. An evicted process will not find its descriptor, as it has been deleted. The evicted process cannot obtain a new process id if it does not hold an exclusive lock on the VARS register. In such a case, the transaction is immediately retried using exclusive locks, see lines 5, and 27 -31. Additionally, the evicted process sets a flag to inform the leader that it was evicted prematurely, see line 30.

The service then checks for changes in the group membership. A process is declared dead if it fails to update its counter in multiple consecutive rounds. The threshold, M ax mhb , determines the number of rounds a process can miss before it is declared dead. The M ax mhb is usually set to ≥ 2. The process elects itself to be the leader if it has the smallest id among the alive processes. The leader process cleans the DESCRIPTORS register by removing the dead processes. If a non-leader process, that holds a shared lock, finds out that it can become the leader then it immediately retries the transaction with exclusive locks. It becomes the leader and removes the dead descriptors. If the process does not have the smallest id then it sets its role to non_leader and stores the descriptor of the process that has smallest id in a local variable, lines 34 -44.

Global Stabilization Time (GST)

The time bounds for communication and processing latencies are not known in advance. For large systems the initial round time for periodically updating the counter may not be sufficiently long enough so that all processes mange to update their counters in a single round. Moreover, the round time must be automatically adjusted to cater for slow processes; otherwise, the system may not stabilize. In our implementation, only the leader process increases the round time by updating the VARS register (which is read by all processes).

Slow processes are evicted by the leader. When a process finds out that it was evicted, it obtains a new id and set a flag in the VARS register to notify the leader that it was wrongfully suspected. When the leader process finds that the evicted flag is set it increases the round time by a constant value ∆, see lines 16 -18.

Leader Lease

Our solution ensures that there is never more than one leader in the system. However, this invariant is difficult to enforce before the GST has reached. Additionally, in order to reduce contention on the registers, methods like isLeader() and getLeader() return information stored in the local variables. On a slow process these variables may contain stale values. For instance, assume a slow process, L a , becomes the leader. After becoming the leader L a fails to update its counter in multiple consecutive rounds. Later, a new process becomes a leader and L a is evicted. However, L a will remain oblivious of the fact that it has been evicted, and its function isLeader() will keep on returning true until L a manages to read new values from the registers.

In order to ensure integrity of the leader election service each leader process stores a local lease. Whenever the leader process updates its counter, it acquires a lease for the duration of (T p * M ax mhb -µ). The constant, µ, is to accommodate for clock drifts. Before committing the transaction, a timestamp is stored in L hbt , which indicates the start of the leader lease time, line 20. The lease is the the maximum time during which the leader cannot be evicted by other processes. If the leader is slow and it fails to update its counter then the lease will eventually expire and the process will voluntarily yield its leader role, line 45 -50. The election of a new leader will happen after the lease of previous leader has expired, see theorem 1 for more details.

Dealing with Failures

Note that the transactions only guarantee the atomicity of the registers. Read committed isolation ensures that, during transaction execution, partial results (changes in the registers) are not be visible to other transactions until the transaction has committed. When a transaction fails the database rollbacks only the partial changes in the registers. However, it is the responsibility of the application to rollback all the local variables, such as role, L hbt , T p , and id. For clarity reasons we do not show code listing to rollback local variables.

In this section we prove the safety (at most one leader invariant) and the liveness (termination) properties of our leader election algorithm.

Theorem 1. There is never more than one leader in the system.

Proof. In order to prove that there cannot be two leaders, L s and L n , in the system at the same time we will prove that (I) two processes cannot declare themselves as leader simultaneously (II) a process cannot become leader while another process still sees itself as leader.

Case I: In order to become a leader both the processes, L s and L n , need to acquire exclusive locks at the beginning of the transaction. As a result the transactions for L s and L n will be serialized. If L s manages to acquire the exclusive lock first, it will update the counter and elect itself as a leader (assuming the transaction commits). L n will wait until L s releases the lock. L n will acquire the locks after L s commits the transaction, and it will find out that L s has already became the leader. As a result L n will not declare itself the leader.

In a case where L s halts after acquiring the exclusive lock, the database will timeout L s 's transaction and release the lock. The database will rollback the transaction and L s will have to re-acquire the exclusive lock in order to become a leader. L s has to reset its local role variable to non_leader.

Case II: When a process becomes the leader it acquires a lease that is valid for (T p * M ax mhb -µ). The process voluntarily gives up the leader role if it fails to renew the lease before it expires. In order to ensure that a process L n cannot become the leader while a slow leader L s still has a valid lease, the protocol needs to ensure that the time needed by L n to suspect L s is higher than the time duration of L s 's lease.

Assume the processing and network latencies of the process L n are zero. Furthermore, the process L n performs a heartbeat (read and update the registers) soon after L s commits an update. The process L n will find out that L s is alive. After that, it will have to read the registers M ax mhb times before it can suspect L s . M ax mhb heartbeat rounds will take (T p * M ax mhb ) seconds, assuming that L n 's clock drift is negligible. Thus, the minimum time that L n needs to suspect L s and elect itself as leader is (T p * M ax mhb ), which is strictly more than the lease time of L s . The assumption that L n does not have any processing and network latencies represents a worst case scenario. In a real system the latencies will always have some positive value which will increase the time needed by L n to declare L s as dead. The constant, µ, should be configured to be higher than the upper bound on clock drift for any process in the system. In practical systems, NTP, GPS, atomic clocks are used to ensure low bounds on clock drift.

An illustration of this worst case scenario is presented in Figure 1 where M ax mhb = 2. The leader L s is faulty and it does not update the counter after t 0 . At time t 2 the process L n detects that L s is faulty and it can become the new leader. As L n does not hold the exclusive lock, it immediately retries the

X

Fig. 1: Black and white circles represent exclusive and shared locks, respectively. Process L s is a slow leader that does not update the counter after t 0 . Process L n becomes leader after the lease for L s expires.

transaction, acquires the exclusive lock and becomes the leader. The lease of L s expires after T p * 2 -µ, which is less than the time L n must wait to detect the failure of the process L s .

Theorem 2. A correct process eventually becomes the leader.

Proof. Assume a system configuration of p 1 , p 2 , p 3 ...p k ...p n processes. Additionally, assume p k is the only correct process that repeatedly manages to update its counter every M ax mhb rounds. All the other processes are incorrect such that these processes do not always manage to update the counter within M ax mhb rounds. A correct process is never suspected by any process in the system. We show that the process p k eventually becomes a leader and retains the leader role in the subsequent rounds. Assume all the processes have just started and the history of each process is empty. The process p 1 will declare itself to be the leader and it will retain the role for T p * M ax mhb -µ seconds. During the first M ax mhb rounds no process will be evicted. If p 1 is an incorrect process which fails to update the counter, its lease for the leadership will expire. In the round (M ax mhb + 1) a process with least id that managed to update the counter, while p 1 was the leader, will become the new leader. The new leader will evict p 1 along with other suspected processes, if any. An evicted process might rejoin the system, but it is will be assigned with a new id by the monotonically increasing function. The eviction of incorrect processes will continue until p k becomes the process with the least id in the system. The process p k will elect itself as the new leader. As the process p k is correct it will not miss any heartbeats and it will retain the leader role in subsequent rounds.

Evaluation

We have implemented the leader election using in-memory, highly-available, distributed database called NDB (Version 7.4.3), the storage engine for MySQL Cluster [START_REF] Ronström | Recovery Principles of MySQL Cluster 5.1[END_REF]. NDB is a real-time, ACID-compliant, relational database with no single point of failure and support for row-level locking. We use the native Java API for NDB, ClusterJ, as it provides lower latency and higher throughput than the SQL API that uses the MySQL Server.

All the experiments were performed on nodes behind a single 1 Gbit switch, where the network round trip time between any two nodes is in single digit millisecond range. The NDB setup consisted of six data nodes (6-core AMD Opteron 2.6 GHz, 32GB RAM) with replication factor of 2. We compare our solution with a leader election solution implemented using ZooKeeper. The ZooKeeper setup consisted of three quorum nodes (6-core AMD Opteron, 32GB RAM). We used the leader election library for ZooKeeper (Version 3.4.6) from the Apache Curator project (Version 2.7.1). Each ZooKeeper client creates a sequential ephemeral node in predetermined directory. Each node registers a watch (callback request) for its predecessor. Upon a node failure its successor is notified. The successor checks if there are any nodes with smaller sequential number. If there are no smaller nodes available then it elects itself as the new leader; otherwise, it registers a new watch for its new predecessor.

In the experiments the initial heartbeat round time was set to 2 seconds and M ax mhb was set to 2. To accurately determine the failover time all the clients were run on a single machine (12-core Intel Xeon 2.8 GHz, 40 GB RAM). All experiments were performed fifteen times and the graphs show the average results, with the error bars showing the standard deviation of the results. In each experiment N processes are started. When all processes have joined the system, the round time is continuously monitored for changes. If it does not change for a certain time (three minutes) then the system is considered stable. After the system has stabilized the leader process is repeatedly killed 50 times to measure failover time.

Figure 2a shows the relation between network size and the time to elect a new leader. Up to 200 processes the service consistently elects a new leader in around five seconds. However, when the network sizes increases beyond 200 nodes the time to elect new leader also increases. This can also be observed in figure 2b which shows the relationship between round time and the network size. For network sizes up to 200 processes, all the processes manage to update the counter before they are suspected by the leader process. However, when the network size increases beyond 200, contention on the registers prevents some processes from writing to the shared register for consecutive heartbeats. The leader processes detects contention on the registers when an evicted process raises the evict flag. The leader process increases the heartbeat delay to release the contention on the registers, which has the side-effect of also increasing the leader failover time. In the experiments, the heartbeat delay increment (∆) was set to 50 milliseconds.

In the implementation of leader election using ZooKeeper, the time to elect a new leader is determined by two configuration parameters: tick time and session timeout. We set these values as low as possible to quickly elect a new leader in case of a leader failure. The lowest allowable values for tick time is 2 seconds, and session timeout is 4 seconds. In order to accurately determine the fail over time all leader election processes were run on one (12-core Intel Xeon 2.8 GHz, 40 GB RAM) machine. Up to 400 processes ZooKeeper constantly elects Fig. 2: Performance of leader election service with the default configuration settings for NDB (MySQL Cluster). Figure 2a shows the average time to elect a new leader when the current leader process fails. Figure 2b shows the increase in the heartbeat round time when the leader detects contention on the registers.

a new leader in six seconds. However the time to elect new leader starts to drop if we increase the number of clients on the same machine. This is because of the contention on the CPU and main memory because of which the processes slowed down. When a leader is killed it may have already skipped a heartbeat. This results in quicker reelection of a new leader. Due to memory limitations we could not add more than 800 processes in the experiment.

Related work

Leader election is a well studied problem. All the related research can be classified into two broad categories: shared memory and message passing based leader election protocols. Guerraoui et al. presented the first failure detector, Ω, that was implemented using shared memory for an eventually synchronous system [START_REF] Guerraoui | A Leader Election Protocol for Eventually Synchronous Shared Memory Systems[END_REF]. The protocol is write optimal, only the leader process writes to the shared memory, and all other non-leader processes only read shared memory. Fernandez et al. further investigated the problem in systems where all processes are not eventually synchronous [START_REF] Fernandez | Electing an eventual leader in an asynchronous shared memory system[END_REF][START_REF] Fernandez | A timing assumption and a t-resilient protocol for implementing an eventual leader service in asynchronous shared memory systems[END_REF]. In [START_REF] Fernandez | Electing an eventual leader in an asynchronous shared memory system[END_REF], they propose solutions for systems which require only one process to eventually behave synchronously. All other process can behave fully asynchronously provided that their timers are well behaved. In [START_REF] Fernandez | A timing assumption and a t-resilient protocol for implementing an eventual leader service in asynchronous shared memory systems[END_REF], two t-resilient protocols are presented that require a single, eventually synchronous, process and t -f processes with well behaved timers, where t is the maximum number of processes that may fail, and f is maximum number of processes that can fail in a single run. For synchronous systems, a leader election algorithm using shared memory is presented in [START_REF] Sanz-Marco | Efficient leader election for synchronous shared-memory systems[END_REF], where a semaphore is used to prevent multiple writers from concurrently updating the counter.

The first leader election protocols using message passing are timer-based. Processes send messages to each other to indicate they are alive. A process is suspected if it fails to send a heartbeat message within a time bound. If a heartbeat is received from a suspected process the timer is increased to accommodate for slow processes. Eventually time bounds for processing and communication latencies are determined for the given system by successively increasing the timer upon receiving a message from a suspected process. Some notable leader election protocols in the message passing paradigm using timers are [START_REF] Aguilera | On implementing omega in systems with weak reliability and synchrony assumptions[END_REF][START_REF] Aguilera | Communicationefficient leader election and consensus with limited link synchrony[END_REF][START_REF] Larrea | Optimal implementation of the weakest failure detector for solving consensus[END_REF].

Mostefaoui et al. presented a time-free implementation of failure detectors [START_REF] Mostefaoui | Asynchronous implementation of failure detectors[END_REF]. It allows the communication and processing times to always increase. The protocol assumes that the query-response messages obey a certain pattern. The protocol requires a correct process p and f + 1 processes from a set Q such that if processes repeatedly wait to receive messages from n -f processes, then eventually the messages from p are always among the first n-f messages received by each process in Q. Here n is the system size and f is the maximum number of processes that can fail. The protocol works for any value of f (i.e., 1 ≤ f < n).

Conclusions

We have shown that a reliable leader election service can be implemented using two phase commit transactions in the NDB storage engine, a NewSQL database. Our solution ensures that there is never more than one leader, and the time taken for leader election is comparable to ZooKeeper for clusters of up to 800 processes. Our algorithm enables distributed systems that already use NewSQL databases to save the operational overhead of deploying a third-party service, such as ZooKeeper, for leader election, as our algorithm can easily be re-implemented for other NewSQL databases.

  Average time to elect a new leader.

  Heartbeat round time after GST.

Algorithm 1 :

 1 Leader Election Our database, NDB, supports two main locking modes: shared (read) and exclusive (write) locks. Multiple transactions can concurrently obtain shared locks on an object. However, only one transaction can obtain an exclusive lock on an object.

	29:	forceExclusiveLock = true
	30:	VARS.setEvictFlag()
	31:	return
	32:	id = VARS.incrementMaxID()
	33:	insertDescriptor(id)
	34: procedure leaderCheck
	35:	Ps = history.getSmallestAliveProcess()
	36:	if Ps.id = id then
	37:	if transaction lock mode is not exclusive then
	38:	forceExclusiveLock = true
	39:	return
	40:	role = leader
	41:	removeDeadNodes()			Evict processes
	42:	else
	43:	role = non_leader
	44:	leader = Ps			Possible leader
	Require: VARS 45: procedure isLeader	Atomic Register. Holds max id,Tp and evict flag
	Require: DESCRIPTORS 46: if role = leader then	Set of atomic registers that stores all descriptors
	47: 1: id = ⊥, role = non_leader, leader = ⊥ elapsed_time = currentTime() -L hbt 48: if elapsed_time < (Tp * M ax mhb -µ) then	Lease check
	2: procedure periodicHeartbeatTask 49: return true
	3: 50:	while true do return false
	4: 5: 51: procedure getLeader begin transaction if role = leader | id = ⊥ | forceExclusiveLock then 6: 52: if role = leader & isLeader() then acquire exclusive lock on VARS register 7: 53: return this forceExclusiveLock = false 8: 54: else if role = non_leader then else 9: acquire shared lock on VARS register 55: return leader	Begin new round
	10:	read all DESCRIPTORS		No locks needed
	11:
	12:	updateCounter()
	13:	leaderCheck()
	14:	DESCRIPTORS	history		Add to history
	15:
	16:	Tp = VARS.getTimePeriod()
	17:	if role = leader & VARS.evictFlag = true then
	18:	Tp = VARS.updateTimePeriod(Tp + ∆)
	19:	VARS.evictFlag = f alse
	20:	L hbt = currentTime()	Leader's lease start time
	21:	commit transaction
	22:	sleep(f orceExclusiveLock ? 0 : Tp)	Immediately retry with higher locks
	23: procedure updateCounter
	24:	if id ∈ DESCRIPTORS then
	25:	updateDescriptor(id, getCurrentCounter()+1)
	26:	else
	27:	if id != ⊥ then			Case: evicted
	28:	if transaction lock mode is not exclusive then

4.1 Shared Memory Registers

We implement shared memory registers using rows in tables. Transactions ensure atomicity of the registers. The atomic register VARS stores global parameters such as the maximum allocated process id, and the duration of heartbeat rounds. The maximum allocated process id is used in monotonic id generation. It also stores a boolean flag that is used to change the heart beat round time to cater for slow processes. VARS is backed by single row in a table that contains all the global variables. DESCRIPTORS represents a set of registers that store information about all the alive process. It is backed by a table where each row contains a process descriptor.

Acknowledgements

This work funded by the EU FP7 project "Scalable, Secure Storage and Analysis of Biobank Data" under Grant Agreement no. 317871, and by Swedish e-Science Research Center (SeRC) as a part of the e-Science for Cancer Prevention and Control (eCPC) project.


Swann Perarnau

Rajeev Thakur

Kamil Iskra

Ken Raffenetti

Franck Cappello

Rinku Gupta

Pete Beckman

Marc Snir

Henry Hoffmann

Martin Schulz

Barry Rountree

Distributed Monitoring and Management of Exascale Systems in the Argo Project

à la diffusion de documents scientifiques de niveau recherche, publiés ou non, émanant des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

Introduction

Disruptive new computing technology has already begun to change the scientific computing landscape. Hybrid CPUs, many-core systems, and low-power systemon-a-chip designs are being used in today's most powerful high-performance computing (HPC) systems. As these technology shifts continue and exascale machines emerge, the Argo research project aims to provide an operating system and runtime (OS/R) designed to support extreme-scale scientific computations. To this end, it seeks to efficiently leverage new chip and interconnect technologies while addressing the new modalities, programming environments, and workflows expected at exascale. At the heart of the project are four key innovations: dynamic reconfiguring of node resources in response to workload changes, allowance for massive concurrency, a hierarchical framework for management of nodes, and a cross-layer communication infrastructure that allows resource managers and optimizers to communicate efficiently across the platform. These innovations will result in an open-source prototype system that is expected to form the basis of production exascale systems deployed in the 2020 timeframe.

Argo is designed with a hierarchical approach. The system is organized in enclaves, a set of resources that share the same configuration and can be controlled as a whole. Enclaves can monitor their performance, respond to failures, and control power usage according to a budget. These enclaves form a hierarchy: the top enclave is acting over the whole system, while jobs are contained in their own enclave and the user can subdivide its job enclave further.

We describe here the early stages of an ongoing effort, as part of Argo, to design the services that will take care of creating, keeping track of, and destroying enclaves, as well as monitoring and controlling the resources and failures happening in those enclaves. Our design is influenced by two factors. First, we expect that future machines will differ significantly from current HPC systems in size, failure rate, and fine-grained access to resources. Second, we don't expect the Argo system to behave as commonly available distributed systems such as peer-to-peer or cloud infrastructures. Indeed, an HPC machine will still be composed of known, dedicated resources over which we have total control (as low level as necessary). As a result, we believe that several services typical of a distributed infrastructure can benefit from this unique setup and from being distributed across the hierarchy of enclaves.

In Section 2 we describe more fully the components of the Argo system. In Section 3 we detail typical services we identified as critical to Argo, while also representing distinct communication and control patterns that could benefit from our unique setup. In Section ?? we review related work, and in Section 4 we briefly discuss future work.

2 The Argo Machine: Architecture, Enclaves and Underlying Services

While predicting the exact architecture of future exascale systems is difficult, the Argo project bases its designs on general trends such as those highlighted in the Exascale Software Project Roadmap [START_REF] Dongarra | The International Exascale Software Project Roadmap[END_REF]. We expect the Argo machine to be composed of hundreds of thousands of compute nodes, with each node containing hundreds of cores. Furthermore, those nodes will be linked together by dedicated and highly efficient networks, integrating smart control and monitoring interfaces. We also expect that, in order to meet the U.S. DOE exascale budget limits, complex power management interfaces will be available at all levels of the machine.

The software stack designed by the Argo project to manage such a machine is divided into four key components. First, each compute node will use a customized operating system derived from Linux (NodeOS ). Second, a runtime taking advantage of massive intranode parallelism (Argobots) will be available. Third, a global information bus or backplane will provide advanced communication services on top of the native network. In particular, a distributed key-value store and a pub-sub system will be available to other components. Fourth, the GlobalOS -the focus of this paper-will manage enclaves and their services.

As stated in Section 1, enclaves are logical groups of nodes sharing the same configuration. They are organized in a tree, whose root is the enclave containing all nodes. Inside an enclave, at least one node (master ) hosts the services specific to this enclave. Masters communicate with each other to distribute control of the nodes in the system. We note that nodes are members of all enclaves above them in the hierarchy. In other words, a master node controls all nodes in its subtree in the hierarchy of enclaves, and not just the masters of its subenclaves. Enclaves cannot be created inside a compute node: they are logical constructs intended to manage only the distributed part of the Argo machine. Figure 1 gives an example of a hierarchy of four enclaves distributed across a system, with each enclave having its master replicated across several nodes. This enclave concept is critical to the design of the distributed services managing the Argo machine. In particular, we organize responsibilities so that as much of the node management as possible is delegated to the deepest master in the hierarchy. Conversely, the higher in the hierarchy, the less the interaction between the master and nodes, and the more coarse-grained this interaction is. We give typical examples in the next section.

A A A A B B B B C C D D A B D C Figure 1.
Example of a hierarchy of 4 enclaves. On the left, the squares represent compute nodes, with the filled ones representing the master and its replica. On the right, the hierarchy of enclaves is represented as a tree.

Typical Services in the Argo Machine

Three services critical to GlobalOS can serve as examples of the different ways the enclave hierarchy is used in Argo.

Hierarchical Control Bus for Enclaves

The Argo system is used like any other HPC system: a user submits to the system a job comprising at least the number of nodes required, the time duration, the configuration (to be deployed across the job), and the script or application to run. In our infrastructure, each job is managed as an enclave, a direct child of the root enclave (which contains all nodes). Furthermore, the user can create enclaves inside the job, if he needs distinct parts of the allocated nodes to have different features or different configurations. This process distributes the responsibilities of node management between the masters at different levels of the hierarchy.

The root master has the exclusive role of creating enclaves at its level, and does so only when receiving commands from the job scheduler. A master at the job enclave level might be responsible only for creating subenclaves; and so on.

Similarly, since the user has control over all nodes in a job enclave, commands to reconfigure an enclave, a subtree of an enclave, or just one node might happen at different levels of the hierarchy.

Thus, the chain of command between the root master and a node can be seen as a bus, where commands flow from master to master, going deeper into the hierarchy, with each master affecting the process or injecting new commands until they reach a node. Some commands might be of higher priority (e.g., the root enclave asking for job termination) or be altered (e.g., a command that must span an entire enclave instead of just one node). Nevertheless, all enclave masters are responsible for the continuous working of their enclaves, from the moment the enclave management is delegated to them until the enclave exits (all nodes gracefully exit the enclave, including the master) or is destroyed (massive failure, forceful exit). We are designing this control bus on top of two mechanisms: a logical naming scheme that allows a message or command to be sent to different stages of the control bus (an enclave, a master, or a specific node) and a message broker dedicated to the control bus, present on each node, and forwarding messages across the bus in the right order.

The naming scheme is similar to paths in a filesystem: each enclave is a directory with the root enclave at the top, and nodes are files in those directories. A few special names also exist: '..' (parent enclave), '.' (closest enclave), and '*' (all enclaves/nodes). This naming scheme simplifies sending a command to a specific part of the system, even from a node having only partial knowledge of the hierarchy: it is enough to know the next master in the path. This knowledge is kept in the key-value store available in the Argo system, so that every node knows its own path and how to contact parent and children enclaves. As an example, a node in the enclave D will have the path /B/D/node.

The message broker infrastructure is also simple. A message broker runs on each node and uses the key-value store to route messages to the various masters on the bus. This broker also inspects each message to decide whether commands need to be triggered locally or whether the message needs to be altered or redirected. This design avoids specifying how those messages should be transported. Indeed, as several communication services are made available by the backplane, several communication channels might be available at the same time. For example, it might be more efficient to distribute a command across all nodes of an enclave by using a reliable publish-subscribe interface rather than using point-to-point communication. Such choices are being evaluated but might depend on the specific architecture our system is deployed on.

Distributed Power Management

Since we expect the size of an exascale machine to involve hundreds of thousands of compute nodes, controlling the total power consumption of the system is critical. Indeed, such systems represent tens of megawatts in consumption and a significant cost to any organization. Therefore, idle parts of the system or underutilized resources should be put in low-power modes as much as possible.

To do so, we expect the architecture to provide meters distributed across the system to measure the current power usage and interfaces, similar to dynamic frequency scaling or Intel's RAPL [START_REF] Rountree | Beyond DVFS: A first look at performance under a hardware-enforced power bound[END_REF] on each node. We therefore are designing a distributed power management service as part of GlobalOS.

This service comprises two components: a reader service, running on each node, that periodically reports power consumption to its enclave master and a power control service, installed on enclave masters, that distributes a global power budget across the enclave hierarchy. The power consumption reader design is straightforward. On each node, periodically the local measuring interface is used to gather information, which is then sent to the closest master. Each master then aggregates the data coming from its enclave and sends it to its parent, and so on. This information gathering is made possible by the publish-subscribe service of the backplane, and likewise for the aggregation. The power control service reacts to information coming from the consumption reader as well as global power budget limits set by the machine administrators. The exact algorithms used to distribute this power budget across the hierarchy will be derived from previous work [START_REF] Hoffmann | PCP: A generalized approach to optimizing performance under power constraints through resource management[END_REF][START_REF] Ellsworth | POW: system-wide dynamic reallocation of limited power in hpc[END_REF].

Managing Failures as Exceptions

Given the expected size of an exascale machine, failures-in both hardware and software-will have a statistically greater chance of occurring than in previous systems. Consequently, all components of Argo are being designed with faults in mind. This effort includes replication of masters across the enclave hierarchy, for example.

While failure detection is a complex issue in itself, our focus here is on reactions to failures. We designed a service distributed across the enclave hierarchy that, when notified of failures, acts like an exception system: each master on the path between the failed component and the root enclave will receive the notification one by one, from the deepest in the hierarchy to the highest, and have a chance to act on it. If a master cannot act or resolve the failure, the parent master will take control, and so on.

The specific action a master takes will depend on the component that failed and on the recovery/reaction strategy active on the failure manager. We plan to design several strategies, ranging from restarting the component to destroying the enclave, that users will be able to configure inside their enclaves. We expect the root enclave to have the most complex strategy, with the additional role of notifying administrators in case of unrecoverable hardware failures, and having sole control of powering off and on nodes, for example.

Related Works and Conclusion

Similar projects on a new software stack for exascale systems have recently emerged, including the Hobbes project directed by Sandia National Laboratories in the United States and the post-K project directed by RIKEN in Japan. We are collaborating with them on the design of several components, and we expect these efforts to result in robust and versatile components to manage an entire machine. Recent cloud technologies, such as OpenStack, have also started to address the issues in orchestrating and monitoring distributed resources. These technologies are, however, targeted to systems with smaller numbers of compute nodes, no dedicated high-performance networks, and no static knowledge of the available hardware.

We are still in the design phase of our distributed framework for the provisioning, management, and monitoring of an exascale machine. Our goal is to build on the features of the future architecture and Argo's communication component, focusing on a hierarchical and lightweight solution that is tuned as issues become apparent as the project advances. We hope that in the coming year, with an integrated Argo prototype implemented, our research will move its focus to the study and design of efficient management strategies to be implemented into the various services of the GlobalOS.

Acknowledgments

This work was supported by the U.S. Dept. of Energy, Office of Science, Advanced Scientific Computing Research Program, under Contract DE-AC02-06CH11357. Part of this work was performed under the auspices of the U.S. Department of Energy by Lawrence Livermore National Laboratory under Contract DE-AC52-07NA27344 (LLNL-CONF-669278). The submitted manuscript has been created by UChicago Argonne, LLC, Operator of Argonne National Laboratory ("Argonne"). Argonne, a U.S. Department of Energy Office of Science laboratory, is operated under Contract No. DE-AC02-06CH11357. The U.S. Government retains for itself, and others acting on its behalf, a paid-up nonexclusive, irrevocable worldwide license in said article to reproduce, prepare derivative works, distribute copies to the public, and perform publicly and display publicly, by or on behalf of the Government.


Yaroslav Hayduk

Anita Sobe

Pascal Felber

Dynamic Message Processing and Transactional Memory in the Actor Model

With the trend of ever growing data centers and scaling core counts, simple programming models for efficient distributed and concurrent programming are required. One of the successful principles for scalable computing is the actor model, which is based on message passing. Actors are objects that hold local state that can only be modified by the exchange of messages. To avoid typical concurrency hazards, each actor processes messages sequentially. However, this limits the scalability of the model. We have shown in former work that concurrent message processing can be implemented with the help of transactional memory, ensuring sequential processing, when required. This approach is advantageous in low contention phases, however, does not scale for high contention phases. In this paper we introduce a combination of dynamic resource allocation and non-transactional message processing to overcome this limitation. This allows for efficient resource utilization as these two mechanisms can be handled in parallel. We show that we can substantially reduce the execution time of high-contention workloads in a micro-benchmark as well as in a real-world application.

Introduction

Recent scaling trends lead to ever growing data centers and cloud computing is gaining attention. Further, the scaling trends at the CPU level let us expect increasing core counts in the following years. This causes limitations of performance gains as it is difficult to program distributed and concurrent applications efficiently. The current methods using shared memory do not keep up with the hardware scaling trends and might need to be abandoned. The actor model, initially proposed by Hewitt [START_REF] Hewitt | A universal modular actor formalism for artificial intelligence[END_REF], is a successful message passing approach that has been integrated into popular local and distributed frameworks [START_REF] Karmani | Actor frameworks for the jvm platform: a comparative analysis[END_REF]. An actor is an independent, asynchronous object with an encapsulated state that can only be modified locally based on the exchange of messages. Received messages are processed sequentially avoiding the necessity of locks. With increasing numbers of actors the model is inherently parallel. To sum up, the actor model introduces desirable properties such as encapsulation, fair scheduling, location transparency, and data consistency to the programmer.

While the data consistency property of the actor model is important for preserving application safety, it is arguably too conservative in concurrent settings as it enforces sequential processing of messages, which limits throughput and hence scalability. With sequential processing, access to the state will be suboptimal when operations do not conflict, e.g., modifications to disjoint parts of the state and multiple read operations.

In previous work [START_REF] Hayduk | Speculative concurrent processing with transactional memory in the actor model[END_REF], we improved the message processing performance of the actor model while being faithful to its semantics. Our key idea was to use transactional memory (TM) to process messages in an actor concurrently as if they were processed sequentially. TM provides automatic conflict resolution by aborting and restarting transactions when required. However, we noticed that in cases of high contention, the performance of parallel processing dropped close to or even below the performance of sequential processing. To improve the performance of such high contention workloads, we propose to parallelize the processing of messages in a transactional context with messages that can run in a non-transactional context.

First, we determine the optimal number of threads that execute transactional operations as the performance of an application is dependent on the level of concurrency [START_REF] Didona | Identifying the optimal level of parallelism in transactional memory systems[END_REF], [START_REF] Heiss | Adaptive load control in transaction processing systems[END_REF]. To avoid high rollback counts in high contention phases fewer threads are desired. In contrast to low contention phases where the number of threads can be increased. Second, we extract messages for which we can relax the atomicity and isolation and process them as non-transactional messages.

Much of the contention in our tests was caused by read-only operations on TM objects [START_REF] Hayduk | Speculative concurrent processing with transactional memory in the actor model[END_REF]. Rollbacks could be avoided by relaxing the semantics of read operations such as proposed by Herlihy et al. [START_REF] Herlihy | Software transactional memory for dynamic-sized data structures[END_REF] who introduced early release, which allows to delete entries from the read set. Another way is to suspend the current transaction temporarily, which is called Escape Action [START_REF] Moravan | Supporting nested transactional memory in logtm[END_REF]. These approaches are only partly realized in current STMs such as the Scala STM (based on CCSTM [START_REF] Bronson | CCSTM: A library-based STM for Scala[END_REF]). Scala STM uses Ref objects to manage and isolate the transactional state. The Ref object does not permit accessing its internal state in a non-transactional context. Instead, Scala STM provides an unrecorded read facility, in which the transactional read does not create an entry in the read set but bundles all meta-data in an object. At commit time the automatic validity check is omitted, but may be done by the caller manually. However, in the presence of concurrent writes on the same value, the unrecorded read might cause conflicts and hence performance would degrade. Our proposal is to break the isolation guarantees of the Ref object in specific cases, i.e., if we limit the reads to specific isolated values, we can omit using the unrecorded read and grant direct access. By using direct access for a number of scenarios in the context of the actor model, we can process a substantial amount of read-only messages while not conflicting with messages processed in regular transactions (TM messages).

Possible candidates for such read-only operations can accept inconsistencies while not interrupting with transactional states. Examples are operations that can be used to make heuristic decisions, or operations that are known to be safe because of algorithm-specific knowledge. Furthermore, debugging and log-ging the current state in long-running applications are candidates for read-only operations.

The read-only and the TM messages may require different levels of concurrency. Following this observation, during high contention phases, we reduce the number of threads processing regular TM messages, which in turn allows us to increase the number of threads processing read-only messages. By handling these two message types separately (i.e., providing a separate queue for each of the message types) we can optimally use all available resources. We show the applicability of our approach by using a micro-benchmark and a real-world application.

This paper is organized as follows: In Section 2 we give an overview on related work. In Section 3 we introduce the basics of our former work and in Section 4 we discuss the proposed extensions. The benchmarks are described and evaluated in Section 5. Finally, we conclude our paper.

Background and Related Work

Actor models are inherently concurrent. They are widely used for implementing parallel, distributed, and mobile systems. An actor is an independent, asynchronous object with an encapsulated state that can only be modified locally based on the exchange of messages. It comprises a mailbox in which messages can be queued, as well as a set of dedicated methods for message processing [START_REF] Agha | A foundation for actor computation[END_REF].

The actor model provides macro-step semantics [START_REF] Karmani | Encyclopedia of Parallel Computing[END_REF] by processing messages sequentially. As a consequence, it also guarantees the following properties:

Atomicity. The state of an actor can only be observed before or after operations took place, therefore changes on the state are perceived either all at once or not at all.

Isolation. The actor model forbids any concurrent access to the local state of an actor. This means that any operation on the state of the actor is done as if it were running alone in the system. These characteristics make actor models particularly attractive and contribute to their popularity. Numerous implementations of actor models exist in popular languages like Java, C, C++, and Python. We decided to use Scala, which is a general-purpose language that runs on top of the JVM and combines functional and object-oriented programming patterns. The recent versions of Scala integrate the Akka Framework [START_REF] Haller | On the integration of the actor model in mainstream technologies: A Scala perspective[END_REF] for implementing actors. Scala also supports transactional memory (TM) [START_REF] Goodman | Software transactional memories for scala[END_REF], a programming model that provides atomicity, isolation, and rollback capabilities within transactional code regions [START_REF] Harris | Transactional Memory[END_REF]. TM provides built-in support for checkpointing and rollback, which we exploit for controlling concurrent message processing. Existing actor frameworks such as those surveyed by Karmani et al. in [START_REF] Karmani | Actor frameworks for the jvm platform: a comparative analysis[END_REF] do not include TM and differ regarding the way they handle parallelism. As an example, implementations of Habanero-Scala and Habanero-Java [START_REF] Imam | Integrating task parallelism with actors[END_REF] introduce parallelism by mixing the actor model with the fork-join model (async-finish model). Actors can start concurrent sub-tasks (async blocks) for the handling of a single message. When all sub-tasks complete their execution, the actor resumes its operation and can process further messages. While this approach avoids concurrent access to the actor's state, it must be used carefully as it provides no protection against synchronization hazards such as data races and deadlocks.

Parallel actor monitors (PAM) [START_REF] Scholliers | Parallel actor monitors[END_REF] support concurrent processing by scheduling multiple messages in actor queues. Using PAM, the programmer must understand the concurrency patterns within the application and define applicationspecific schedulers. This may prove particularly challenging for applications where concurrency patterns vary during execution. In contrast, our approach (see Section 4) removes any programmer intervention and automatically allows concurrent executions when possible.

To optimally use the resources and to improve performance, researchers proposed several mechanisms to match the level of concurrency with the current workload. Heiss and Wagner [START_REF] Heiss | Adaptive load control in transaction processing systems[END_REF] discuss the problem of thrashing in concurrent transactional programs. Thrashing is a phenomenon that takes place in phases of high contention in which it is likely that the throughput suddenly drops. They propose three ways to avoid thrashing. First, they propose to set an upper bound that sets the maximum number of concurrent transactions; second, they propose to use analytical models for preventing high contention phases; and third, they propose to monitor the current load and decide dynamically on the best level of concurrency. The last approach is seen as a dynamic optimization problem that considers the relationship of concurrency level and throughput. Similarly, Didona et al. [START_REF] Didona | Identifying the optimal level of parallelism in transactional memory systems[END_REF] propose to dynamically adjust the level of concurrency according to the number of commits and aborts. The optimal number of threads is found with the help of two phases: (1) measurement phase and (2) decision phase. In the first phase the application is profiled with a fixed number of threads, in the second phase a hill-climbing approach is applied to increase and decrease the number of threads according to the given workload, maximizing the transaction's throughput based on successful commits. While the basic idea is interesting, the variation of the thread count based on the throughput might be disadvantageous for transactions with different granularity.

Concurrent Message Processing

To motivate our proposed work, we relate to the enhancements of the Actor Model as presented in our former work [START_REF] Hayduk | Speculative concurrent processing with transactional memory in the actor model[END_REF]. There, we reduced the execution time without violating the main characteristics of the Actor Model. Our main idea was based on the observation that we can guarantee atomicity and isolation if we encapsulate the handling of messages inside transactions. Thanks to the rollback and restart capability of transactions, several messages can be processed concurrently, even if they access the same state. The concurrent message processing only changes the message handling provided by the Akka framework as integrated in Scala 2.10.0. Specifically, we altered the behavior of the actor's mailbox processing code. In the original Akka implementation a dispatcher is responsible for ensuring that the same mailbox is not scheduled for processing messages more than once at a given time. We adapted the dispatcher to assign each message processing to a thread of the thread pool. Further, each message processing is handled in a transaction for which we use the default Scala STM. Our work performed well for read-dominated as well as write-dominated workloads and we outperformed the state-of-the-art Habanero Scala [START_REF] Imam | Integrating task parallelism with actors[END_REF] in their distributed list benchmark shared amongst list actors with 97% of reads, 2% writes and 1% sum operations created by request actors. Figure 1 shows performance improvements over sequential processing and Habanero Scala for all numbers of list actors in a shared linked list. On the x-axis we show the effects of increasing the number of request actors (125-500), while the y-axis displays the execution time in seconds (log scale), i.e., the time needed to finish processing all requests. The lower the execution time, the better. A higher number of list actors also leads to higher contention, increased rollback counts and hence decreased performance. With 16 list actors, concurrent processing as well as Habanero Scala perform close to sequential processing.

Dynamic Concurrent Message Processing

For improving the performance in high contention workloads we propose the following combination of methods. First, we adapt the level of concurrency for processing actor messages according to the current contention level. Second, we extract read-only message processing from the transactional context. And third, we exploit the fact that the two types of messages do not interfere and occupy the existing threads with both types of messages according to the current contention level. As a result, we occupy all threads with work. To differentiate between these two types of messages, we adapted the concurrent mailbox implementation in Akka (as part of Scala 2.10) to handle two different queues as shown in Figure 2.

One queue collects the messages to be processed in a transactional context (STM messages), and another one for the processing of read-only messages. We further adapted the actor message dispatcher such that it picks messages from both queues and forwards them for processing to the thread pool. Our new dispatcher automatically adapts the number of STM messages and read-only messages to state Mailbox Fig. 2. The different handling of STM messages and read-only messages. Dispatcher assigns a dynamic number of threads for message processing.

be processed according to the current contention of the workload. The detailed principles of processing STM messages and read-only messages are described in the following sections.

STM Message Processing

At the beginning of an application we start from a random number of threads from the thread pool to process transactional messages (STM messages) and then, driven by a predefined threshold α, the level of concurrency is adapted.

The dispatcher assigns the rest of the threads to process read-only messages. α is dependent on the knowledge of the current number of commits and rollbacks of transactionally processed messages and their ratio. Instead of focusing on the throughput such as Didona et al. [START_REF] Didona | Identifying the optimal level of parallelism in transactional memory systems[END_REF], we are able to support transactions of any granularity by steering α with commit-to-rollback ratio. If the current commit-to-rollback ratio is lower than α, we the number of threads processing transactional messages by two; if it is higher, we multiply them by two.

We chose the commit-to-rollback ratio in combination with the fixed α threshold due to its simplicity. To find the right α value for the current workload, we consider a short profiling phase monitoring the relation of commits and rollbacks.

Read-only Message Processing

The second category of messages are read-only messages. They are handled in a separate queue from the messages that require TM context. Scala STM [16] is based on CCSTM [START_REF] Bronson | CCSTM: A library-based STM for Scala[END_REF] (which extends SwissTM [START_REF] Dragojević | Stretching transactional memory[END_REF]). It is a write-back TM, where writes are cached and written to the memory on commit. Further, it provides eager conflict detection for writes and lazy conflict detection for reads. Validation is done based on a global time stamp. In Scala STM transactional objects are encapsulated in so-called Ref s. This implies that any access to a transactional object has to be within an atomic block, which ensures strong atomicity and isolation.

We argue that in some cases we can relax isolation, e.g., for performing approximate read-only operations. A ubiquitous example is a sequential data structure such as a linked list. While traversing the list in read-only mode, a concurrent write of a value, which has been already read, causes a conflict. For such cases, Scala STM provides a mechanism called unrecorded read. An unrecorded read is a transactional read, which returns an UnrecordedRead object containing the value, the version number before read, and a validity field of the read. The validity field returns true when there were no changes to the value, which helps to resolve the ABA problem [START_REF] Bronson | CCSTM: A library-based STM for Scala[END_REF].

In Scala STM the unrecorded read can be accessed through calling the re-laxedGet method. By using it, we can perform a read that will not be validated during commit. The relaxedGet method has to be executed inside of an atomic block:

atomic{ val unvalidatedValue = ref.relaxedGet({(_,_) => true}) }
Unrecorded reads do not yield new entries in the read set, but still need to ensure reading the latest version of a TM object. Scala STM checks for concurrent writes and forces eager conflict detection, which, in turn, causes a rollback of the writing transaction. As we see later this resolution strategy leads to similar runtime behavior as of the regular transactions.

To remove the overhead associated with unrecorded reads, we propose to provide direct access to the Ref object's data (which is safely possible due to the write-back characteristic of Scala STM). We extended Scala STM's Ref implementation with a singleRelaxedGet method and provide an example for generic references below. As the data variable in GenericRef is marked as volatile, the singleRelaxed-Get read-only operation can therefore safely interleave with other transactional write operations while guaranteeing to see the latest result. Moreover, singleRe-laxedGet does not interfere with other transactional operations, i.e., it cannot force another transaction to roll back.

Evaluation

Our optimizations are expected to be most useful in applications where state is shared among many actors. Hence, to evaluate our approach, we use a benchmark application provided by Imam and Sarkar [START_REF] Imam | Integrating task parallelism with actors[END_REF] that implements a stateful distributed sorted integer linked-list. It is the same benchmark as in our former work and shown in Section 3 in comparison to Habanero Scala. This benchmark is relevant as sequential data structures are typical applications for relaxed atomicity and isolation (see, e.g., early release). The read-only operation is a non-consistent sum that traverses each list element.

To show wider applicability we also consider a real-world scientific application that is used to simulate the hydraulic subsurface. The read-only operations in this application are used to control progress and debugging and hence should not interfere with any regular operation.

The thread pool is configured to support two scenarios. First, we specify a static ratio, in which 90% of threads are assigned to process STM messages and the rest processes read-only messages. In the second case we consider a dynamic ratio, but ensure that the number of threads assigned to process any message type never drops below the ratio of 10%. We compare the performance of our proposed singleRelaxedGet to the default atomic block and the relaxedGet method provided by Scala STM.

We execute the benchmarks on a 48-core machine equipped with four 12-core AMD Opteron 6172 CPUs running at 2.1GHz. Each core has private L1 and L2 caches and a shared L3 cache. The sizes of both instruction and data caches are 64KB at L1, 512KB at L2, and 5MB at L3.

List Benchmark

The list benchmark comprises two actor types: request and list actors. Request actors send requests such as lookup, insert, remove, and sum. List actors are responsible for handling a range of values (buckets) of a distributed linked list. We implemented a list element as an object containing a value field and a next field, which is wrapped in a Scala STM Ref object.

In a list with l list actors, where each actor stores at most n elements representing consecutive integer values, the i th list actor is responsible for elements in the [(i -1) • n, (i • n) -1] range, e.g., in a list with 4 actors and 8 entries in total, each actor is responsible for two values. A request forwarder matches the responsible list actors with the incoming requests. For the sum operation, we traverse each list element in every list actor. This operation is read-only and does not necessarily report a consistent value. It should not conflict with other accesses to the list.

We run the benchmark with 32 threads (to be able to divide and multiply the thread count by two) in 7 runs from which we take the median throughput and number of rollbacks for the results. Also, we create 8 list actors that are in responsible for 41,216 list elements. We create 500 request actors, where each actor sends 1,000 messages to the list. After each of the request actors finished their 1,000 requests the benchmark terminates.

To read a value from the list, we consider three different options: (1) the regular transactional read (node.next.get()), (2) the unrecorded read accessed via (node.next.relaxedGet()) and (3) our direct access (node.next.singleRelaxedGet()) method.

In our experiments we consider a write-dominated and a mixed workload. The write-dominated workload is configured as follows: Each request actor sends 98% of requests to modify the list (insert or remove), 1% of lookup requests and 1% of sum requests.

In the first experiments, we evaluate the impact of different access approaches to the sum operation if the threads are assigned statically. The static approach (90% STM:10% read-only) reserves 3 threads (10%) for the processing of readonly messages and 29 threads (90%) for processing STM messages. Figure 3 demonstrates the message throughput (left) and the rollback count over time (right). The shorter the line, the better the execution time. The singleRelaxed-Get() outperforms the other operations with respect to both, execution time (50%) and throughput (40%). However, we observe a drastic increase of rollbacks (90%). This observation is counter intuitive, as one would expect to have a lower number of rollbacks to achieve higher throughput. In fact, when we use atomic and relaxedGet() to implement the sum operation, we cause significantly more read-write conflicts. Scala STM resolves them by waiting for the write operations to finish to follow up with the execution of the read operations. On the contrary, when we use the singleRelaxedGet() operation, we remove transactional read operations, which increases the likelihood of concurrent write operations. As a result, we get more write-write conflicts, which are typically resolved eagerly by rolling back one of the transactions.

Since the performance of transactional operations can be further improved, we dynamically determine the optimal number of threads as described in Section 4.1. We schedule a thread, which obtains the total number of commits and rollbacks every second, and decides the optimal number of threads to schedule for the processing of STM and read-only messages. Initially, the thread counts are randomly chosen and within the first two seconds the number of threads converges to the optimal values with the help of α. For determining α we profile the application for a short time. In the case of the list benchmark this results in α = 0.21, hence if the commit-to-rollback ratio is below α the number of threads will be reduced else increased (multiplied or divided by two).

In Figure 4 we see that the throughput and rollback values for the atomic and relaxedGet() operations are not significantly different when compared to the static approach. This behavior is expected as the operations interfere with the concurrently executing transactional write operations. This causes more contention as both the STM messages and the read-only messages are conflicting. On the contrary, the singleRelaxedGet() operation never conflicts with other list operations. Hence, we can efficiently use the threads not processing STM messages resulting in increased throughput (65%) and reduced runtime (70%).

In the mixed read-write workload, consisting of 50% of requests to modify the list, 49% lookup requests and 1% sum requests, we show the applicability of singleRelaxedGet in a more read-dominant scenario. With the help of a profiling phase, we set α to 0.08. Figures 5 and6 show similar results as the write-dominated list benchmark, amplifying the benefits of singleRelaxedGet. The dynamic thread assignment further reduces the rollback counts in compar- ison to the static assignment. In order to compare to Figure 1 the request ratio has to be increased, leading to results below 1 second for the singleRelaxedGet. It would hence clearly outperform the default implementation of Habanero Scala.

Simulation of the Hydraulic Subsurface

The aim of multiple-point geostatistical simulations is to simulate the hydraulic properties of the subsurface. A number of techniques has been developed for executing the simulation, the most popular of them is multiple-point geostatistics [START_REF] Guardiano | Multivariate geostatistics: Beyond bivariate moments[END_REF], which analyzes the relationship between multiple variables in several locations at a time. For its implementation, we can use the Direct Sampling simulation introduced by Mariethoz et al [START_REF] Mariethoz | The Direct Sampling method to perform multiple-point geostatistical simulations[END_REF]. A simulation consists of a Training Image (TI) and a Simulation Grid (SG). The task is to fill unknown points of the simulation grid according to the known points from the training image (see Figure 7(a)). The algorithm starts with selecting a random point x to be filled in the SG. Then, it locates n closest neighbors, which are already filled. The neighbors and x are considered as a pattern, for which the algorithm searches in the TI. If found, x can be filled. Given that in the field of hydraulic subsurface simulation the simulation grid and the training image are of a very large size, sequential processing is not efficient. Two cases of parallelization are possible: (1) the parallelization of searching within the training image, (2) the parallelization of the node filling within the simulation grid. In this paper we concentrate on the parallelization of the node filling within the SG as parallelizing the TI would only comprise independent read operations. Our implementation of the actor-based simulation considers a main actor that stores the SG and the TI. Consider Figure 7(b) having two points T 1 and T 2 to be simulated concurrently. We can see that these are close and if simulated, are likely to be part of the n closest neighbor pattern. Thus, the point that finishes first invalidates the current simulation of the other point and a synchronization mechanism is required. In our implementation, all SG points are protected by a Scala STM Ref object, which causes a rollback once a SG point is simulated that has been visited during a nearest neighbor search. Besides the main actor we implemented a number of worker actors, either sim- ulating the SG points (simulation actors) or responsible for logging the current state (log actors). Simulation actors claim a number of points to be simulated and initiate the closest neighbor search and the TI matching for each of the points at the main actor. We run the benchmark with 32 worker actors, with each simulation actor requesting to process 30 simulation points at once, finally sending 167 messages during the benchmark. Researchers using the multiple-point geostatistics simulation usually validate the results manually by visualizing the simulated result. Since the simulation can take several hours, intermediate results are of interest; we use them to see whether the simulation is on the right track. This case is especially of interest, because once the points are simulated, they do not change anymore. Therefore, these intermediate results do not require full consistency and should not interfere with the main simulation. We select these messages to be handled as read-only messages sent by log actors. The list benchmark considered a constant number of sum messages while here, each log actor sends a request to the main actor, repeating it upon receiving a response. In the experiments we consider 8 log actors. The size of the SG is 750 times 750 elements and we investigate the 15 closest neighbors for each point, finally we profiled α = 1.2. We use the sample image provided by the MPDS [START_REF] Mariethoz | The Direct Sampling method to perform multiple-point geostatistical simulations[END_REF] distribution as the TI.

As shown in Figure 8 the static thread allocation results for throughput suggest that it is possible to increase the number of processed query messages with the help of singleRelaxedGet, while reducing the execution time in comparison to the atomic case by approximately 40 seconds. The total number of rollbacks, however, remains stable. Figure 9 (left) demonstrates that the dynamic thread allocation improves the execution time of the singleRelaxedGet by another 20 seconds. The throughput seems to decrease in comparison to the static scenario, which is not true considering the total throughput. While the amount of work performed is in both scenarios the same, we are able to successfully process more messages in the period of second 20-50 in the dynamic scenario. This can be seen by the low rollback counts shown in Figure 9 (right) for the same period. In the beginning of the execution, however, the rollback counts reflect the effects of adjusting the concurrency level. In total the dynamic scenario results in lower rollback counts than the static scenario. In all of the cases singleRelaxedGet in combination with the dynamic handling of STM messages is able to outperform concurrent message processing.

Conclusions

In this paper we introduce dynamic concurrent message processing in the actor model for high contention workloads. We propose to extract read-only messages from the transactional context if consistency is not required as well as adapt the number of threads to the workload. By handling transactional messages with a different level of concurrency we can efficiently use the remaining resources (threads in a thread pool) for processing read-only messages. We showed the applicability of our approach as well as candidates for messages processed with relaxed consistency. In a list benchmark and a real-world application we demonstrated that our approach helps reducing the execution time, while also decreasing the rollback counts. In future work we target to improve the performance by introducing a learning phase (e.g., with the help of hill-climbing) for guiding the level of concurrency.

Fig. 1 .

 1 Fig. 1. Concurrent message processing in a read-dominated workload within a shared linked list.

class

  GenericRef[A](@volatile var data: A) extends BaseRef[A] { ... def singleRelaxedGet(): A = data } Its usage is then straightforward as shown in the next example: val relaxedValue = ref.singleRelaxedGet()

Fig. 3 .Fig. 4 .Fig. 5 .

 345 Fig. 3. Throughput (left) and rollback count (right) for the list write-dominated workload: static thread allocation.

Fig. 6 .Fig. 7 .

 67 Fig.[START_REF] Herlihy | Software transactional memory for dynamic-sized data structures[END_REF]. Throughput (left) and rollback count (right) for the list mixed workload: dynamic thread allocation.

Fig. 8 .Fig. 9 .

 89 Fig. 8. Throughput (left) and rollbacks (right) of STM messages and read-only messages over time: static thread allocation.

Acknowledgement

The research leading to these results has received funding from the European Community's Seventh Framework Programme [FP7/2007-2013] under the ParaDIME Project (www.paradime-project.eu), grant agreement no 318693.