Java 类org.apache.hadoop.hdfs.server.datanode.BPServiceActor.Scheduler 实例源码

/**
 * Tests the case when a block report was delayed past its scheduled time.
 * In that case the next block report should not be delayed for a full interval.
 */
@Test
public void testScheduleNextBlockReport3() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.resetBlockReportTime = false;

    // Make it look like the block report was scheduled to be sent between 1-3
    // intervals ago but sent just now.
    final long blockReportDelay =
        BLOCK_REPORT_INTERVAL_MS + random.nextInt(2 * (int) BLOCK_REPORT_INTERVAL_MS);
    final long origBlockReportTime = now - blockReportDelay;
    scheduler.nextBlockReportTime = origBlockReportTime;
    scheduler.scheduleNextBlockReport();
    assertTrue(scheduler.nextBlockReportTime - now < BLOCK_REPORT_INTERVAL_MS);
    assertTrue(((scheduler.nextBlockReportTime - origBlockReportTime) % BLOCK_REPORT_INTERVAL_MS) == 0);
  }
}

/**
 * Tests the case when a block report was delayed past its scheduled time.
 * In that case the next block report should not be delayed for a full interval.
 */
@Test
public void testScheduleNextBlockReport3() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.resetBlockReportTime = false;

    // Make it look like the block report was scheduled to be sent between 1-3
    // intervals ago but sent just now.
    final long blockReportDelay =
        BLOCK_REPORT_INTERVAL_MS + random.nextInt(2 * (int) BLOCK_REPORT_INTERVAL_MS);
    final long origBlockReportTime = now - blockReportDelay;
    scheduler.nextBlockReportTime = origBlockReportTime;
    scheduler.scheduleNextBlockReport();
    assertTrue(scheduler.nextBlockReportTime - now < BLOCK_REPORT_INTERVAL_MS);
    assertTrue(((scheduler.nextBlockReportTime - origBlockReportTime) % BLOCK_REPORT_INTERVAL_MS) == 0);
  }
}

/**
 * Regression test for HDFS-9305.
 * Delayed processing of a heartbeat can cause a subsequent heartbeat
 * storm.
 */
@Test
public void testScheduleDelayedHeartbeat() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleNextHeartbeat();
    assertFalse(scheduler.isHeartbeatDue(now));

    // Simulate a delayed heartbeat e.g. due to slow processing by NN.
    scheduler.nextHeartbeatTime = now - (HEARTBEAT_INTERVAL_MS * 10);
    scheduler.scheduleNextHeartbeat();

    // Ensure that the next heartbeat is not due immediately.
    assertFalse(scheduler.isHeartbeatDue(now));
  }
}

/**
 * Tests the case when a block report was delayed past its scheduled time.
 * In that case the next block report should not be delayed for a full interval.
 */
@Test
public void testScheduleNextBlockReport3() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.resetBlockReportTime = false;

    // Make it look like the block report was scheduled to be sent between 1-3
    // intervals ago but sent just now.
    final long blockReportDelay =
        BLOCK_REPORT_INTERVAL_MS + random.nextInt(2 * (int) BLOCK_REPORT_INTERVAL_MS);
    final long origBlockReportTime = now - blockReportDelay;
    scheduler.nextBlockReportTime = origBlockReportTime;
    scheduler.scheduleNextBlockReport();
    assertTrue(scheduler.nextBlockReportTime - now < BLOCK_REPORT_INTERVAL_MS);
    assertTrue(((scheduler.nextBlockReportTime - origBlockReportTime) % BLOCK_REPORT_INTERVAL_MS) == 0);
  }
}

@Test
public void testInit() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    assertTrue(scheduler.isHeartbeatDue(now));
    assertTrue(scheduler.isBlockReportDue());
  }
}

@Test
public void testScheduleBlockReportImmediate() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleBlockReport(0);
    assertTrue(scheduler.resetBlockReportTime);
    assertThat(scheduler.nextBlockReportTime, is(now));
  }
}

@Test
public void testScheduleBlockReportDelayed() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    final long delayMs = 10;
    scheduler.scheduleBlockReport(delayMs);
    assertTrue(scheduler.resetBlockReportTime);
    assertTrue(scheduler.nextBlockReportTime - now >= 0);
    assertTrue(scheduler.nextBlockReportTime - (now + delayMs) < 0);
  }
}

/**
 * If resetBlockReportTime is true then the next block report must be scheduled
 * in the range [now, now + BLOCK_REPORT_INTERVAL_SEC).
 */
@Test
public void testScheduleNextBlockReport() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    assertTrue(scheduler.resetBlockReportTime);
    scheduler.scheduleNextBlockReport();
    assertTrue(scheduler.nextBlockReportTime - (now + BLOCK_REPORT_INTERVAL_MS) < 0);
  }
}

/**
 * If resetBlockReportTime is false then the next block report must be scheduled
 * exactly at (now + BLOCK_REPORT_INTERVAL_SEC).
 */
@Test
public void testScheduleNextBlockReport2() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.resetBlockReportTime = false;
    scheduler.scheduleNextBlockReport();
    assertThat(scheduler.nextBlockReportTime, is(now + BLOCK_REPORT_INTERVAL_MS));
  }
}

@Test
public void testScheduleHeartbeat() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleNextHeartbeat();
    assertFalse(scheduler.isHeartbeatDue(now));
    scheduler.scheduleHeartbeat();
    assertTrue(scheduler.isHeartbeatDue(now));
  }
}

private Scheduler makeMockScheduler(long now) {
  LOG.info("Using now = " + now);
  Scheduler mockScheduler = spy(new Scheduler(HEARTBEAT_INTERVAL_MS, BLOCK_REPORT_INTERVAL_MS));
  doReturn(now).when(mockScheduler).monotonicNow();
  mockScheduler.nextBlockReportTime = now;
  mockScheduler.nextHeartbeatTime = now;
  return mockScheduler;
}

@Test
public void testInit() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    assertTrue(scheduler.isHeartbeatDue(now));
    assertTrue(scheduler.isBlockReportDue(scheduler.monotonicNow()));
  }
}

@Test
public void testScheduleBlockReportImmediate() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleBlockReport(0);
    assertTrue(scheduler.resetBlockReportTime);
    assertThat(scheduler.nextBlockReportTime, is(now));
  }
}

@Test
public void testScheduleBlockReportDelayed() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    final long delayMs = 10;
    scheduler.scheduleBlockReport(delayMs);
    assertTrue(scheduler.resetBlockReportTime);
    assertTrue(scheduler.nextBlockReportTime - now >= 0);
    assertTrue(scheduler.nextBlockReportTime - (now + delayMs) < 0);
  }
}

/**
 * If resetBlockReportTime is true then the next block report must be scheduled
 * in the range [now, now + BLOCK_REPORT_INTERVAL_SEC).
 */
@Test
public void testScheduleNextBlockReport() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    assertTrue(scheduler.resetBlockReportTime);
    scheduler.scheduleNextBlockReport();
    assertTrue(scheduler.nextBlockReportTime - (now + BLOCK_REPORT_INTERVAL_MS) < 0);
  }
}

/**
 * If resetBlockReportTime is false then the next block report must be scheduled
 * exactly at (now + BLOCK_REPORT_INTERVAL_SEC).
 */
@Test
public void testScheduleNextBlockReport2() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.resetBlockReportTime = false;
    scheduler.scheduleNextBlockReport();
    assertThat(scheduler.nextBlockReportTime, is(now + BLOCK_REPORT_INTERVAL_MS));
  }
}

@Test
public void testScheduleHeartbeat() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleNextHeartbeat();
    assertFalse(scheduler.isHeartbeatDue(now));
    scheduler.scheduleHeartbeat();
    assertTrue(scheduler.isHeartbeatDue(now));
  }
}

private Scheduler makeMockScheduler(long now) {
  LOG.info("Using now = " + now);
  Scheduler mockScheduler = spy(new Scheduler(HEARTBEAT_INTERVAL_MS, BLOCK_REPORT_INTERVAL_MS));
  doReturn(now).when(mockScheduler).monotonicNow();
  mockScheduler.nextBlockReportTime = now;
  mockScheduler.nextHeartbeatTime = now;
  return mockScheduler;
}

@Test
public void testInit() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    assertTrue(scheduler.isHeartbeatDue(now));
    assertTrue(scheduler.isBlockReportDue());
  }
}

@Test
public void testScheduleBlockReportImmediate() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleBlockReport(0);
    assertTrue(scheduler.resetBlockReportTime);
    assertThat(scheduler.nextBlockReportTime, is(now));
  }
}

@Test
public void testScheduleBlockReportDelayed() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    final long delayMs = 10;
    scheduler.scheduleBlockReport(delayMs);
    assertTrue(scheduler.resetBlockReportTime);
    assertTrue(scheduler.nextBlockReportTime - now >= 0);
    assertTrue(scheduler.nextBlockReportTime - (now + delayMs) < 0);
  }
}

/**
 * If resetBlockReportTime is true then the next block report must be scheduled
 * in the range [now, now + BLOCK_REPORT_INTERVAL_SEC).
 */
@Test
public void testScheduleNextBlockReport() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    assertTrue(scheduler.resetBlockReportTime);
    scheduler.scheduleNextBlockReport();
    assertTrue(scheduler.nextBlockReportTime - (now + BLOCK_REPORT_INTERVAL_MS) < 0);
  }
}

/**
 * If resetBlockReportTime is false then the next block report must be scheduled
 * exactly at (now + BLOCK_REPORT_INTERVAL_SEC).
 */
@Test
public void testScheduleNextBlockReport2() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.resetBlockReportTime = false;
    scheduler.scheduleNextBlockReport();
    assertThat(scheduler.nextBlockReportTime, is(now + BLOCK_REPORT_INTERVAL_MS));
  }
}

@Test
public void testScheduleHeartbeat() {
  for (final long now : getTimestamps()) {
    Scheduler scheduler = makeMockScheduler(now);
    scheduler.scheduleNextHeartbeat();
    assertFalse(scheduler.isHeartbeatDue(now));
    scheduler.scheduleHeartbeat();
    assertTrue(scheduler.isHeartbeatDue(now));
  }
}

private Scheduler makeMockScheduler(long now) {
  LOG.info("Using now = " + now);
  Scheduler mockScheduler = spy(new Scheduler(HEARTBEAT_INTERVAL_MS, BLOCK_REPORT_INTERVAL_MS));
  doReturn(now).when(mockScheduler).monotonicNow();
  mockScheduler.nextBlockReportTime = now;
  mockScheduler.nextHeartbeatTime = now;
  return mockScheduler;
}