Function get_num_cache_requests

tests/query_test/test_hdfs_caching.py:329–369

Returns the number of outstanding cache requests. Due to race conditions in the way cache requests are added/dropped/reported (see IMPALA-3040), this function tries to return a stable result by making several attempts to stabilize it within a reasonable timeout.

()

def get_num_cache_requests():
  """Returns the number of outstanding cache requests. Due to race conditions in the
    way cache requests are added/dropped/reported (see IMPALA-3040), this function tries
    to return a stable result by making several attempts to stabilize it within a
    reasonable timeout."""
  def get_num_cache_requests_util():
    rc, stdout, stderr = exec_process("hdfs cacheadmin -listDirectives -stats")
    assert rc == 0, 'Error executing hdfs cacheadmin: %s %s' % (stdout, stderr)
    # remove blank new lines from output count
    lines = [line for line in stdout.split('\n') if line.strip()]
    count = None
    for line in lines:
      if line.startswith("Found "):
        # the line should say "Found <int> entries"
        # if we find this line we parse the number of entries
        # from this line.
        count = int(re.search(r'\d+', line).group())
        break
    # if count is available we return it else we just
    # return the total number of lines
    if count is not None:
      return count
    else:
      return len(stdout.split('\n'))

  # IMPALA-3040: This can take time, especially under slow builds like ASAN.
  wait_time_in_sec = build_flavor_timeout(5, slow_build_timeout=20)
  num_stabilization_attempts = 0
  max_num_stabilization_attempts = 10
  num_requests = None
  LOG.info("{0} Entered get_num_cache_requests()".format(time.time()))
  while num_stabilization_attempts < max_num_stabilization_attempts:
    new_requests = get_num_cache_requests_util()
    if new_requests == num_requests: break
    LOG.info("{0} Waiting to stabilise: num_requests={1} new_requests={2}".format(
        time.time(), num_requests, new_requests))
    num_requests = new_requests
    num_stabilization_attempts = num_stabilization_attempts + 1
    time.sleep(wait_time_in_sec)
  LOG.info("{0} Final num requests: {1}".format(time.time(), num_requests))
  return num_requests