Fast General Distributed Transactions with Opacity
Shamis, Alex
Renzelmann, Matthew
Novakovic, Stanko
Chatzopoulos, Georgios
Dragojević, Aleksandar
Dushyanth, Narayanan
Castro, Miguel
Transactions can simplify distributed applications by hiding data distribution, concurrency, and failures from the application developer. Ideally the developer would see the abstraction of a single large machine that runs transactions sequentially and never fails. This requires the transactional subsystem to provide opacity (strict serializability for both committed and aborted transactions), as well as transparent fault tolerance with high availability. As even the best abstractions are unlikely to be used if they perform poorly, the system must also provide high performance. Existing distributed transactional designs either weaken this abstraction or are not designed for the best performance within a data center. This paper extends the design of FaRM - which provides strict serializability only for committed transactions - to provide opacity while maintaining FaRM's high throughput, low latency, and high availability within a modern data center. It uses timestamp ordering based on real time with clocks synchronized to within tens of microseconds across a cluster, and a failover protocol to ensure correctness across clock master failures. FaRM with opacity can commit 5.4 million neworder transactions per second when running the TPC-C transaction mix on 90 machines with 3-way replication.
 The distributed applications are hard to build. In the distributed system, you have to deal with failure, concurrence, concurrency where multiple operations can access the data at the same time and data that's distributed all over the cluster. It would be really nice to have an abstraction of a giant machine that pulls all the resources in the cluster, never fails and gives the illusion of running transactions sequentially. Farm provided such an abstraction. We had a system that had strict serializability, high availability, had great performance and scalability. We then published this work and deployed it in the Microsoft production data centers. We were able to do this by leveraging new hardware trends in data centers such as cheap and plentiful DRAM. We used lithium ion batteries to make this DRAM persistent. And also, we leveraged new networking in the form of one-sided RDMA. And after we did this, developers at Microsoft started building on top of our platform. However, programmability was still an issue. Developers found that some of the variants they put in, they could not rely on. And we kept getting bug reports, which I'm going to show you a sample of. So firstly, what they would do is they would start a transaction. Inside of the same transaction, inside of this transaction, they would insert a key into a table and then assert that the key was inserted correctly. Next, they would check in the same transaction, in the same table, if this key existed and get a result back, which they expected that it would. However, sometimes the assert highlight in yellow would fire. One could imagine that it would be possible to, instead of asserting, you could abort such a transaction. However, that would conflate a certain abort and make debugging this out system or the program that the user wrote much more difficult. So as I said, they had issues with programmability. Farm executed transactions optimistically. There was no locking during execution. And the commit phase was responsible for enforcing strict serializability. And this was done by aborting transactions that could not be serialized during the commit phase. Because of this, you could see inconsistent state during the execution phase. Farm lacked the property called opacity. Opacity means having all transactions being strictly serialized, even those that eventually abort. So our challenge was to add opacity to Farm while maintaining its great performance. We also knew that Farm's performance was dependent on using one-sided RDMA, which meant that we had to use optimistic concurrency control. So we rose to this challenge by creating a timestamp ordering service with scalable timestamp generation. We used local CPU clocks. And we had low overheads in the transaction protocol. And additionally, we created a clockmaster failover service. So the basis of our algorithm is based on Mozilla's algorithm. And just as an intuition of how this works, you have a clockmaster and a server. And the server wants to work out what the current time is from the clockmaster. The server would send a message to the clockmaster. And the clockmaster would respond with a timestamp. And at this point, the server would know when these two messages are sent in the form of a round-trip time. What the server was able to do, according to Mozilla's algorithm, is get an upper bound and a lower bound on the time, as it knew that the timestamp was generated sometime between when it sent the request and when it received the response. In addition, if we needed to work out what the time was some time after the response was received, we needed to take into account the elapsed time. So here is the algorithm that works out the bounds from Mozilla's algorithm. As you can see, that we include we have the upper and lower band. And then there's also an epsilon parameter here. Epsilon represents the drift between different CPUs. This drift is measured, is defined in the hardware spec of a CPUs. And we also measured it ourselves in our production services to work out what was the maximum possible drift. And to make sure that our algorithm did not go outside of this bound, we multiplied this maximum possible drift factor by 10. Now what I told you was just was an upper lower bound for time. I didn't provide a scale of value. So the way we did get a scale of value is we would take the upper bound of this, of the time that we got from the clock master, which is great in the possible time at the clock master. We waited for this upper bound to be lower than the new lower bound. And we use this upper bound as our timestamp for our transaction protocol. So what does this look like inside of our transaction protocol? So the very first thing we would do is the user would start a new transaction and they would get a read time stamp. This write stamp would have to wait out this bound, which we call the uncertainty. So once we have this bound, you have your R, which is your read time stamp. Then we would read an object in farm. All objects in farm exist of a payload and some metadata. And the metadata includes the write time stamp for when that object was last written. So when we read the object, we would get the write time stamp. We would check if this time stamp was greater than what our read time stamp was. And if so, we would bought the transaction, as this would mean that we would have to, we would lose strict serializability if we continued. Next we would write an object and in farm we store objects locally until they're committed. Then we would read another object B, again reading its time stamp also and doing the same check to make sure that we don't violate opacity. Finally, the transaction finishes and we try to commit. We lock all of the objects that we are writing to. We then get a write time stamp and this write time stamp is the time stamp that will be written with the new objects that are going to be updated and this is our serialization time stamp. And again, since it is getting a single vector time stamp, we have to wait out on certainty again. Next we would validate every single object that we just read and assuming the locking validation completed successfully, we would write the object committed and tell the user that the transaction committed successfully. So just to provide an intuition of why do we need these write time stamps and read time stamps. So we must say that the locks are held whenever the write time stamp is equal to the time stamp at the clock master. This ensures that we see any locks that anybody else takes on an object so we know if there's any conflicts with the transaction and in addition the lock must be held at this time to make sure that anyone else sees our locks so that we can serialize both transactions properly. Next the read time stamp. For the read time stamp, again the read time stamp cannot be before ahead of the clock master and this is to ensure that all locks or updates are seen by the transaction that is reading. However, it is possible to use a write time stamp that's in the past. For read only transaction what you would see is a consistent state but it would not be strictly serializable, it would just be a serializable state. And if this was a read write transaction we would enforce serializability at the commit phase and abort if one of these objects was updated. However, if you do take a read time stamp in the future, wait out uncertainty, you get strictness and serializability. Next let's look at our clock master failover. We make sure that any machine can be the clock master and we do this by just using CPU clocks to take the time. On our clock master algorithm has the following high level intuition. First, when a clock master fails we elect a new clock master, the clock master disables its own and all the other clocks, it will then advance time such that it is higher than any other, any uncertain times that we have seen and finally it will re-enable clocks. Importantly our protocol is able to survive failures during failover. So just have a look at the specifics of how it works. You have a new clock master that was just elected and you have another server that is attempting to synchronize. So the very first thing is a new clock master does is it disables its own clock. It sends a message to all the other servers saying disable your clocks. While the other servers disable the clocks, it works out a variable ff which is the maximum of all the upper bounds of the uncertainty window that it has and responds to the clock master with this ff variable. The clock master has to wait out lease expiry to make sure that any machines that are not part of this new configuration are no longer attempting to act on transactions or requests that are being sent to it. Once this is done, we calculate a new ff which is the maximum of all the ff's in the cluster. We send this ff to all the other servers such that time has been fast forwarded now and then we enable the clock on the clock master and regular synchronization can occur. To evaluate our system, we had a 19 machine cluster and we ran TPCC with 3-way replication and all our experiments, unless otherwise specified, were strictly serializable. This graph represents the scalability of farm. On the y-axis, you see the throughput in millions of new order transactions per second and on the x-axis is the number of servers. We are able to achieve at the farm master's right when there's 19 machines 5.4 million new order transactions per second which is as far as we're aware of the highest published number at this time. Now when we look at the uncertainty rate as we scale out the cluster, you can see the blue line that I introduced with the new access on the right-hand side being the mean uncertainty in microseconds. You can see here as we increase the cluster size from 3 machines to 90 machines, the uncertainty goes from 7.5 microseconds to 12 microseconds which is only an increase of 3.5 microseconds for a 30x increase in the number of machines. Now it was also interesting to see what happens when the cluster gets bigger. Unfortunately, we did not have more than 90 machines to run these experiments on. So what we did is we had an experiment to sample this. We said that we are using 19 machines and we would take every x sample to represent what happens if we get a slow back. So if we're saying that there's 19 machines and we're taking every second sample, it's as if there are 1,180 machines synchronizing and we get the response back half of the times we would maintain the same load on the clockmaster. So in this diagram on the left-hand side, you can see the mean uncertainty rate again in microseconds and the sampling ratio at the bottom. And here when we go from 19 machines to 900 machines with this ratio, we only increase uncertainty by 5 microseconds. Next we looked at the TPCC throughput versus latency graph. And for this experiment, we were using 57 machines. On the y-axis is the latency of the new order transactions in milliseconds and on the x-axis is through the throughput in millions of new order transactions per second. The main takeaway from this graph is that as we increase the load on the system, latency still is less than 1 millisecond up to a maximum of 3.5 million new order transactions per second. Now we want to look at what the cost of opacity was. So we were able to run our system without the opacity feature. The throughput overhead for opacity was 3.6% for the TPCC experiment I just showed you with 57 machines. And the latency overhead at low load at the 50th percentile was 39 microseconds and at the 99th percentile, 69 microseconds. So transactions that had microsecond latency would be affected by this overhead. The next thing we wanted to look at is what happens if we remove the strictness or the serializability properties from our system. And while this would be bad for programmability, we believe that this is just an interesting experiment to understand how the system functions. So if we remove either strictness or serializability, we can get a 3% increase of TPCC throughput. However, if we remove both, we get a 5% performance increase, a throughput increase. We believe that this cost is low enough that the advantage of having opacity and helping developers with their programmability is what the price that we pay. Finally we looked at recovery. This diagram on the left-hand side shows a throughput again in millions of new order transactions per second. And the x-axis is the walk clock time starting at arbitrary point zero and increasing in milliseconds. Here we kill two machines at the same time. And one of those machines is the clock master. So we can see that we stop handing out the time for when the clock master is paused is 19 milliseconds. And we return to the same throughput within 67 milliseconds. This is similar to the performance we had with Farm before opacity. And in itself, this blip is lower than most tail latencies for some production systems. We had a lot more other work in the paper, which I don't have enough time to go into. We implemented a distributed MBCC protocol. We talk about how the scalability of our time stamp generation, we go into a lot of detail of what happens when we increase load, and optimizations to get very low microsecond uncertainty among many other things. So in conclusion, we built a new system with opacity, high throughput, low latency, high availability, and strict serializability. Thank you. Thank you.
 The distributed applications are hard to build.  In the distributed system, you have to deal with failure, concurrence, concurrency where  multiple operations can access the data at the same time and data that's distributed  all over the cluster.  It would be really nice to have an abstraction of a giant machine that pulls all the resources  in the cluster, never fails and gives the illusion of running transactions sequentially.  Farm provided such an abstraction.  We had a system that had strict serializability, high availability, had great performance and  scalability.  We then published this work and deployed it in the Microsoft production data centers.  We were able to do this by leveraging new hardware trends in data centers such as cheap  and plentiful DRAM.  We used lithium ion batteries to make this DRAM persistent.  And also, we leveraged new networking in the form of one-sided RDMA.  And after we did this, developers at Microsoft started building on top of our platform.  However, programmability was still an issue.  Developers found that some of the variants they put in, they could not rely on.  And we kept getting bug reports, which I'm going to show you a sample of.  So firstly, what they would do is they would start a transaction.  Inside of the same transaction, inside of this transaction, they would insert a key  into a table and then assert that the key was inserted correctly.  Next, they would check in the same transaction, in the same table, if this key existed and  get a result back, which they expected that it would.  However, sometimes the assert highlight in yellow would fire.  One could imagine that it would be possible to, instead of asserting, you could abort such  a transaction.  However, that would conflate a certain abort and make debugging this out system or the  program that the user wrote much more difficult.  So as I said, they had issues with programmability.  Farm executed transactions optimistically.  There was no locking during execution.  And the commit phase was responsible for enforcing strict serializability.  And this was done by aborting transactions that could not be serialized during the commit  phase.  Because of this, you could see inconsistent state during the execution phase.  Farm lacked the property called opacity.  Opacity means having all transactions being strictly serialized, even those that eventually  abort.  So our challenge was to add opacity to Farm while maintaining its great performance.  We also knew that Farm's performance was dependent on using one-sided RDMA, which meant that  we had to use optimistic concurrency control.  So we rose to this challenge by creating a timestamp ordering service with scalable  timestamp generation.  We used local CPU clocks.  And we had low overheads in the transaction protocol.  And additionally, we created a clockmaster failover service.  So the basis of our algorithm is based on Mozilla's algorithm.  And just as an intuition of how this works, you have a clockmaster and a server.  And the server wants to work out what the current time is from the clockmaster.  The server would send a message to the clockmaster.  And the clockmaster would respond with a timestamp.  And at this point, the server would know when these two messages are sent in the form of  a round-trip time.  What the server was able to do, according to Mozilla's algorithm, is get an upper bound  and a lower bound on the time, as it knew that the timestamp was generated sometime between  when it sent the request and when it received the response.  In addition, if we needed to work out what the time was some time after the response  was received, we needed to take into account the elapsed time.  So here is the algorithm that works out the bounds from Mozilla's algorithm.  As you can see, that we include we have the upper and lower band.  And then there's also an epsilon parameter here.  Epsilon represents the drift between different CPUs.  This drift is measured, is defined in the hardware spec of a CPUs.  And we also measured it ourselves in our production services to work out what was the maximum  possible drift.  And to make sure that our algorithm did not go outside of this bound, we multiplied this  maximum possible drift factor by 10.  Now what I told you was just was an upper lower bound for time.  I didn't provide a scale of value.  So the way we did get a scale of value is we would take the upper bound of this, of  the time that we got from the clock master, which is great in the possible time at the  clock master.  We waited for this upper bound to be lower than the new lower bound.  And we use this upper bound as our timestamp for our transaction protocol.  So what does this look like inside of our transaction protocol?  So the very first thing we would do is the user would start a new transaction and they  would get a read time stamp.  This write stamp would have to wait out this bound, which we call the uncertainty.  So once we have this bound, you have your R, which is your read time stamp.  Then we would read an object in farm.  All objects in farm exist of a payload and some metadata.  And the metadata includes the write time stamp for when that object was last written.  So when we read the object, we would get the write time stamp.  We would check if this time stamp was greater than what our read time stamp was.  And if so, we would bought the transaction, as this would mean that we would have to,  we would lose strict serializability if we continued.  Next we would write an object and in farm we store objects locally until they're committed.  Then we would read another object B, again reading its time stamp also and doing the  same check to make sure that we don't violate opacity.  Finally, the transaction finishes and we try to commit.  We lock all of the objects that we are writing to.  We then get a write time stamp and this write time stamp is the time stamp that will be  written with the new objects that are going to be updated and this is our serialization  time stamp.  And again, since it is getting a single vector time stamp, we have to wait out on certainty  again.  Next we would validate every single object that we just read and assuming the locking  validation completed successfully, we would write the object committed and tell the user  that the transaction committed successfully.  So just to provide an intuition of why do we need these write time stamps and read time  stamps.  So we must say that the locks are held whenever the write time stamp is equal to the time stamp  at the clock master.  This ensures that we see any locks that anybody else takes on an object so we know if there's  any conflicts with the transaction and in addition the lock must be held at this time  to make sure that anyone else sees our locks so that we can serialize both transactions  properly.  Next the read time stamp.  For the read time stamp, again the read time stamp cannot be before ahead of the clock  master and this is to ensure that all locks or updates are seen by the transaction that  is reading.  However, it is possible to use a write time stamp that's in the past.  For read only transaction what you would see is a consistent state but it would not be  strictly serializable, it would just be a serializable state.  And if this was a read write transaction we would enforce serializability at the commit  phase and abort if one of these objects was updated.  However, if you do take a read time stamp in the future, wait out uncertainty, you get  strictness and serializability.  Next let's look at our clock master failover.  We make sure that any machine can be the clock master and we do this by just using CPU clocks  to take the time.  On our clock master algorithm has the following high level intuition.  First, when a clock master fails we elect a new clock master, the clock master disables  its own and all the other clocks, it will then advance time such that it is higher than  any other, any uncertain times that we have seen and finally it will re-enable clocks.  Importantly our protocol is able to survive failures during failover.  So just have a look at the specifics of how it works.  You have a new clock master that was just elected and you have another server that is  attempting to synchronize.  So the very first thing is a new clock master does is it disables its own clock.  It sends a message to all the other servers saying disable your clocks.  While the other servers disable the clocks, it works out a variable ff which is the maximum  of all the upper bounds of the uncertainty window that it has and responds to the clock  master with this ff variable.  The clock master has to wait out lease expiry to make sure that any machines that are not  part of this new configuration are no longer attempting to act on transactions or requests  that are being sent to it.  Once this is done, we calculate a new ff which is the maximum of all the ff's in the cluster.  We send this ff to all the other servers such that time has been fast forwarded now and  then we enable the clock on the clock master and regular synchronization can occur.  To evaluate our system, we had a 19 machine cluster and we ran TPCC with 3-way replication  and all our experiments, unless otherwise specified, were strictly serializable.  This graph represents the scalability of farm.  On the y-axis, you see the throughput in millions of new order transactions per second and on  the x-axis is the number of servers.  We are able to achieve at the farm master's right when there's 19 machines 5.4 million  new order transactions per second which is as far as we're aware of the highest published  number at this time.  Now when we look at the uncertainty rate as we scale out the cluster, you can see the  blue line that I introduced with the new access on the right-hand side being the mean uncertainty  in microseconds.  You can see here as we increase the cluster size from 3 machines to 90 machines, the uncertainty  goes from 7.5 microseconds to 12 microseconds which is only an increase of 3.5 microseconds  for a 30x increase in the number of machines.  Now it was also interesting to see what happens when the cluster gets bigger.  Unfortunately, we did not have more than 90 machines to run these experiments on.  So what we did is we had an experiment to sample this.  We said that we are using 19 machines and we would take every x sample to represent what  happens if we get a slow back.  So if we're saying that there's 19 machines and we're taking every second sample, it's  as if there are 1,180 machines synchronizing and we get the response back half of the times  we would maintain the same load on the clockmaster.  So in this diagram on the left-hand side, you can see the mean uncertainty rate again  in microseconds and the sampling ratio at the bottom.  And here when we go from 19 machines to 900 machines with this ratio, we only increase  uncertainty by 5 microseconds.  Next we looked at the TPCC throughput versus latency graph.  And for this experiment, we were using 57 machines.  On the y-axis is the latency of the new order transactions in milliseconds and on the x-axis  is through the throughput in millions of new order transactions per second.  The main takeaway from this graph is that as we increase the load on the system, latency  still is less than 1 millisecond up to a maximum of 3.5 million new order transactions per  second.  Now we want to look at what the cost of opacity was.  So we were able to run our system without the opacity feature.  The throughput overhead for opacity was 3.6% for the TPCC experiment I just showed you  with 57 machines.  And the latency overhead at low load at the 50th percentile was 39 microseconds and at  the 99th percentile, 69 microseconds.  So transactions that had microsecond latency would be affected by this overhead.  The next thing we wanted to look at is what happens if we remove the strictness or the  serializability properties from our system.  And while this would be bad for programmability, we believe that this is just an interesting  experiment to understand how the system functions.  So if we remove either strictness or serializability, we can get a 3% increase of TPCC throughput.  However, if we remove both, we get a 5% performance increase, a throughput increase.  We believe that this cost is low enough that the advantage of having opacity and helping  developers with their programmability is what the price that we pay.  Finally we looked at recovery.  This diagram on the left-hand side shows a throughput again in millions of new order  transactions per second.  And the x-axis is the walk clock time starting at arbitrary point zero and increasing in  milliseconds.  Here we kill two machines at the same time.  And one of those machines is the clock master.  So we can see that we stop handing out the time for when the clock master is paused is  19 milliseconds.  And we return to the same throughput within 67 milliseconds.  This is similar to the performance we had with Farm before opacity.  And in itself, this blip is lower than most tail latencies for some production systems.  We had a lot more other work in the paper, which I don't have enough time to go into.  We implemented a distributed MBCC protocol.  We talk about how the scalability of our time stamp generation, we go into a lot of detail  of what happens when we increase load, and optimizations to get very low microsecond  uncertainty among many other things.  So in conclusion, we built a new system with opacity, high throughput, low latency, high  availability, and strict serializability.  Thank you.  Thank you.
Shamis_Renzlmann_navakovic_Chatzopoulos_0_47.png Shamis_Renzlmann_navakovic_Chatzopoulos_48_131.png Shamis_Renzlmann_navakovic_Chatzopoulos_132_179.png Shamis_Renzlmann_navakovic_Chatzopoulos_180_431.png Shamis_Renzlmann_navakovic_Chatzopoulos_432_767.png Shamis_Renzlmann_navakovic_Chatzopoulos_768_1787.png Shamis_Renzlmann_navakovic_Chatzopoulos_1788_2135.png Shamis_Renzlmann_navakovic_Chatzopoulos_2136_2231.png Shamis_Renzlmann_navakovic_Chatzopoulos_2232_3275.png Shamis_Renzlmann_navakovic_Chatzopoulos_3276_4331.png Shamis_Renzlmann_navakovic_Chatzopoulos_4332_5267.png Shamis_Renzlmann_navakovic_Chatzopoulos_5268_5711.png Shamis_Renzlmann_navakovic_Chatzopoulos_5712_6551.png Shamis_Renzlmann_navakovic_Chatzopoulos_6552_7571.png Shamis_Renzlmann_navakovic_Chatzopoulos_7572_8315.png Shamis_Renzlmann_navakovic_Chatzopoulos_8316_8459.png Shamis_Renzlmann_navakovic_Chatzopoulos_8460_8891.png Shamis_Renzlmann_navakovic_Chatzopoulos_8892_10211.png Shamis_Renzlmann_navakovic_Chatzopoulos_10212_11063.png Shamis_Renzlmann_navakovic_Chatzopoulos_11064_12035.png Shamis_Renzlmann_navakovic_Chatzopoulos_12036_13223.png Shamis_Renzlmann_navakovic_Chatzopoulos_13224_14255.png Shamis_Renzlmann_navakovic_Chatzopoulos_14256_14603.png Shamis_Renzlmann_navakovic_Chatzopoulos_14604_16307.png Shamis_Renzlmann_navakovic_Chatzopoulos_16308_17759.png Shamis_Renzlmann_navakovic_Chatzopoulos_17760_19283.png Shamis_Renzlmann_navakovic_Chatzopoulos_19284_20183.png Shamis_Renzlmann_navakovic_Chatzopoulos_20184_21023.png Shamis_Renzlmann_navakovic_Chatzopoulos_21024_22163.png Shamis_Renzlmann_navakovic_Chatzopoulos_22164_23531.png Shamis_Renzlmann_navakovic_Chatzopoulos_23532_24131.png Shamis_Renzlmann_navakovic_Chatzopoulos_24132_24419.png Shamis_Renzlmann_navakovic_Chatzopoulos_24420_24539.png
The distributed applications are hard to build. In the distributed system, you have to deal with failure, concurrence, concurrency where multiple operations can access the data at the same time and data that's distributed all over the cluster. It would be really nice to have an abstraction of a giant machine that pulls all the resources in the cluster, never fails and gives the illusion of running transactions sequentially. Farm provided such an abstraction. We had a system that had strict serializability, high availability, had great performance and scalability. We then published this work and deployed it in the Microsoft production data centers. We were able to do this by leveraging new hardware trends in data centers such as cheap and plentiful DRAM. We used lithium ion batteries to make this DRAM persistent. And also, we leveraged new networking in the form of one-sided RDMA. And after we did this, developers at Microsoft started building on top of our platform. However, programmability was still an issue. Developers found that some of the variants they put in, they could not rely on. And we kept getting bug reports, which I'm going to show you a sample of. So firstly, what they would do is they would start a transaction. Inside of the same transaction, inside of this transaction, they would insert a key into a table and then assert that the key was inserted correctly. Next, they would check in the same transaction, in the same table, if this key existed and get a result back, which they expected that it would. However, sometimes the assert highlight in yellow would fire. One could imagine that it would be possible to, instead of asserting, you could abort such a transaction. However, that would conflate a certain abort and make debugging this out system or the program that the user wrote much more difficult. So as I said, they had issues with programmability. Farm executed transactions optimistically. There was no locking during execution. And the commit phase was responsible for enforcing strict serializability. And this was done by aborting transactions that could not be serialized during the commit phase. Because of this, you could see inconsistent state during the execution phase. Farm lacked the property called opacity. Opacity means having all transactions being strictly serialized, even those that eventually abort. So our challenge was to add opacity to Farm while maintaining its great performance. We also knew that Farm's performance was dependent on using one-sided RDMA, which meant that we had to use optimistic concurrency control. So we rose to this challenge by creating a timestamp ordering service with scalable timestamp generation. We used local CPU clocks. And we had low overheads in the transaction protocol. And additionally, we created a clockmaster failover service. So the basis of our algorithm is based on Mozilla's algorithm. And just as an intuition of how this works, you have a clockmaster and a server. And the server wants to work out what the current time is from the clockmaster. The server would send a message to the clockmaster. And the clockmaster would respond with a timestamp. And at this point, the server would know when these two messages are sent in the form of a round-trip time. What the server was able to do, according to Mozilla's algorithm, is get an upper bound and a lower bound on the time, as it knew that the timestamp was generated sometime between when it sent the request and when it received the response. In addition, if we needed to work out what the time was some time after the response was received, we needed to take into account the elapsed time. So here is the algorithm that works out the bounds from Mozilla's algorithm. As you can see, that we include we have the upper and lower band. And then there's also an epsilon parameter here. Epsilon represents the drift between different CPUs. This drift is measured, is defined in the hardware spec of a CPUs. And we also measured it ourselves in our production services to work out what was the maximum possible drift. And to make sure that our algorithm did not go outside of this bound, we multiplied this maximum possible drift factor by 10. Now what I told you was just was an upper lower bound for time. I didn't provide a scale of value. So the way we did get a scale of value is we would take the upper bound of this, of the time that we got from the clock master, which is great in the possible time at the clock master. We waited for this upper bound to be lower than the new lower bound. And we use this upper bound as our timestamp for our transaction protocol. So what does this look like inside of our transaction protocol? So the very first thing we would do is the user would start a new transaction and they would get a read time stamp. This write stamp would have to wait out this bound, which we call the uncertainty. So once we have this bound, you have your R, which is your read time stamp. Then we would read an object in farm. All objects in farm exist of a payload and some metadata. And the metadata includes the write time stamp for when that object was last written. So when we read the object, we would get the write time stamp. We would check if this time stamp was greater than what our read time stamp was. And if so, we would bought the transaction, as this would mean that we would have to, we would lose strict serializability if we continued. Next we would write an object and in farm we store objects locally until they're committed. Then we would read another object B, again reading its time stamp also and doing the same check to make sure that we don't violate opacity. Finally, the transaction finishes and we try to commit. We lock all of the objects that we are writing to. We then get a write time stamp and this write time stamp is the time stamp that will be written with the new objects that are going to be updated and this is our serialization time stamp. And again, since it is getting a single vector time stamp, we have to wait out on certainty again. Next we would validate every single object that we just read and assuming the locking validation completed successfully, we would write the object committed and tell the user that the transaction committed successfully. So just to provide an intuition of why do we need these write time stamps and read time stamps. So we must say that the locks are held whenever the write time stamp is equal to the time stamp at the clock master. This ensures that we see any locks that anybody else takes on an object so we know if there's any conflicts with the transaction and in addition the lock must be held at this time to make sure that anyone else sees our locks so that we can serialize both transactions properly. Next the read time stamp. For the read time stamp, again the read time stamp cannot be before ahead of the clock master and this is to ensure that all locks or updates are seen by the transaction that is reading. However, it is possible to use a write time stamp that's in the past. For read only transaction what you would see is a consistent state but it would not be strictly serializable, it would just be a serializable state. And if this was a read write transaction we would enforce serializability at the commit phase and abort if one of these objects was updated. However, if you do take a read time stamp in the future, wait out uncertainty, you get strictness and serializability. Next let's look at our clock master failover. We make sure that any machine can be the clock master and we do this by just using CPU clocks to take the time. On our clock master algorithm has the following high level intuition. First, when a clock master fails we elect a new clock master, the clock master disables its own and all the other clocks, it will then advance time such that it is higher than any other, any uncertain times that we have seen and finally it will re-enable clocks. Importantly our protocol is able to survive failures during failover. So just have a look at the specifics of how it works. You have a new clock master that was just elected and you have another server that is attempting to synchronize. So the very first thing is a new clock master does is it disables its own clock. It sends a message to all the other servers saying disable your clocks. While the other servers disable the clocks, it works out a variable ff which is the maximum of all the upper bounds of the uncertainty window that it has and responds to the clock master with this ff variable. The clock master has to wait out lease expiry to make sure that any machines that are not part of this new configuration are no longer attempting to act on transactions or requests that are being sent to it. Once this is done, we calculate a new ff which is the maximum of all the ff's in the cluster. We send this ff to all the other servers such that time has been fast forwarded now and then we enable the clock on the clock master and regular synchronization can occur. To evaluate our system, we had a 19 machine cluster and we ran TPCC with 3-way replication and all our experiments, unless otherwise specified, were strictly serializable. This graph represents the scalability of farm. On the y-axis, you see the throughput in millions of new order transactions per second and on the x-axis is the number of servers. We are able to achieve at the farm master's right when there's 19 machines 5.4 million new order transactions per second which is as far as we're aware of the highest published number at this time. Now when we look at the uncertainty rate as we scale out the cluster, you can see the blue line that I introduced with the new access on the right-hand side being the mean uncertainty in microseconds. You can see here as we increase the cluster size from 3 machines to 90 machines, the uncertainty goes from 7.5 microseconds to 12 microseconds which is only an increase of 3.5 microseconds for a 30x increase in the number of machines. Now it was also interesting to see what happens when the cluster gets bigger. Unfortunately, we did not have more than 90 machines to run these experiments on. So what we did is we had an experiment to sample this. We said that we are using 19 machines and we would take every x sample to represent what happens if we get a slow back. So if we're saying that there's 19 machines and we're taking every second sample, it's as if there are 1,180 machines synchronizing and we get the response back half of the times we would maintain the same load on the clockmaster. So in this diagram on the left-hand side, you can see the mean uncertainty rate again in microseconds and the sampling ratio at the bottom. And here when we go from 19 machines to 900 machines with this ratio, we only increase uncertainty by 5 microseconds. Next we looked at the TPCC throughput versus latency graph. And for this experiment, we were using 57 machines. On the y-axis is the latency of the new order transactions in milliseconds and on the x-axis is through the throughput in millions of new order transactions per second. The main takeaway from this graph is that as we increase the load on the system, latency still is less than 1 millisecond up to a maximum of 3.5 million new order transactions per second. Now we want to look at what the cost of opacity was. So we were able to run our system without the opacity feature. The throughput overhead for opacity was 3.6% for the TPCC experiment I just showed you with 57 machines. And the latency overhead at low load at the 50th percentile was 39 microseconds and at the 99th percentile, 69 microseconds. So transactions that had microsecond latency would be affected by this overhead. The next thing we wanted to look at is what happens if we remove the strictness or the serializability properties from our system. And while this would be bad for programmability, we believe that this is just an interesting experiment to understand how the system functions. So if we remove either strictness or serializability, we can get a 3% increase of TPCC throughput. However, if we remove both, we get a 5% performance increase, a throughput increase. We believe that this cost is low enough that the advantage of having opacity and helping developers with their programmability is what the price that we pay. Finally we looked at recovery. This diagram on the left-hand side shows a throughput again in millions of new order transactions per second. And the x-axis is the walk clock time starting at arbitrary point zero and increasing in milliseconds. Here we kill two machines at the same time. And one of those machines is the clock master. So we can see that we stop handing out the time for when the clock master is paused is 19 milliseconds. And we return to the same throughput within 67 milliseconds. This is similar to the performance we had with Farm before opacity. And in itself, this blip is lower than most tail latencies for some production systems. We had a lot more other work in the paper, which I don't have enough time to go into. We implemented a distributed MBCC protocol. We talk about how the scalability of our time stamp generation, we go into a lot of detail of what happens when we increase load, and optimizations to get very low microsecond uncertainty among many other things. So in conclusion, we built a new system with opacity, high throughput, low latency, high availability, and strict serializability. Thank you. Thank you.
The distributed applications are hard to build. It would be really nice to have an abstraction of the cluster, never fails and gives the illusion of running transactions sequentially. Farm for enforcing strict serializability. And this opacity to Farm while maintaining its great a timestamp ordering service with scalable timestamp generation. We used local CPU clocks. algorithm is based on Mozilla's algorithm. And from Mozilla's algorithm. As you can see, that we of this, of the time that we got from the clock new transaction and they would get a read time committed successfully. So just to provide an clock master and this is to ensure that all locks or updates are seen by the transaction that is failures during failover. So just have a look at that any machines that are not part of this new our system, we had a 19 machine cluster and we ran TPCC with 3-way replication and all our master's right when there's 19 machines 5.4 million new order transactions per second which 90 machines, the uncertainty goes from 7.5 microseconds to 12 microseconds which is only an throughput. However, if we remove both, we get a machines is the clock master. So we can see that milliseconds. This is similar to the performance high throughput, low latency, high availability,