通过本文可以获得如下知识：
- StripeReader#readStripe源码逻辑。
- readDataForDecoding、readParityChunks方法源码详细分析。
- prepareDecodeInputs、prepareParityChunk方法源码详细分析。
- readChunk、getNextCompletedStripedRead方法源码详细分析。
- decode方法简要逻辑介绍
一、背景
读EC文件的时候，假设我们使用RS(k，m)的算法，则最多允许m个data cell的读取失败，然后可以通过k-m个数据块和m个校验块对丢失的（missing）data cell进行解码还原。
本文就是介绍EC读流程中，如果遇到数据块missing的情况下的EC解码（decode）过程。主要就是对StripeReader#readStripe方法及其内部调用的关键方法进行源码解析。
二、StripeReader#readStripe整体流程
为什么要从这个readStripe方法开始呢? 因为在StripedInputStream的read的方法内部，有两个用来读stripe的方法：
DFSStripedInputStream#fetchBlockByteRange和DFSStripedInputStream#readOneStripe，内部就是使用readStripe方法来对条带数据进行读取。
StripeReader#readStripe方法的主要功能：
/**
* read the whole stripe. do decoding if necessary。
* 翻译：读取整个条带、如果有需要的话需要做解码操作（data chunk miss掉的情况，需要读parity校验块，然后解码出miss掉的数据块）。
**/
readStripe方法代码如下，这里我先把所有代码粘贴出来并给出大概的注释，后面我们会对readStripe方法按步骤进行拆解并详细注释。
/**
* read the whole stripe. do decoding if necessary
*/
void readStripe() throws IOException {
// 第一部分：读数据块，如果读失败了则把missingChunksNum + 1，记录一下missing的块数
for (int i = 0; i < dataBlkNum; i++) {
if (alignedStripe.chunks[i] != null &&
alignedStripe.chunks[i].state != StripingChunk.ALLZERO) {
if (!readChunk(targetBlocks[i], i)) {
alignedStripe.missingChunksNum++;
}
}
}
// There are missing block locations at this stage. Thus we need to read
// the full stripe and one more parity block.
// 第二部分：如果missingChunk的个数大于0，则我们就需要读整个条带和对应个数的校验块。
if (alignedStripe.missingChunksNum > 0) {
checkMissingBlocks();
// 构造decodeInputs这个变量的data chunk部分。
// 从不同datanode上读取条带中的data chunk数据。
readDataForDecoding();
// read parity chunks
// 构造decodeInputs这个变量的parity chunk部分。
// 从不同datanode上读取条件中的parity chunk数据
readParityChunks(alignedStripe.missingChunksNum);
}
// TODO: for a full stripe we can start reading (dataBlkNum + 1) chunks
// Input buffers for potential decode operation, which remains null until
// first read failure
// 在上面readChunk时或者readParityChunks里，会创建一些异步的读任务加到futures这个Map里，这里就是去异步的获取每一个读请求的状态的。
while (!futures.isEmpty()) {
try {
// 获取下一个已经完成的Striped Read任务结果。
// 这个方法内部会去调用Java的CompletionService框架的take和get方法，如果遇到读任务异常，会catch住ExecutionException异常，然后返回的读结果状态是FAILED
StripingChunkReadResult r = StripedBlockUtil
.getNextCompletedStripedRead(service, futures, 0);
dfsStripedInputStream.updateReadStats(r.getReadStats());
if (DFSClient.LOG.isDebugEnabled()) {
DFSClient.LOG.debug("Read task returned: " + r + ", for stripe "
+ alignedStripe);
}
StripingChunk returnedChunk = alignedStripe.chunks[r.index];
Preconditions.checkNotNull(returnedChunk);
Preconditions.checkState(returnedChunk.state == StripingChunk.PENDING);
if (r.state == StripingChunkReadResult.SUCCESSFUL) {
returnedChunk.state = StripingChunk.FETCHED;
alignedStripe.fetchedChunksNum++;
updateState4SuccessRead(r);
// fetch到的chunk数等于数据块数，证明读取未发生异常。
if (alignedStripe.fetchedChunksNum == dataBlkNum) {
clearFutures();
break;
}
} else {
// 如果读请求发生了异常，返回的读结果状态就是FAILED，就会走到这个else里
returnedChunk.state = StripingChunk.MISSING;
// close the corresponding reader
dfsStripedInputStream.closeReader(readerInfos[r.index]);
final int missing = alignedStripe.missingChunksNum;
alignedStripe.missingChunksNum++;
checkMissingBlocks();
// 有异常的时候，需要读数据块进行解码。准备解码所需的decodeInput数组的数据块部分。
readDataForDecoding();
// 有异常的时候，需要读parity块进行解码。准备解码所需的decodeInput数组的parity块部分。
readParityChunks(alignedStripe.missingChunksNum - missing);
}
} catch (InterruptedException ie) {
String err = "Read request interrupted";
DFSClient.LOG.error(err);
clearFutures();
// Don't decode if read interrupted
throw new InterruptedIOException(err);
}
}
if (alignedStripe.missingChunksNum > 0) {
decode();
}
}
接下来拆解readStripe方法：
网友评论