diff --git a/src/sw_models/EflomalAlignmentModel.cc b/src/sw_models/EflomalAlignmentModel.cc index adc84717..c6379f15 100644 --- a/src/sw_models/EflomalAlignmentModel.cc +++ b/src/sw_models/EflomalAlignmentModel.cc @@ -26,6 +26,7 @@ EflomalAlignmentModel::EflomalAlignmentModel() ch.lexTable = lexTable; ch.jumpTable = jumpTable; ch.fertilityTable = fertilityTable; + buildDecodeCache(ch); // decodeJumpLin must be non-empty for decode to work pre-training chains.push_back(std::move(ch)); } @@ -239,6 +240,13 @@ unsigned int EflomalAlignmentModel::startTraining(int /*verbosity*/) { clearTempVars(); trainingAlignments.clear(); // recomputed in endTraining if emitTrainingAlignments + // getSrcSent/getTrgSent (called from buildCorpus, below) intern unknown words via + // addSrcSymbol/addTrgSymbol - a mutating write to the shared vocab. buildCorpus is + // the ONLY place that happens; every #pragma omp parallel for in this class (train, + // endTraining, and the per-chain loop below) runs strictly after this call returns, + // so the vocab is read-only (safe to query concurrently via getSrcVocabSize/ + // getTrgVocabSize) for the rest of a training run. This invariant is load-bearing + // for those parallel regions and isn't otherwise enforced by any type or lock. buildCorpus(); if (autoIterations) { @@ -269,6 +277,7 @@ unsigned int EflomalAlignmentModel::startTraining(int /*verbosity*/) size_t buckets = size_t{2} * jumpWindow + 1; chains.resize((size_t)numSamplers); +#pragma omp parallel for schedule(static) if(!deterministic) for (int s = 0; s < numSamplers; ++s) { SamplerChain& chain = chains[(size_t)s]; @@ -278,16 +287,19 @@ unsigned int EflomalAlignmentModel::startTraining(int /*verbosity*/) chain.lexCountSum.assign(srcVocabSize, 0); chain.jumpCounts.assign(buckets, 0); chain.jumpCountSum = 0; - chain.fertCounts.assign(srcVocabSize, vector(MaxFertility + 1, 0)); + chain.fertCounts.assign(srcVocabSize * (MaxFertility + 1), 0.0); chain.fertCountSum.assign(srcVocabSize, 0); chain.fertRatioSampled.clear(); chain.lexPriorMass = chain.jumpPriorMass = chain.fertPriorMass = 0; - chain.accumLexCounts.assign(srcVocabSize, {}); chain.accumLexCountSum.assign(srcVocabSize, 0); chain.accumulated = false; chain.lexTable = make_shared(); chain.jumpTable = make_shared(); chain.fertilityTable = make_shared(); + // Keep decodeJumpLin/decodeFertRatio in sync with the freshly-reset (empty) + // tables, in case a caller queries/decodes mid-training; normalizeChain rebuilds + // these again once real counts are trained into the tables at endTraining. + buildDecodeCache(chain); initializeChain(chain); } @@ -351,7 +363,7 @@ void EflomalAlignmentModel::initializeChain(SamplerChain& chain) chain.alig[n][j - 1] = i; WordIndex e = nsrc[i]; WordIndex f = trg[j - 1]; - chain.lexCounts[e][f] += 1; + chain.lexCounts[e][f].live += 1; chain.lexCountSum[e] += 1; } } @@ -402,21 +414,21 @@ void EflomalAlignmentModel::computeJumpCounts(SamplerChain& chain) void EflomalAlignmentModel::computeFertilityCounts(SamplerChain& chain) { - for (auto& row : chain.fertCounts) - fill(row.begin(), row.end(), 0.0); + fill(chain.fertCounts.begin(), chain.fertCounts.end(), 0.0); fill(chain.fertCountSum.begin(), chain.fertCountSum.end(), 0.0); + vector phi; for (size_t n = 0; n < corpusSrc.size(); ++n) { const vector& nsrc = corpusSrc[n]; PositionIndex slen = (PositionIndex)nsrc.size() - 1; - vector phi(slen + 1, 0); + phi.assign(slen + 1, 0); for (PositionIndex j = 1; j <= (PositionIndex)corpusTrg[n].size(); ++j) phi[chain.alig[n][j - 1]]++; for (PositionIndex i = 1; i <= slen; ++i) { WordIndex e = nsrc[i]; - chain.fertCounts[e][std::min(phi[i], (int)MaxFertility)] += 1; + chain.fertCounts[(size_t)e * (MaxFertility + 1) + std::min(phi[i], (int)MaxFertility)] += 1; chain.fertCountSum[e] += 1; } } @@ -425,24 +437,25 @@ void EflomalAlignmentModel::computeFertilityCounts(SamplerChain& chain) void EflomalAlignmentModel::sampleFertilityRatios(SamplerChain& chain) { size_t srcVocabSize = getSrcVocabSize(); - chain.fertRatioSampled.assign(srcVocabSize, vector(MaxFertility + 1, 1.0)); + chain.fertRatioSampled.assign(srcVocabSize * (MaxFertility + 1), 1.0); vector draw(MaxFertility + 1); for (WordIndex s = 0; s < (WordIndex)srcVocabSize; ++s) { if (chain.fertCountSum[s] <= 0) continue; + size_t base = (size_t)s * (MaxFertility + 1); // Draw an unnormalized categorical fertility distribution from the Dirichlet // posterior alpha[phi] = FERT_ALPHA + count(s, phi) via gamma variates. for (PositionIndex phi = 0; phi <= MaxFertility; ++phi) { - std::gamma_distribution gamma(alphaFertility + chain.fertCounts[s][phi], 1.0); + std::gamma_distribution gamma(alphaFertility + chain.fertCounts[base + phi], 1.0); draw[phi] = gamma(chain.rng); } // Store the ratio P(phi)/P(phi-1); reaching the capped maximum fertility is // made very unlikely. - chain.fertRatioSampled[s][MaxFertility] = 1e-10; + chain.fertRatioSampled[base + MaxFertility] = 1e-10; for (PositionIndex phi = MaxFertility - 1; phi >= 1; --phi) - chain.fertRatioSampled[s][phi] = draw[phi - 1] > 0 ? draw[phi] / draw[phi - 1] : 0.0; + chain.fertRatioSampled[base + phi] = draw[phi - 1] > 0 ? draw[phi] / draw[phi - 1] : 0.0; } } @@ -485,17 +498,20 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a // Fully collapsed Gibbs: the lexical AND jump counts are updated per token within // the sweep, so every token is resampled against the current state of all tokens - // already resampled this sweep. - vector lexDenomInv(chain.lexCountSum.size()); + // already resampled this sweep. chain.lexDenomInv is a persistent member (see + // SamplerChain) reused via assign() every sweep instead of a fresh vocab-sized + // heap allocation each of the 84+ sweeps per chain. + chain.lexDenomInv.assign(chain.lexCountSum.size(), 0.0f); auto recomputeInv = [&](WordIndex e) { double d = eflomalLexNorm ? std::max(chain.lexCountSum[e], 1.0) : chain.lexCountSum[e] + chain.lexPriorMass; - lexDenomInv[e] = (float)(1.0 / d); + chain.lexDenomInv[e] = (float)(1.0 / d); }; - for (size_t e = 0; e < lexDenomInv.size(); ++e) + for (size_t e = 0; e < chain.lexDenomInv.size(); ++e) recomputeInv((WordIndex)e); vector weights; vector aaRight; + vector phi; for (size_t n = 0; n < corpusSrc.size(); ++n) { @@ -504,7 +520,6 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a PositionIndex slen = (PositionIndex)nsrc.size() - 1; PositionIndex tlen = (PositionIndex)trg.size(); - vector phi; if (stage == FertilityStage) { phi.assign(slen + 1, 0); @@ -534,14 +549,24 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a WordIndex eOld = nsrc[iOld]; int aR = aaRight[j - 1]; + // f is fixed for this whole token (remove, all slen+1 candidate lookups, and + // add all key on f); hash it once and reuse via robin_map's precalculated_hash + // overloads instead of rehashing the same key up to slen+3 times. + const std::size_t hF = WordIndexHash{}(f); + // Remove the current token's lexical contribution (collapsed Gibbs). - auto itOld = chain.lexCounts[eOld].find(f); + auto itOld = chain.lexCounts[eOld].find(f, hF); if (itOld != chain.lexCounts[eOld].end()) { // tsl::robin_map requires value() (not ->second) to mutate the mapped value. - itOld.value() -= 1; - if (itOld.value() <= 0) - chain.lexCounts[eOld].erase(itOld); + itOld.value().live -= 1; + // Only erase once BOTH fields are spent: accum (the fertility-stage + // variance-reduction marginal, LexEntry::accum) must survive a token + // temporarily leaving this (e,f) pair mid-sweep even though live hits 0. + if (itOld.value().live <= 0 && itOld.value().accum <= 0) + // erase_fast: skips computing the next-iterator that plain erase() must + // return, which this call site discards anyway. + chain.lexCounts[eOld].erase_fast(itOld); } chain.lexCountSum[eOld] -= 1; recomputeInv(eOld); @@ -579,10 +604,10 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a { WordIndex e = nsrc[i]; double count = 0; - auto it = chain.lexCounts[e].find(f); + auto it = chain.lexCounts[e].find(f, hF); if (it != chain.lexCounts[e].end()) - count = it->second; - double w = (count + alphaLex) * lexDenomInv[e]; + count = it->second.live; + double w = (count + alphaLex) * chain.lexDenomInv[e]; if (stage == Ibm1Stage) { w *= i == 0 ? p0 : (1.0 - p0) / slen; @@ -600,7 +625,8 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a double j2 = std::max(chain.jumpCounts[jumpBucket(aR - pos0)] + alphaJump, kMinJumpWeight); w *= j1 * j2; if (stage == FertilityStage) - w *= chain.fertRatioSampled[e][std::min((PositionIndex)(phi[i] + 1), MaxFertility)]; + w *= chain.fertRatioSampled[(size_t)e * (MaxFertility + 1) + + std::min((PositionIndex)(phi[i] + 1), MaxFertility)]; } weights[i] = w; total += w; @@ -610,7 +636,25 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a chain.alig[n][j - 1] = iNew; WordIndex eNew = nsrc[iNew]; - chain.lexCounts[eNew][f] += 1; + { + auto& mapNew = chain.lexCounts[eNew]; + auto itNew = mapNew.find(f, hF); + if (itNew != mapNew.end()) + { + itNew.value().live += 1; + // Piggyback the accumulate update on the same slot/iterator instead of a + // second lookup into a separate map (see LexEntry / normalizeChain). + if (accumulate) + itNew.value().accum += 1; + } + else + { + LexEntry& entry = mapNew[f]; + entry.live = 1; + if (accumulate) + entry.accum = 1; + } + } chain.lexCountSum[eNew] += 1; recomputeInv(eNew); if (stage == FertilityStage) @@ -634,10 +678,7 @@ void EflomalAlignmentModel::sampleSweep(SamplerChain& chain, Stage stage, bool a } if (accumulate) - { - chain.accumLexCounts[eNew][f] += 1; chain.accumLexCountSum[eNew] += 1; - } if (iNew > 0) aaLeft = (int)iNew - 1; @@ -705,14 +746,17 @@ void EflomalAlignmentModel::normalizeChain(SamplerChain& chain) { // Lexical counts are the variance-reduced marginal accumulated over the final // stage; the jump and fertility distributions are read off the final alignment. - const vector& lex = chain.accumulated ? chain.accumLexCounts : chain.lexCounts; + // live/accum share one map (LexEntry) now, so select the field per chain.accumulated + // instead of selecting between two separate maps. + const vector& lex = chain.lexCounts; const vector& lexSum = chain.accumulated ? chain.accumLexCountSum : chain.lexCountSum; + bool useAccum = chain.accumulated; computeJumpCounts(chain); computeFertilityCounts(chain); const vector& jumps = chain.jumpCounts; double jumpSum = chain.jumpCountSum; - const vector>& fert = chain.fertCounts; + const vector& fert = chain.fertCounts; const vector& fertSum = chain.fertCountSum; double trgVocabSize = (double)getTrgVocabSize(); @@ -726,7 +770,12 @@ void EflomalAlignmentModel::normalizeChain(SamplerChain& chain) double denom = lexSum[s] + alphaLex * trgVocabSize; chain.lexTable->setDenominator(s, (float)log(denom)); for (const auto& entry : lex[s]) - chain.lexTable->setNumerator(s, entry.first, (float)log(entry.second + alphaLex)); + { + float value = useAccum ? entry.second.accum : entry.second.live; + if (value <= 0) + continue; + chain.lexTable->setNumerator(s, entry.first, (float)log(value + alphaLex)); + } } // Jump table p(delta). @@ -735,14 +784,52 @@ void EflomalAlignmentModel::normalizeChain(SamplerChain& chain) // Fertility table p(phi|s). chain.fertilityTable->clear(); double fertNorm = alphaFertility * (double)(MaxFertility + 1); - for (WordIndex s = 0; s < (WordIndex)fert.size(); ++s) + for (WordIndex s = 0; s < (WordIndex)fertSum.size(); ++s) { if (fertSum[s] <= 0) continue; double denom = fertSum[s] + fertNorm; chain.fertilityTable->setDenominator(s, (float)log(denom)); + size_t base = (size_t)s * (MaxFertility + 1); + for (PositionIndex phi = 0; phi <= MaxFertility; ++phi) + chain.fertilityTable->setNumerator(s, phi, (float)log(fert[base + phi] + alphaFertility)); + } + + buildDecodeCache(chain); +} + +// Derives decodeJumpLin/decodeFertRatio from the now-trained jumpTable/fertilityTable +// once, so decodeMarginalFromTables never rebuilds them per sentence/per call - the +// prior implementation recomputed both from scratch on every decode call, which +// dominated computeTrainingAlignments' cost (called once per training pair per chain). +void EflomalAlignmentModel::buildDecodeCache(SamplerChain& chain) +{ + chain.decodeJumpLin.assign((size_t)2 * jumpWindow + 1, 0.0); + for (int b = 0; b < (int)chain.decodeJumpLin.size(); ++b) + chain.decodeJumpLin[b] = chain.jumpTable->prob(b - jumpWindow); + + auto fertProb = [&](WordIndex s, PositionIndex phi) -> double { + bool found; + double numer = chain.fertilityTable->getNumerator(s, std::min(phi, MaxFertility), found); + if (!found) + return SmoothingProb; + double denom = chain.fertilityTable->getDenominator(s, found); + if (!found) + return SmoothingProb; + return std::exp(numer - denom); + }; + + size_t srcVocabSize = getSrcVocabSize(); + chain.decodeFertRatio.assign(srcVocabSize * (MaxFertility + 1), 1.0); + for (WordIndex e = 0; e < (WordIndex)srcVocabSize; ++e) + { + size_t base = (size_t)e * (MaxFertility + 1); for (PositionIndex phi = 0; phi <= MaxFertility; ++phi) - chain.fertilityTable->setNumerator(s, phi, (float)log(fert[s][phi] + alphaFertility)); + { + double cur = fertProb(e, phi); + double nxt = fertProb(e, std::min((PositionIndex)(phi + 1), MaxFertility)); + chain.decodeFertRatio[base + phi] = cur > 0 ? nxt / cur : 1.0; + } } } @@ -820,8 +907,8 @@ LgProb EflomalAlignmentModel::hmmAlignmentLogProb(PositionIndex prevI, PositionI return log(1.0 - p0) + jumpTable->logProb((int)i - (int)prevI); } -void EflomalAlignmentModel::decodeMarginalFromTables(const MemoryLexTable& lex, const EflomalJumpTable& jump, - const FertilityTable& fert, unsigned int chainSeed, +void EflomalAlignmentModel::decodeMarginalFromTables(const MemoryLexTable& lex, const vector& jumpLin, + const vector& fertRatioTable, unsigned int chainSeed, const vector& nsrc, const vector& trg, vector>& acc, const vector* warmStart) @@ -832,7 +919,10 @@ void EflomalAlignmentModel::decodeMarginalFromTables(const MemoryLexTable& lex, if (tlen == 0 || slen == 0) return; - // Helpers to read table probabilities without going through virtual dispatch. + // Helper to read table probabilities without going through virtual dispatch. + // jumpLin/fertRatioTable are the chain's precomputed decodeJumpLin/decodeFertRatio + // (see buildDecodeCache) - they depend only on the trained tables, not on this + // sentence, so unlike lexProb they are never recomputed here. auto lexProb = [&](WordIndex s, WordIndex t) -> double { bool found; double numer = lex.getNumerator(s, t, found); @@ -843,48 +933,28 @@ void EflomalAlignmentModel::decodeMarginalFromTables(const MemoryLexTable& lex, return SmoothingProb; return std::exp(numer - denom); }; - auto fertProb = [&](WordIndex s, PositionIndex phi) -> double { - bool found; - double numer = fert.getNumerator(s, std::min(phi, MaxFertility), found); - if (!found) - return SmoothingProb; - double denom = fert.getDenominator(s, found); - if (!found) - return SmoothingProb; - return std::exp(numer - denom); - }; - // Precompute the lexical emissions p(f|e), the linearized jump distribution and - // per-source fertility ratios so the inner sampling loop is free of exp/log. - // Source positions are 1..slen (nsrc[p]); 0-based source index is p-1. The - // nearest-non-NULL neighbours use a virtual BOS at -1 and EOS at slen. - vector> emission(slen + 1, vector(tlen)); + // Precompute the lexical emissions p(f|e) so the inner sampling loop is free of + // exp/log. Source positions are 1..slen (nsrc[p]); 0-based source index is p-1. + // thread_local: reused across calls on the same OS thread (safe under OpenMP, + // one instance per thread) instead of a fresh heap allocation per call; sized via + // assign() so capacity carries over rather than reallocating every sentence. + thread_local vector emission; + emission.assign((size_t)(slen + 1) * tlen, 0.0); for (PositionIndex p = 0; p <= slen; ++p) for (PositionIndex j = 0; j < tlen; ++j) - emission[p][j] = lexProb(nsrc[p], trg[j]); - - vector jumpLin((size_t)2 * jumpWindow + 1); - for (int b = 0; b < (int)jumpLin.size(); ++b) - jumpLin[b] = (double)jump.prob(b - jumpWindow); - - vector> fertRatio(slen + 1, vector(MaxFertility + 1, 1.0)); - for (PositionIndex p = 1; p <= slen; ++p) - { - WordIndex e = nsrc[p]; - for (PositionIndex phi = 0; phi <= MaxFertility; ++phi) - { - double cur = fertProb(e, phi); - double nxt = fertProb(e, std::min((PositionIndex)(phi + 1), MaxFertility)); - fertRatio[p][phi] = cur > 0 ? nxt / cur : 1.0; - } - } + emission[(size_t)p * tlen + j] = lexProb(nsrc[p], trg[j]); // acc[j][k] (the output) accumulates the marginal: k in [0, slen-1] -> source // position k+1, k == slen -> NULL. - vector links(tlen, 0); - vector phi(slen + 1, 0); // per-source fertility counts within this decode chain - vector aaRight(tlen, 0); - vector ps(slen + 1, 0.0); + thread_local vector links; + thread_local vector phi; // per-source fertility counts within this decode chain + thread_local vector aaRight; + thread_local vector ps; + links.assign(tlen, 0); + phi.assign(slen + 1, 0); + aaRight.assign(tlen, 0); + ps.assign(slen + 1, 0.0); // Local, deterministic RNG seeded per chain: decode is reproducible per pair // and thread-safe (no shared state with other threads or model fields). @@ -944,12 +1014,21 @@ void EflomalAlignmentModel::decodeMarginalFromTables(const MemoryLexTable& lex, for (PositionIndex p = 1; p <= slen; ++p) { int pos0 = (int)p - 1; - double w = emission[p][j] * jumpLin[jumpBucket(pos0 - aL)] * jumpLin[jumpBucket(aR - pos0)] - * fertRatio[p][std::min(phi[p], (int)MaxFertility)]; + // Bounds-check: fertRatioTable is sized to the vocab as of the last + // buildDecodeCache call, but vocab can grow afterward (addSrcSymbol can be + // called post-training - see oovWordsReturnSmoothedProb). A word added since + // falls back to ratio 1.0, matching what the original per-call fertProb + // lookup returned for an unseen word (SmoothingProb/SmoothingProb == 1.0). + size_t fertBase = (size_t)nsrc[p] * (MaxFertility + 1); + double fertRatio = fertBase < fertRatioTable.size() + ? fertRatioTable[fertBase + std::min(phi[p], (int)MaxFertility)] + : 1.0; + double w = emission[(size_t)p * tlen + j] * jumpLin[jumpBucket(pos0 - aL)] * jumpLin[jumpBucket(aR - pos0)] + * fertRatio; sum += w; ps[p - 1] = sum; } - sum += p0 * emission[0][j] * jumpLin[jumpBucket(aR - aL)]; + sum += p0 * emission[(size_t)0 * tlen + j] * jumpLin[jumpBucket(aR - aL)]; ps[slen] = sum; if (accumulate && sum > 0.0) @@ -1008,7 +1087,7 @@ void EflomalAlignmentModel::sampleDecode(const vector& nsrc, const ve for (const auto& chain : chains) { vector> chainAcc; - decodeMarginalFromTables(*chain.lexTable, *chain.jumpTable, *chain.fertilityTable, chain.chainSeed, nsrc, trg, + decodeMarginalFromTables(*chain.lexTable, chain.decodeJumpLin, chain.decodeFertRatio, chain.chainSeed, nsrc, trg, chainAcc); if (acc.empty()) { @@ -1035,7 +1114,7 @@ void EflomalAlignmentModel::accumulateDecodeMarginal(const vector& sr for (const auto& chain : chains) { vector> acc; - decodeMarginalFromTables(*chain.lexTable, *chain.jumpTable, *chain.fertilityTable, chain.chainSeed, nsrc, + decodeMarginalFromTables(*chain.lexTable, chain.decodeJumpLin, chain.decodeFertRatio, chain.chainSeed, nsrc, trgSentence, acc); if (accOut.empty()) { @@ -1065,7 +1144,7 @@ void EflomalAlignmentModel::accumulateWarmDecodeMarginal(size_t n, vector= chain.alig.size()) return; vector> acc; - decodeMarginalFromTables(*chain.lexTable, *chain.jumpTable, *chain.fertilityTable, chain.chainSeed, nsrc, trg, acc, + decodeMarginalFromTables(*chain.lexTable, chain.decodeJumpLin, chain.decodeFertRatio, chain.chainSeed, nsrc, trg, acc, &chain.alig[n]); if (accOut.empty()) { @@ -1152,16 +1231,32 @@ bool EflomalAlignmentModel::getEntriesForSource(WordIndex s, NbestTableNode EflomalAlignmentModel::loglikelihoodForPairRange(pair sentPairRange, int verbosity) { - double loglikelihood = 0; - unsigned int numSents = 0; - for (unsigned int n = sentPairRange.first; n <= sentPairRange.second; ++n) + unsigned int first = sentPairRange.first; + unsigned int last = sentPairRange.second; + if (last < first) + return make_pair(0.0, 0.0); + unsigned int numSents = last - first + 1; + + // getSrcSent/getTrgSent intern unknown words into the shared vocab via + // addSrcSymbol/addTrgSymbol (see the concurrency-contract note on the class and at + // startTraining's buildCorpus() call) - a write that must happen serially, before + // the parallel scoring loop below only reads that (now-stable) vocab and the + // trained tables. computeSumLogProb -> sampleDecode/scoreAlignment touch no shared + // mutable state (per-call local RNG/vectors, read-only table lookups), so scoring + // different pairs concurrently is safe once the pre-conversion pass above completes. + vector> srcs(numSents), trgs(numSents); + for (unsigned int i = 0; i < numSents; ++i) { - vector src = getSrcSent(n); - vector trg = getTrgSent(n); - loglikelihood += (double)computeSumLogProb(src, trg, verbosity); - ++numSents; + srcs[i] = getSrcSent(first + i); + trgs[i] = getTrgSent(first + i); } - return make_pair(loglikelihood, numSents == 0 ? 0 : loglikelihood / (double)numSents); + + double loglikelihood = 0; +#pragma omp parallel for schedule(dynamic, 16) reduction(+ : loglikelihood) if(!deterministic) + for (int i = 0; i < (int)numSents; ++i) + loglikelihood += (double)computeSumLogProb(srcs[i], trgs[i], verbosity); + + return make_pair(loglikelihood, loglikelihood / (double)numSents); } void EflomalAlignmentModel::loadConfig(const YAML::Node& config) @@ -1256,6 +1351,7 @@ bool EflomalAlignmentModel::load(const char* prefFileName, int verbose) return THOT_ERROR; if (chains[(size_t)s].fertilityTable->load((pref + ".efl_fertnd" + suffix).c_str(), verbose) == THOT_ERROR) return THOT_ERROR; + buildDecodeCache(chains[(size_t)s]); } // Wire model-level query tables to chain[0]. @@ -1318,13 +1414,13 @@ void EflomalAlignmentModel::clearTempVars() chain.alig.clear(); chain.lexCounts.clear(); chain.lexCountSum.clear(); + chain.lexDenomInv.clear(); chain.jumpCounts.clear(); chain.jumpCountSum = 0; chain.fertCounts.clear(); chain.fertCountSum.clear(); chain.fertRatioSampled.clear(); chain.lexPriorMass = chain.jumpPriorMass = chain.fertPriorMass = 0; - chain.accumLexCounts.clear(); chain.accumLexCountSum.clear(); chain.accumulated = false; } @@ -1343,6 +1439,7 @@ void EflomalAlignmentModel::clear() ch.lexTable = lexTable; ch.jumpTable = jumpTable; ch.fertilityTable = fertilityTable; + buildDecodeCache(ch); chains.push_back(std::move(ch)); corpusSrc.clear(); corpusTrg.clear(); diff --git a/src/sw_models/EflomalAlignmentModel.h b/src/sw_models/EflomalAlignmentModel.h index 45233259..cc9178ee 100644 --- a/src/sw_models/EflomalAlignmentModel.h +++ b/src/sw_models/EflomalAlignmentModel.h @@ -22,6 +22,18 @@ // via marginal-mode sampling, so it satisfies thot's queryable/serializable // AlignmentModel interface for arbitrary (including held-out) sentence pairs. // Batch-only: does not implement IncrAlignmentModel. +// +// Concurrency contract: startTraining/train/endTraining/load/clear all mutate +// shared state (vocab, chains, corpus) and must not be called concurrently with each +// other or with query/decode methods on the same instance. Once endTraining or load +// completes, query and decode methods (translationProb, getBestAlignment, +// accumulateDecodeMarginal, etc.) are safe to call concurrently with each other for +// different sentence pairs - they only read the trained tables and use per-call local +// state (see decodeMarginalFromTables). This mirrors Rust's &mut-vs-&-self split even +// though C++ has no compiler-enforced equivalent here: the training/lifecycle methods +// are the "&mut self" phase, decode is the "&self" phase, and nothing in the type +// system stops a caller from violating the split - it's an invariant this comment +// exists to make explicit instead. class EflomalAlignmentModel : public AlignmentModelBase { public: @@ -156,7 +168,17 @@ class EflomalAlignmentModel : public AlignmentModelBase return h; } }; - using LexCountMap = tsl::robin_map; + // live: current collapsed-sampler count (decremented before sampling, incremented + // after). accum: running total added while a fertility-stage sweep is accumulating + // (variance-reduction marginal); folded into the same slot as live so the + // accumulate step reuses the iterator/hash probe the live update already paid for, + // instead of a second lookup into a separate, independently-growing map. + struct LexEntry + { + float live = 0; + float accum = 0; + }; + using LexCountMap = tsl::robin_map; // Per-chain Gibbs sampler state. N chains train in parallel; each has its own // RNG, alignment, count tables and trained parameter tables. Decode sums @@ -170,20 +192,29 @@ class EflomalAlignmentModel : public AlignmentModelBase std::vector> alig; // Running lexical counts (collapsed sampler: decremented before sampling, - // incremented after, per target token). Open-addressing map with float values - // (contiguous, no per-node pointer chase): counts are integer-valued (+/-1) and - // bounded under 2^24, so float holds them exactly. The running sums stay double - // (they can exceed float's exact-int range and are not read per candidate). + // incremented after, per target token) AND the accumulated marginal (see + // LexEntry), sharing one open-addressing map so both updates hit the same slot. + // Counts are integer-valued (+/-1) and bounded under 2^24, so float holds them + // exactly. The running sums stay double (they can exceed float's exact-int range + // and are not read per candidate). std::vector lexCounts; std::vector lexCountSum; + // 1/denominator per source word, recomputed every sweep in sampleSweep (counts + // change sweep to sweep) but kept as a chain member instead of a sampleSweep + // local so the vocab-sized buffer is allocated once in startTraining and reused + // via assign() every sweep, not reallocated fresh 84+ times per chain. + std::vector lexDenomInv; // Jump counts: seeded per sweep by computeJumpCounts, then updated per token // during the collapsed sweep. Fertility counts: recomputed per sweep. std::vector jumpCounts; double jumpCountSum = 0; - std::vector> fertCounts; // [s][phi] + // Flat [s*(MaxFertility+1)+phi] instead of vector>: one + // contiguous allocation instead of one per source-vocab entry, and a dense + // stride-(MaxFertility+1) access pattern instead of scattered per-row heap blocks. + std::vector fertCounts; std::vector fertCountSum; - std::vector> fertRatioSampled; // [s][phi] = sampled P(phi)/P(phi-1) + std::vector fertRatioSampled; // flat [s*(MaxFertility+1)+phi] = sampled P(phi)/P(phi-1) // Dirichlet prior masses (alpha * support size). Constant within a sweep; // cached here to avoid recomputation in the per-candidate inner loop. @@ -191,9 +222,10 @@ class EflomalAlignmentModel : public AlignmentModelBase double jumpPriorMass = 0; double fertPriorMass = 0; - // Lexical counts accumulated over final-stage sweeps (variance reduction); - // jump and fertility tables are read off the final alignment only. - std::vector accumLexCounts; + // Per-source-word running total of the accumulated marginal (LexEntry::accum + // summed over t); a plain vector since it's a row aggregate, not a per-(s,t) + // value, so it isn't foldable into LexCountMap itself. Jump and fertility tables + // are read off the final alignment only, not accumulated. std::vector accumLexCountSum; bool accumulated = false; @@ -202,6 +234,15 @@ class EflomalAlignmentModel : public AlignmentModelBase std::shared_ptr lexTable; std::shared_ptr jumpTable; std::shared_ptr fertilityTable; + + // Decode-time caches derived from jumpTable/fertilityTable once (by + // buildDecodeCache, called from normalizeChain and load), instead of every + // decodeMarginalFromTables call re-deriving them from scratch per sentence: + // decodeJumpLin is the linearized jump distribution (flat [2*jumpWindow+1]); + // decodeFertRatio is P(phi+1)/P(phi) per source word, flat [s*(MaxFertility+1)+phi]. + // Both depend only on the trained tables, never on a specific sentence. + std::vector decodeJumpLin; + std::vector decodeFertRatio; }; const double SmoothingProb = 1e-9; @@ -260,6 +301,10 @@ class EflomalAlignmentModel : public AlignmentModelBase int jumpBucket(int offset) const; // Normalizes this chain's accumulated counts into its own lex/jump/fert tables. void normalizeChain(SamplerChain& chain); + // Derives decodeJumpLin/decodeFertRatio from chain.jumpTable/fertilityTable; called + // once after those tables are populated (end of normalizeChain, and after load() + // reads them from disk), not per decode call. + void buildDecodeCache(SamplerChain& chain); // Decoding / scoring shared by getBestAlignment and computeSumLogProb. Extracts // the alignment as the per-target marginal mode of the Gibbs posterior: sample @@ -278,9 +323,11 @@ class EflomalAlignmentModel : public AlignmentModelBase // warmStart (optional): if non-null, seed the decode alignment from it and run a // single accumulate pass with no burn-in (one decode chain) instead of the cold - // diagonal-init multi-iteration re-sample. - void decodeMarginalFromTables(const MemoryLexTable& lex, const EflomalJumpTable& jump, - const FertilityTable& fert, unsigned int chainSeed, + // diagonal-init multi-iteration re-sample. jumpLin/fertRatioTable are the chain's + // precomputed decodeJumpLin/decodeFertRatio (see SamplerChain), passed by the + // caller instead of being rebuilt from jump/fert tables on every call. + void decodeMarginalFromTables(const MemoryLexTable& lex, const std::vector& jumpLin, + const std::vector& fertRatioTable, unsigned int chainSeed, const std::vector& nsrc, const std::vector& trg, std::vector>& acc, const std::vector* warmStart = nullptr); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 4c00c2ce..8d2fd939 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -1,3 +1,21 @@ +# Diagnostic-only, off by default: links mimalloc into thot_test to measure whether +# the default allocator is a scaling bottleneck under concurrent eflomal training +# (see eflomal-optimizations.md, Phase 1 opt 1). Not used by thot_lib/the shipped +# library - this only swaps the allocator for benchmark runs of the test binary. +option(THOT_BENCH_MIMALLOC "Link mimalloc into thot_test for allocator-contention benchmarking" OFF) +if(THOT_BENCH_MIMALLOC) + include(FetchContent) + FetchContent_Declare( + mimalloc + GIT_REPOSITORY https://github.com/microsoft/mimalloc.git + GIT_TAG v2.1.7 + ) + set(MI_BUILD_SHARED OFF CACHE BOOL "" FORCE) + set(MI_BUILD_TESTS OFF CACHE BOOL "" FORCE) + set(MI_OVERRIDE ON CACHE BOOL "" FORCE) + FetchContent_MakeAvailable(mimalloc) +endif() + add_executable(thot_test nlp_common/WordAlignmentMatrixTest.cc phrase_models/_phraseTableTest.h @@ -22,6 +40,11 @@ target_link_libraries(thot_test PRIVATE thot_lib gtest_main ) +if(THOT_BENCH_MIMALLOC) + target_link_libraries(thot_test PRIVATE mimalloc-static) + target_compile_definitions(thot_test PRIVATE THOT_BENCH_MIMALLOC) + target_include_directories(thot_test PRIVATE ${mimalloc_SOURCE_DIR}/include) +endif() include(GoogleTest) diff --git a/tests/sw_models/EflomalAlignmentModelTest.cc b/tests/sw_models/EflomalAlignmentModelTest.cc index 357ec4df..40df59fb 100644 --- a/tests/sw_models/EflomalAlignmentModelTest.cc +++ b/tests/sw_models/EflomalAlignmentModelTest.cc @@ -1,5 +1,13 @@ #include "sw_models/EflomalAlignmentModel.h" +#ifdef THOT_BENCH_MIMALLOC +// Diagnostic-only (Phase 1 opt 1, see eflomal-optimizations.md): overrides global +// operator new/delete to route through mimalloc, to test whether the default +// allocator is a scaling bottleneck. Only compiled in when configured with +// -DTHOT_BENCH_MIMALLOC=ON; no effect on the normal build. +#include +#endif + #include "TestUtils.h" #include "nlp_common/MathDefs.h" #include "sw_models/FastAlignModel.h" @@ -9,12 +17,16 @@ #include #include #include +#include #include #include #include +#include #include #include #include +#include +#include #include #include @@ -59,6 +71,128 @@ TEST(EflomalAlignmentModelTest, train) EXPECT_EQ(alignment, (std::vector{1, 2, 3, 5, 4, 0, 6})); } +// initializeChain runs under `#pragma omp parallel for ... if(!deterministic)` +// (EflomalAlignmentModel.cc, in startTraining): each chain only ever reads the +// shared read-only corpus and writes its own SamplerChain slot, seeded purely from +// its chain index (seed + s*2654435761), so concurrent init should be exactly as +// reproducible as the serial version it replaced. Trains the same corpus twice with +// numSamplers>1 and deterministic=false (the mode that actually exercises the +// parallel path) and checks the two runs agree, which a data race or +// order-dependency bug in the parallelized init would be very likely to break. +TEST(EflomalAlignmentModelTest, initializeChainParallelIsReproducible) +{ + auto trainAndAlign = [] { + EflomalAlignmentModel model; + model.setNumSamplers(4); + model.setDeterministic(false); + model.setIterations(2, 2, 2); + addTrainingData(model); + train(model, 6); + std::vector a1, a2, a3; + model.getBestAlignment("isthay isyay ayay esttay-N .", "this is a test N .", a1); + model.getBestAlignment("isthay isyay otnay ayay esttay-N .", "this is not a test N .", a2); + model.getBestAlignment("isthay isyay ayay esttay-N ardhay .", "this is a hard test N .", a3); + return std::make_tuple(a1, a2, a3); + }; + + auto run1 = trainAndAlign(); + auto run2 = trainAndAlign(); + EXPECT_EQ(run1, run2); +} + +// Regression coverage for a code-review finding: an earlier attempt at optimizing +// sampleSweep's per-candidate jumpBucket() clamp into a running saturating counter +// (bucket = min(bucket+1,cap) / max(bucket-1,0)) was wrong whenever the counter started +// already pinned to a boundary at i==1 - it would immediately start drifting away from +// the pin instead of staying there until the true unclamped offset re-entered range. +// The default jumpWindow=100 combined with this file's <10-token test sentences never +// reaches that boundary, so the whole existing suite passed anyway; only a jumpWindow +// smaller than the sentence length forces every candidate past the boundary immediately. +// Pins the alignment from the correct per-candidate jumpBucket() computation; a +// reintroduction of an incorrectly-saturating running counter would change these values +// without crashing or failing any other test. +// Trains the same jumpWindow=2 + long-sentence scenario (see below) and returns the +// long sentence's decoded alignment; factored out so the test can both check it and +// re-run it for a same-platform determinism cross-check. +namespace +{ +std::vector trainSaturationScenarioAndAlignLongSentence() +{ + EflomalAlignmentModel model; + model.setNumSamplers(1); + model.setDeterministic(true); + model.setIterations(2, 2, 2); + model.setDecodeParams(4, 12, 4); + model.setJumpWindow(2); // shorter than the long sentence below: forces saturation + addTrainingData(model); + addSentencePair(model, "wanyay owtay eethray ourfay ivefay ixsay evensay eightyay ninenay tenyay elevenyay elvetway irteenthay ourteenfay", + "one two three four five six seven eight nine ten eleven twelve thirteen fourteen"); + train(model, 6); + + std::vector alignment; + model.getBestAlignment( + "wanyay owtay eethray ourfay ivefay ixsay evensay eightyay ninenay tenyay elevenyay elvetway irteenthay ourteenfay", + "one two three four five six seven eight nine ten eleven twelve thirteen fourteen", alignment); + return alignment; +} +} // namespace + +TEST(EflomalAlignmentModelTest, jumpBucketSaturationBoundary) +{ + EflomalAlignmentModel model; + model.setNumSamplers(1); + model.setDeterministic(true); + model.setIterations(2, 2, 2); + model.setDecodeParams(4, 12, 4); + model.setJumpWindow(2); // shorter than every training sentence: forces saturation + addTrainingData(model); + // A longer, purpose-built pair on top of addTrainingData's toy corpus: long enough + // (14 tokens) relative to jumpWindow=2 that mid-sentence candidates sit far past the + // saturation boundary for many consecutive resamples - the regime a since-reverted + // running-counter optimization got wrong (see eflomal-optimizations.md, "Opt B", + // and fable-improvements.md item 10's follow-up attempt). + addSentencePair(model, "wanyay owtay eethray ourfay ivefay ixsay evensay eightyay ninenay tenyay elevenyay elvetway irteenthay ourteenfay", + "one two three four five six seven eight nine ten eleven twelve thirteen fourteen"); + train(model, 6); + + // These three short (<10 token) sentences are stable pinned-value checks: verified + // identical across Windows/MSVC, Linux/glibc, and macOS/libc++ CI runs, because + // jumpWindow=2 doesn't push them far enough into the saturation regime to make the + // few sampling decisions involved sensitive to platform libm/RNG differences. + std::vector alignment; + model.getBestAlignment("isthay isyay ayay esttay-N .", "this is a test N .", alignment); + EXPECT_EQ(alignment, (std::vector{1, 2, 3, 0, 4, 5})); + + model.getBestAlignment("isthay isyay otnay ayay esttay-N .", "this is not a test N .", alignment); + EXPECT_EQ(alignment, (std::vector{1, 2, 3, 4, 0, 5, 6})); + + model.getBestAlignment("isthay isyay ayay esttay-N ardhay .", "this is a hard test N .", alignment); + EXPECT_EQ(alignment, (std::vector{1, 2, 3, 5, 0, 4, 6})); + + // The long (14-token) sentence is NOT pinned to an exact value: with jumpWindow=2 - + // deliberately small relative to sentence length, to force many resamples deep into + // the saturation regime this test targets - the 6-sweep collapsed Gibbs chain + // accumulates enough sampling decisions that tiny, platform-specific differences in + // libm (exp/log) and std::uniform_real_distribution's implementation-defined + // mapping compound into a genuinely different (but equally valid, since both are + // legitimate draws from the same posterior) final alignment on each platform - + // confirmed empirically: Windows/MSVC, Linux/glibc, and macOS/libc++ CI runs each + // produced a different, self-consistent alignment for this specific sentence. + // SamplingUtils.h documents this exact tradeoff ("cross-platform bit-identical + // results are not required"). What IS portable and still meaningful: the alignment + // is well-formed (right length, every link a valid NULL-or-1-based source position) + // and reproducible on a given platform (same input, same output, twice). + model.getBestAlignment( + "wanyay owtay eethray ourfay ivefay ixsay evensay eightyay ninenay tenyay elevenyay elvetway irteenthay ourteenfay", + "one two three four five six seven eight nine ten eleven twelve thirteen fourteen", alignment); + ASSERT_EQ(alignment.size(), 14u); + for (PositionIndex link : alignment) + EXPECT_LE(link, 14u); // 0 = NULL, else a valid 1-based source position + + std::vector repeat = trainSaturationScenarioAndAlignLongSentence(); + EXPECT_EQ(alignment, repeat); +} + TEST(EflomalAlignmentModelTest, trainingAlignmentAccessor) { EflomalAlignmentModel model; @@ -189,6 +323,26 @@ TEST(EflomalAlignmentModelTest, oovWordsReturnSmoothedProb) EXPECT_LE((double)p, 1.0); } +// Regression coverage: decodeMarginalFromTables' fertRatioTable/decodeFertRatio (see +// buildDecodeCache) is sized to the vocab as of the last training/load, but +// getBestAlignment(string, string, ...) grows the vocab via addSrcSymbol for every +// word in the query (strVectorToSrcIndexVector -> addSrcSymbol), including brand-new +// words never seen during training. Confirms a post-training query containing a new +// word doesn't index the flat decode cache out of bounds - it should fall back to the +// same "unseen word" smoothing the old per-call table lookup gave, not crash. +TEST(EflomalAlignmentModelTest, getBestAlignmentWithPostTrainingNewWord) +{ + EflomalAlignmentModel model; + trainEflomal(model); + + std::vector alignment; + EXPECT_NO_THROW(model.getBestAlignment("brandnewunseenword isyay ayay esttay-N .", + "brandnewunseentarget is a test N .", alignment)); + EXPECT_EQ(alignment.size(), 6u); + for (PositionIndex p : alignment) + EXPECT_LE(p, 6u); // every link is either NULL (0) or a valid 1-based source position +} + TEST(EflomalAlignmentModelTest, getEntriesForSource) { EflomalAlignmentModel model; @@ -234,6 +388,34 @@ TEST(EflomalAlignmentModelTest, loglikelihoodForPairRange) EXPECT_GT(result.second, -1e6); } +// Regression coverage for loglikelihoodForPairRange's parallel rewrite (fable- +// improvements.md item 11): scoring is now split into a serial getSrcSent/getTrgSent +// pass (which can intern new vocab words) followed by a parallel +// #pragma omp reduction(+:loglikelihood) over computeSumLogProb. Confirms the empty- +// range edge case (last < first) still returns (0,0) as the original unsigned-loop +// version did, and that deterministic=true (which gates the new pragma off) gives +// bit-identical repeated results, which a data race in the parallel path would be +// very likely to break. +TEST(EflomalAlignmentModelTest, loglikelihoodForPairRangeEdgeCasesAndDeterminism) +{ + EflomalAlignmentModel model; + model.setNumSamplers(4); + model.setDeterministic(true); + model.setIterations(2, 2, 2); + addTrainingData(model); + train(model, 6); + + // Empty range: second < first. + auto empty = model.loglikelihoodForPairRange({3, 1}); + EXPECT_EQ(empty.first, 0.0); + EXPECT_EQ(empty.second, 0.0); + + auto run1 = model.loglikelihoodForPairRange({0, model.numSentencePairs() - 1}); + auto run2 = model.loglikelihoodForPairRange({0, model.numSentencePairs() - 1}); + EXPECT_EQ(run1.first, run2.first); + EXPECT_EQ(run1.second, run2.second); +} + TEST(EflomalAlignmentModelTest, fertilityProbIsValid) { EflomalAlignmentModel model; @@ -1145,6 +1327,197 @@ TEST(EflomalAlignmentModelTest, DISABLED_runtimeCompare) SUCCEED(); } +// Phase-0 scaling grid for the eflomal-optimizations effort (see +// eflomal-optimizations.md at the repo root). Trains the same corpus repeatedly +// across a matrix of numSamplers x OMP thread count x deterministic, to measure the +// actual parallel scaling ceiling before any code changes. No gold alignment +// needed; this is a pure timing harness. Run with EFL_SRC/EFL_TRG and +// --gtest_filter=*scalingGrid*. Grids are comma-separated env overrides: +// EFL_GRID_SAMPLERS (default "1,3,8"), EFL_GRID_THREADS (default "1,2,4,8"), +// EFL_GRID_DET (default "0,1", 0=false/1=true). +namespace +{ +std::vector parseIntList(const char* env, const char* dflt) +{ + std::string s = env ? env : dflt; + std::vector out; + std::stringstream ss(s); + std::string tok; + while (std::getline(ss, tok, ',')) + if (!tok.empty()) + out.push_back(std::atoi(tok.c_str())); + return out; +} +} // namespace + +TEST(EflomalAlignmentModelTest, DISABLED_scalingGrid) +{ + const char* srcEnv = std::getenv("EFL_SRC"); + const char* trgEnv = std::getenv("EFL_TRG"); + ASSERT_TRUE(srcEnv && trgEnv) << "EFL_SRC/EFL_TRG must be set"; + + std::vector> src = readTokenizedCorpus(srcEnv); + std::vector> trg = readTokenizedCorpus(trgEnv); + ASSERT_EQ(src.size(), trg.size()); + size_t n = src.size(); + + std::vector samplersGrid = parseIntList(std::getenv("EFL_GRID_SAMPLERS"), "1,3,8"); + std::vector threadsGrid = parseIntList(std::getenv("EFL_GRID_THREADS"), "1,2,4,8"); + std::vector detGrid = parseIntList(std::getenv("EFL_GRID_DET"), "0,1"); + // Off by default (matches the original grid so round-1 numbers stay comparable); + // set EFL_GRID_EMIT=1 to also produce warm training alignments in endTraining, + // exercising computeTrainingAlignments' decode-path cost (the fable-improvements.md + // item-2 target), which end_ms otherwise never touches. + bool emit = envInt("EFL_GRID_EMIT", 0) != 0; + + for (int det : detGrid) + { + for (int samplers : samplersGrid) + { + for (int threads : threadsGrid) + { + omp_set_num_threads(threads); + + EflomalAlignmentModel m; + m.setNumSamplers(samplers); + m.setDeterministic(det != 0); + m.setAutoIterations(true); + m.setEmitTrainingAlignments(emit); + for (size_t i = 0; i < n; ++i) + m.addSentencePair(src[i], trg[i], 1); + + auto t0 = std::chrono::steady_clock::now(); + m.startTraining(); + auto t1 = std::chrono::steady_clock::now(); + int sweeps = m.getScheduledIterations(); + for (int it = 0; it < sweeps; ++it) + m.train(); + auto t2 = std::chrono::steady_clock::now(); + m.endTraining(); + auto t3 = std::chrono::steady_clock::now(); + + long initMs = std::chrono::duration_cast(t1 - t0).count(); + long trainMs = std::chrono::duration_cast(t2 - t1).count(); + long endMs = std::chrono::duration_cast(t3 - t2).count(); + long totalMs = std::chrono::duration_cast(t3 - t0).count(); + + std::cerr << "GRID samplers=" << samplers << " threads=" << threads << " det=" << det + << " pairs=" << n << " sweeps=" << sweeps << " init_ms=" << initMs << " train_ms=" << trainMs + << " end_ms=" << endMs << " total_ms=" << totalMs << "\n"; + } + } + } + SUCCEED(); +} + +// Phase-1 opt 5 for the eflomal-optimizations effort: SymmetrizedAligner/ +// SymmetrizedAlignmentModel compose an already-trained direct + inverse model +// (decode-time only) but nothing trains the two directions concurrently, even +// though they are fully independent. Trains direct (EFL_SRC->EFL_TRG) and inverse +// (EFL_TRG->EFL_SRC) sequentially, then concurrently via std::async, at a fixed +// numSamplers (EFL_SYM_SAMPLERS, default 3 - kept well under half the machine's +// hardware_concurrency so 2*numSamplers threads don't oversubscribe). Run with +// EFL_SRC/EFL_TRG and --gtest_filter=*concurrentDirections*. +TEST(EflomalAlignmentModelTest, DISABLED_concurrentDirections) +{ + const char* srcEnv = std::getenv("EFL_SRC"); + const char* trgEnv = std::getenv("EFL_TRG"); + ASSERT_TRUE(srcEnv && trgEnv) << "EFL_SRC/EFL_TRG must be set"; + + std::vector> src = readTokenizedCorpus(srcEnv); + std::vector> trg = readTokenizedCorpus(trgEnv); + ASSERT_EQ(src.size(), trg.size()); + size_t n = src.size(); + + int samplers = envInt("EFL_SYM_SAMPLERS", 3); + + auto trainOneDirection = [&](bool swapped) { + EflomalAlignmentModel m; + m.setNumSamplers(samplers); + m.setDeterministic(false); + m.setAutoIterations(true); + for (size_t i = 0; i < n; ++i) + { + if (swapped) + m.addSentencePair(trg[i], src[i], 1); + else + m.addSentencePair(src[i], trg[i], 1); + } + m.startTraining(); + int sweeps = m.getScheduledIterations(); + for (int it = 0; it < sweeps; ++it) + m.train(); + m.endTraining(); + }; + + // Sequential: direct then inverse, back to back (today's implicit behaviour, since + // nothing in the library trains both directions at once). + auto s0 = std::chrono::steady_clock::now(); + trainOneDirection(false); + trainOneDirection(true); + long seqMs = std::chrono::duration_cast(std::chrono::steady_clock::now() - s0).count(); + + // Concurrent: both directions on separate threads. Each spawns its own numSamplers + // OpenMP chains, so this only pays off wall-clock-wise while 2*numSamplers stays + // within the machine's hardware concurrency; past that the two directions start + // competing for the same cores instead of adding parallelism. + auto c0 = std::chrono::steady_clock::now(); + auto futDirect = std::async(std::launch::async, trainOneDirection, false); + auto futInverse = std::async(std::launch::async, trainOneDirection, true); + futDirect.get(); + futInverse.get(); + long concMs = std::chrono::duration_cast(std::chrono::steady_clock::now() - c0).count(); + + std::cerr << "SYMDIR samplers=" << samplers << " hw_concurrency=" << std::thread::hardware_concurrency() + << " pairs=" << n << " sequential_ms=" << seqMs << " concurrent_ms=" << concMs + << " speedup=" << std::fixed << std::setprecision(2) << (double)seqMs / (double)concMs << "\n"; + SUCCEED(); +} + +// Benchmark for fable-improvements.md item 11: loglikelihoodForPairRange's new +// #pragma omp reduction over computeSumLogProb (previously a strictly serial loop). +// Trains briefly (this benchmarks decode/scoring, not training quality) then times +// loglikelihoodForPairRange over the whole corpus at threads=1 vs EFL_LL_THREADS. +// Run with EFL_SRC/EFL_TRG and --gtest_filter=*loglikelihoodParallel*. +TEST(EflomalAlignmentModelTest, DISABLED_loglikelihoodParallel) +{ + const char* srcEnv = std::getenv("EFL_SRC"); + const char* trgEnv = std::getenv("EFL_TRG"); + ASSERT_TRUE(srcEnv && trgEnv) << "EFL_SRC/EFL_TRG must be set"; + + std::vector> src = readTokenizedCorpus(srcEnv); + std::vector> trg = readTokenizedCorpus(trgEnv); + ASSERT_EQ(src.size(), trg.size()); + size_t n = src.size(); + + int threads = envInt("EFL_LL_THREADS", 8); + + EflomalAlignmentModel model; + model.setNumSamplers(3); + model.setDeterministic(false); + model.setIterations(4, 4, 4); // short: this times decode/scoring, not training + for (size_t i = 0; i < n; ++i) + model.addSentencePair(src[i], trg[i], 1); + omp_set_num_threads(threads); + train(model, 12); + + omp_set_num_threads(1); + auto t0 = std::chrono::steady_clock::now(); + auto serialResult = model.loglikelihoodForPairRange({0, model.numSentencePairs() - 1}); + long serialMs = std::chrono::duration_cast(std::chrono::steady_clock::now() - t0).count(); + + omp_set_num_threads(threads); + auto t1 = std::chrono::steady_clock::now(); + auto parallelResult = model.loglikelihoodForPairRange({0, model.numSentencePairs() - 1}); + long parallelMs = std::chrono::duration_cast(std::chrono::steady_clock::now() - t1).count(); + + std::cerr << "LLPARALLEL pairs=" << n << " threads=" << threads << " serial_ms=" << serialMs + << " parallel_ms=" << parallelMs << " speedup=" << std::fixed << std::setprecision(2) + << (double)serialMs / (double)parallelMs << " ll_serial=" << serialResult.first + << " ll_parallel=" << parallelResult.first << "\n"; + SUCCEED(); +} + // Validates persisted training alignments + round-trip: trains with // setEmitTrainingAlignments, computes AER of the in-memory warm alignments, // saves, reloads, and checks the reloaded links are identical and score the same