共计 9191 个字符，预计需要花费 23 分钟才能阅读完成。
序
本文主要研究一下 hazelcast 的 PhiAccrualFailureDetector
FailureDetector
hazelcast-3.12-sources.jar!/com/hazelcast/internal/cluster/fd/FailureDetector.java
/**
* Failure detector tracks heartbeats of a member and decides liveness/availability of the member.
*/
public interface FailureDetector {
/**
* Notifies this failure detector about received heartbeat message from the tracked member.
*
* @param timestamp timestamp of heartbeat message in milliseconds
*/
void heartbeat(long timestamp);
/**
* Returns true if the tracked member is considered as alive/available.
* @param timestamp timestamp in milliseconds
* @return true if the member is alive
*/
boolean isAlive(long timestamp);
/**
* Returns the last heartbeat timestamp for the tracked member.
* @return heartbeat timestamp in milliseconds
*/
long lastHeartbeat();
/**
* Returns suspicion level about the tracked member. Returned value is mostly implementation dependent.
* <code>0</code> indicates no suspicion at all.
* @param timestamp timestamp in milliseconds
* @return suspicion level
*/
double suspicionLevel(long timestamp);
}- FailureDetector 接口定义了 heartbeat、isAlive、lastHeartbeat、suspicionLevel 方法
PhiAccrualFailureDetector
hazelcast-3.12-sources.jar!/com/hazelcast/internal/cluster/fd/PhiAccrualFailureDetector.java
/**
* Port of Akka's PhiAccrualFailureDetector.scala
* <p>
* Implementation of 'The Phi Accrual Failure Detector' by Hayashibara et al. as defined in their paper.
* <p>
* The suspicion level of failure is given by a value called φ (phi).
* The basic idea of the φ failure detector is to express the value of φ on a scale that
* is dynamically adjusted to reflect current network conditions. A configurable
* threshold is used to decide if <code>φ</code> is considered to be a failure.
* <p>
* The value of <code>φ</code> is calculated as:
* <p>
* <code>
* <pre>
* φ = -log10(1 - F(timeSinceLastHeartbeat)
* </pre>
* </code>
* where F is the cumulative distribution function of a normal distribution with mean
* and standard deviation estimated from historical heartbeat inter-arrival times.
*/
public class PhiAccrualFailureDetector implements FailureDetector {
static final long NO_HEARTBEAT_TIMESTAMP = -1;
private final double threshold;
private final double minStdDeviationMillis;
private final long acceptableHeartbeatPauseMillis;
private final HeartbeatHistory heartbeatHistory;
private volatile long lastHeartbeatMillis = NO_HEARTBEAT_TIMESTAMP;
/**
* @param threshold A low threshold is prone to generate many wrong suspicions but ensures
* a quick detection in the event of a real crash. Conversely, a high threshold
* generates fewer mistakes but needs more time to detect actual crashes
* @param maxSampleSize Number of samples to use for calculation of mean and standard deviation of
* inter-arrival times.
* @param minStdDeviationMillis Minimum standard deviation to use for the normal distribution used when
* calculating phi. Too low standard deviation might result in too much sensitivity
* for sudden, but normal, deviations in heartbeat inter arrival times.
* @param acceptableHeartbeatPauseMillis Duration corresponding to number of potentially lost/delayed
* heartbeats that will be accepted before considering it to be an anomaly.
* This margin is important to be able to survive sudden, occasional, pauses
* in heartbeat arrivals, due to for example garbage collect or network drop.
* @param firstHeartbeatEstimateMillis Bootstrap the stats with heartbeats that corresponds to this duration,
* with a with rather high standard deviation (since environment is unknown
* in the beginning)
*/
public PhiAccrualFailureDetector(double threshold, int maxSampleSize, double minStdDeviationMillis,
long acceptableHeartbeatPauseMillis, long firstHeartbeatEstimateMillis) {this.threshold = checkPositive(threshold, "Threshold must be positive:" + threshold);
this.minStdDeviationMillis = checkPositive(minStdDeviationMillis, "Minimum standard deviation must be positive:"
+ minStdDeviationMillis);
this.acceptableHeartbeatPauseMillis = checkNotNegative(acceptableHeartbeatPauseMillis,
"Acceptable heartbeat pause millis must be >= 0:" + acceptableHeartbeatPauseMillis);
checkPositive(firstHeartbeatEstimateMillis, "First heartbeat value must be > 0:" + firstHeartbeatEstimateMillis);
heartbeatHistory = new HeartbeatHistory(maxSampleSize);
firstHeartbeat(firstHeartbeatEstimateMillis);
}
// guess statistics for first heartbeat,
// important so that connections with only one heartbeat becomes unavailable
// bootstrap with 2 entries with rather high standard deviation
@SuppressWarnings("checkstyle:magicnumber")
private void firstHeartbeat(long firstHeartbeatEstimateMillis) {
long stdDeviationMillis = firstHeartbeatEstimateMillis / 4;
heartbeatHistory.add(firstHeartbeatEstimateMillis - stdDeviationMillis);
heartbeatHistory.add(firstHeartbeatEstimateMillis + stdDeviationMillis);
}
private double ensureValidStdDeviation(double stdDeviationMillis) {return Math.max(stdDeviationMillis, minStdDeviationMillis);
}
/**
* The suspicion level of the accrual failure detector.
*
* If a connection does not have any records in failure detector then it is
* considered healthy.
*/
private double phi(long timestampMillis) {
long timeDiffMillis;
double meanMillis;
double stdDeviationMillis;
synchronized (heartbeatHistory) {
long lastTimestampMillis = lastHeartbeatMillis;
if (lastTimestampMillis == NO_HEARTBEAT_TIMESTAMP) {return 0.0;}
timeDiffMillis = timestampMillis - lastTimestampMillis;
meanMillis = heartbeatHistory.mean();
stdDeviationMillis = ensureValidStdDeviation(heartbeatHistory.stdDeviation());
}
return phi(timeDiffMillis, meanMillis + acceptableHeartbeatPauseMillis, stdDeviationMillis);
}
/**
* Calculation of phi, derived from the Cumulative distribution function for
* N(mean, stdDeviation) normal distribution, given by
* 1.0 / (1.0 + math.exp(-y * (1.5976 + 0.070566 * y * y)))
* where y = (x - mean) / standard_deviation
* This is an approximation defined in β Mathematics Handbook (Logistic approximation).
* Error is 0.00014 at +- 3.16
* The calculated value is equivalent to -log10(1 - CDF(y))
*/
@SuppressWarnings("checkstyle:magicnumber")
private static double phi(long timeDiffMillis, double meanMillis, double stdDeviationMillis) {double y = (timeDiffMillis - meanMillis) / stdDeviationMillis;
double e = Math.exp(-y * (1.5976 + 0.070566 * y * y));
if (timeDiffMillis > meanMillis) {return -Math.log10(e / (1.0 + e));
} else {return -Math.log10(1.0 - 1.0 / (1.0 + e));
}
}
@Override
public boolean isAlive(long timestampMillis) {double phi = phi(timestampMillis);
return phi < threshold;
}
@Override
public void heartbeat(long timestampMillis) {synchronized (heartbeatHistory) {long lastTimestampMillis = getAndSetLastHeartbeat(timestampMillis);
if (lastTimestampMillis == NO_HEARTBEAT_TIMESTAMP) {return;}
if (isAlive(timestampMillis)) {heartbeatHistory.add(timestampMillis - lastTimestampMillis);
}
}
}
private long getAndSetLastHeartbeat(long timestampMillis) {
long lastTimestampMillis = lastHeartbeatMillis;
lastHeartbeatMillis = timestampMillis;
return lastTimestampMillis;
}
@Override
public long lastHeartbeat() {return lastHeartbeatMillis;}
@Override
public double suspicionLevel(long timestamp) {return phi(timestamp);
}
/**
* Holds the heartbeat statistics for a specific member.
* It is capped by the number of samples specified in `maxSampleSize`.
*
* The stats (mean, variance, stdDeviation) are not defined for
* for empty HeartbeatHistory, i.e. throws ArithmeticException.
*/
private static class HeartbeatHistory {
private final int maxSampleSize;
private final LinkedList<Long> intervals = new LinkedList<Long>();
private long intervalSum;
private long squaredIntervalSum;
HeartbeatHistory(int maxSampleSize) {if (maxSampleSize < 1) {throw new IllegalArgumentException("Sample size must be >= 1 :" + maxSampleSize);
}
this.maxSampleSize = maxSampleSize;
}
double mean() {return (double) intervalSum / intervals.size();}
double variance() {double mean = mean();
return ((double) squaredIntervalSum / intervals.size()) - (mean * mean);
}
double stdDeviation() {return Math.sqrt(variance());
}
void add(long interval) {if (intervals.size() >= maxSampleSize) {dropOldest();
}
intervals.add(interval);
intervalSum += interval;
squaredIntervalSum += pow2(interval);
}
private void dropOldest() {long dropped = intervals.pollFirst();
intervalSum -= dropped;
squaredIntervalSum -= pow2(dropped);
}
private static long pow2(long x) {return x * x;}
}
}- PhiAccrualFailureDetector 实现了 FailureDetector 接口，其实现是 akka 的 PhiAccrualFailureDetector.scala 的 java 版本
- φ (phi)为指定值被认定为 failure 的 suspicion level，其计算公式为 φ = -log10(1 – CDF(timeSinceLastHeartbeat)，其中 CDF 函数为 normal distribution 的 cumulative distribution function，即正态分布的累积分布函数
- phi 方法使用了 β Mathematics Handbook 中定义的 Logistic approximation 公式来近似计算 CDF(y)(
Error is 0.00014 at +- 3.16)，即 CDF(y)=1.0 / (1.0 + math.exp(-y (1.5976 + 0.070566 y * y)))，其中 y = (x – mean) / standard_deviation
CDF(y)在 x > mean 的时候 (
e<1) 将公式转换为 φ = -Math.log10(e / (1.0 + e))；在 x <= mean 的时候 (e>=1) 的采用公式 -Math.log10(1.0 – 1.0 / (1.0 + e))，目前还不清楚为何这样区分计算
- isAlive 方法会计算该 timestampMillis 的 phi 值，然后与 threshold 值 (
hazelcast 中默认为 10) 判断，小于 threshold 值才判断为 live - heartbeat 方法会先判断该 timestampMillis 是否 live，是的话，再将 timestampMillis – lastTimestampMillis 值添加到 heartbeatHistory
- 这里的实现增加了 acceptableHeartbeatPauseMillis 参数，即在最后计算 phi 值时传入的 meanMillis 为 meanMillis + acceptableHeartbeatPauseMillis
小结
- FailureDetector 接口定义了 heartbeat、isAlive、lastHeartbeat、suspicionLevel 方法；PhiAccrualFailureDetector 实现了 FailureDetector 接口，其实现是 akka 的 PhiAccrualFailureDetector.scala 的 java 版本
- φ (phi)为指定值被认定为 failure 的 suspicion level，其计算公式为 φ = -log10(1 – CDF(timeSinceLastHeartbeat)，其中 CDF 函数为 normal distribution 的 cumulative distribution function，即正态分布的累积分布函数
- akka 的实现中，其 phi 方法使用了 β Mathematics Handbook 中定义的 Logistic approximation 公式来近似计算 CDF(y)(
Error is 0.00014 at +- 3.16)，即 CDF(y)=1.0 / (1.0 + math.exp(-y (1.5976 + 0.070566 y * y)))，其中 y = (x – mean) / standard_deviation；另外也增加了 acceptableHeartbeatPauseMillis 参数，即在最后计算 phi 值时传入的 meanMillis 为 meanMillis + acceptableHeartbeatPauseMillis - isAlive 方法会计算该 timestampMillis 的 phi 值，然后与 threshold 值 (
hazelcast 中默认为 10) 判断，小于 threshold 值才判断为 live - heartbeat 方法会先判断该 timestampMillis 是否 live，是的话，再将 timestampMillis – lastTimestampMillis 值添加到 heartbeatHistory
doc
- PhiAccrualFailureDetector.scala
- PhiAccrualFailureDetector.java
- A logistic approximation to the cumulative normal distribution – Core
正文完
