more red diffs, remove hevc framereader stuff and two unused classes

5 years ago · d4f8943075
parent 8b448a7c16
commit d4f8943075
3 changed files with 0 additions and 584 deletions
--- a/tools/lib/async_generator.py
+++ b/tools/lib/async_generator.py
@ -1,352 +0,0 @@
 import functools
 import threading
 import inspect
 import sys
 import select
 import struct
 from math import sqrt
 from collections import OrderedDict, deque
 from time import time
 from tools.lib.pollable_queue import PollableQueue, Empty, Full, ExistentialError
 EndSentinel = object()
 def _sync_inner_generator(input_queue, *args, **kwargs):
  func = args[0]
  args = args[1:]
  get = input_queue.get
  while True:
    item = get()
    if item is EndSentinel:
      return
    cookie, value = item
    yield cookie, func(value, *args, **kwargs)
 def _async_streamer_async_inner(input_queue, output_queue, generator_func, args, kwargs):
  put = output_queue.put
  put_end = True
  try:
    g = generator_func(input_queue, *args, **kwargs)
    for item in g:
      put((time(), item))
    g.close()
  except ExistentialError:
    put_end = False
    raise
  finally:
    if put_end:
      put((None, EndSentinel))
 def _running_mean_var(ltc_stats, x):
  old_mean, var = ltc_stats
  mean = min(600., 0.98 * old_mean + 0.02 * x)
  var = min(5., max(0.1, 0.98 * var + 0.02 * (mean - x) * (old_mean - x)))
  return mean, var
 def _find_next_resend(sent_messages, ltc_stats):
  if not sent_messages:
    return None, None
  oldest_sent_idx = sent_messages._OrderedDict__root[1][2]
  send_time, _ = sent_messages[oldest_sent_idx]
  # Assume message has been lost if it is >10 standard deviations from mean.
  mean, var = ltc_stats
  next_resend_time = send_time + mean + 40. * sqrt(var)
  return oldest_sent_idx, next_resend_time
 def _do_cleanup(input_queue, output_queue, num_workers, sentinels_received, num_outstanding):
  input_fd = input_queue.put_fd()
  output_fd = output_queue.get_fd()
  poller = select.epoll()
  poller.register(input_fd, select.EPOLLOUT)
  poller.register(output_fd, select.EPOLLIN)
  remaining_outputs = []
  end_sentinels_to_send = num_workers - sentinels_received
  while sentinels_received < num_workers:
    evts = dict(poller.poll(-1 if num_outstanding > 0 else 10.))
    if not evts:
      # Workers aren't responding, crash.
      break
    if output_fd in evts:
      _, maybe_sentinel = output_queue.get()
      if maybe_sentinel is EndSentinel:
        sentinels_received += 1
      else:
        remaining_outputs.append(maybe_sentinel[1])
        num_outstanding -= 1
    if input_fd in evts:
      if end_sentinels_to_send > 0:
        input_queue.put_nowait(EndSentinel)
        end_sentinels_to_send -= 1
      else:
        poller.modify(input_fd, 0)
  # TODO: Raise an exception when a queue thread raises one.
  assert sentinels_received == num_workers, (sentinels_received, num_workers)
  assert output_queue.empty()
  return remaining_outputs
 def _generate_results(input_stream, input_queue, worker_output_queue, output_queue,
                      num_workers, max_outstanding):
  pack_cookie = struct.pack
  # Maps idx -> (send_time, input)
  sent_messages = OrderedDict()
  oldest_sent_idx = None
  next_resend_time = None
  ltc_stats = 5., 10.
  # Maps idx -> result
  received_messages = {}
  next_out = 0
  # Start things off by pulling the first value.
  next_in_item = next(input_stream, EndSentinel)
  inputs_remain = next_in_item is not EndSentinel
  sentinels_received = 0
  input_fd = input_queue.put_fd()
  worker_output_fd = worker_output_queue.get_fd()
  output_fd = output_queue.put_fd()
  poller = select.epoll()
  poller.register(input_fd, select.EPOLLOUT)
  poller.register(worker_output_fd, select.EPOLLIN)
  poller.register(output_fd, 0)
  # Keep sending/retrying until the input stream and sent messages are all done.
  while sentinels_received < num_workers and (inputs_remain or sent_messages):
    if max_outstanding:
      can_send_new = (len(sent_messages) < max_outstanding and
                      len(received_messages) < max_outstanding and inputs_remain)
    else:
      can_send_new = inputs_remain
    if (next_resend_time and now >= next_resend_time) or can_send_new:
      poller.modify(input_fd, select.EPOLLOUT)
    else:
      poller.modify(input_fd, 0)
    if next_resend_time:
      t = max(0, next_resend_time - now)
      evts = dict(poller.poll(t))
    else:
      evts = dict(poller.poll())
    now = time()
    if output_fd in evts:
      output_queue.put_nowait(received_messages.pop(next_out))
      next_out += 1
      if next_out not in received_messages:
        poller.modify(output_fd, 0)
    if worker_output_fd in evts:
      for receive_time, maybe_sentinel in worker_output_queue.get_multiple_nowait():
        # Check for EndSentinel in case of worker crash.
        if maybe_sentinel is EndSentinel:
          sentinels_received += 1
          continue
        idx_bytes, value = maybe_sentinel
        idx = struct.unpack("<Q", idx_bytes)[0]
        # Don't store duplicates.
        sent_message = sent_messages.pop(idx, None)
        if sent_message is not None:
          received_messages[idx] = value
          ltc_stats = _running_mean_var(ltc_stats, receive_time - sent_message[0])
          if idx == oldest_sent_idx:
            oldest_sent_idx, next_resend_time = _find_next_resend(sent_messages, ltc_stats)
          if idx == next_out:
            poller.modify(output_fd, select.EPOLLOUT)
        else:
          if oldest_sent_idx is not None:
            # The message was resent, so it is at least old as the oldest tracked message.
            ltc_stats = _running_mean_var(ltc_stats, now - sent_messages[oldest_sent_idx][0])
    elif input_fd in evts:
      if can_send_new:
        # We're under the limit, get the next input.
        send_idx, send_value = next_in_item
        input_queue.put_nowait((pack_cookie("<Q", send_idx), send_value))
        sent_messages[next_in_item[0]] = now, next_in_item[1]
        next_in_item = next(input_stream, EndSentinel)
        inputs_remain = next_in_item is not EndSentinel
        if oldest_sent_idx is None:
          oldest_sent_idx, next_resend_time = _find_next_resend(sent_messages, ltc_stats)
      else:
        # Move the resent item to the end.
        send_time, resend_input = sent_messages.pop(oldest_sent_idx)
        sys.stdout.write("Resending {} (ltc, mean, var) = ({}, {}, {})\n".format(
          oldest_sent_idx, now - send_time, ltc_stats[0], ltc_stats[1]))
        input_queue.put_nowait((pack_cookie("<Q", oldest_sent_idx), resend_input))
        sent_messages[oldest_sent_idx] = now, resend_input
        oldest_sent_idx, next_resend_time = _find_next_resend(sent_messages, ltc_stats)
  # Return remaining messages.
  while next_out in received_messages:
    output_queue.put(received_messages.pop(next_out))
    next_out += 1
  _do_cleanup(input_queue, worker_output_queue, num_workers, sentinels_received, 0)
  output_queue.put(EndSentinel)
 def _generate_results_unreliable(input_stream, input_queue, worker_output_queue,
                                 output_queue, num_workers, max_outstanding_unused):
  # Start things off by pulling the first value.
  next_in_item = next(input_stream, EndSentinel)
  inputs_remain = next_in_item is not EndSentinel
  # TODO: Use heapq to return oldest message.
  received_messages = deque()
  pack_cookie = struct.pack
  input_fd = input_queue.put_fd()
  worker_output_fd = worker_output_queue.get_fd()
  output_fd = output_queue.put_fd()
  poller = select.epoll()
  poller.register(input_fd, select.EPOLLOUT)
  poller.register(worker_output_fd, select.EPOLLIN)
  poller.register(output_fd, 0)
  # Keep sending/retrying until the input stream and sent messages are all done.
  num_outstanding = 0
  sentinels_received = 0
  while sentinels_received < num_workers and (inputs_remain or received_messages):
    # Avoid poll() if we can easily detect that there is work to do.
    evts = (input_fd if inputs_remain and not input_queue.full() else 0, output_fd
            if not output_queue.full() and len(received_messages) else 0, worker_output_fd
            if not worker_output_queue.empty() else 0)
    if all(evt == 0 for evt in evts):
      evts = dict(poller.poll())
    if output_fd in evts:
      output_queue.put(received_messages.pop())
      if len(received_messages) == 0:
        poller.modify(output_fd, 0)
    if worker_output_fd in evts:
      for receive_time, maybe_sentinel in worker_output_queue.get_multiple():
        # Check for EndSentinel in case of worker crash.
        if maybe_sentinel is EndSentinel:
          sentinels_received += 1
          continue
        received_messages.appendleft(maybe_sentinel[1])
        num_outstanding -= 1
      poller.modify(output_fd, select.EPOLLOUT)
    if input_fd in evts:
      # We're under the limit, get the next input.
      send_idx, send_value = next_in_item
      input_queue.put((pack_cookie("<Q", send_idx), send_value))
      next_in_item = next(input_stream, EndSentinel)
      inputs_remain = next_in_item is not EndSentinel
      num_outstanding += 1
      if not inputs_remain:
        poller.modify(input_fd, 0)
  # TODO: Track latency even though we don't retry.
  for value in _do_cleanup(input_queue, worker_output_queue, num_workers,
                           sentinels_received, num_outstanding):
    output_queue.put(value)
  output_queue.put(EndSentinel)
 def _async_generator(func, max_workers, in_q_size, out_q_size, max_outstanding,
                     async_inner, reliable):
  if async_inner:
    assert inspect.isgeneratorfunction(
      func), "async_inner == True but {} is not a generator".format(func)
  @functools.wraps(func)
  def wrapper(input_sequence_or_self, *args, **kwargs):
    # HACK: Determine whether the first arg is "self". ismethod returns False here.
    if inspect.getargspec(func).args[0] == "self":
      inner_func = func.__get__(input_sequence_or_self, type(input_sequence_or_self))
      input_sequence = args[0]
      args = args[1:]
    else:
      inner_func = func
      input_sequence = input_sequence_or_self
    input_stream = enumerate(iter(input_sequence))
    if reliable:
      generate_func = _generate_results
    else:
      generate_func = _generate_results_unreliable
    input_queue = PollableQueue(in_q_size)
    worker_output_queue = PollableQueue(8 * max_workers)
    output_queue = PollableQueue(out_q_size)
    # Start the worker threads.
    if async_inner:
      generator_func = inner_func
    else:
      args = (inner_func,) + args
      generator_func = _sync_inner_generator
    worker_threads = []
    for _ in range(max_workers):
      t = threading.Thread(
        target=_async_streamer_async_inner,
        args=(input_queue, worker_output_queue, generator_func, args, kwargs))
      t.daemon = True
      t.start()
      worker_threads.append(t)
    master_thread = threading.Thread(
      target=generate_func,
      args=(input_stream, input_queue, worker_output_queue, output_queue, max_workers,
            max_outstanding))
    master_thread.daemon = True
    master_thread.start()
    try:
      while True:
        for value in output_queue.get_multiple():
          if value is EndSentinel:
            return
          else:
            yield value
    finally:
      # Make sure work is done and the threads are stopped.
      for t in worker_threads:
        t.join(1)
      master_thread.join(1)
      input_queue.close()
      worker_output_queue.close()
      output_queue.close()
  return wrapper
 def async_generator(max_workers=1,
                    in_q_size=10,
                    out_q_size=12,
                    max_outstanding=10000,
                    async_inner=False,
                    reliable=True):
  return (
    lambda f: _async_generator(f, max_workers, in_q_size, out_q_size, max_outstanding, async_inner, reliable)
  )
--- a/tools/lib/framereader.py
+++ b/tools/lib/framereader.py
@ -413,131 +413,6 @@ def index_pstream(fns, typ, cache_prefix=None):
    with atomic_write_in_dir(cache_path, mode="wb", overwrite=True) as cache_file:
      pickle.dump(segment_index, cache_file, -1)
 def gpu_info():
  ret = []
  for fn in glob.glob("/proc/driver/nvidia/gpus/*/information"):
    with open(fn, "r") as f:
      dat = f.read()
      kvs = dat.strip().split("\n")
      kv = {}
      for s in kvs:
        k, v = s.split(":", 1)
        kv[k] = v.strip()
      ret.append(kv)
  return ret
 def gpu_supports_hevc(gpuinfo):
  return ("GTX 10" in gpuinfo['Model'] or "GTX 20" in gpuinfo['Model'] or gpuinfo['Model'] == "Graphics Device")
 def find_hevc_gpu():
  for gpuinfo in gpu_info():
    if gpu_supports_hevc(gpuinfo):
      return int(gpuinfo['Device Minor'])
  return None
 def _ffmpeg_fcamera_input_for_frame_info(frame_info):
  st = time.time()
  fn, num, count, cache_prefix = frame_info
  assert fn.endswith('.hevc')
  sindex = index_stream(fn, "hevc", cache_prefix=cache_prefix)
  index = sindex['index']
  prefix = sindex['global_prefix']
  probe = sindex['probe']
  frame_e = num + count
  frame_b = num
  # must start decoding on an i-frame
  while index[frame_b, 0] != HEVC_SLICE_I:
    frame_b -= 1
  offset_b = index[frame_b, 1]
  offset_e = index[frame_e, 1]
  assert frame_b <= num < frame_e
  skip = num - frame_b
  w = probe['streams'][0]['width']
  h = probe['streams'][0]['height']
  assert (h, w) == (874, 1164)
  st2 = time.time()
  with FileReader(fn) as f:
    f.seek(offset_b)
    input_data = f.read(offset_e - offset_b)
  et = time.time()
  get_time = et-st
  get_time2 = et-st2
  if get_time > 10.0:
    print("TOOK OVER 10 seconds to fetch %r %f %f" % (frame_info, get_time, get_time2))
  return prefix, input_data, skip, count
 def _ffmpeg_fcamera_input_for_frame(pair):
  cookie, frame_info = pair
  try:
    return cookie, _ffmpeg_fcamera_input_for_frame_info(frame_info)
  except Exception as e:
    # Let the caller handle exceptions.
    return cookie, e
 def _feed_ffmpeg_fcamera_input_work_loop(frames, proc_stdin, select_pipe_fd, cookie_queue):
  last_prefix = None
  """
  with ThreadPoolExecutor(64) as pool:
    futures = []
    for f in frames:
      futures.append(pool.submit(_ffmpeg_fcamera_input_for_frame, f))
    for f in as_completed(futures):
      cookie, data = f.result()
      if isinstance(data, Exception):
        # Just print exceptions for now.
        print(data)
        continue
      prefix, input_data, skip, count = data
      cookie_queue.put((cookie, count))
      # Write zeros for skipped frames, ones for keep frames.
      os.write(select_pipe_fd, b"\x00" * skip + b"\x01" * count)
      if prefix != last_prefix:
        proc_stdin.write(prefix)
        last_prefix = prefix
      proc_stdin.write(input_data)
  """
  num_threads = 64
  for cookie, data in async_generator(
      num_threads, 8 * num_threads, 8 * num_threads,
      reliable=False)(_ffmpeg_fcamera_input_for_frame)(frames):
    if isinstance(data, Exception):
      # Just print exceptions for now.
      print(data)
      continue
    prefix, input_data, skip, count = data
    cookie_queue.put((cookie, count))
    # Write zeros for skipped frames, ones for keep frames.
    os.write(select_pipe_fd, b"\x00" * skip + b"\x01" * count)
    if prefix != last_prefix:
      proc_stdin.write(prefix)
      last_prefix = prefix
    proc_stdin.write(input_data)
 _FCAMERA_FEED_SUCCESS = object()
 def feed_ffmpeg_fcamera_input(frames, proc_stdin, select_pipe_fd, cookie_queue):
  print("Feed started on {}".format(threading.current_thread().name))
  try:
    _feed_ffmpeg_fcamera_input_work_loop(frames, proc_stdin, select_pipe_fd, cookie_queue)
    cookie_queue.put((_FCAMERA_FEED_SUCCESS, None))
  finally:
    # Always close ffmpeg input.
    proc_stdin.close()
 def read_file_check_size(f, sz, cookie):
  buff = bytearray(sz)
  bytes_read = f.readinto(buff)
--- a/tools/lib/pollable_queue.py
+++ b/tools/lib/pollable_queue.py
@ -1,107 +0,0 @@
 import os
 import select
 import fcntl
 from itertools import count
 from collections import deque
 Empty = OSError
 Full = OSError
 ExistentialError = OSError
 class PollableQueue(object):
  """A Queue that you can poll().
     Only works with a single producer.
  """
  def __init__(self, maxlen=None):
    with open("/proc/sys/fs/pipe-max-size") as f:
      max_maxlen = int(f.read().rstrip())
    if maxlen is None:
      maxlen = max_maxlen
    else:
      maxlen = min(maxlen, max_maxlen)
    self._maxlen = maxlen
    self._q = deque()
    self._get_fd, self._put_fd = os.pipe()
    fcntl.fcntl(self._get_fd, fcntl.F_SETFL, os.O_NONBLOCK)
    fcntl.fcntl(self._put_fd, fcntl.F_SETFL, os.O_NONBLOCK)
    fcntl.fcntl(self._get_fd, fcntl.F_SETLEASE + 7, self._maxlen)
    fcntl.fcntl(self._put_fd, fcntl.F_SETLEASE + 7, self._maxlen)
    get_poller = select.epoll()
    put_poller = select.epoll()
    get_poller.register(self._get_fd, select.EPOLLIN)
    put_poller.register(self._put_fd, select.EPOLLOUT)
    self._get_poll = get_poller.poll
    self._put_poll = put_poller.poll
  def get_fd(self):
    return self._get_fd
  def put_fd(self):
    return self._put_fd
  def put(self, item, block=True, timeout=None):
    if block:
      while self._put_poll(timeout if timeout is not None else -1):
        try:
          # TODO: This is broken for multiple push threads when the queue is full.
          return self.put_nowait(item)
        except OSError as e:
          if e.errno != 11:
            raise
        raise Full()
    else:
      return self.put_nowait(item)
  def put_nowait(self, item):
    self._q.appendleft(item)
    os.write(self._put_fd, b"\x00")
  def get(self, block=True, timeout=None):
    if block:
      while self._get_poll(timeout if timeout is not None else -1):
        try:
          return self.get_nowait()
        except OSError as e:
          if e.errno != 11:
            raise
      raise Empty()
    else:
      return self.get_nowait()
  def get_nowait(self):
    os.read(self._get_fd, 1)
    return self._q.pop()
  def get_multiple(self, block=True, timeout=None):
    if block:
      if self._get_poll(timeout if timeout is not None else -1):
        return self.get_multiple_nowait()
      else:
        raise Empty()
    else:
      return self.get_multiple_nowait()
  def get_multiple_nowait(self, max_messages=None):
    num_read = len(os.read(self._get_fd, max_messages or self._maxlen))
    return [self._q.pop() for _ in range(num_read)]
  def empty(self):
    return len(self._q) == 0
  def full(self):
    return len(self._q) >= self._maxlen
  def close(self):
    os.close(self._get_fd)
    os.close(self._put_fd)
  def __len__(self):
    return len(self._q)