parallelize kORE outer (k, n) trials via ProcessPoolExecutor

Add _kore_trial() module-level worker for pickling. infer() accepts
n_workers param — >1 runs all (k x N) trials concurrently instead
of serial. Default 1 preserves existing behavior.

Thread n_workers through _run_kore() and infer_ensemble() to support
--kore flag with parallel kORE inference.
This commit is contained in:
tobjend 2026-07-04 03:04:54 +02:00
parent 5906adfc95
commit 0b5b0e623b
2 changed files with 52 additions and 18 deletions

View file

@ -386,18 +386,18 @@ _ALGORITHMS = {
} }
def _run_kore(sequences, kmax, N): def _run_kore(sequences, kmax, N, n_workers=1):
"""Run kOREInference, return (grammar, score) or (None, inf).""" """Run kOREInference, return (grammar, score) or (None, inf)."""
from .kore import kOREInference from .kore import kOREInference
kore = kOREInference(k_max=kmax, N=N) kore = kOREInference(k_max=kmax, N=N)
result = kore.infer(sequences) result = kore.infer(sequences, n_workers=n_workers)
if result: if result:
_, expr, _ = result _, expr, _ = result
return expr, mdl_score_simple(expr, sequences) return expr, mdl_score_simple(expr, sequences)
return None, float('inf') return None, float('inf')
def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, include_kore=False): def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, include_kore=False, n_workers=1):
"""Run all applicable algorithms and return the best by MDL score. """Run all applicable algorithms and return the best by MDL score.
Args: Args:
@ -450,7 +450,7 @@ def infer_ensemble(sequences, kmax=2, N=3, prefer=None, min_coverage=1.0, includ
# 3. kOREInference (opt-in via include_kore=True) # 3. kOREInference (opt-in via include_kore=True)
if include_kore: if include_kore:
kore_g, kore_score = _run_kore(sequences, kmax, N) kore_g, kore_score = _run_kore(sequences, kmax, N, n_workers=n_workers)
if kore_g: if kore_g:
results.append(('kOREInference', kore_g, kore_score)) results.append(('kOREInference', kore_g, kore_score))

View file

@ -13,6 +13,9 @@ Unlike the PTA→Shrink→Repair approach from Bex 2008, this follows
the journal paper (arXiv 1004.2372) exactly. the journal paper (arXiv 1004.2372) exactly.
""" """
import os
from concurrent.futures import ProcessPoolExecutor, as_completed
from .ikoa import ikoa from .ikoa import ikoa
from .rwrsq import rwr_sq from .rwrsq import rwr_sq
from .idregex import is_deterministic from .idregex import is_deterministic
@ -53,6 +56,28 @@ def validate_k_ore(expr, k, alphabet_set=None):
return True, "OK" return True, "OK"
def _kore_trial(args):
"""Run one (k, n) trial. Module-level for ProcessPoolExecutor.
Args:
args: (sequences, k, trial_index) trial_index unused, for diagnostics
Returns:
(koa_automaton, expression_string, k) or None
"""
sequences, k, _ = args
G = ikoa(sequences, k, num_trials=1)
if G is None:
return None
expr = rwr_sq(G)
if expr and expr not in ('', 'ε'):
if is_deterministic(expr):
valid, _ = validate_k_ore(expr, k)
if valid:
return (G, expr, k)
return None
class kOREInference: class kOREInference:
""" """
| Algorithm 4: iDRegEx | | Algorithm 4: iDRegEx |
@ -72,9 +97,12 @@ class kOREInference:
self.k_max = k_max self.k_max = k_max
self.N = N self.N = N
def infer(self, sequences): def infer(self, sequences, n_workers=1):
""" """Infer the best k-ORE for the given sequences.
Infer the best k-ORE for the given sequences.
Args:
sequences: list of token sequences
n_workers: parallel workers. >1 runs (k, n) trials concurrently.
Returns: Returns:
(koa_automaton, expression_string, best_k) or None if no valid (koa_automaton, expression_string, best_k) or None if no valid
@ -84,19 +112,25 @@ class kOREInference:
if not sequences: if not sequences:
return None return None
trials = [(sequences, k, i)
for k in range(1, self.k_max + 1)
for i in range(self.N)]
candidates = [] candidates = []
for k in range(1, self.k_max + 1): if n_workers <= 1 or len(trials) < 2:
for _ in range(self.N): for t in trials:
G = ikoa(sequences, k, num_trials=1) result = _kore_trial(t)
if G is None: if result is not None:
continue candidates.append(result)
expr = rwr_sq(G) else:
if expr and expr not in ('', 'ε'): nw = min(n_workers, len(trials))
if is_deterministic(expr): with ProcessPoolExecutor(max_workers=nw) as ex:
valid, _ = validate_k_ore(expr, k) futures = [ex.submit(_kore_trial, t) for t in trials]
if valid: for f in as_completed(futures):
candidates.append((G, expr, k)) result = f.result()
if result is not None:
candidates.append(result)
if not candidates: if not candidates:
return None return None