mirror of
https://github.com/littlefs-project/littlefs.git
synced 2025-10-18 09:21:29 +08:00
4fe0738ff4
These are really just different flavors of test.py and test_runner.c
without support for power-loss testing, but with support for measuring
the cumulative number of bytes read, programmed, and erased.
Note that the existing define parameterization should work perfectly
fine for running benchmarks across various dimensions:
./scripts/bench.py \
runners/bench_runner \
bench_file_read \
-gnor \
-DSIZE='range(0,131072,1024)'
Also added a couple basic benchmarks as a starting point.
122 lines
3.2 KiB
Python
Executable File
122 lines
3.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
#
|
|
# Efficiently displays the last n lines of a file/pipe.
|
|
#
|
|
# Example:
|
|
# ./scripts/tailpipe.py trace -n5
|
|
#
|
|
# Copyright (c) 2022, The littlefs authors.
|
|
# SPDX-License-Identifier: BSD-3-Clause
|
|
#
|
|
|
|
import os
|
|
import sys
|
|
import threading as th
|
|
import time
|
|
|
|
|
|
def openio(path, mode='r'):
|
|
if path == '-':
|
|
if mode == 'r':
|
|
return os.fdopen(os.dup(sys.stdin.fileno()), 'r')
|
|
else:
|
|
return os.fdopen(os.dup(sys.stdout.fileno()), 'w')
|
|
else:
|
|
return open(path, mode)
|
|
|
|
def main(path='-', *, lines=1, sleep=0.01, keep_open=False):
|
|
ring = [None] * lines
|
|
i = 0
|
|
count = 0
|
|
lock = th.Lock()
|
|
event = th.Event()
|
|
done = False
|
|
|
|
# do the actual reading in a background thread
|
|
def read():
|
|
nonlocal i
|
|
nonlocal count
|
|
nonlocal done
|
|
while True:
|
|
with openio(path) as f:
|
|
for line in f:
|
|
with lock:
|
|
ring[i] = line
|
|
i = (i + 1) % lines
|
|
count = min(lines, count + 1)
|
|
event.set()
|
|
if not keep_open:
|
|
break
|
|
# don't just flood open calls
|
|
time.sleep(sleep or 0.1)
|
|
done = True
|
|
|
|
th.Thread(target=read, daemon=True).start()
|
|
|
|
try:
|
|
last_count = 1
|
|
while not done:
|
|
time.sleep(sleep)
|
|
event.wait()
|
|
event.clear()
|
|
|
|
# create a copy to avoid corrupt output
|
|
with lock:
|
|
ring_ = ring.copy()
|
|
i_ = i
|
|
count_ = count
|
|
|
|
# first thing first, give ourself a canvas
|
|
while last_count < count_:
|
|
sys.stdout.write('\n')
|
|
last_count += 1
|
|
|
|
for j in range(count_):
|
|
# move cursor, clear line, disable/reenable line wrapping
|
|
sys.stdout.write('\r')
|
|
if count_-1-j > 0:
|
|
sys.stdout.write('\x1b[%dA' % (count_-1-j))
|
|
sys.stdout.write('\x1b[K')
|
|
sys.stdout.write('\x1b[?7l')
|
|
sys.stdout.write(ring_[(i_-count_+j) % lines][:-1])
|
|
sys.stdout.write('\x1b[?7h')
|
|
if count_-1-j > 0:
|
|
sys.stdout.write('\x1b[%dB' % (count_-1-j))
|
|
|
|
sys.stdout.flush()
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
sys.stdout.write('\n')
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
import argparse
|
|
parser = argparse.ArgumentParser(
|
|
description="Efficiently displays the last n lines of a file/pipe.")
|
|
parser.add_argument(
|
|
'path',
|
|
nargs='?',
|
|
help="Path to read from.")
|
|
parser.add_argument(
|
|
'-n',
|
|
'--lines',
|
|
type=lambda x: int(x, 0),
|
|
help="Number of lines to show, defaults to 1.")
|
|
parser.add_argument(
|
|
'-s',
|
|
'--sleep',
|
|
type=float,
|
|
help="Seconds to sleep between reads, defaults to 0.01.")
|
|
parser.add_argument(
|
|
'-k',
|
|
'--keep-open',
|
|
action='store_true',
|
|
help="Reopen the pipe on EOF, useful when multiple "
|
|
"processes are writing.")
|
|
sys.exit(main(**{k: v
|
|
for k, v in vars(parser.parse_intermixed_args()).items()
|
|
if v is not None}))
|