Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加 kvstart kvend kvindex 转换方法，转换成kvbuffer对应的坐标；方便理解

This commit is contained in:
zhaoshengwolf 2022-12-01 07:10:54 +00:00
parent a48e8c9beb
commit cccd504557
1 changed files with 1980 additions and 1970 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/MapTask.java
View File

@ -1106,8 +1106,8 @@ public class MapTask extends Task {
try {
do {
if (!spillInProgress) {
final int kvbidx = 4 * kvindex;
final int kvbend = 4 * kvend;
final int kvbidx = convertKvmetaIndexToKvbufferIndex(kvindex);
final int kvbend = convertKvmetaIndexToKvbufferIndex(kvend);
// serialized, unspilled bytes always lie between kvindex and
// bufindex, crossing the equator. Note that any void space
// created by a reset must be included in "used" bytes
@ -1138,7 +1138,7 @@ public class MapTask extends Task {
% kvbuffer.length;
setEquator(newPos);
bufmark = bufindex = newPos;
final int serBound = 4 * kvend;
final int serBound = convertKvmetaIndexToKvbufferIndex(kvend);
// bytes remaining before the lock must be held and limits
// checked is the minimum of three arcs: the metadata space, the
// serialization space, and the soft limit
@ -1218,7 +1218,7 @@ public class MapTask extends Task {
kvindex = (int)
(((long)aligned - METASIZE + kvbuffer.length) % kvbuffer.length) / 4;
LOG.info("(EQUATOR) " + pos + " kvi " + kvindex +
"(" + (kvindex * 4) + ")");
"(" + (convertKvmetaIndexToKvbufferIndex(kvindex)) + ")");
}
/**
@ -1235,7 +1235,7 @@ public class MapTask extends Task {
kvstart = kvend = (int)
(((long)aligned - METASIZE + kvbuffer.length) % kvbuffer.length) / 4;
LOG.info("(RESET) equator " + e + " kv " + kvstart + "(" +
(kvstart * 4) + ")" + " kvi " + kvindex + "(" + (kvindex * 4) + ")");
(convertKvmetaIndexToKvbufferIndex(kvstart)) + ")" + " kvi " + kvindex + "(" + (convertKvmetaIndexToKvbufferIndex(kvindex)) + ")");
}
/**
@ -1337,8 +1337,8 @@ public class MapTask extends Task {
// spillLock unnecessary; both kvend and kvindex are current
int headbytelen = bufvoid - bufmark;
bufvoid = bufmark;
final int kvbidx = 4 * kvindex;
final int kvbend = 4 * kvend;
final int kvbidx = convertKvmetaIndexToKvbufferIndex(kvindex);
final int kvbend = convertKvmetaIndexToKvbufferIndex(kvend);
final int avail =
Math.min(distanceTo(0, kvbidx), distanceTo(0, kvbend));
if (bufindex + headbytelen < avail) {
@ -1388,8 +1388,8 @@ public class MapTask extends Task {
do {
checkSpillException();
final int kvbidx = 4 * kvindex;
final int kvbend = 4 * kvend;
final int kvbidx = convertKvmetaIndexToKvbufferIndex(kvindex);
final int kvbend = convertKvmetaIndexToKvbufferIndex(kvend);
// ser distance to key index
final int distkvi = distanceTo(bufindex, kvbidx);
// ser distance to spill end index
@ -1487,7 +1487,7 @@ public class MapTask extends Task {
}
checkSpillException();
final int kvbend = 4 * kvend;
final int kvbend = convertKvmetaIndexToKvbufferIndex(kvend);
if ((kvbend + METASIZE) % kvbuffer.length !=
equator - (equator % METASIZE)) {
// spill finished
@ -1499,8 +1499,8 @@ public class MapTask extends Task {
LOG.info("Spilling map output");
LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +
"; bufvoid = " + bufvoid);
LOG.info("kvstart = " + kvstart + "(" + (kvstart * 4) +
"); kvend = " + kvend + "(" + (kvend * 4) +
LOG.info("kvstart = " + kvstart + "(" + (convertKvmetaIndexToKvbufferIndex(kvstart)) +
"); kvend = " + kvend + "(" + (convertKvmetaIndexToKvbufferIndex(kvend)) +
"); length = " + (distanceTo(kvend, kvstart,
kvmeta.capacity()) + 1) + "/" + maxRec);
sortAndSpill();
@ -1596,8 +1596,8 @@ public class MapTask extends Task {
LOG.info("Spilling map output");
LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +
"; bufvoid = " + bufvoid);
LOG.info("kvstart = " + kvstart + "(" + (kvstart * 4) +
"); kvend = " + kvend + "(" + (kvend * 4) +
LOG.info("kvstart = " + kvstart + "(" + (convertKvmetaIndexToKvbufferIndex(kvstart)) +
"); kvend = " + kvend + "(" + (convertKvmetaIndexToKvbufferIndex(kvend)) +
"); length = " + (distanceTo(kvend, kvstart,
kvmeta.capacity()) + 1) + "/" + maxRec);
spillReady.signal();
@ -2094,4 +2094,14 @@ public class MapTask extends Task {
}
}
}
/**
* 将kvmeta对应的 kvstart kvend kvindex 转换成对应kvbuffer的index，便于理解
* 1 int = 4 byte
* @param kvindex
* @return
*/
private static int convertKvmetaIndexToKvbufferIndex(int kvindex) {
return kvindex * 4;
}
}