deps/libchdr/deps/zstd-1.5.5/tests/rateLimiter.py

   1 #!/usr/bin/env python3
   2
   3 # ################################################################
   4 # Copyright (c) Meta Platforms, Inc. and affiliates.
   5 # All rights reserved.
   6 #
   7 # This source code is licensed under both the BSD-style license (found in the
   8 # LICENSE file in the root directory of this source tree) and the GPLv2 (found
   9 # in the COPYING file in the root directory of this source tree).
  10 # You may select, at your option, one of the above-listed licenses.
  11 # ##########################################################################
  12
  13 # Rate limiter, replacement for pv
  14 # this rate limiter does not "catch up" after a blocking period
  15 # Limitations:
  16 # - only accepts limit speed in MB/s
  17
  18 import sys
  19 import time
  20
  21 MB = 1024 * 1024
  22 rate = float(sys.argv[1]) * MB
  23 start = time.time()
  24 total_read = 0
  25
  26 # sys.stderr.close()  # remove error message, for Ctrl+C
  27
  28 try:
  29   buf = " "
  30   while len(buf):
  31     now = time.time()
  32     to_read = max(int(rate * (now - start)), 1)
  33     max_buf_size = 1 * MB
  34     to_read = min(to_read, max_buf_size)
  35     start = now
  36
  37     buf = sys.stdin.buffer.read(to_read)
  38     sys.stdout.buffer.write(buf)
  39
  40 except (KeyboardInterrupt, BrokenPipeError) as e:
  41     pass