simon987/hexlib
1
0
Fork 0
mirror of https://github.com/simon987/hexlib.git synced 2025-04-21 18:46:42 +00:00
Code Issues Projects Releases Wiki Activity

Compare commits

base: simon987:cd5a1ac50c09c48c3b1feee20f64a4af71a89bb1
Branches Tags
simon987:master
simon987:1.89
..
compare: simon987:084acbe1849792074842e569a86d5f2e41816e34
Branches Tags
simon987:master
simon987:1.89

2 Commits
cd5a1ac50c ... 084acbe184

Author SHA1 Message Date
simon987
084acbe184 Set max_window_size=2147483648 for zstd 2022-01-29 10:44:38 -05:00
simon987
d578be3218 Increase timeout 2022-01-29 10:38:23 -05:00
3 changed files with 5 additions and 4 deletions
Show Stats Expand all files Collapse all files

5
hexlib/concurrency.py
View File

@ -34,7 +34,7 @@ class StatelessStreamWorker:
class StatelessStreamProcessor: class StatelessStreamProcessor:
def __init__(self, worker_factory, chunk_size=128, processes=1): def __init__(self, worker_factory, chunk_size=128, processes=1, timeout=60):
self._chunk_size = 128 self._chunk_size = 128
self._queue = MPQueue(maxsize=chunk_size) self._queue = MPQueue(maxsize=chunk_size)
self._queue_out = MPQueue(maxsize=processes * 2) self._queue_out = MPQueue(maxsize=processes * 2)
@ -42,6 +42,7 @@ class StatelessStreamProcessor:
self._processes = [] self._processes = []
self._factory = worker_factory self._factory = worker_factory
self._workers = [] self._workers = []
self._timeout = timeout
if processes > 1: if processes > 1:
for _ in range(processes): for _ in range(processes):
@ -67,7 +68,7 @@ class StatelessStreamProcessor:
ingest_thread = Thread(target=self._ingest, args=(iterable,)) ingest_thread = Thread(target=self._ingest, args=(iterable,))
ingest_thread.start() ingest_thread.start()
for results in queue_iter(self._queue_out, joinable=False, timeout=10): for results in queue_iter(self._queue_out, joinable=False, timeout=self._timeout):
yield from results yield from results
ingest_thread.join() ingest_thread.join()

2
hexlib/files.py
View File

@ -85,7 +85,7 @@ def ndjson_iter(*files, compression=""):
line_iter = BufferedReader(gzip.open(file)) line_iter = BufferedReader(gzip.open(file))
elif compression == COMPRESSION_ZSTD: elif compression == COMPRESSION_ZSTD:
fp = open(file, "rb") fp = open(file, "rb")
dctx = zstandard.ZstdDecompressor() dctx = zstandard.ZstdDecompressor(max_window_size=2147483648)
reader = dctx.stream_reader(fp) reader = dctx.stream_reader(fp)
line_iter = BufferedReader(reader) line_iter = BufferedReader(reader)

2
setup.py
View File

@ -2,7 +2,7 @@ from setuptools import setup
setup( setup(
name="hexlib", name="hexlib",
version="1.73", version="1.75",
description="Misc utility methods", description="Misc utility methods",
author="simon987", author="simon987",
author_email="me@simon987.net", author_email="me@simon987.net",