abstractdog commented on code in PR #530:
URL: https://github.com/apache/tez/pull/530#discussion_r3795449421
##########
tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java:
##########
@@ -189,6 +189,11 @@ public enum TaskCounter {
*/
SHUFFLE_BYTES_DISK_DIRECT,
+ /**
+ * Time spent waiting on network I/O during shuffle. Represented in
milliseconds.
+ */
+ SHUFFLE_IO_TIME_MILLISECONDS,
+
Review Comment:
let me be a bit picky here, and elaborate on the counter: my motivation is
that we have so many, sometimes underdocumented counters, a few ideas:
1. Javadoc to mention this is only used if
"tez.runtime.shuffle.measure.io.time" is enabled, otherwise it's not created at
all, feel free to mention possible overhead here
2. connection to `SHUFFLE_BYTES`: are these used at the very same codepaths
(I guess)? if so, if we count and aggregate the `ret` values in
`MeasureDataInputStream`, would we get the same amount as "SHUFFLE_BYTES"
counter?
3. new idea from 2), whether to introduce SHUFFLE_IO_STREAM_BYTES, which is
the actual streamed bytes? this could be interesting, as "SHUFFLE_BYTES" are
simply aggregated from shuffle headers, so these should be equal theoretically
(might reveal some bug if they are not)
4. if 3) is introduced, consider introducing new names like:
SHUFFLE_IO_STREAM_TIME_MILLISECONDS, SHUFFLE_IO_STREAM_BYTES (or if you can
think of any better), because "IO_TIME" already feels too general
##########
tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java:
##########
@@ -798,4 +799,62 @@ private InputContext createMockInputContext() {
return inputContext;
}
+
+ @Test
+ public void testShuffleMeasureIOTime() throws Exception {
+ Configuration conf = new TezConfiguration();
+
conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEASURE_IO_TIME,
true);
+
+ ShuffleScheduler scheduler = mock(ShuffleScheduler.class);
+ MergeManager merger = mock(MergeManager.class);
+ Shuffle shuffle = mock(Shuffle.class);
+
+ final MapHost host = new MapHost(HOST, PORT, 1, 1);
+ InputContext inputContext = createMockInputContext();
+ FetcherOrderedGrouped mockFetcher =
+ new FetcherOrderedGrouped(null, scheduler, merger, shuffle, null,
false, 0, null, conf, getRawFs(conf), false,
+ HOST, PORT, host, ioErrsCounter, wrongLengthErrsCounter,
badIdErrsCounter, wrongMapErrsCounter,
+ connectionErrsCounter, wrongReduceErrsCounter, false, false, true,
false, inputContext);
+ final FetcherOrderedGrouped fetcher = spy(mockFetcher);
+
+ final List<InputAttemptIdentifier> srcAttempts =
+ List.of(new InputAttemptIdentifier(0, 1,
InputAttemptIdentifier.PATH_PREFIX + "pathComponent_0"));
+ doReturn(srcAttempts).when(scheduler).getMapsForHost(host);
+
+ URL url =
+ ShuffleUtils.constructInputURL("http" + "://" + HOST + ":" + PORT +
"/mapOutput?job=job_123&&reduce=1&map=",
+ srcAttempts, false);
+ fetcher.httpConnection = new FakeHttpConnection(url, null, "", null) {
+ @Override
+ public DataInputStream getInputStream() {
+ ByteArrayInputStream bin = new ByteArrayInputStream(new byte[1024]) {
+ @Override
+ public int read(byte[] b, int off, int len) {
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return super.read(b, off, len);
+ }
+ };
+ return new DataInputStream(bin);
+ }
+ };
+
+ fetcher.setupConnectionInternal(host, srcAttempts);
+
+ // Read some bytes to trigger the elapsed time measurement
+ byte[] buffer = new byte[10];
+ int bytesRead = fetcher.input.read(buffer, 0, buffer.length);
+ assertEquals(10, bytesRead);
+
+ // shutDown will update the counter
+ fetcher.shutDown();
+
+ // Check if io time counter is updated
+ TezCounter ioTimeCounter =
inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_IO_TIME_MILLISECONDS);
+ long ioTime = ioTimeCounter.getValue();
+ assertTrue(ioTime >= 10, "IO Time should be at least 10ms, but was " +
ioTime);
Review Comment:
10ms is a wild guess, even if it's working for 99% of the cases: I would use
`ioTime >= 0`, because it already proves that the IO time was indeed measure
##########
tez-runtime-library/src/main/java/org/apache/tez/runtime/library/api/TezRuntimeConfiguration.java:
##########
@@ -416,6 +416,13 @@ private TezRuntimeConfiguration() {}
public static final float TEZ_RUNTIME_SHUFFLE_FETCH_BUFFER_PERCENT_DEFAULT =
0.90f;
+ /**
+ * Enables measuring network IO time in shuffle fetchers.
Review Comment:
emphasize overhead here, feel free refer Javadoc to `MeasuredDataInputStream`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
