Stabilize large word2vec exports
Bound streaming entry work globally across open ZIPs and release output claims when each ZIP closes, preventing corpus-sized scheduler state and per-ZIP backlog multiplication.
Separate ZIP-reader parallelism from entry-worker parallelism, avoid redundant annotation ZIP processing for surface output, and keep irrelevant late layers from restoring released document state.
Preserve caller ZIP order by default and add --largest-first as an explicit throughput-oriented option. Also make custom CoNLL-U foundry selection independent of concurrent base-entry completion order.
Change-Id: I1ca15de53bf2f500ee748d2478939b99a3ee77e4
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 650ff91..f4bf0cd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,8 +4,14 @@
### Added
+- Added `--zip-parallelism` for controlling open/read ZIP concurrency independently from XML entry worker count.
+- Added opt-in `--largest-first` ZIP scheduling. Word2vec/NOW exports now keep argument order by default, avoiding an automatic size-based corpus-domain bias.
- Krill tars can now be used as *input* and updated in place of a full re-export ("merge mode"): listing an existing `.krill.tar` among the inputs with `-t krill` streams it to a new tar (`<name>.updated.krill.tar`, or `-o`), updated from the remaining inputs. Two use cases: (1) adding or fixing metadata (stand-off `<standOff>` files, `<xenoData>`, header ZIPs) when the original KorAP-XML ZIPs are no longer available, and (2) replacing or adding single annotation foundries (e.g. an improved wiki-taxonomy topic-domain classification) without touching the rest. Texts unaffected by the new inputs are copied through byte-identically without being decompressed; affected texts are patched surgically at the JSON level and recompressed in parallel, and end up identical to a full re-export with the same inputs. New metadata fields replace same-key fields (including legacy/corrected name pairs such as `textClass`/`dmozDomain`) and are appended otherwise; `creationDate`/`pubDate` are only replaced when a new text-level header is supplied, so corpus-level headers cannot clobber per-text dates. Texts present in the new inputs but missing from the tar are ignored with a warning; base tokenization/structure entries are ignored as well. Tars produced by korapxml2krill (Perl) are supported. The input tar is never modified, and the regular Krill export path is unchanged.
+### Fixed
+
+- Stabilized large `word2vec`/`NOW` exports: ZIP readers now share one bounded XML-entry backlog instead of multiplying the backlog by every open ZIP, plain-output document claims are released when each ZIP closes instead of growing for the whole corpus, annotation ZIPs are not re-reading the same base ZIP for surface output, and unused late annotation layers no longer repopulate cleaned document state.
+
## [v4.1.0] - 2026-06-19
### Added
diff --git a/Readme.md b/Readme.md
index 74ac4c1..da07587 100644
--- a/Readme.md
+++ b/Readme.md
@@ -47,6 +47,8 @@
- `-t FORMAT`, `--to FORMAT`: Output format (`zip`, `conllu`, `w2v`, `now`, `krill`)
- `-j N`, `--jobs N`, `--threads N`: Number of threads/jobs to use
+- `--zip-parallelism N`: Maximum ZIP files read concurrently (default: up to 8); XML entry work still uses `--threads`
+- `--largest-first`: Schedule larger input ZIPs first (opt-in; the default is argument order)
- `-T TAGGER[:MODEL]`, `--tag-with TAGGER[:MODEL]`: POS tagger and optional model
- `-P PARSER[:MODEL]`, `--parse-with PARSER[:MODEL]`: Parser and optional model
- `-f`, `--force`: Overwrite existing output files
@@ -134,20 +136,21 @@
- `--lemma-only`: For `-t w2v` and `-t now`, skip loading `data.xml` and output only lemmas from `morpho.xml`. This reduces memory and speeds up throughput.
- `--sequential`: Process entries inside each zip sequentially (zips can still run in parallel). Recommended for `w2v`/`now` to keep locality and lower memory.
+- Parallel `w2v`/`now` runs schedule ZIPs in argument order and use a single bounded entry backlog across all open ZIPs. Completion and output order can still vary when multiple ZIPs run concurrently. Use `--largest-first` to opt into size-descending scheduling when minimizing the low-parallelism tail matters more than corpus order. For reproducible server runs, set both `--threads` and `--zip-parallelism` explicitly.
- `--exclude-zip-glob GLOB` (repeatable): Skip zip basenames that match the glob (e.g., `--exclude-zip-glob 'w?d24.tree_tagger.zip'`).
Example for large NOW export with progress and exclusions:
```
KORAPXMLTOOL_XMX=64g KORAPXMLTOOL_MODELS_PATH=/data/models KORAPXMLTOOL_JAVA_OPTS="-XX:+UseG1GC -Djdk.util.zip.disableMemoryMapping=true -Djdk.util.zip.reuseInflater=true" \
- ./build/bin/korapxmltool -l info -j 100 \
+ ./build/bin/korapxmltool -l info -j 100 --zip-parallelism 8 \
--lemma-only --sequential -t now \
--exclude-zip-glob 'w?d24.tree_tagger.zip' \
/vol/corpora/DeReKo/current/KorAP/zip/*24.tree_tagger.zip | pv > dach2024.lemma.txt
```
At INFO level the tool logs:
-- The zip processing order with file sizes (largest-first in `--lemma-only`).
+- The ZIP processing order with file sizes (argument order by default, or size-descending with `--largest-first`).
- For each zip: start message including its size and a completion line with cumulative progress, ETA and average MB/s.
### Conversion to Krill (KoralQuery) JSON format
diff --git a/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt b/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt
index 520ad94..4002198 100644
--- a/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt
+++ b/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt
@@ -265,6 +265,31 @@
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", threads.toString())
}
+ var zipParallelism: Int = 0 // 0 = auto-detect after --threads is resolved
+
+ @Option(
+ names = ["--zip-parallelism"],
+ paramLabel = "ZIPS",
+ description = [
+ "Maximum number of ZIP files read concurrently. Default: up to 8; entry processing still uses --threads."
+ ]
+ )
+ fun configureZipParallelism(parallelism: Int) {
+ if (parallelism < 1) {
+ throw ParameterException(spec.commandLine(), String.format(Locale.ROOT,
+ "Invalid value `%d' for option '--zip-parallelism': must be at least 1", parallelism))
+ }
+ zipParallelism = parallelism
+ }
+
+ @Option(
+ names = ["--largest-first"],
+ description = [
+ "Schedule larger input ZIPs first. By default ZIPs are scheduled in argument order."
+ ]
+ )
+ var largestFirst: Boolean = false
+
@Option(
names = ["--sequential"],
@@ -693,6 +718,8 @@
if (outputFile != null) sb.appendLine(" Output file: $outputFile")
if (outputDir != ".") sb.appendLine(" Output dir: $outputDir")
sb.appendLine(" Threads: $maxThreads")
+ sb.appendLine(" ZIP parallelism:${if (zipParallelism > 0) " $zipParallelism" else " auto"}")
+ sb.appendLine(" ZIP order: ${if (largestFirst) "largest first" else "argument order"}")
sb.appendLine(" Log level: $logLevel")
// Annotation options
@@ -1070,6 +1097,21 @@
else -> outputFormat.name
}
+ internal fun effectiveZipParallelism(zipCount: Int): Int {
+ if (zipCount <= 0) return 1
+ val requested = if (zipParallelism > 0) zipParallelism else minOf(maxThreads, 8)
+ return requested.coerceIn(1, zipCount)
+ }
+
+ internal fun orderZipInputsForProcessing(zips: Array<String>): Array<String> =
+ if (largestFirst) {
+ zips.sortedByDescending { zipSizes[it] ?: 0L }.toTypedArray()
+ } else {
+ zips.copyOf()
+ }
+
+ internal fun streamingOutputClaimCount(): Int = streamingOutputClaims.values.sumOf { it.size }
+
internal fun canUseStaxTextParsing(): Boolean =
outputFormat == OutputFormat.CONLLU ||
outputFormat == OutputFormat.WORD2VEC ||
@@ -1171,6 +1213,12 @@
// Single priority-based executor for all entry processing
private var entryExecutor: java.util.concurrent.ExecutorService? = null
+ private var streamEntryPermits: java.util.concurrent.Semaphore? = null
+ private val streamEntriesInFlight = AtomicInteger(0)
+ private val streamEntriesPeak = AtomicInteger(0)
+ private val activeZipWorkers = AtomicInteger(0)
+ private val peakActiveZipWorkers = AtomicInteger(0)
+ private val textStreamWorkerNumber = AtomicInteger(0)
private val MONTH_ORDER = mapOf(
"JAN" to 1, "FEB" to 2, "MAR" to 3, "MRZ" to 3, "APR" to 4,
@@ -1389,6 +1437,11 @@
// Track which texts have been output to avoid counting duplicates (thread-safe)
val outputTexts: MutableSet<String> = ConcurrentHashMap.newKeySet()
+ // Plain streaming output only needs duplicate protection while one ZIP is open.
+ // Keeping every emitted ID in outputTexts made memory grow with total corpus size.
+ private val streamingOutputClaims: ConcurrentHashMap<String, MutableSet<String>> = ConcurrentHashMap()
+ private val streamingZipUsers: ConcurrentHashMap<String, AtomicInteger> = ConcurrentHashMap()
+
// Scheduled executor for periodic scanning
var incrementalOutputScheduler: java.util.concurrent.ScheduledExecutorService? = null
@@ -1413,8 +1466,31 @@
return if (File(baseZip).exists()) baseZip else null
}
+ internal fun planPlainSurfaceTextInputs(zips: Array<String>): Array<String> {
+ if ((outputFormat != OutputFormat.WORD2VEC && outputFormat != OutputFormat.NOW) || useLemma) {
+ return zips
+ }
+ val planned = LinkedHashMap<String, String>()
+ zips.forEach { input ->
+ val usefulInput = input.correspondingBaseZip() ?: input
+ val identity = try {
+ File(usefulInput).canonicalPath
+ } catch (_: Exception) {
+ File(usefulInput).absolutePath
+ }
+ planned.putIfAbsent(identity, usefulInput)
+ }
+ return planned.values.toTypedArray()
+ }
+
fun korapxml2conllu(args: Array<String>) {
outputTexts.clear()
+ streamingOutputClaims.clear()
+ streamingZipUsers.clear()
+ streamEntriesInFlight.set(0)
+ streamEntriesPeak.set(0)
+ activeZipWorkers.set(0)
+ peakActiveZipWorkers.set(0)
firstTextOutputLogged.set(false)
// Reset Krill state for fresh run (important for tests)
if (outputFormat == OutputFormat.KRILL) {
@@ -1468,12 +1544,16 @@
LOGGER.info("Initialized work-stealing scheduler with $maxThreads worker threads for Krill output")
} else if (canUseArchiveOrderTextStreaming()) {
entryExecutor = java.util.concurrent.Executors.newFixedThreadPool(maxThreads) { r ->
- Thread(r, "TextStreamWorker-${Thread.currentThread().threadId()}")
+ Thread(r, "TextStreamWorker-${textStreamWorkerNumber.incrementAndGet()}")
}
+ // One global budget bounds all open ZIPs together. The old per-ZIP budget
+ // grew quadratically with --threads (128 * 512 queued entries at -j 128).
+ val maxStreamEntries = maxOf(maxThreads * 2, 32)
+ streamEntryPermits = java.util.concurrent.Semaphore(maxStreamEntries, true)
val textParserMode = if (shouldParseDataXmlWithStax()) "StAX" else "DOM"
LOGGER.info(
"Initialized ${textStreamingModeLabel()} streaming mode: archive-order entries, parallel entry processing, " +
- "no text-ID scheduling, data.xml via $textParserMode"
+ "global in-flight limit=$maxStreamEntries, data.xml via $textParserMode"
)
} else {
// For other formats, use priority-based executor
@@ -1691,6 +1771,17 @@
LOGGER.info("Excluded $excluded of $before zip(s) by glob(s): ${excludeZipGlobs.joinToString(", ")}")
}
}
+ if (canUseArchiveOrderTextStreaming() && !useLemma) {
+ val before = zips.size
+ zips = planPlainSurfaceTextInputs(zips)
+ val skipped = before - zips.size
+ if (skipped > 0) {
+ LOGGER.info(
+ "Surface ${textStreamingModeLabel()} output reduced $before inputs to ${zips.size} unique base ZIP(s); " +
+ "annotation ZIPs do not affect surface tokens"
+ )
+ }
+ }
// Initialize zip progress tracking and sizes
startTimeMillis = System.currentTimeMillis()
processedZipBytes.set(0)
@@ -1702,10 +1793,10 @@
registerZipProgress(zip, try { File(zip).length() } catch (_: Exception) { 0L })
}
totalZipBytes = zipSizes.values.sum()
- // In lemma-only mode, process largest zips first
- if (lemmaOnly) {
- zips = zips.sortedByDescending { zipSizes[it] ?: 0L }.toTypedArray()
- }
+ // Keep caller-controlled order by default. Size-descending scheduling can
+ // improve worker utilization near the end, but may put one large corpus
+ // domain at the beginning of word2vec training data, so it is opt-in.
+ zips = orderZipInputsForProcessing(zips)
zips.forEachIndexed { index, zip -> zipOrdinals[zip] = index + 1 }
// Log zip order with sizes so the user can verify sorting
@@ -1752,9 +1843,13 @@
if (maxThreads > 1) {
val foundry = getFoundryFromZipFileNames(zips)
- val parallelism = maxThreads.coerceAtLeast(1)
+ val parallelism = effectiveZipParallelism(zips.size)
if (canUseArchiveOrderTextStreaming()) {
- LOGGER.info("Processing zips in ${textStreamingModeLabel()} streaming mode; zip parallelism=$parallelism; entry order=archive")
+ LOGGER.info(
+ "Processing zips in ${textStreamingModeLabel()} streaming mode; " +
+ "zip parallelism=$parallelism; entry threads=$maxThreads; " +
+ "ZIP order=${if (largestFirst) "largest first" else "argument order"}"
+ )
} else {
LOGGER.info("Processing zips with ordered queue; parallelism=$parallelism; entries ${if (sequentialInZip) "sequential" else "parallel"}")
}
@@ -1766,6 +1861,13 @@
}
}
+ if (canUseArchiveOrderTextStreaming()) {
+ LOGGER.info(
+ "Text streaming scheduler summary: peak ZIP workers=${peakActiveZipWorkers.get()}, " +
+ "peak entries in flight=${streamEntriesPeak.get()}, retained claims=${streamingOutputClaimCount()}"
+ )
+ }
+
} finally {
// Signal work-stealing scheduler that all foundries have been submitted
if (workStealingSchedulerActive) {
@@ -2027,6 +2129,8 @@
repeat(parallelism) {
executor.submit {
active.incrementAndGet()
+ val activeNow = activeZipWorkers.incrementAndGet()
+ peakActiveZipWorkers.accumulateAndGet(activeNow, ::maxOf)
try {
while (true) {
val zipPath = queue.poll(100, java.util.concurrent.TimeUnit.MILLISECONDS)
@@ -2048,6 +2152,7 @@
}
} finally {
active.decrementAndGet()
+ activeZipWorkers.decrementAndGet()
}
}
}
@@ -2863,14 +2968,16 @@
} else {
foundry // Keep original foundry for non-krill formats
}
+ val requireRelatedMorpho =
+ if (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) useLemma else true
if (useJavaZipForTextStreaming()) {
openJavaZipFile(zip).use { zipFile ->
LOGGER.info("Using ${textStreamingModeLabel()} streaming mode for $zip: archive-order entries, no text-ID sorting")
- processZipEntriesStreaming(zipFile, zip, zipFoundry, true)
+ processZipEntriesStreaming(zipFile, zip, zipFoundry, requireRelatedMorpho)
}
} else {
openZipFile(zip).use { zipFile ->
- processZipEntriesWithPool(zipFile, zip, zipFoundry, true)
+ processZipEntriesWithPool(zipFile, zip, zipFoundry, requireRelatedMorpho)
}
}
}
@@ -2929,10 +3036,12 @@
} else {
foundry // Keep original foundry for non-krill formats
}
+ val requireRelatedMorpho =
+ if (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) useLemma else true
if (useJavaZipForTextStreaming()) {
openJavaZipFile(zip).use { zipFile ->
LOGGER.info("Using ${textStreamingModeLabel()} streaming mode for $zip: archive-order entries, no text-ID sorting")
- processZipEntriesStreaming(zipFile, zip, zipFoundry, true)
+ processZipEntriesStreaming(zipFile, zip, zipFoundry, requireRelatedMorpho)
}
} else {
openZipFile(zip).use { zipFile ->
@@ -2941,7 +3050,7 @@
.filter { extractMetadataRegex.isNotEmpty() || !it.name.contains("header.xml") }
.sortedBy { getTextIdFromPath(it.name) }
.forEach { zipEntry ->
- processZipEntry(zipFile, zip, zipFoundry, zipEntry, true)
+ processZipEntry(zipFile, zip, zipFoundry, zipEntry, requireRelatedMorpho)
}
}
}
@@ -2990,10 +3099,18 @@
val pct = (done * 100.0 / total).coerceIn(0.0, 100.0)
val humanSpeed = String.format(Locale.ROOT, "%.2f MB/s", speedBytesPerSec / (1024.0 * 1024.0))
val etaStr = if (etaSeconds >= 0) formatDuration(etaSeconds) else "unknown"
+ val entryPool = entryExecutor as? java.util.concurrent.ThreadPoolExecutor
+ val schedulerState = if (canUseArchiveOrderTextStreaming()) {
+ ", workers{zip=${activeZipWorkers.get()},entry=${entryPool?.activeCount ?: 0}," +
+ "queued=${entryPool?.queue?.size ?: 0},inFlight=${streamEntriesInFlight.get()}," +
+ "claims=${streamingOutputClaims.values.sumOf { it.size }}}"
+ } else {
+ ""
+ }
LOGGER.info(
"Finished zip ${if (ord>0) ord else "?"}/$totalZips: ${zipFilePath} " +
"(${humanBytes(size)}). Progress: ${String.format(Locale.ROOT, "%.1f", pct)}%, " +
- "ETA ${etaStr} at ${humanSpeed}"
+ "ETA ${etaStr} at ${humanSpeed}$schedulerState"
)
} catch (e: Exception) {
LOGGER.fine("Failed to log zip progress for $zipFilePath: ${e.message}")
@@ -3042,74 +3159,118 @@
}
}
- private fun tryProcessReadyText(docId: String, foundry: String): Boolean {
- if (!outputTexts.add(docId)) return false
+ private fun tryProcessReadyText(zipPath: String, docId: String, foundry: String): Boolean {
+ val claims = if (canUseArchiveOrderTextStreaming()) {
+ streamingOutputClaims.computeIfAbsent(zipPath) { ConcurrentHashMap.newKeySet() }
+ } else {
+ outputTexts
+ }
+ if (!claims.add(docId)) return false
return try {
processText(docId, foundry)
noteFirstTextOutput(docId)
true
} catch (t: Throwable) {
- outputTexts.remove(docId)
+ claims.remove(docId)
+ throw t
+ }
+ }
+
+ private fun beginStreamingZip(zipPath: String) {
+ streamingZipUsers.computeIfAbsent(zipPath) { AtomicInteger(0) }.incrementAndGet()
+ }
+
+ private fun endStreamingZip(zipPath: String) {
+ val users = streamingZipUsers[zipPath] ?: return
+ if (users.decrementAndGet() == 0) {
+ streamingZipUsers.remove(zipPath, users)
+ streamingOutputClaims.remove(zipPath)
+ }
+ }
+
+ private fun isRelevantStreamingEntry(name: String, zipFoundry: String): Boolean {
+ if (name.contains("header.xml")) return extractMetadataRegex.isNotEmpty()
+ if (outputFormat != OutputFormat.WORD2VEC && outputFormat != OutputFormat.NOW) return true
+ return when {
+ name.endsWith("data.xml") -> !lemmaOnly
+ name.endsWith("tokens.xml") || name.endsWith("structure.xml") -> true
+ // A base ZIP may use morpho.xml as its tokenization layer. Surface-form
+ // output extracts only those spans and does not retain morpho features.
+ name.endsWith("morpho.xml") -> useLemma || zipFoundry == "base"
+ else -> false
+ }
+ }
+
+ private fun submitStreamingEntry(phaser: java.util.concurrent.Phaser, task: () -> Unit) {
+ val permits = streamEntryPermits
+ permits?.acquireUninterruptibly()
+ phaser.register()
+ val inFlight = streamEntriesInFlight.incrementAndGet()
+ streamEntriesPeak.accumulateAndGet(inFlight) { current, update -> maxOf(current, update) }
+ try {
+ entryExecutor!!.execute {
+ try {
+ task()
+ } finally {
+ streamEntriesInFlight.decrementAndGet()
+ permits?.release()
+ phaser.arriveAndDeregister()
+ }
+ }
+ } catch (t: Throwable) {
+ streamEntriesInFlight.decrementAndGet()
+ permits?.release()
+ phaser.arriveAndDeregister()
throw t
}
}
private fun processZipEntriesStreaming(zipFile: ApacheZipFile, zipPath: String, foundry: String, waitForMorpho: Boolean) {
- LOGGER.fine("Streaming NOW entries in archive order for $zipPath without text-ID sorting")
+ LOGGER.fine("Streaming ${textStreamingModeLabel()} entries in archive order for $zipPath")
val enumEntries = zipFile.entries
+ val zipFoundry = getFoundryFromZipFileName(zipPath)
val phaser = java.util.concurrent.Phaser(1)
- val maxInFlight = maxOf(maxThreads * 4, 32)
- val permits = java.util.concurrent.Semaphore(maxInFlight)
-
- while (enumEntries.hasMoreElements()) {
- val entry = enumEntries.nextElement()
- if (extractMetadataRegex.isEmpty() && entry.name.contains("header.xml")) continue
- if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
- permits.acquireUninterruptibly()
- phaser.register()
- entryExecutor!!.execute {
- try {
+ beginStreamingZip(zipPath)
+ try {
+ while (enumEntries.hasMoreElements()) {
+ val entry = enumEntries.nextElement()
+ if (!isRelevantStreamingEntry(entry.name, zipFoundry)) continue
+ if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
+ submitStreamingEntry(phaser) {
processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
- } finally {
- permits.release()
- phaser.arriveAndDeregister()
}
+ } else {
+ processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
}
- } else {
- processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
}
+ phaser.arriveAndAwaitAdvance()
+ } finally {
+ endStreamingZip(zipPath)
}
-
- phaser.arriveAndAwaitAdvance()
}
private fun processZipEntriesStreaming(zipFile: ZipFile, zipPath: String, foundry: String, waitForMorpho: Boolean) {
- LOGGER.fine("Streaming NOW entries in archive order for $zipPath without text-ID sorting")
+ LOGGER.fine("Streaming ${textStreamingModeLabel()} entries in archive order for $zipPath")
val enumEntries = zipFile.entries()
+ val zipFoundry = getFoundryFromZipFileName(zipPath)
val phaser = java.util.concurrent.Phaser(1)
- val maxInFlight = maxOf(maxThreads * 4, 32)
- val permits = java.util.concurrent.Semaphore(maxInFlight)
-
- while (enumEntries.hasMoreElements()) {
- val entry = enumEntries.nextElement()
- if (extractMetadataRegex.isEmpty() && entry.name.contains("header.xml")) continue
- if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
- permits.acquireUninterruptibly()
- phaser.register()
- entryExecutor!!.execute {
- try {
+ beginStreamingZip(zipPath)
+ try {
+ while (enumEntries.hasMoreElements()) {
+ val entry = enumEntries.nextElement()
+ if (!isRelevantStreamingEntry(entry.name, zipFoundry)) continue
+ if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
+ submitStreamingEntry(phaser) {
processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
- } finally {
- permits.release()
- phaser.arriveAndDeregister()
}
+ } else {
+ processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
}
- } else {
- processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
}
+ phaser.arriveAndAwaitAdvance()
+ } finally {
+ endStreamingZip(zipPath)
}
-
- phaser.arriveAndAwaitAdvance()
}
private fun processZipEntriesWithPool(zipFile: ApacheZipFile, zipPath: String, foundry: String, waitForMorpho: Boolean) {
@@ -3507,8 +3668,12 @@
}
"morpho.xml" -> {
- waitForMorpho = true
- fnames[docId] = zipEntry.name
+ if (useLemma || (outputFormat != OutputFormat.WORD2VEC && outputFormat != OutputFormat.NOW)) {
+ waitForMorpho = true
+ }
+ val surfaceTextOutput =
+ (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma
+ if (!surfaceTextOutput) fnames[docId] = zipEntry.name
LOGGER.info("Processing morpho.xml for $docId with foundry=$annotationFoundry from ${zipEntry.name}")
val fsSpans: NodeList = doc.getElementsByTagName("span")
val morphoSpans = extractMorphoSpans(fsSpans)
@@ -3522,6 +3687,13 @@
tokens[docId] = morphoTokens
collectKrillTokensFromMorpho(docId, morphoFoundry, morphoTokens)
}
+ } else if ((outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma) {
+ // Surface-form text output needs morpho.xml only as a token-span
+ // fallback for base ZIPs without tokens.xml. Do not retain the much
+ // larger lemma/POS map after the text may already have been emitted.
+ if (_foundry == "base" && tokens[docId] == null) {
+ tokens[docId] = extractSpans(fsSpans, docId)
+ }
} else {
// For other formats, use the shared morpho map
// Merge with existing morpho data (e.g., from dependency.xml)
@@ -3646,7 +3818,7 @@
&& (extractMetadataRegex.isEmpty() || metadata[docId] != null)
) {
LOGGER.fine("All data ready for $docId, calling processText")
- tryProcessReadyText(docId, annotationFoundry)
+ tryProcessReadyText(zipPath, docId, annotationFoundry)
} else {
LOGGER.fine("NOT ready to process $docId yet: textOK=${texts[docId] != null || !textRequired}, sentencesOK=${sentences[docId] != null}, tokensOK=${tokens[docId] != null}, morphoOK=${!morphoRequired || morpho[docId] != null}")
}
@@ -3716,7 +3888,7 @@
&& (!morphoRequired || morpho[docId] != null)
) {
LOGGER.info("Processing text (meta-ready): $docId in thread ${Thread.currentThread().threadId()}")
- tryProcessReadyText(docId, foundry)
+ tryProcessReadyText(zipPath, docId, foundry)
}
}
}
@@ -3786,7 +3958,14 @@
}
}
"morpho.xml" -> {
- fnames[docId] = zipEntry.name
+ val surfaceTextOutput =
+ (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma
+ if (surfaceTextOutput && tokens[docId] != null) {
+ // Standard base ZIPs already supplied tokens.xml. Avoid parsing and
+ // allocating an unused lemma/POS map from a later morpho layer.
+ return
+ }
+ if (!surfaceTextOutput) fnames[docId] = zipEntry.name
val (morphoSpans, allSpans) = extractMorphoSpansStax(reader)
if (outputFormat == OutputFormat.KRILL) {
@@ -3796,6 +3975,11 @@
tokens[docId] = allSpans
collectKrillTokensFromMorpho(docId, morphoFoundry, allSpans)
}
+ } else if (surfaceTextOutput) {
+ // See the DOM path above: surface output uses only token offsets.
+ if (_foundry == "base" && tokens[docId] == null) {
+ tokens[docId] = allSpans
+ }
} else {
val morphoMap = synchronized(morpho) {
morpho.getOrPut(docId) { morphoSpans }
@@ -3854,7 +4038,9 @@
processedTextsPerZip.getOrPut(zipPath) { mutableSetOf() }.add(docId)
}
- val effectiveWaitForMorpho = if (fileName == "morpho.xml") true else waitForMorpho
+ val surfaceTextOutput =
+ (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma
+ val effectiveWaitForMorpho = if (fileName == "morpho.xml" && !surfaceTextOutput) true else waitForMorpho
val finalMorphoRequired = when {
taggerName != null || parserName != null -> false
@@ -3875,7 +4061,7 @@
&& (!finalMorphoRequired || morpho[docId] != null)
&& (extractMetadataRegex.isEmpty() || metadata[docId] != null)
) {
- tryProcessReadyText(docId, annotationFoundry)
+ tryProcessReadyText(zipPath, docId, annotationFoundry)
}
} catch (e: Exception) {
@@ -3956,7 +4142,18 @@
if (outputFormat == OutputFormat.KORAP_XML && annotationWorkerPool == null) {
formatKorapXmlOutput(getMorphoFoundry(), docId)
} else {
- formatConlluOutput(foundry, docId)
+ // In archive-order mode, data/structure/morpho entries for one text
+ // finish concurrently. Derive an input annotation foundry from the
+ // retained annotation filename so the last base-layer task cannot
+ // nondeterministically relabel custom morpho data as "base".
+ val inputFile = fnames[docId]
+ val outputFoundry = if (inputFile != null &&
+ (inputFile.endsWith("morpho.xml") || inputFile.endsWith("dependency.xml"))) {
+ annotationFoundryFor(inputFile, File(inputFile).name, foundry)
+ } else {
+ foundry
+ }
+ formatConlluOutput(outputFoundry, docId)
}
}
@@ -4142,7 +4339,9 @@
val max = rt.maxMemory() / (1024 * 1024)
LOGGER.info(
"MEM-STATS docs=${count} usedMB=${used} totalMB=${total} maxMB=${max} " +
- "maps{texts=${texts.size},tokens=${tokens.size},sentences=${sentences.size},morpho=${morpho.size}}"
+ "maps{texts=${texts.size},tokens=${tokens.size},sentences=${sentences.size},morpho=${morpho.size}} " +
+ "stream{zipActive=${activeZipWorkers.get()},entriesInFlight=${streamEntriesInFlight.get()}," +
+ "entriesPeak=${streamEntriesPeak.get()},claims=${streamingOutputClaims.values.sumOf { it.size }}}"
)
} catch (e: Exception) {
LOGGER.warning("Failed to log memory stats: ${e.message}")
diff --git a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt
index 5a7bf73..e307009 100644
--- a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt
+++ b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt
@@ -158,6 +158,57 @@
}
@Test
+ fun textStreamingSeparatesZipAndEntryParallelism() {
+ val tool = KorapXmlTool()
+ tool.maxThreads = 128
+
+ assertEquals(8, tool.effectiveZipParallelism(2600))
+ tool.zipParallelism = 3
+ assertEquals(3, tool.effectiveZipParallelism(2600))
+ assertEquals(2, tool.effectiveZipParallelism(2))
+ }
+
+ @Test
+ fun zipParallelismMustBePositive() {
+ val exitCode = debug(arrayOf(
+ "-t", "w2v", "--zip-parallelism", "0", loadResource("wdf19.zip").path
+ ))
+
+ assertTrue(exitCode != 0)
+ assertContains(errContent.toString(), "--zip-parallelism': must be at least 1")
+ }
+
+ @Test
+ fun zipSchedulingKeepsArgumentOrderByDefaultAndLargestFirstIsOptIn() {
+ val tool = KorapXmlTool()
+ val small = "/tmp/small.zip"
+ val large = "/tmp/large.zip"
+ tool.registerZipProgress(small, 10L)
+ tool.registerZipProgress(large, 100L)
+
+ assertEquals(
+ listOf(small, large),
+ tool.orderZipInputsForProcessing(arrayOf(small, large)).toList()
+ )
+
+ tool.largestFirst = true
+ assertEquals(
+ listOf(large, small),
+ tool.orderZipInputsForProcessing(arrayOf(small, large)).toList()
+ )
+ }
+
+ @Test
+ fun surfaceWord2VecUsesEachBaseZipOnlyOnce() {
+ val tool = KorapXmlTool()
+ tool.outputFormat = OutputFormat.WORD2VEC
+ val base = loadResource("goe.zip").path
+ val annotation = loadResource("goe.tree_tagger.zip").path
+
+ assertEquals(listOf(base), tool.planPlainSurfaceTextInputs(arrayOf(annotation, base)).toList())
+ }
+
+ @Test
fun singleBaseConlluOutputCanUseArchiveOrderStreaming() {
val tool = KorapXmlTool()
tool.outputFormat = OutputFormat.CONLLU
diff --git a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt
index 0e676bc..507bdc3 100644
--- a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt
+++ b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt
@@ -5,6 +5,7 @@
import java.io.ByteArrayOutputStream
import java.io.PrintStream
import java.net.URL
+import picocli.CommandLine
import kotlin.test.Test
import kotlin.test.assertContains
import kotlin.test.assertFalse
@@ -116,4 +117,29 @@
assertContains(out, "automatique")
assertFalse(out.contains("Gedanken"))
}
+
+ @Test
+ fun streamingClaimsAreReleasedAfterZipCompletes() {
+ val tool = KorapXmlTool()
+ val exitCode = CommandLine(tool).execute(
+ "-t", "w2v", "-j", "4", loadResource("wdf19.zip").path
+ )
+
+ assertTrue(exitCode == 0)
+ assertTrue(tool.outputTexts.isEmpty(), "Plain w2v must not retain all emitted document IDs")
+ assertTrue(tool.streamingOutputClaimCount() == 0, "Per-ZIP claims must be released when the ZIP closes")
+ }
+
+ @Test
+ fun surfaceWord2VecSupportsMorphoTokenizationWithoutRetainingMorpho() {
+ val tool = KorapXmlTool()
+ val exitCode = CommandLine(tool).execute(
+ "-t", "w2v", "-j", "2", loadResource("dck_sample.zip").path
+ )
+
+ assertTrue(exitCode == 0)
+ assertTrue(outContent.size() > 0, "Custom morpho tokenization should produce surface output")
+ assertTrue(tool.morpho.isEmpty(), "Surface w2v must not retain unused morpho maps")
+ assertTrue(tool.fnames.isEmpty(), "Late optional layers must not repopulate per-document state")
+ }
}