diff --git a/src/openptv2/algorithms/correspondences.py b/src/openptv2/algorithms/correspondences.py index 021a809e..1511c67f 100644 --- a/src/openptv2/algorithms/correspondences.py +++ b/src/openptv2/algorithms/correspondences.py @@ -20,6 +20,30 @@ PT_UNUSED = -999 +@cython.cfunc +@cython.nogil +@cython.boundscheck(False) +@cython.wraparound(False) +def _bisect_left(arr: cython.double[:], n: cython.int, value: cython.double) -> cython.int: + """Index of the first element >= ``value`` in an ascending array of length ``n``. + + Same result as ``int(np.searchsorted(arr, value))``, but as a plain C loop + instead of numpy's per-call Python/ufunc dispatch. Called once per source + target inside ``_build_adjacency_for_pair``'s hot loop -- thousands of + times per frame -- where the numpy call's own dispatch overhead dominated. + """ + lo: cython.int = 0 + hi: cython.int = n + mid: cython.int + while lo < hi: + mid = (lo + hi) // 2 + if arr[mid] < value: + lo = mid + 1 + else: + hi = mid + return lo + + # --------------------------------------------------------------------------- # Output data type — NTupel is the external result of the matching pipeline. # It is only created / consumed once per frame (not in hot loops), so the @@ -146,9 +170,11 @@ def _build_adjacency_for_pair( Speedups applied: * :func:`epi_mm_batch` computes all N epipolar bounding boxes in one vectorised call (replacing N individual :func:`epi_mm` calls). - * :func:`numpy.searchsorted` on the x-sorted destination array finds - the epipolar-band start in O(log M) instead of the manual bisection - previously inside :func:`find_candidate`. + * :func:`_bisect_left`, a typed C loop, finds the epipolar-band start on + the x-sorted destination array in O(log M) -- replacing both the manual + bisection previously inside :func:`find_candidate` and (later) + :func:`numpy.searchsorted`, whose per-call Python/ufunc dispatch + dominated this loop when profiled. * Quality-ratio checks and distance filtering are applied inline, avoiding Python function-call overhead for :func:`find_candidate`. """ @@ -267,7 +293,7 @@ def _build_adjacency_for_pair( sqrt_m2_1: cython.double = np.sqrt(m_line * m_line + 1.0) # Binary-search for x-range start (replaces manual bisection) - lo = int(np.searchsorted(dst_x, xa - eps)) + lo = _bisect_left(dst_x, n2, xa - eps) n_i: cython.double = frm.targets[i1][src_pnr[i]].n nx_i: cython.double = frm.targets[i1][src_pnr[i]].nx @@ -361,29 +387,47 @@ def match_pairs( ) return - # Multi-threaded: each camera pair is independent - from concurrent.futures import ThreadPoolExecutor, as_completed - - with ThreadPoolExecutor(max_workers=len(pairs)) as pool: - futures = { - pool.submit( - _build_adjacency_for_pair, - i1, - i2, - n_arr, - p2_arr, - corr_arr, - dist_arr, - corrected, - frm, - vpar, - cpar, - calib, - ): (i1, i2) - for i1, i2 in pairs - } + # Multi-threaded: each camera pair is independent. Uses the shared, + # process-wide pool (openptv2.thread_pool) instead of opening a fresh + # ThreadPoolExecutor here -- this runs once per frame, and a fresh pool + # per frame measured at ~22% of wall time in a profiled batch run purely + # in thread create/teardown, not in the work itself. + from concurrent.futures import as_completed, wait + + from openptv2.thread_pool import get_executor + + pool = get_executor() + futures = { + pool.submit( + _build_adjacency_for_pair, + i1, + i2, + n_arr, + p2_arr, + corr_arr, + dist_arr, + corrected, + frm, + vpar, + cpar, + calib, + ): (i1, i2) + for i1, i2 in pairs + } + try: for future in as_completed(futures): future.result() # propagate exceptions + except BaseException: + # Unlike the removed `with ThreadPoolExecutor() as pool:`, this pool + # is shared and outlives this call, so it won't wait for stragglers + # on its own. Cancel what hasn't started and wait for what has, + # so nothing keeps writing into n_arr/p2_arr/corr_arr/dist_arr after + # we've raised (a caller retrying or reusing those buffers would + # otherwise race with them). + for f in futures: + f.cancel() + wait(futures) + raise # --------------------------------------------------------------------------- diff --git a/src/openptv2/thread_pool.py b/src/openptv2/thread_pool.py new file mode 100644 index 00000000..a7f0b51a --- /dev/null +++ b/src/openptv2/thread_pool.py @@ -0,0 +1,34 @@ +"""A single reusable thread pool, shared across per-frame calls. + +``match_pairs`` used to open a fresh ``ThreadPoolExecutor`` (and tear it +down) on every frame -- real OS thread creation/teardown 25+ times per batch +run, measured at ~22% of wall time in a profiled run. One pool, sized to the +machine and created once, removes that churn: submitting work to it costs +only a queue push. +""" + +import atexit +import os +from concurrent.futures import ThreadPoolExecutor + +_executor: ThreadPoolExecutor | None = None +_creator_pid: int | None = None + + +def get_executor() -> ThreadPoolExecutor: + """The process-wide shared executor, created on first use. + + Recreated whenever the current PID doesn't match the one that created + it: pyptv_batch_parallel forks worker processes on Unix, and a fork + only clones the calling thread, so a child inheriting an + already-initialized executor would have a pool object whose worker + threads never actually exist in that process -- submissions there + would have no consumer and hang forever. + """ + global _executor, _creator_pid + pid = os.getpid() + if _executor is None or _creator_pid != pid: + _executor = ThreadPoolExecutor(max_workers=os.cpu_count() or 4) + _creator_pid = pid + atexit.register(_executor.shutdown, wait=False, cancel_futures=True) + return _executor