From 3c303bf3fbd6da35249c999c447182790e7e6bdf Mon Sep 17 00:00:00 2001 From: autoresearch Date: Thu, 23 Jul 2026 08:05:51 -0400 Subject: [PATCH 1/2] Fix PoseHeaderCache race under concurrent Pose.read PoseHeaderCache was global mutable state updated non-atomically: check_cache could match the old hash while another thread's set_cache had already replaced the header, and PoseHeader.read re-read end_offset after the check. Concurrent Pose.read calls on files with different headers could crash ("buffer is too small for requested array") or silently parse a pose with another file's header. The cache is now an immutable (hash, header, start_offset, end_offset) tuple swapped atomically; readers take a single snapshot. The legacy class attributes are kept in sync for backward compatibility. Co-Authored-By: Claude Fable 5 --- src/python/pose_format/pose_header.py | 34 +++++++++++++++++++-------- src/python/tests/pose_test.py | 25 ++++++++++++++++++++ 2 files changed, 49 insertions(+), 10 deletions(-) diff --git a/src/python/pose_format/pose_header.py b/src/python/pose_format/pose_header.py index 3640573..7467cf6 100644 --- a/src/python/pose_format/pose_header.py +++ b/src/python/pose_format/pose_header.py @@ -230,22 +230,30 @@ def __str__(self): class PoseHeaderCache: + # The individual attributes are kept for backward compatibility (and torn reads of + # them are harmless where they are used as hints), but the authoritative cache is + # `entry`: an immutable (hash, header, start_offset, end_offset) tuple swapped + # atomically, so concurrent Pose.read calls never observe a half-updated cache. start_offset: int = None end_offset: int = None hash: str = None header: 'PoseHeader' = None + entry: tuple = None @staticmethod - def calc_hash(buffer: bytes): - return hashlib.md5(buffer[PoseHeaderCache.start_offset:PoseHeaderCache.end_offset]).hexdigest() + def calc_hash(buffer: bytes, start_offset: int, end_offset: int): + return hashlib.md5(buffer[start_offset:end_offset]).hexdigest() @staticmethod - def check_cache(buffer: bytes) -> 'PoseHeader': - if PoseHeaderCache.hash is None: + def check_cache(buffer: bytes) -> Optional[Tuple['PoseHeader', int]]: + entry = PoseHeaderCache.entry # single atomic snapshot; class attrs may change under us + if entry is None: return None - if PoseHeaderCache.hash == PoseHeaderCache.calc_hash(buffer): - return PoseHeaderCache.header + cached_hash, header, start_offset, end_offset = entry + if cached_hash == PoseHeaderCache.calc_hash(buffer, start_offset, end_offset): + return header, end_offset + return None @staticmethod def clear_cache(): @@ -253,13 +261,16 @@ def clear_cache(): PoseHeaderCache.end_offset = None PoseHeaderCache.hash = None PoseHeaderCache.header = None + PoseHeaderCache.entry = None @staticmethod def set_cache(header: 'PoseHeader', buffer: bytes, start_offset: int, end_offset: int): + new_hash = PoseHeaderCache.calc_hash(buffer, start_offset, end_offset) PoseHeaderCache.start_offset = start_offset PoseHeaderCache.end_offset = end_offset PoseHeaderCache.header = header - PoseHeaderCache.hash = PoseHeaderCache.calc_hash(buffer) + PoseHeaderCache.hash = new_hash + PoseHeaderCache.entry = (new_hash, header, start_offset, end_offset) class PoseHeader: @@ -317,9 +328,12 @@ def read(reader: BufferReader) -> 'PoseHeader': PoseHeader An instance of PoseHeader. """ - cached_header = PoseHeaderCache.check_cache(reader.buffer) - if cached_header is not None: - reader.read_offset = PoseHeaderCache.end_offset + cached = PoseHeaderCache.check_cache(reader.buffer) + if cached is not None: + # header and end_offset come from the same atomic cache snapshot -- reading + # PoseHeaderCache.end_offset here instead would race with concurrent set_cache + cached_header, end_offset = cached + reader.read_offset = end_offset return cached_header start_offset = reader.read_offset diff --git a/src/python/tests/pose_test.py b/src/python/tests/pose_test.py index 514e27d..a05841d 100644 --- a/src/python/tests/pose_test.py +++ b/src/python/tests/pose_test.py @@ -1,5 +1,6 @@ import random import string +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Optional, Tuple from unittest import TestCase @@ -340,6 +341,30 @@ def test_read_empty_pose_body_shape_matches_numpy_pose_body(self): self.assertEqual(empty_pose.body.confidence.shape, numpy_pose.body.confidence.shape) self.assertEqual(empty_pose.body.fps, numpy_pose.body.fps) + def test_concurrent_read_of_different_files_is_safe(self): + # Regression test: PoseHeaderCache used to be updated non-atomically, so + # concurrent Pose.read calls on files with different headers could crash + # ("buffer is too small for requested array") or, worse, silently return a + # pose parsed with another file's header. + data_dir = Path(__file__).parent / "data" + buffers = {} + expected = {} + for name in ["mediapipe.pose", "openpose.pose"]: + with open(data_dir / name, 'rb') as f: + buffers[name] = f.read() + pose = Pose.read(buffers[name]) + expected[name] = ([c.name for c in pose.header.components], pose.body.data.shape) + + def read_one(name): + pose = Pose.read(buffers[name]) + return name, ([c.name for c in pose.header.components], pose.body.data.shape) + + names = list(buffers.keys()) * 4 + with ThreadPoolExecutor(max_workers=8) as executor: + for _ in range(30): + for name, got in executor.map(read_one, names): + self.assertEqual(expected[name], got) + def test_pose_bbox(self): data_dir = Path(__file__).parent / "data" with open(data_dir / 'mediapipe.pose', 'rb') as f: From 7e886f44b359c86863eea819513926862ec33dbd Mon Sep 17 00:00:00 2001 From: autoresearch Date: Thu, 23 Jul 2026 08:42:17 -0400 Subject: [PATCH 2/2] Use a lock for PoseHeaderCache instead of an atomic entry tuple Per review: a plain threading.Lock over check_cache/set_cache/clear_cache is easier to follow than the snapshot tuple. check_cache still returns (header, end_offset) -- the offset must come from the same critical section, otherwise the caller re-reading the class attribute races again. Co-Authored-By: Claude Fable 5 --- src/python/pose_format/pose_header.py | 49 +++++++++++++-------------- 1 file changed, 24 insertions(+), 25 deletions(-) diff --git a/src/python/pose_format/pose_header.py b/src/python/pose_format/pose_header.py index 7467cf6..c334c3b 100644 --- a/src/python/pose_format/pose_header.py +++ b/src/python/pose_format/pose_header.py @@ -1,6 +1,7 @@ import hashlib import math import struct +import threading from typing import BinaryIO, List, Tuple, Optional, Union from .utils.reader import BufferReader, ConstStructs @@ -230,47 +231,45 @@ def __str__(self): class PoseHeaderCache: - # The individual attributes are kept for backward compatibility (and torn reads of - # them are harmless where they are used as hints), but the authoritative cache is - # `entry`: an immutable (hash, header, start_offset, end_offset) tuple swapped - # atomically, so concurrent Pose.read calls never observe a half-updated cache. + # All access goes through _lock: an unsynchronized reader races with set_cache and + # can observe a half-updated cache (e.g. the new header with the old hash/offsets). + # check_cache therefore also returns end_offset, so callers position their reader + # from the same consistent snapshot instead of re-reading the class attribute. start_offset: int = None end_offset: int = None hash: str = None header: 'PoseHeader' = None - entry: tuple = None + _lock = threading.Lock() @staticmethod - def calc_hash(buffer: bytes, start_offset: int, end_offset: int): - return hashlib.md5(buffer[start_offset:end_offset]).hexdigest() + def calc_hash(buffer: bytes): + return hashlib.md5(buffer[PoseHeaderCache.start_offset:PoseHeaderCache.end_offset]).hexdigest() @staticmethod def check_cache(buffer: bytes) -> Optional[Tuple['PoseHeader', int]]: - entry = PoseHeaderCache.entry # single atomic snapshot; class attrs may change under us - if entry is None: - return None + with PoseHeaderCache._lock: + if PoseHeaderCache.hash is None: + return None - cached_hash, header, start_offset, end_offset = entry - if cached_hash == PoseHeaderCache.calc_hash(buffer, start_offset, end_offset): - return header, end_offset - return None + if PoseHeaderCache.hash == PoseHeaderCache.calc_hash(buffer): + return PoseHeaderCache.header, PoseHeaderCache.end_offset + return None @staticmethod def clear_cache(): - PoseHeaderCache.start_offset = None - PoseHeaderCache.end_offset = None - PoseHeaderCache.hash = None - PoseHeaderCache.header = None - PoseHeaderCache.entry = None + with PoseHeaderCache._lock: + PoseHeaderCache.start_offset = None + PoseHeaderCache.end_offset = None + PoseHeaderCache.hash = None + PoseHeaderCache.header = None @staticmethod def set_cache(header: 'PoseHeader', buffer: bytes, start_offset: int, end_offset: int): - new_hash = PoseHeaderCache.calc_hash(buffer, start_offset, end_offset) - PoseHeaderCache.start_offset = start_offset - PoseHeaderCache.end_offset = end_offset - PoseHeaderCache.header = header - PoseHeaderCache.hash = new_hash - PoseHeaderCache.entry = (new_hash, header, start_offset, end_offset) + with PoseHeaderCache._lock: + PoseHeaderCache.start_offset = start_offset + PoseHeaderCache.end_offset = end_offset + PoseHeaderCache.header = header + PoseHeaderCache.hash = PoseHeaderCache.calc_hash(buffer) class PoseHeader: