
\subsection{Analysis of Throughput}\label{subsec:result-analysis-throughput}

\textbf{Comparison of Eventual Consistency with Monitors vs. Sequential Consistency}.
As discussed in the introduction, one of the problems faced by the designers is that they have access to an algorithm that is correct under sequential consistency but the underlying key-value store provides a weaker consistency. In this case, one of the choices is to pretend as if sequential consistency is available but monitor the critical predicate $P$. If this predicate is violated, we need to rollback to an earlier state and resume the computation from there. Clearly, this approach would be feasible if the monitored computation with eventual consistency provides sufficient benefit compared with sequential consistency. In this section, we evaluate this benefit. 

Figure \ref{fig:coloring-benefit} compares the performance of our algorithms for eventual consistency with monitors and sequential consistency without monitors in the \textit{Social Media Analysis} application on the AWS environment. Using our approach, the client throughput was increased by \SI{57}{\percent} (for N3R1W3) and \SI{78}{\percent} (for N3R2W2). Note that the cost of a GET request is more expensive in N3R2W2 (the required number of positive acknowledgment is 2) than in N3R1W3 (the required acknowledgment is 1). Since in the \textit{Social Media Analysis} application GET requests dominates, the application performs better in N3R1W3 than in N3R2W2.

\begin{figure*}[t]
   
    \begin{center}
        \subfigure[Benefit=57\% and 78\%]{%
            \label{fig:coloring-benefit}
            [width=0.46\textwidth]{{Benefit-app-coloring-N3R1W3-N3R1W1-C15-put0.50-uni-gaF-phdT}.eps}
        }
        \subfigure[Overhead on R1W1=1.4\%]{%
            \label{fig:coloring-overhead-1}
            [width=0.46\textwidth]{{OH-server-coloring-N3R1W1-C15-put0.50-uni-gaF-phdT}.eps}
        } \\%
        \subfigure[Overhead on R1W3=1.7\%]{%
           \label{fig:coloring-overhead-2}
            [width=0.46\textwidth]{{OH-server-coloring-N3R1W3-C15-put0.50-uni-gaF-phdT}.eps}
        }
        \subfigure[Overhead on R2W2=1.3\%]{%
            \label{fig:coloring-overhead-3}
            [width=0.46\textwidth]{{OH-server-coloring-N3R2W2-C15-put0.50-uni-gaF-phdT}.eps}
        }
    \end{center}
    \vspace{-10pt}
    \caption{(AWS) \textit{Social Media Analysis} application, 3 servers, 15 clients. The benefit of eventual consistency with monitors vs. sequential consistency without monitors (throughput improvement compared to R1W3 and R2W2 is 57\% and 78\%, respectively), and the overhead of running monitors on each consistency setting (the overhead is less than $2\%$).}
    \label{fig:coloring-benefit-overhead}
\end{figure*}

\textbf{Overhead of monitoring.} A weaker consistency model allows the application to increase the performance on a key-value store as illustrated above. To ensure correctness, a weaker consistency model needs monitors to detect violations and trigger rollback recovery when such violations happen. As a separate tool, the monitors are useful in debugging to ensure that the program satisfies the desired property throughout the execution. In all cases, it is desirable that the overhead of the monitors is small so that they would not curtail the benefit of weaker consistency or make the debugging cost expensive.

Figures \ref{fig:coloring-overhead-1} \ref{fig:coloring-overhead-2}, and \ref{fig:coloring-overhead-3} show the overhead of the monitors on different consistency settings in the \textit{Social Media Analysis} application. The overhead was between \SIrange{1}{2}{\percent}. At its peak, the number of active predicates being monitored reached \num{20 000} predicates. Thus, the overhead remains reasonable even with monitoring many predicates simultaneously.  


\subsection{Analysis of System and Application Factors}

\textbf{Impact of workload characteristics.} In order to evaluate the impact of workload on our algorithms we ran the \textit{Weather Monitoring} application where the proportional of PUT and GET was configurable. 
The number of servers was 5 and the number of clients was 10. The machines hosting the servers and clients were in the same AWS region (North Virginia, U.S.) but in 5 different availability zones. We choose machines in the same region to reduce the latency (to less than \SI{2}{\ms}), thus increasing the throughput measure and stressing the servers and the monitors. If we put the clients and servers in different regions (e.g., Frankfurt Germany, Oregon USA, Ohio USA) then the throughput for 15 clients is low. To stress it further, we would have to add hundreds of clients which is very expensive. Hence, for the stress test, we put the servers and clients in the same region.


\begin{figure*}[th]
    \begin{center}
        \subfigure[PUT=25\%. Benefit=18\%]{%
            \label{fig:graph-aws-1-benefit}
            [width=0.3\textwidth]{{New-Fig-graph-N5R1W5-N5R1W1-C10-put0.25}.eps}
        }%
        \subfigure[PUT=50\%. Benefit=37\%]{%
           \label{fig:graph-aws-2-benefit}
           [width=0.3\textwidth]{{New-Fig-graph-N5R1W5-N5R1W1-C10-put0.50}.eps}
        }
        \subfigure[PUT=50\%. Overhead=4\%]{%
            \label{fig:graph-aws-3-overhead}
            [width=0.3\textwidth]{{Overhead-graph-N5R1W1-C10-put0.50}.eps}
        }
    \end{center}
    \caption{Benefit and overhead of monitors in \textit{Weather Monitoring} application. Percentage of PUT requests is 25\% and 50\% Number of servers =$5$. Number of clients = $10$. Machines are on the AWS North Virginia region but in different availability zones.}
    \label{fig:graph-aws}
\end{figure*}


From Figures \ref{fig:graph-aws-1-benefit} and \ref{fig:graph-aws-2-benefit}, we find that when the percentage of PUT request increased from \SIrange{25}{50}{\percent}, the benefit over sequential consistency (N5R1W5 in this case) increased from \SIrange{18}{37}{\percent}. 
This is because the cost for a PUT request is expensive in N5R1W5 as a PUT request is successful only when it is confirmed by all 5 servers. Thus, when the proportion of PUT increases, the performance of N5R1W5 decreases. In such cases, sequential settings that balance R and W (e.g. N5R3W3) will perform better than sequential settings that emphasize W (e.g. N5R1W5). When GET requests dominate, it is vice versa (cf. Figure \ref{fig:coloring-benefit}). 
We also observe that, when PUT percentage increased and other parameters were unchanged, the aggregated throughput measured at clients decreased. That is because a PUT request consists of a GET\_VERSION request (which is as expensive as a GET request) and an actual PUT request, therefore a PUT request takes a longer time to complete than a GET request does.


Regarding overhead, Figure \ref{fig:graph-aws-3-overhead} shows that the overhead was \SI{4}{\percent} when PUT percentage was \SI{50}{\percent}. Note that in \textit{Weather Monitoring} application, the number of predicates being monitored is proportional to the number of clients. Thus, the overhead remains reasonable even when monitoring several predicates simultaneously and the servers are stressed.

The number of violations detected in this experiment was only one instance in executions with a total time of \SI{18 000}{\ms}. The violation was detected within \SI{20}{\ms}. 



\textbf{Impact of network latency.}
We ran experiments on the local lab network (cf. Section \ref{subsec:experiment-setup}) where the one-way latency within a region (cf. Figure \ref{fig:local-lab-network}) was \SI{1}{\ms} and one-way latency between regions varied from \SIrange{50}{100}{\ms}. 
The number of clients per each server varied between $10$ and $20$. The values in sub-columns ``server'' and ``app'' are the aggregate throughput measured at the servers and at the applications (unit is \si{ops}).
In Table \ref{table:local-overhead-benefit}, the overhead is computed by comparing server measurements when the monitors are enabled and disabled. The benefit is computed by comparing application measurements on sequential consistency without monitoring to those on eventual consistency with monitoring. 
For example, when one-way latency is \SI{50}{\ms}, if we run the \textit{Weather Monitoring} application on N3R1W3 without monitoring, the aggregate server throughput is \SI{649}{ops} (Table \ref{table:local-overhead-benefit}, column 12 (N3R1W3 $\rightarrow$ server) and row 6 (\SI{50}{\ms} $\rightarrow$ \textit{Weather Monitoring} $\rightarrow$ Monitor = no)) and the aggregate client throughput is \SI{313}{ops}.
If we run the same application on N3R1W3 with monitoring, the server throughput is \SI{628}{ops} (Table \ref{table:local-overhead-benefit}, column 12 and row 5). The overhead of monitoring \textit{Weather Monitoring} application on N3R1W3 is $(649-628)/649 = 3.2\%$. The client throughput when run the same application on N3R1W1 with monitoring is \SI{454}{ops} (Table \ref{table:local-overhead-benefit}, column 7, row 5). Thus, the benefit of eventual consistency with monitoring vs. sequential consistency N3R1W3 is $(454-313)/313 = 45\%$.

From Table \ref{table:local-overhead-benefit}, as latency increases, the benefit of eventual consistency with monitoring vs. sequential consistency increases. For example when one-way latency increased from \SIrange{50}{100}{\ms}, in \textit{Social Media Analysis} application, the benefit of eventual consistency with monitoring vs. sequential consistency R1W3 increased from \SIrange{47}{60}{\percent}. In the case of R2W2, the increase was from \SIrange{65}{80}{\percent}.
This increase is expected because when latency increases, the chance for a request to be successful at a remote server decreases. Due to strict replication requirement of sequential consistency, the client will have to repeat the request again. On the other hand, on eventual consistency, requests are likely to be successfully served a local server and the client can continue regardless of results at remote servers. Hence, as servers are distributed in more geographically disperse locations, the benefit of eventual consistency is more noticeable. Regarding overhead, it was generally less than \SI{4}{\percent}. In all cases, the overhead was at most \SI{8}{\percent}.


\begin{table*}[htbp]
\caption{Overhead and benefit of monitors in local lab network. For \textit{Conjunctive} and \textit{Weather Monitoring}, PUT percentage is 50\%.}
\vspace{-10pt}
\begin{center}
\begin{tabular}{|c|l|c|c|r|l|r|r|l|r|l|r|l|r|l|}
\hline
\multirow{2}{*}{\parbox{0.8cm}{Latency (ms)}} & \multirow{2}{*}{Application}  & \multirow{2}{*}{\parbox{0.8cm}{Client/ Server}}  & \multirow{2}{*}{\parbox{0.8cm}{Monitor}} & \multicolumn{ 3}{c|}{N3R1W1  } & \multicolumn{ 4}{c|}{N3R2W2  } & \multicolumn{ 4}{c|}{N3R1W3  } \\ \cline{5-15}
&  &  &  & \multicolumn{1}{l|}{server} & {\parbox{0.9cm}{overhead}} & \multicolumn{1}{l|}{app} & \multicolumn{1}{l|}{server} & {\parbox{0.9cm}{overhead}} & \multicolumn{1}{l|}{app} & benefit & \multicolumn{1}{l|}{server} & {\parbox{0.9cm}{overhead}} & \multicolumn{1}{l|}{app} & benefit \\ \hline
\multirow{6}{*}{50} & \multirow{2}{*}{Conjunctive} & \multirow{2}{*}{20} & yes & 821 & \multicolumn{1}{r|}{-0.2\%} & 470 & 842 & \multicolumn{1}{r|}{0.6\%} & 375 & \multicolumn{1}{r|}{25.3\%} & 588 & \multicolumn{1}{r|}{3.3\%} & 337 & \multicolumn{1}{r|}{40.7\%} \\ \cline{4-15}
 & &  & no & 819 &  & 470 & 847 &  & 375 &  & 608 &  & 334 &  \\ \cline{2-15}
 & \multirow{2}{*}{\parbox{1.5cm}{Weather Monitoring}} & \multirow{2}{*}{20} & yes & 924 & \multicolumn{1}{r|}{0.2\%} & 454 & 795 & \multicolumn{1}{r|}{7.1\%} & 345 & \multicolumn{1}{r|}{27.2\%} & 628 & \multicolumn{1}{r|}{3.2\%} & 312 & \multicolumn{1}{r|}{45.0\%} \\ \cline{4-15}
 &  &  & no & 926 &  & 453 & 856 &  & 357 &  & 649 &  & 313 &  \\ \cline{2-15}
 & \multirow{2}{*}{\parbox{1.5cm}{Social Media Analysis}} & \multirow{2}{*}{10} & yes & 560 & \multicolumn{1}{r|}{0.2\%} & 258 & 367 & \multicolumn{1}{r|}{0.5\%} & 156 & \multicolumn{1}{r|}{65.4\%} & 344 & \multicolumn{1}{r|}{7.8\%} & 174 & \multicolumn{1}{r|}{47.4\%} \\ \cline{4-15}
 &  &  & no & 561 &  & 267 & 369 &  & 156 &  & 373 &  & 175 &  \\ \hline
\multirow{6}{*}{100} & \multirow{2}{*}{Conjunctive} & \multirow{2}{*}{20} & yes & 476 & \multicolumn{1}{r|}{0.4\%} & 270 & 491 & \multicolumn{1}{r|}{-0.2\%} & 218 & \multicolumn{1}{r|}{23.3\%} & 354 & \multicolumn{1}{r|}{0.0\%} & 191 & \multicolumn{1}{r|}{42.1\%} \\ \cline{4-15}
 &  &  & no & 478 &  & 271 & 490 &  & 219 &  & 354 &  & 190 &  \\ \cline{2-15}
 & \multirow{2}{*}{\parbox{1.5cm}{Weather Monitoring}} & \multirow{2}{*}{20} & yes & 544 & \multicolumn{1}{r|}{0.7\%} & 266 & 500 & \multicolumn{1}{r|}{1.0\%} & 209 & \multicolumn{1}{r|}{28.5\%} & 371 & \multicolumn{1}{r|}{0.8\%} & 176 & \multicolumn{1}{r|}{49.4\%} \\ \cline{4-15}
 &  &  & no & 548 &  & 273 & 505 &  & 207 &  & 374 &  & 178 &  \\ \cline{2-15}
 & \multirow{2}{*}{\parbox{1.5cm}{Social Media Analysis}} & \multirow{2}{*}{10} & yes & 287 & \multicolumn{1}{r|}{0.0\%} & 135 & 236 & \multicolumn{1}{r|}{0.0\%} & 74 & \multicolumn{1}{r|}{80\%} & 185 & \multicolumn{1}{r|}{-0.5\%} & 86 & \multicolumn{1}{r|}{60.7\%} \\ \cline{4-15}
 &  &  & no & 287 &  & 133 & 236 &  & 75 &  & 184 &  & 84 &  \\ \hline
\end{tabular}
\end{center}
\label{table:local-overhead-benefit}
\end{table*}

\subsection{Analysis of Violations and Detection Latency}

\textit{Detection latency} is the time elapsed between the violation of the predicate being monitored and the time when the monitors detect it. 
In our experiment with \textit{Social Media Analysis} applications on eventual consistency (N3R1W1), in several executions of total \SI{9 000}{\s}, we detected only 2 instances of mutual exclusion violations. Detection latency for those violations were \SIlist{2 238;2 213}{\ms}. So for \textit{Social Media Analysis} application, violations could happen on eventual consistency every \SI{4 500}{\s} on average.






In order to evaluate the detection latency of monitors with higher statistical reliability, we need experiments where violations are more frequent.
In these experiments, the clients ran \textit{Conjunctive} application in the same AWS configuration as \textit{Weather Monitoring} application above.
The monitors have to detect violations of conjunctive predicates of the form $P = P_1 \wedge P_2 \wedge \cdots P_{10}$. Furthermore, we can control how often these predicates become true by changing when local predicates are true. In these experiments, the rate of local predicate being true ($\beta$) was \SI{1}{\percent}, which was chosen based on the time breakdown of some MapReduce applications \cite{RRPBK07HPCA, BPERST10SIGMOD}. 
The PUT percentage was \SI{50}{\percent}. The \textit{Conjunctive} application is designed so that the number of predicate violations is large and to stress the monitors. We considered both eventual consistency and sequential consistency. 
Table \ref{table:response-time-freq} shows detection latency distribution of more than \num{20 000} violations recorded in the \textit{Conjunctive} experiments. Predicate violations are generally detected promptly. Specifically, \SI{99.93}{\percent} of violations were detected in \SI{50}{\ms}, \SI{99.97}{\percent} of violations were detected in \SI{1}{\s}.
There were rare cases where detection latency was greater than ten seconds. Among all the runs, the maximum detection latency recorded was \SI{17}{\s}, the average was \SI{8}{\ms}.


\begin{table}[htbp]
\caption{Response time in $20,647$ conjunctive predicate violations}
\begin{center}
\begin{tabular}{|l|l|l|}
    \hline
    Response time (milliseconds) & Count & Percentage \\
    \hline
    $< 50$ & 20,632 & $99.927\%$ \\
    \hline
    $50 - 1,000$ & 6 & $0.029\%$ \\
    \hline
    $1,000 - 10,000$ & 3 & $0.015\%$ \\
    \hline
    $10,000 - 17,000$ & 6 & $0.029\%$ \\
    \hline
\end{tabular}
\label{table:response-time-freq}
\end{center}
\end{table}


Regarding overhead and benefit, the overhead of monitors on N5R1W1, N5R1W5, and N5R3W3 was \SIlist{7.81; 6.50; 4.66}{\percent}, respectively. The benefit of N5R1W1 over N5R1W5 and N5R3W3 was \SIlist{27.90; 20.16}{\percent}, respectively.




\subsection{Evaluating Strategies for Handling Livelocks}


In this section, we evaluate the effect of rollback mechanisms. We consider the evaluation of the \textit{Social Media Analysis} with a power-law graph and \textit{Weather Monitoring} with grid-based graph (cf. Section \ref{subsec:experiment-setup} for description of the graphs). We consider the execution with sequential consistency, eventual consistency with rollback but no mechanism for dealing with livelocks, and eventual consistency with one or more mechanism for dealing with livelocks. The results are shown in Figure \ref{fig:livelock}.

\begin{figure*}[t]
    \begin{center}
        \subfigure[]{%
           \label{fig:fig:livelock-social}
           [width=0.46\textwidth]{{node-progress-graph-20ms-livelock}.eps}
        }
        \subfigure[]{%
           \label{fig:livelock-weather}
           [width=0.46\textwidth]{{node-progress-wea-grid-put0.50-d5ms-size300-livelock}.eps}
        }
    \end{center}
    \caption{Effectiveness of livelock handling mechanisms. Number of servers=3, number of clients=30. We observed that adaptive mechanism worked best for \textit{Social Media Analysis} (Figure \ref{fig:fig:livelock-social}), and backoff mechanism worked best for \textit{Weather Monitoring} (Figure \ref{fig:livelock-weather}).}
    \label{fig:livelock}
\end{figure*}

From this figure, we observe that the impact of livelocks is not the same in different applications. In particular, for terminating applications like \textit{Social Media Analysis}, if the livelock issue is ignored, the computation does not terminate. Likewise, computation does not terminate with the mechanism of reordering of remaining tasks upon rollback. This is anticipated, in part, because recurrence of rollback happens in end-stages where the number of remaining tasks is low. 
On the other hand, for non-terminating application like \textit{Weather Monitoring}, livelocks do not cause the computation to stall. Except for adaptive consistency, the effectiveness of different livelock handling strategies are almost similar. 
From Figure \ref{fig:livelock}, we observe that rollback with adaptive consistency works best for terminating applications, and rollback with backoff works best for non-terminating applications. 
Therefore, we choose these mechanisms to handle livelocks in the detailed analysis of applications in Section \ref{subsec:applications}. 





\subsection{Analysis of Applications}
\label{subsec:applications}


In this section, to illustrate the benefit of our approach, we run the recovery algorithm described in section \ref{subsec:rollback-mechanism} for two applications: \textit{Weather Monitoring} and \textit{Social Media Analysis}. We do not consider \textit{Conjunctive}, as it was designed explicitly to cause too many violations for the purpose of detecting latency of violations. The analysis was performed in our local lab network with the round-trip latency varying between \SIrange{5}{50}{\ms}. We use the approach in Section \ref{subsec:experiment-setup} to add additional delays to evaluate the behavior of the application in a realistic setting where replicas are not physically co-located. 
In order to deal with livelocks, we utilize the backoff mechanism for \textit{Weather Monitoring} application, and adaptive mechanism for \textit{Social Media Analysis} application.
The number of servers was 3 and the number of clients was 30.

\textbf{Weather Monitoring.} 
When running the \textit{Weather Monitoring} application with eventual consistency, first, we consider the nodes organized in a line. In this case, the application progressed \SI{47.2}{\percent} faster than running on sequential consistency (cf. Figure \ref{fig:node-progress-weather-line-put0.50-delay10ms-lineSize1000}).
Even if we extend it to a grid graph, the results are similar. In Figure \ref{fig:node-progress-weather-grid-gridSize500}, we find that in the grid graph, the application progressed \SI{46.8}{\percent} faster under eventual consistency than in sequential consistency. In both of these executions, no violations were detected in the \SIlist{500; 1000}{\s} window, respectively.

To evaluate the effect of rollbacks, we increase the chance of conflicts by reducing the coverage of each client (i.e. the number of nodes in the graph assigned to each client) so that the clients work on bordering nodes more frequently. In that setting, on a line graph, eventual consistency still progressed about \SI{45}{\percent} faster than running on sequential consistency (cf. Figure \ref{fig:node-progress-weather-line-put0.50-delay10ms-lineSize500}), even though we had a substantial number of rollbacks (\num{36} in \SI{500}{\s}). The detection latency for violation was on average \SI{18}{\ms}. The worst case detection latency was \SI{55}{\ms}. We note that the application motivated by \textit{Weather Monitoring} is a non-terminating application which keeps running without termination. Hence, the number of nodes processed measured in stable phase reflects the overall progress of the application. For this reason, in order to compare the progress of different experiment configurations, we measure the progress made by the clients after the same execution duration. For example, in Figure \ref{fig:node-progress-weather-line-put0.50-delay10ms-lineSize1000}, the larger points on each line are where we measure the progress after the execution has run for \SI{490}{\s}.
Figure \ref{fig:node-progress-weather-line-put0.50-delay10ms-lineSize500} also considers the progress made by the application on eventual consistency without rollback or monitoring. 
Thus, the resulting answer may be incorrect. The reason for this analysis is to evaluate the cost of monitoring and rollback. As shown in Figure \ref{fig:node-progress-weather-line-put0.50-delay10ms-lineSize500}, the cost of rollback is very small. Specifically, with rollback, the number of nodes processed decreased by about \SI{1.4}{\percent}.

\begin{figure*}[ht]
    \begin{center}
        \subfigure[]{%
           \label{fig:node-progress-weather-line-put0.50-delay10ms-lineSize1000}
           [width=0.45\textwidth]{{node-progress-weather-line-put0.50-delay10ms-lineSize1000}.eps}
        }
        \subfigure[]{%
           \label{fig:node-progress-weather-line-put0.50-delay10ms-lineSize500}
           [width=0.45\textwidth]{{node-progress-weather-line-put0.50-delay10ms-lineSize500}.eps}
        } \\
        \subfigure[]{%
            \label{fig:node-progress-weather-grid-gridSize500}
            [width=0.45\textwidth]{{node-progress-weather-grid-put0.50-delay5ms-gridSize500}.eps}
        }
        \subfigure[]{%
            \label{fig:node-progress-weather-grid-gridSize300}
            [width=0.45\textwidth]{{node-progress-weather-grid-put0.50-delay5ms-gridSize300}.eps}
        }
    \end{center}
    \caption{The benefit and overhead of Eventual consistency+Rollback vs. Sequential consistency in \textit{Weather Monitoring} application.  The inset figure within Figure \ref{fig:node-progress-weather-line-put0.50-delay10ms-lineSize500} is a close-up view showing the impact of rollback. The larger points near the end of each data sequence are where we choose the representative values for the data sequences.}
    \label{fig:node-progress-weather}
\end{figure*}

In grid-based graphs, eventual consistency progressed \SI{45.1}{\percent} faster than sequential consistency did (cf. Figure \ref{fig:node-progress-weather-grid-gridSize300}) even though it had to rollback a number of times (68 times in \SI{1000}{\s}). The detection latency was \SI{10}{\ms} on average, and \SI{41}{\ms} in the worst case. The cost of rollback was \SI{1.4}{\percent}.



\textbf{Social Media Analysis. }
Since the \textit{Weather Monitoring} task is a non-terminating task, its behavior remains the same throughout the execution. Hence, to evaluate the effect of termination, we evaluate our approach in the \textit{Social Media Analysis} application. Terminating computation suffer from the following when compared with non-terminating computations: (1) At the end, some clients may have completed their task thereby reducing the level of concurrency, and (2) The chance of rollback resulting in the same conflict increases, as the tasks remaining are very small. Therefore, the computation after rollback is more likely to be similar to the one before the rollback. In other words, the conflict is likely to recur. 




We evaluate the effect of termination in two types of graph: (1) Power-law clustering (cf. Figure \ref{fig:node-progess-50ms}), and (2) Regular graphs (cf. Figure \ref{fig:node-progress-regular-graph-50ms}) where degrees of all nodes are \textit{close}. (The details of these graphs is given in Section \ref{subsec:experiment-setup}.)  

On power-law clustering graphs, as shown in Figure \ref{fig:node-progess-50ms}, before the execution reached $90\%$ completion of the work, eventual consistency -- even with the cost of monitoring and rolling back -- progressed about $18.5\%$ faster than sequential consistency. However, in the remaining $10\%$ of the work, when there were a few nodes to be colored, the chance of conflict increased. Furthermore, the same conflict occurred after rollback as well. Hence, in the final phase, execution under eventual consistency almost stalled due to frequent rollbacks. When the clients utilized adaptive consistency then they could make progress through the final phase and finished about $9.5\%$ faster than sequential consistency. We note that the decline in computation rate in the final phase is also true for sequential consistency, and that is related to a property of power-law cluster graph that some nodes are high degree nodes. In regular random graph, we do not observe this decline as shown in Figure \ref{fig:node-progress-regular-graph-50ms}. The main reason for this is that the likelihood of conflict in the power-law graph is high since there are several nodes with a high degree. Furthermore, it is difficult to distribute the workload of power-law clustering graph to the clients evenly. Therefore, in the final phase, some clients have completed before the others, thus reducing the parallelism. By contrast, in the regular graph, the likelihood of conflict in end stages remains the same and the workload can be evenly distributed among the clients. On a regular graph, eventual consistency with monitoring and rollback was $26\%$ faster than sequential consistency before $90\%$ of the nodes were processed, and $20.8\%$ faster overall (cf. Figure \ref{fig:node-progress-regular-graph-50ms}).



\begin{figure*}[ht]
    \begin{center}
       
       
       
       
        \subfigure[]{%
            \label{fig:node-progess-50ms}
            [width=0.45\textwidth]{{node-progress-50ms-adaptive}.eps}
        }
        \subfigure[]{%
           \label{fig:node-progress-regular-graph-50ms}
           [width=0.45\textwidth]{{node-progress-50ms-regular-adaptive}.eps}
        } 
    \end{center}
    \caption{Comparing the completion time of Sequential Consistency (R1W3) vs. Eventual Consistency with rollback and adaptive consistency (R1W1+adaptive) in \textit{Social Media Analysis} application. On a power-law clustering graph, before 90\% of the nodes are processed, R1W1+adaptive progresses about 18\% faster than R1W3. Overall, R1W1+adaptive is $9.5\%$ faster than R1W3. On a regular random graph, the benefit before $90\%$ of the nodes are processed is $26\%$ and the overall benefit is $20.8\%$.}
    \label{fig:computation-progress}
\end{figure*}










\subsection{Discussion}
\label{subsec:discussion}

In this section, we consider some of the questions raised by this work including questions raised by the reviewers of LADC 2018 and JBCS.

\textbf{What is the likely effect of the number of clients on the probability of rollback?}
First, we note that for linear predicates (e.g. conjunctive predicates), when the number of clients increases, the number of violations decreases as it is less likely to find a consistent snapshot where the local predicate at every client is true \cite{YNVSD16RV}. As a result, if we increase the concurrency level, the probability of rollback decreases.
Hence in the following discussion, we limit the context to semi-linear predicates (e.g. mutual exclusion).

In general, the probability of rollback depends upon the probability that two clients are updating conflicting data. Thus, if the number of clients is too large when compared with the size of the graph (i.e. the coverage of a client is too small), the probability of conflict/rollback is high. We have validated this with experimental analysis of applications motivated by \textit{Weather Monitoring} in section \ref{subsec:applications}. However, in a typical deployment, the coverage of a client is usually large enough (e.g. thousands of nodes) that the chance of two clients concurrently working on neighboring nodes is small. Furthermore, when working on neighboring nodes, clients utilize mutual exclusion mechanism such as Peterson locks to prevent conflicts. Conflicts/rollbacks only happen when there is some data inconsistency related to the mutual exclusion mechanism that causes the clients concurrently updating neighboring nodes. In eventual consistency, data inconsistencies exist but are rare \cite{dynamo} and usually involve hardware and/or network failures. Hence, from our analysis, we anticipate that the probability of rollback is small given that each client is assigned a reasonable workload.

\textbf{How do the observations in this paper relate to the CAP theorem?}
When latency increases, we are simulating pseudo network partition. In this case, which consistency level is better depends upon the configuration of the Voldemort servers. 

As an illustration, consider the example where we have 5 servers and we use R2W4. Furthermore, suppose that one of the servers is partitioned from the other 4 servers.

With sequential consistency, the 4 servers and their associated clients can still make progress correctly. The partitioned server and its clients would not make progress. Hence, they could be considered as being dead. By some detection mechanism, we can detect such partitioning and assign the tasks of dead clients to other clients. And the computation could progress to the end.

With eventual consistency, all 5 servers and their clients make progress but the clients will process based on stale data. When the network is recovered, the data inconsistencies will invalidate the computation results of both sub-networks.
When the monitor detects such partition and inconsistencies, we will have to rollback the whole systems, including the 4 servers and their clients (even though their results are correct, given that the partitioned server has been rolled back).


While the above discussion applies to R2W4, if the system used R1W5 then neither eventual consistency nor sequential consistency could make progress. This is because eventual consistency would result in inconsistently updated replicas. These inconsistencies would be resolved based on the implementation of Voldemort (e.g., latest write wins, minority replicas follow the majority replicas, etc). However, this conflict resolution may not be consistent with the needs of the application. And, in sequential consistency, no write operation would succeed. 
However, if the nodes are not partitioned but rather suffer from a high delay (but no partition), eventual consistency may be able to make progress. However, it would need to rollback frequently. By contrast, in sequential consistency, it is likely that most write operations fail (as they take too long to complete). Consequently, sequential consistency will not be able to make progress. 

What the above discussion suggests is that when the delays are very high, the above approach would work for some configurations (e.g., R2W4) but not for others (e.g. R1W5). 
Hence, one of the future work in this area is to allow only certain clients to rollback while allowing others to continue without rollback. 









\textbf{Applications that cannot be rolled back.}
In this paper, we assume the application has exclusive access to its data. Specifically, before the application finishes, other applications will not read this application results. If the data is shared and used by multiple applications, then the rollback approach is not suitable since it is almost impossible to rollback other applications. For instance, the results of computing shortest paths, routing information can be produced by one application and used by other applications. In this case, other approaches such as self-stabilization can be useful.


\subsection{Experimental Setup}
\label{subsec:experiment-setup}

\textbf{System configurations.} We ran experiments on Amazon AWS EC2 instances. The servers ran on M5.xlarge instances with 4 vCPUs, 16 GB RAM, and a GP2 general-purpose solid-state drive storage volume. The clients ran on M5.large instances with 2 vCPUs and 8 GB RAM. The EC2 instances were located in three AWS regions: Ohio, U.S; Oregon, U.S; Frankfurt, Germany.

We also ran experiments on our local lab network which is set up so that we can control network latency. We used 9 commodity PCs, 3 for servers, 6 for clients, with configurations as in Table \ref{table:local-lab-hardware}. Each client machine hosted multiple client processes, while each server machine hosted one Voldemort server process.

\begin{table}[ht]
\caption{Machine configuration in local lab experiments}
\begin{center}
\begin{tabular}{|l|l|l|}
    \hline
    Machine & CPU & RAM \\
    \hline
    Server machine 1, 2 & 4 Intel Core i5 3.33 GHz  & 4 GB \\
    \hline
    Server machine 3 & 4 Intel Core i3 3.70 GHz & 8 GB \\
    \hline
    Client machine 1, 2  & 4 Intel Core i5 3.33 GHz& 4 GB \\
    \hline
    Client machine 3, 4& Intel Core Duo 3.00 GHz & 4 GB \\
    \hline
    Client machine 5& 4 AMD Athlon II 2.8 GHz & 6 GB \\
    \hline
    Client machine 6 & 4 Intel Core i5 2.30 GHz & 4 GB \\
    \hline
\end{tabular}
\label{table:local-lab-hardware}
\end{center}
\end{table}


On the local network, we control the delay by placing proxies between the clients and the servers. For all clients on the same physical machine, there is one proxy process for those clients. All communication between those clients and any server is relayed through that proxy (cf. Figure \ref{fig:proxy}). Due to the proxy delays, machines are virtually arranged into three regions as in Figure \ref{fig:local-lab-network}. Latency within a region is small (\SI{2}{\ms}) while those across regions are high and tunable (e.g. \SIrange{50}{100}{\ms}). Since Voldemort uses active replication, we do not place proxies between servers. The latency in the proxies is simulated to follow the Gamma distribution \cite{BMHUV2002PAM,gammaDist2013}.



\color{black}

We considered replication factors ($N$) of 3 and 5. The parameters $R$ (required reads) and $W$ (required writes) are chosen to achieve different consistency models as shown in Table \ref{table:consistency-models}.
The number of servers is equal to the replication factor $N$. The number of clients is varied between 15 and 90.

\begin{table}[!h
\begin{center}
\caption{Setup of consistency models with $N$ (replication factor), $R$ (required reads), and $W$ (required writes)}
\begin{tabular}{|l|l|l|l|l|}
    \hline
    N & R & W  & Abbreviation & Consistency model \\
    \hline
    3 & 1 & 3 & N3R1W3 & Sequential \\
      & 2 & 2 & N2R2W2 & Sequential \\
     & 1 & 1 & N3R1W1 & Eventual \\
    \hline
    5 & 1 & 5 & N5R1W5 & Sequential \\
     & 3 & 3 & N5R3W3 & Sequential \\
     & 1 & 1 & N5R1W1 & Eventual \\
    \hline
\end{tabular}
\label{table:consistency-models}
\end{center}
\end{table}

\begin{figure}[!t
\centering
[width=0.46\textwidth]{ClientProxyDiagram.png}
\caption{Simulating network delay using proxies}
\label{fig:proxy}
\end{figure}

\begin{figure}[!t
\centering
[width=0.46\textwidth]{proxyDelayNetwork.png}
\caption{Network arrangement with proxies}
\label{fig:local-lab-network}
\end{figure}



\textbf{Test cases.} In our experiments, we used 3 case studies: \textit{Social Media Analysis}, \textit{Weather Monitoring}, and \textit{Conjunctive}.

The application motivated by \textit{Social Media Analysis} considers a large graph representing users and their connections. The goal of clients is to update the state of each user (node) based on its connections. For the sake of illustration in our analysis, the attribute associated with each user is a color and the task is to assign each node a color that is different from its neighbors.
We use the tool \textit{networkx} \cite{networkx} to generate input graphs. There are two types of graph: (1) Power-law clustering graph that simulates the power-law degree and clustering characteristics of social networks, and (2) Random 6-regular graph in which each node has 6 adjacent edges and the edges are selected randomly. The reason we use random regular graphs is that they are the test cases where the workload is distributed evenly between clients and throughout the execution. The graphs have \num{50 000} nodes with about \num{150 000} edges.
Each client is assigned a set of nodes to be colored and run a distributed coloring algorithm \cite{Raynal2013distributed}. 

Since the color of a node is chosen based on its neighbors' colors, while a client $C_1$ is coloring node $v_1$, no other client is updating the colors of $v_1$'s neighbors. 
The goal of the monitors is to detect violation of this requirement. 
This requirement can be viewed as a mutual exclusion (semi-linear) predicate where a client going to update the color of $v_1$ has to obtain all the exclusive locks associated with the edges incident to $v_1$.
Mutual exclusion is guaranteed if clients use Peterson's algorithm and the system provides sequential consistency \cite{BW02PPAM}. However, it may be violated in the eventual consistency model.
To avoid deadlock, clients obtain locks in a consistent order.
For example, let $A\_B$ and $C\_D$ are the locks associated with the edges between nodes $A$ and $B$, and $C$ and $D$ respectively. Assume $A < B$ and $C < D$. Then lock $A\_B$ is obtained before $C\_D$ when $A < C$ or when $A = C$ and $B < D$.


The number of predicates being monitored in this test case is proportional to the number of edges.

We note that the task performed by each client (i.e., choosing the color of a node) is just used as an example. It is easily generalized for other analysis of Social Media Graph (e.g., finding clusters, collaborative learning, etc.) 

The application motivated by \textit{Weather Monitoring} task considers a planar graph (e.g. a line or a grid) where the state of each node is affected by the state of its neighbors. 
In a line-based graph, all the nodes of the graph are arranged on a line and each client is assigned a segment of the line.
In a grid-based graph, the graph nodes are arranged on a grid. The clients are also organized as a grid and each client is responsible for a section of the grid of nodes.
In this application, we model a client that updates the state of each node by reading the state of its neighbors and updating its own state. This application can be tailored to vary the ratio of GET/PUT request.
This application is relevant to several practical planar graph problem such as weather forecasting \cite{FBMA2013PCS}, radio-coloring in wireless and sensor network \cite{PSNS1995PODC}, computing Voronoi diagram \cite{NXIA2008CCCG}.

Finally, the \textit{Conjunctive} application is an instance of distributed debugging where the predicate being detected (i.e., $\neg P$) is of the form $P_1 \wedge P_2 \wedge \cdots \wedge P_l$. Each local predicate $P_i$ becomes true with a probability $\beta$ and the goal of the monitors is to determine if the global conjunctive predicate $\neg P$ becomes true.
In this application, we monitor multiple conjunctive predicates simultaneously.
Since we can control how frequently these predicates become true by varying $\beta$, we can use it mainly to assess monitoring latency and stress the monitors. Conjunctive predicates are also useful in distributed testing such as to specify breakpoints.


\textbf{Performance metrics and measurement.} We use throughput as the performance metrics in our experiments. Throughput can be measured at two perspectives: application, and Voldemort server. 
The two perspectives are not the same but related. One application request triggers multiple requests at Voldemort client. For example, one application PUT request is translated into one GET\_VERSION request (to obtain the last version of the key) and one PUT request (with a new incremented version) at the Voldemort client library. Then each Voldemort client request causes multiple requests at servers due to replication. Failures and timeout also make the counts at the applications and the servers differ. For example, an application request is served and counted at a server but if the server response is lost or arrives after the timeout, the request is considered unsuccessful and thus not counted at the application. Generally, servers' counts are greater than applications' counts.
In our experiments, we use the aggregated measurement at servers to assess the overhead of our approach since the monitors directly interfere with the operation of the server, and use aggregated measurement at applications to assess the benefit of our approach because that measurement is close to users' perspective. Hence, in the following sections, for the same experiment, we note that the measurements used for overhead and benefit evaluation are different.

\textbf{Results stabilization.} We ran each experiment three times and used the average as the representative results for that experiment. Figure \ref{fig:rundiff} shows the stabilization of different runs of an experiment. Note that the values are aggregated from all applications. We observe that in every run, after a short period of initialization, the measurements converge on a stable value. When evaluating our approach, we use the values measured at the stable phase.
We also note that the aggregated throughput in Figure \ref{fig:rundiff} is not very high but expected. The pairwise round-trip latency between three AWS regions (Ohio, Oregon, Frankfurt) were \SIlist{76;103;163}{\ms}. The average round-trip latency was \SI{114}{\ms}. On M5.xlarge EC2 instances with a GP2 storage volume, the average I/O latency for a read and a write operation was roughly \SI{0.3}{\ms} and \SI{0.5}{\ms}, respectively. We will roughly estimate the cost of a GET request since in \textit{Social Media Analysis}, most operations are GET requests to read lock availability and colors of neighbors. Assume eventual consistency R1W1 is used, a GET request is executed by Voldemort client in two steps:

\begin{itemize}
\item[1.] Perform parallel request: client simultaneously sends GET requests to all servers (N = 3) and wait for responses with a timeout of \SI{500}{\ms}. The wait is over when either client gets responses from all servers or the timeout expires. In this case, the client will get all responses in about \SI{114.3}{\ms} (\SI{114}{\ms} for communication delay, and \SI{0.3}{\ms} for the read operation processing time at the server). 
\item[2.] Perform serial request: client checks if it has received enough required responses. If not, it has to send addition GET requests to servers to get enough number of responses. If after the additional requests, the required number of responses is not met, the GET request is considered unsuccessful. Otherwise, the result is returned. In the current case, the number of responses received (3) is greater than the required (R = 1). Thus this step is skipped.
\end{itemize}

From this discussion, a GET request takes roughly \SI{115}{\ms} to complete, on average. Since GET is the dominating operation in the \textit{Social Media Analysis} application, with 15 clients, the expected aggregated throughput is $\frac{15}{0.1143} \approx$ \SI{131}{ops}. The average throughput measured in experiments was \SI{132}{ops} (cf. Figure \ref{fig:rundiff}). 

If we run experiments where all machines are in the same region but in different availability zones, the aggregated throughput will be higher (cf. Figure \ref{fig:graph-aws}). For example, in the AWS North Virginia region, the average round-trip latency within an availability zone was about \SI{0.5}{\ms}, and between different availability zones was about \SI{1.4}{\ms}. Based on the discussion about GET request above, a GET request takes roughly \SI{0.8}{\ms} (\SI{0.5}{\ms} for network latency within an availability zone plus \SI{0.3}{\ms} for processing read request at the server). Similarly, a GET\_VERSION request takes \SI{0.8}{\ms}. Since we are using R1W1 configuration, an actual PUT request can be satisfied by the server within the same availability zone. Thus, an actual PUT request takes roughly \SI{1}{\ms} (\SI{0.5}{\ms} for network latency within an availability zone plus \SI{0.5}{\ms} for write operation processing time at the server).
A PUT request (consisting of a GET\_VERSION request and an actual PUT request) takes roughly \SI{1.8}{\ms}.
Assume the workload consists of \SI{50}{\percent} GETs and \SI{50}{\percent} PUTs, then on average, a request takes $0.5 \times 0.8 + 0.5 \times 1.8 =$ \SI{1.3}{\ms} = \SI[group-separator={}]{0.0013}{\s}. With 10 clients, the expected aggregate throughput is $\frac{10}{0.0013} =$ \SI{7692}{ops}. If the workload consists of \SI{75}{\percent} GETs and \SI{25}{\percent} PUTs, a request takes $0.75 \times 0.8 + 0.25 \times 1.8 =$ \SI{1.05}{\ms}  =  \SI[group-separator={}]{0.00105}{\s}, and the expected aggregate throughput is $\frac{10}{0.00105} =$ \SI{9524}{ops}. In our experiments, the aggregate throughput measured for \SI{50}{\percent} PUT and \SI{25}{\percent} PUT was \SI{7782}{ops} and \SI{9593}{ops}, respectively (cf. Figure \ref{fig:graph-aws-1-benefit} and \ref{fig:graph-aws-2-benefit}).

\begin{figure}[t]
\centering
[width=0.46\textwidth]{{Run-differ-app-coloring-N3R1W1-C15-put0.50-predyes}.eps}
    \caption{Illustration of result stabilization. The \textit{Social Media Analysis} application is run three times on Amazon AWS with monitoring enabled. Number of servers ($N$) = $3$. Number of clients per server ($C/N$) = $5$. Aggregated throughput measured by \textit{Social Media Analysis} application in three different runs and their average is shown. This average is used to represent the stable value of the application throughput.}
    \label{fig:rundiff}
\end{figure}



\section{Introduction}\label{sec:intro}
\input{Intro.tex}

\section{System Architecture}\label{sec:sys-arch}
\input{SystemArchitecture.tex}


\section{The Problem of Predicate Detection in Distributed Systems}
\label{sec:problem}
\input{pred}


\section{A Framework for Optimistic Execution}
\label{sec:framework}
\input{framework}


\section{Monitoring Module}\label{sec:pred-detect-module}
\input{PredicateDetectionModule.tex}

\section{Rollback from Violations}
\label{sec:rollback}
\input{rollback.tex}

\section{Evaluation Results and Discussion}\label{sec:evaluations}
\input{Experiment-setup.tex}
\input{Experiment-results.tex}


\section{Related Work}\label{sec:related-work}
\input{related.tex}

\section{Conclusion}\label{sec:conclusion}
\input{conclusion.tex}


\begin{backmatter}

\section*{Abbreviation}
AWS: Amazon Web Service.  
CAP: CAP theorem, also known as Brewer's theorem; C=Consistency, A = Availability, P = Partition tolerance.
CPU: Central Processing Unit. 
EC2: Amazon Elastic Compute Cloud.  
GP2: Amazon General Purpose SSD storage volume. 
HDFS: Hadoop Distributed File System. 
HLC: Hybrid Logical Clock. 
HVC: Hybrid Vector Clock. 
I/O: Input/Output.
JBCS: Journal of the Brazilian Computer Society.
LADC: Latin-American Symposium on Dependable Computing. 
\si{\ms}: The time unit millisecond.
NP in NP-complete and NP-hard: Non-deterministic Polynomial time. 
\si{ops}: Operations per second.
$P$: we often denote a predicate as $P$ in this manuscript (this is not an abbreviation). 
PC: Personal Computer. 
PT: Physical Time. 
RAM: Random Access Memory. 
\si{\s}: The time unit second.
TAO: Facebook's The Associations and Objects distributed data store. 
VC: Vector Clock. 
VLS: Virtual Lightweight Snapshots. 
XML: Extensible Markup Language. 

\section*{Availability of data and material}
      The source code and experimental results supporting the conclusions of this article are at https://doi.org/10.5281/zenodo.3338381.
      
\section*{Competing interests}
   The authors declare that they have no competing interests.

\section*{Author's contributions}
    All authors have contributed to the methodological and research aspects of the research. The authors have also read and approved the final manuscript.


\section*{Acknowledgement}
    We thank the reviewers of the 8th Latin-American Symposium on Dependable Computing (LADC 2018) for their suggestions on our work.

\section*{Authors' information}
    Duong Nguyen, Michigan State University, nguye476@msu.edu; Aleksey Charapko, University at Buffalo, SUNY, acharapk@buffalo.edu; Sandeep S. Kulkarni, Michigan State University, sandeep@cse.msu.edu; Murat Demirbas, University at Buffalo, SUNY, demirbas@buffalo.edu.

    

\bibliographystyle{bmc-mathphys}


\subsection{Vector Clocks and Hybrid Vector Clocks}

To determine whether state $a$ happened before state $b$, we can utilize vector clocks or hybrid vector clocks. Vector clocks, defined by Fidge and Mattern \cite{Fidge87, Mattern89PDA}, are designed for asynchronous distributed systems that make no assumption about underlying speed of processes or about message delivery. Hybrid vector clocks \cite{DK13LADIS} are designed for systems where clocks of processes are synchronized within a given synchronization error (denoted as parameter $\epsilon$ in this paper). 
While the size of vector clocks is always $n$, the number of processes in the system, hybrid vector clocks have the potential to reduce the size to less than $n$.

Our predicate detection module can work with either of these clocks. For simplicity, we recall hybrid vector clocks (HVC) below. 

Every process maintains its own HVC. HVC at process $i$, denoted as $HVC_i$, is a vector with $n$ elements such that $HVC_i[j]$ is the most recent information process $i$ knows about the physical clock of process $j$. $HVC_i[i] = PT_i$, the physical time at process $i$. Other elements $HVC_i[j], j \neq i$ is learned through the communication between processes. When process $i$ sends a message, it updates its HVC as follows: $HVC_i[i] = PT_i$, $HVC_i[j] = max(HVC_i[j], PT_i - \epsilon)$ for $j \neq i$. Then $HVC_i$ is piggy-backed with the outgoing message. Upon reception of a message $msg$, process $i$ will use the piggy-backed hybrid vector clock $HVC_{msg}$ to update its HVC: $HVC_i[i] = PT_i$, $HVC_i[j] = max(HVC_{msg}[j], PT_i - \epsilon)$ for $j \neq i$.

Hybrid vector clocks are vectors and can be compared as usual. Given two hybrid vector clock $HVC_i$ and $HVC_j$, we say $HVC_i$ is smaller than $HVC_j$, denoted as $HVC_i < HVC_j$, iff $HVC_i[k] \leq HVC_j[k] \forall k$ and $\exists l: HVC_i[l] < HVC_j[l]$. If $\neg (HVC_i < HVC_j) \wedge \neg(HVC_j < HVC_i)$, then the two hybrid vector clocks are concurrent, denoted as $HVC_i || HVC_j$.

If we set $\epsilon = \infty$, then hybrid vector clocks have the same properties as vector clocks. If $\epsilon$ is finite, certain entries in $HVC_i$ can have the default value $PT_i - \epsilon$ and their representation can be compressed. For example, if $n = 10, \epsilon = 20$, a hybrid vector clock $HVC_0 = [100, 80, 80, 95, 80, 80, 100, 80, 80, 80]$ could be represented by $n$($10$) bits $10010010001$ and a list of three integers \numlist{100;95;100}, instead of a list of ten integers.

We use HVC in our implementation to facilitate its use when the number of processes is very large. However, in the experimental results, we ignore this optimization and treat as if $\epsilon$ is $\infty$.


\subsection{Different Types of Predicate Involved in Predicate Detection}

In the most general form, predicate $P$ is an arbitrary boolean function on the global state and the problem of detecting $\neg P$ is NP-complete \cite{CG98DC}. However, for some classes of predicates such as linear predicates, semilinear predicates, and bounded sum predicates, there exist efficient detection algorithms \cite{GC95ICDCS, Garg96, CG98DC}. In this paper, we adapt these algorithms for monitoring applications running on key-value stores. Since the correctness of our algorithms follows from the existing algorithms, we omit the detailed discussion of the algorithms and focus on their effectiveness in key-value stores.

\subsection{Predicate Detection in Distributed Systems}

Predicate detection is an important task in distributed debugging. An algorithm for capturing consistent global snapshots and detecting stable predicates was proposed by Chandy and Lamport \cite{CL85TCS}.
A framework for general predicate detection is introduced by Marzullo and Neiger \cite{MN91WDAG} for asynchronous systems, and Stollers \cite{Stoller00DC} for partially synchronous systems.
These general frameworks face the challenge of state explosion as the predicate detection problem is NP-hard in general \cite{CG98DC}.
However, there exist efficient detection algorithms for several classes of practical predicates such as unstable predicates \cite{GW92ICFSTTCS, GW94TPDS, GW96TPDS}, conjunctive predicates \cite{GC95ICDCS, GCMK95HICSS}, linear predicates, semilinear predicates, bounded sum predicates \cite{CG98DC}.
Some techniques such as partial-order method \cite{SUL00CAV} and computation slicing \cite{MG05DC, CGNM13SRDS} are also approaches to address the NP-Completeness of predicate detection.
Those works use vector clocks to determine causality and the monitors receive states directly from the constituent processes. Furthermore, the processes are static. \cite{WMHG09ICPADS, WMH10ICPADS} address the predicate detection in dynamic distributed systems. However, the class of predicate is limited to the conjunctive predicate.
In this paper, our algorithms are adapted for detecting the predicate from only the states of the servers in the key-value store, not from the clients. The servers are static (except failure), but the clients can be dynamics. The predicates supported include linear (including conjunctive) predicates and semilinear predicates.

In \cite{VK2018SIROCCO, VYKTD2017RV}, the monitors use Hybrid Logical Clock (HLC) to determine causality between events in a distributed execution. HLC has the advantage of low overhead but suffers from false negatives (some valid violations are not detected). In contrast, we use hybrid vector clocks to determine causality in our algorithms. In \cite{YNVSD16RV}, the authors discussed the impact of various factors, among which is clock synchronization error, on the precision of the monitors. In this paper, we set epsilon at a safe upper bound for practical clock synchronization error to avoid missing potential violations. In other words, a hybrid vector clock is practically a vector clock. Furthermore, this paper focuses on the efficiency and effectiveness of the monitors. 
Bloom Clock \cite{Ramabaja2019CoRR} is another alternative to vector clock. Due to the overhead of the counting Bloom filter, the benefit of Bloom clock only payoffs on very large distributed systems.

\subsection{Distributed data-stores}
Many NoSQL data-stores exist on the market today, and a vast portion of these systems provide eventual consistency. The eventual consistency model is especially popular among key-value and column-family databases. The original Dynamo~\cite{DHJKLPSVV07SOSP} was one of the pioneers in the eventual consistency movement and served as the basis for Voldemort key-value store. Dynamo introduced the idea of hash-ring for data-sharding and distribution, but unlike Voldemort it relied on server-side replication instead of active client replication.  Certain modern databases, such as Cosmos DB and DynamoDB ~\cite{CosmosDB, DynamoDB} offer tunable consistency guarantees, allowing operators to balance consistency and performance. This flexibility would enable some applications to take advantage of optimistic execution while allowing other applications to operate under stronger guarantees if needed.  However, many data-stores ~\cite{CDEFFFGGHH13TOCS,CDGHWBCFG08TOCS} are designed to provide strong consistency and may not benefit from optimistic execution module. 

Aside from general purpose databases, a variety of specialized solutions exist. For instance, TAO ~\cite{BACCDDFGKL13ATC} handles social graph data at Facebook. TAO is not strongly consistent, as its main goal is performance and high scalability, even across datacenters and geographical regions. Gorilla ~\cite{PFTCHMV15VLDB} is another Facebook's specialized store. It operates on performance time-series data and highly tuned for Facebook's global architecture. Gorilla also favors availability over consistency in regards to the CAP theorem. 
Crail-KV \cite{BCCBK2018IPCC} is Samsung's extension for Apache Crail data storage system \cite{STPSMIK2017DEB} that leverages recent advances in hardware technology, especially key-value solid state drive, to provide higher I/O performance for distributed data store.

Various consistency models in distributed system are presented in the survey \cite{ADMG2019Corr}. In \cite{KKW2019Corr}, the authors introduce the notion of \textit{Fluctuating Eventual Consistency} which is the mix of eventual consistency and strong consistency in order to provide stronger guarantee for eventual consistency. However, this correctness property is not suitable for the adaptive behaviour of application since it is not sufficient to prevent violations as sequential consistency does, and it has more extra synchronization effort than eventual consistency. Consistify \cite{SMG2019ICDCN} is a framework that supports tuning the consistency level of a distributed data store. However, Consistify has to statically analyzes the semantics of the application.


\subsection{Snapshots and Reset}
The problem of acquiring past snapshots of a system state and rolling back to these snapshots has been studied extensively. Freeze-frame file system ~\cite{SGBCX16SoCC} uses Hybrid Logical Clock (HLC) to implement a multi-version Apache HDFS. Retroscope ~ \cite{CADK17ICDCS} takes advantage of HLC to find consistent cuts in the system\textquotesingle s global state by examining the state-history logs independently on each node of the system. The snapshots produced by Retroscope can later be used for node reset by simple swapping of data-files.
Eidetic systems ~\cite{DCDFC14OSDI} take a different approach and do not record all prior state changes. Instead, the eidetic system records any non-deterministic changes at the operating system level and constructing a model to navigate deterministic state mutations. This allows the system to revert the state of an entire machine, including the operating system, data and applications, to some prior point. 
Certain applications may not require past snapshots and instead need to quickly identify consistent snapshots in the presence of concurrent requests affecting the data. VLS ~\cite{CSHF16ICDE} is one such example designed to provide snapshots for data-analytics applications while supporting high throughput of requests executing against the system.

\subsection{Distributed Data Processing}
MapReduce \cite{MapReduce2008CACM}, DataFlow \cite{DataFlow2015VLDB} are general-purpose distributed data processing frameworks. In the realm of distributed graph processing, many frameworks are available such as Pregel \cite{Pregel2010SIGMOD}, GraphLab \cite{GraphLab2012VLDB}, GraphX \cite{GraphX2014OSDI}, and PowerGraph \cite{PowerGraph2012OSDI}. In those works, data is persisted in semi-structural storages such Google File System, Hadoop Distributed File Systems \cite{GFS2003ACM}, BigTable \cite{BigTable2008TOCS}, or in in-memory storage such as Spark \cite{Spark2012Usenix}. Our work focuses on the no-structure key-value stores and the impact of different consistency models on key-value store performance. Our approach's usefulness is also not limited to graph applications.


\subsection{Rollback Mechanism}
\label{subsec:rollback-mechanism}

While inconsistency is possible with eventual consistency, it is rare \cite{DHJKLPSVV07SOSP} given that networks are reliable and data conflicts are infrequent. However, such inconsistencies and data conflicts can arise and, hence, one needs to deal with these conflicts if we are using an application that relies on sequential consistency. We discuss the rollback approaches for such scenarios. 



One possible approach for rollback, especially if violations can be detected quickly is as follows: 
We partition the work assigned to each client in terms of several tasks. Each task consists of two phases (cf. Figure \ref{fig:detection-latency}): (1) \textit{Read} phase: the client obtains all necessary locks for all nodes in the task, reading the necessary data, and identify the values that need to be changed. However, all updates in this phase are done in local memory. (2) \textit{Write} phase: the client writes the data that they are expected to change and reflect it in the data store. 

In such a system, a violation could occur if clients C1 and C2 are accessing the same data simultaneously. For sake of discussion, suppose that client C1 started accessing the data before C2. Now, if the detection of violation is quick then detection would occur before client C2 enters the write phase. In this case, client C2 has not performed any changes to the key-value store. In other words, client C2 can re-start its task (that involves reading the data from the key-value store) to recover from the violation. 

With this intuition, we can provide recovery as follows: When a violation is detected, if the client causing the violation is in the read phase, it aborts that task and starts that task again. On the other hand, if a client is in write phase (and this can happen to at most one task if detection is quick enough) then it continues its task normally. Note that with this approach, it is possible that two clients that result in a violation are both in the read phase. While one of the clients could be allowed to continue normally, this requires clients to know the status of other clients. We do not consider this option as it is expected that in most applications clients do not communicate directly. Rather, they communicate only via the key-value store. We utilize this approach in our rollback mechanism. In particular, when detection is quick, we use the Algorithm \ref{algo:rollback} for rollback (cf. Figure \ref{fig:detection-latency} and Algorithm \ref{algo:rollback}). 


\begin{figure}[tbp
\centering
[width=0.45\textwidth]{DetectionLatency.png}
\caption{Two client tasks involved in a violation. Since detection latency is much smaller than the \textit{Read} phase time, violation will be notified within \textit{Read} phase of the current task of at least one client.}
\label{fig:detection-latency}
\end{figure}


\iffalse
\begin{itemize}
\item If the client has not issued PUT requests in the current task, it will ignore the values read from the key-value store and restart the task.


\item If the client has issued some PUT requests, it will continue to finish all PUT requests and release the locks it holds. Thus, the current task is completed as normal. We note that if the violation is detected quickly then at most one client will continue along this path, thereby satisfying mutual exclusion requirement.
\end{itemize}
\fi 

Other approaches for rollback are as follows:

\begin{itemize}


    \item \textbf{Rollback via Retroscope\cite{CADK17ICDCS}.  } The most general approach is to utilize an algorithm such as RetroScope \cite{CADK17ICDCS}. Specifically, it allows one to rollback the state of the key-value store to an earlier state. The time, $t$, of rollback is chosen in such a way that there are no violations before time $t$. Upon such a rollback, we can determine the phases the clients are in at time $t$. If a client is in the read phase at time $t$, it will abort its current task and begin it again. And, if the client is in a write phase, it will finish that phase. Note that since there are no conflicts until time $t$, such write phases will not result in conflicts. 
    
    While this approach is most general, it is also potentially expensive. Hence, some alternate approaches are as follows:
    
    \item
    \textbf{Use of Self-Stabilizing Algorithms. }
    One possibility is if we are using a self-stabilizing algorithm. An algorithm is self-stabilizing if it is guaranteed to recover to a legitimate state even in the presence of arbitrary state perturbation. In \cite{NKD2019ICDCN}, it is shown that if the underlying algorithm is self-stabilizing then we can simply ignore the violations as we can treat it as a state perturbation and the algorithm is already designed to handle it. In this case, there is neither a need for monitoring or rollback.  
    
    \item
    \textbf{Use of Application-Specific Rollbacks. }
    Another possibility is application specific rollback. To illustrate this, consider an example of graph coloring. For sake of illustration, consider that we have three nodes A, B, C, arranged in a line with node B in the middle. Each node may have additional neighbors as well.  Node A chooses its color based on the colors of its neighbors. Subsequently, node B chooses its color based on node A (and other neighbors of B). Afterward, C chooses its color based on B (and other neighbors of C). At this point, node B is required to rollback, it can still choose its color based on the new color of node C while still satisfying the constraints of graph coloring. In other words, in this application, we do not need to worry about cascading rollback. 
    
\end{itemize} 

\begin{algorithm}[t
\caption{Rollback algorithm at a client}\label{algo:rollback}
\begin{algorithmic}[1]
\For{taskId = clientFirstTask to clientLastTask}
    \While{(performTask(taskId) == False)}
    \EndWhile
\EndFor
\\
\Function{performTask}{$taskId$}
    \State Obtain relevant locks
    \State Read information from data-store
    \State Compute new values
    \If{Violation is received}
        \State Release locks
        \State \Return False
        \Comment{abort}
    \EndIf
    \State Write new values to data-store
    \State \Return True
    \Comment{success}
\EndFunction
\end{algorithmic}
\end{algorithm}



\begin{comment}


----


Since nodes have different connectivity, the processing time for each task varies. In our experiments, the minimum, average, and maximum processing time for tasks of size 10 are $22,645\ ms$, $45,136\ ms$, and $217,369\ ms$, respectively.



Furthermore, in some graph processing applications such as \textit{Social Media Analysis}, recovery could be achieved efficiently as follows.
The clients process graph nodes in tasks. Each task contains some configurable number of nodes.
The computation of a task at a client is divided into two phases (cf. Figure \ref{fig:detection-latency}): (1) \textit{Read} phase: the client obtains all necessary locks for all nodes in the task, reading their neighbors' colors, and compute the colors of nodes in the task, (2) \textit{Write} phase: the client writes the colors of nodes in the task onto the data store. Since nodes have different connectivity, the processing time for each task varies. In our experiments, the minimum, average, and maximum processing time for tasks of size 10 are $22,645\ ms$, $45,136\ ms$, and $217,369\ ms$, respectively. 
We also observe that the \textit{Read} phase constitutes the majority of task time.
Since the \textit{Read} phase time of a task is far beyond the detection latency (less than $1 \ s$ in general), if a violation occurs, clients will be notified by the monitors within the \textit{Read} phase of the current task in at least one client.

When a client receives a violation notification, it will determine whether it was involved in the violation or not (whether the client ever obtained the lock reported in the violation notification). If the client was not involved, it ignores the notification and continues its task. If the client was involved in the violation, there are two possibilities:
\begin{itemize}
\item If the client has not issued PUT requests to update colors of nodes in the task yet, it will skip those updates and releases all the locks it is currently holding. Thus, the current task is aborted without the need for state rollback since no PUT request has been committed. 

\item If the client has issued some PUT requests, it will continue to finish all PUT requests and release the locks it holds. Thus, the current task is completed as normal. We note that if the violation is detected quickly then at most one client will continue along this path, thereby satisfying mutual exclusion requirement.
\end{itemize}


\LARGE
An observation on rollback for distributed graph coloring:
\normalsize
\begin{itemize}
    \item The coloring constraint is local, i.e. only within an edge (the two nodes of the edge have different colors)
    \item When a ME violation is detected on an edge, that edge needs rollback but the adjacent edges do not need rollback as long as the adjacent edges have no ME violation, even if the coloring of adjacent edges occur later.
    \item If this argument is correct, then the rollback for graph coloring is greatly simplified. We do not need actual rollback like Retroscope for graph coloring. We just need to re-do any task that is detected to have ME violations.
    
    For example, suppose the client has finished task-1, task-2 and is working on task-3. Then it receives a violation report for task-1. It does not have to undo task-2 and task-3. It just needs to finish task-3 and then redo task-1.
    \item This observation is like ``local idempotent'', i.e. we can repeat doing a task without affecting the results of other tasks, while we can correct the result of the current task.
    \item This characteristic is not applicable to distributed computations where the results are linked/chained together such as computing shortest path, semi-clustering, etc. In this case, we could either use self-stabilization algorithms \cite{ICDCNpaper} or use the general rollback framework with Retroscope if self-stabilization algorithms are not available.
\end{itemize}
\normalsize

Rollback text here. We discuss different choices. And, say that we are going to work with the one we are using. 
\end{comment}

\subsection{Dealing with Potential of Livelocks}


One potential issue with rollback is a possibility of livelocks. Specifically, if two clients C1 and C2 rollback and continue their execution then the same violation is likely to happen again. We consider the following choices for dealing with such livelocks. 

\begin{itemize}
    \item \textbf{Random Backoff. } Upon rollback, clients perform a random backoff. 
   
    With backoff, the requests for locks from clients arrive at different times in the key-value store. Hence, the second client is likely to observe locks obtained by the first client in a consistent manner. In turn, this will reduce the possibility of the same violation to recur. 
    \item \textbf{Reordering of Tasks. } If the work assigned to clients consists of several independent tasks, then clients can reorder the tasks upon detecting a violation. In this case, the clients involved in the rollback are likely to access different data and, hence, the possibility of another violation is reduced. 
    \item \textbf{Moving to Sequential Consistency. }
    If the number of violations is beyond a certain threshold, clients may conclude that the cost of rollback is too high and, hence, they can move to sequential consistency. While this causes one to lose the benefits of an eventual consistent key-value store, there would be no need for rollback or monitoring. 
\end{itemize}



\subsection{Distributed Key-Value Store}

We utilize the standard architecture for key-value stores. Specifically, the data consists of (one or more) tables with two fields, a unique key and the corresponding value. 
The field value consists of a list of $<version, value>$ pairs. A version is a vector clock that describes the origin of the associated value. It is possible that a key has multiple versions when different clients issue PUT (write) requests for that key independently.
When a client issues a GET (read) request for a key, all existing versions of that key will be returned. The client could resolve multiple versions for the same key on its own or use the resolver function provided from the library.
To provide efficient access to this table, it is divided into multiple partitions. Furthermore, to provide redundancy and ease of access, the table is replicated across multiple replicas.

To access the entries in this table, the client utilizes two operations, GET and PUT. The operation GET($x$) provides the client with the value (or values if multiple versions exist) associated with key $x$.  The operation PUT($x, val$) changes the value associated with key $x$ to $val$. The state of the servers can be changed only by PUT requests from clients.

\subsection{Voldemort Key Store}

Voldemort is LinkedIn's open source equivalence of Amazon's Dynamo key-value store. In Voldemort, clients are responsible for handling replication. When connecting to a server for the first time, a client receives meta-data from the server. The meta-data contains the list of servers and their addresses, the replication factor ($N$), required reads ($R$), required writes ($W$), and other configuration information.

When a client wants to perform a PUT (or GET) operation, it sends PUT (GET) requests to $N$ servers and waits for the responses for a predefined amount of time (timeout). If at least $W$ ($R$) acknowledgments (responses) are received before the timeout, the PUT (GET) operation is considered successful. If not, the client performs one more round of requests to other servers to get the necessary number of acknowledgments (responses). After the second round, if still less than $W$ ($R$) replies are received, the PUT (GET) operation is considered unsuccessful.

Since the clients do the task of replication, the values $N$, $R$, $W$ specified in the meta-data is only a suggestion. The clients can change those values for their needs.
By adjusting the value of $W$, $R$, and $N$, the client can tune the consistency model. For example, if $W + R > N$ and $W > \frac{N}{2}$ for every client, then they run on sequential consistency. On the other hand, if $W + R \le N$ then they have eventual consistency.

\begin{comment}
Describe the overall architecture of the distributed key-value store ({\color{blue} I am not sure whether we need this part since this content may be considered basics by some readers}). A brief overview of Voldemort, and it is tunable between different consistency model.
\begin{itemize}
\item The distributed system model we are interested in here is the distributed data store. It consists of several servers and much more number of clients. The state of the systems is stored at the servers. The state of the systems can only be changed by PUT/WRITE requests from the clients.
\item Each data item will be replicated at several servers to provide safe data storage. Whether different replicas (versions) of the same data item stored at different servers are the same or not is determined by the consistency model. In sequential consistency model, every data item must be the same at different servers. In eventual consistency model, a data item could be temporarily different at the servers, and the conflicts would be resolved eventually either by the clients or automatically by the servers.
\item In Voldemort data store, the consistency model is configured via the parameters: 

\begin{itemize}
\item {\texttt{replication-factor}} specifies the number of copies of any data item that must be
stored at different Voldemort servers. This allows Voldemort to tolerate replication-factor
- 1 failures.
\item {\texttt{required-writes}} specifies the minimum number of confirmations from different servers
so that a write request can successfully commit.
\item {\texttt{required-reads}} specifies the minimum number of responses from different servers so
that a read request can complete.
\item If required-reads + required-writes > replication-factor then the Voldemort
store instance is using sequential consistency model. Otherwise, eventual consistency model is
being used
\end{itemize}

\end{itemize}

\end{comment}

\begin{comment}
\subsection{Hybrid Vector Clock}
The difference between HVC and VC (refer to Prof. Kulkarni's paper for this content)

Advantages of HVC: increase the efficiency and scalability, and avoid false positives.
\end{comment}