src/similarity-engine-parallel.lua

64local ffi = require("ffi")

65ffi.cdef[[

66 typedef void (*sighandler_t)(int);

67 sighandler_t signal(int signum, sighandler_t handler);

69 // Native sleep functions - avoids subprocess that eats Ctrl+C signals

70 int usleep(unsigned int usec);

71 unsigned int sleep(unsigned int seconds);

72]]

74local SIGINT = 2 -- Standard POSIX signal number for Ctrl+C

75local interrupted = false

78local function on_interrupt(sig)

79 interrupted = true

83local interrupt_handler = ffi.cast("sighandler_t", on_interrupt)

84ffi.C.signal(SIGINT, interrupt_handler)

87local function is_interrupted()

88 return interrupted

92local function reset_interrupt()

93 interrupted = false

110local function cosine_similarity(vec1, vec2)

111 if #vec1 ~= #vec2 then

112 error("Vector dimensions must match")

113 end

114

115 local dot_product = 0

116 local norm_a = 0

117 local norm_b = 0

118

119 for i = 1, #vec1 do

120 dot_product = dot_product + vec1[i] * vec2[i]

121 norm_a = norm_a + vec1[i] * vec1[i]

122 norm_b = norm_b + vec2[i] * vec2[i]

123 end

124

125 norm_a = math.sqrt(norm_a)

126 norm_b = math.sqrt(norm_b)

127

128 if norm_a == 0 or norm_b == 0 then

129 return 0

130 end

131

132 return dot_product / (norm_a * norm_b)

137local function get_cpu_count()

138 local handle = io.popen("nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || echo 4")

139 local result = handle:read("*a")

140 handle:close()

141 return tonumber(result:match("%d+")) or 4

146local function get_similarity_output_dir(model_name)

147 -- Issue 10-054: route through embeddings_dir() so the per-poem similarity

148 -- cache follows the RAM/disk switch with every other movable cache. While the

149 -- switch is off this is identical to the old get_assets_root()/embeddings/

150 -- <safe>/similarities (embeddings_dir applies the same model-name sanitizing).

151 return utils.embeddings_dir(model_name) .. "/similarities/"

158local function get_poem_similarity_file(output_dir, poem_id, poem_index)

159 -- Use poem_index for file naming - guaranteed unique across categories

160 local filename = "poem_index_" .. (poem_index or poem_id) .. ".json"

161 return output_dir .. filename

171local function count_completed_poems(output_dir, total_poems, requested_threads)

172 local completed = 0

173 local completed_poems = {}

175 if not utils.directory_exists(output_dir) then

176 return 0, completed_poems

177 end

179 -- Clean up any temporary files from interrupted runs

180 local cleanup_handle = io.popen("find '" .. output_dir .. "' -name '*.tmp' 2>/dev/null")

181 if cleanup_handle then

182 for temp_file in cleanup_handle:lines() do

183 utils.log_warn("Cleaning up temporary file: " .. temp_file)

184 os.remove(temp_file)

185 end

186 cleanup_handle:close()

187 end

189 -- Get list of all files to validate

190 local all_files = {}

191 local list_handle = io.popen("find '" .. output_dir .. "' -name 'poem_*.json' 2>/dev/null")

192 if list_handle then

193 for filepath in list_handle:lines() do

194 table.insert(all_files, filepath)

195 end

196 list_handle:close()

197 end

199 local total_files = #all_files

200 if total_files == 0 then

201 return 0, completed_poems

202 end

204 utils.log_info("Found " .. total_files .. " files to validate...")

206 -- Determine thread count (use requested, or default to CPU count)

207 local cpu_count = get_cpu_count()

208 local thread_count = requested_threads or cpu_count

209 thread_count = math.min(thread_count, total_files) -- Don't use more threads than files

211 -- Split files into batches

212 local files_per_thread = math.ceil(total_files / thread_count)

213 local batches = {}

214 for t = 1, thread_count do

215 local start_idx = (t - 1) * files_per_thread + 1

216 local end_idx = math.min(t * files_per_thread, total_files)

217 if start_idx <= total_files then

218 local batch = {}

219 for i = start_idx, end_idx do

220 table.insert(batch, all_files[i])

221 end

222 batches[t] = batch

223 end

224 end

226 -- Create progress channel

227 local progress_channel = effil.channel()

229 -- Create validation threads

230 local threads = {}

231 local batch_sizes = {}

233 for thread_id, batch in pairs(batches) do

234 batch_sizes[thread_id] = #batch

236 local thread_func = effil.thread(function(file_batch, thread_id, prog_channel)

237 -- Load required modules in thread context

238 package.path = package.path .. ';./libs/?.lua;./src/?.lua'

239 local dkjson = require('libs.dkjson')

241 local valid_count = 0

242 local valid_files = {}

243 local corrupted = {}

244 local checked = 0

246 for _, filepath in ipairs(file_batch) do

247 local filename = filepath:match("([^/]+)%.json$")

248 if filename then

249 -- Read and validate file

250 local file = io.open(filepath, "r")

251 if file then

252 local content = file:read("*a")

253 file:close()

255 local ok, file_data = pcall(dkjson.decode, content)

256 if ok and file_data and file_data.metadata and file_data.similarities and

257 file_data.metadata.total_comparisons and

258 #file_data.similarities == file_data.metadata.total_comparisons then

259 -- File is valid

260 valid_count = valid_count + 1

261 table.insert(valid_files, filename)

262 else

263 -- File is corrupted

264 table.insert(corrupted, filepath)

265 end

266 end

267 end

269 checked = checked + 1

270 -- Send progress update

271 prog_channel:push(thread_id, checked)

272 end

274 return valid_count, valid_files, corrupted

275 end)

277 threads[thread_id] = thread_func(batch, thread_id, progress_channel)

278 end

280 -- Progress display

281 local thread_progress = {}

282 for tid = 1, thread_count do

283 thread_progress[tid] = 0

284 end

285 local first_display = true

286 local start_time = os.time()

288 local function drain_channel()

289 while true do

290 local tid, count = progress_channel:pop(0)

291 if tid == nil then break end

292 thread_progress[tid] = count

293 end

294 end

296 local function display_progress()

297 drain_channel()

299 -- Move cursor up to overwrite previous display

300 -- After initial blank lines: cursor is on line N+2, need to move up N+1

301 -- After subsequent displays: cursor is on summary line, need to move up N

302 -- Also add \r to go to column 1 (cursor up doesn't change column position)

303 if first_display then

304 io.write(string.format("\027[%dA\r", thread_count + 1))

305 first_display = false

306 else

307 io.write(string.format("\027[%dA\r", thread_count))

308 end

310 local total_done = 0

311 local total_size = 0

313 for tid = 1, thread_count do

314 local done = thread_progress[tid] or 0

315 local size = batch_sizes[tid] or 0

316 total_done = total_done + done

317 total_size = total_size + size

319 local pct = (done / math.max(size, 1)) * 100

320 local bar_width = 20

321 local filled = math.floor((done / math.max(size, 1)) * bar_width)

322 local bar = string.rep("#", filled) .. string.rep("-", bar_width - filled)

323 io.write(string.format("\r Thread %2d: [%s] %4d/%4d (%5.1f%%)\027[K\n",

324 tid, bar, done, size, pct))

325 end

327 local elapsed = os.time() - start_time

328 local rate = total_done / math.max(elapsed, 1)

329 local total_pct = (total_done / math.max(total_size, 1)) * 100

330 io.write(string.format("\r --- Total: %d/%d (%.1f%%) | %.1f files/s\027[K",

331 total_done, total_size, total_pct, rate))

332 io.flush()

333 end

335 -- Print initial blank lines (thread_count + 1 for summary)

336 for _ = 1, thread_count + 1 do

337 print("")

338 end

340 -- Poll progress while threads run

341 local all_done = false

342 while not all_done do

343 display_progress()

344 all_done = true

345 for _, task in pairs(threads) do

346 local status = task:status()

347 if status ~= "completed" and status ~= "failed" then

348 all_done = false

349 end

350 end

351 if not all_done then

352 native_sleep(0.2)

353 end

354 end

356 -- Final display

357 display_progress()

358 print("") -- newline after progress

360 -- Collect results

361 local total_valid = 0

362 local all_corrupted = {}

364 for thread_id, task in pairs(threads) do

365 local valid_count, valid_files, corrupted = task:get()

366 if valid_count then

367 total_valid = total_valid + valid_count

368 for _, filename in ipairs(valid_files) do

369 completed_poems[filename] = true

370 end

371 for _, filepath in ipairs(corrupted) do

372 table.insert(all_corrupted, filepath)

373 end

374 end

375 end

377 -- Remove corrupted files

378 if #all_corrupted > 0 then

379 utils.log_warn("Removing " .. #all_corrupted .. " corrupted files...")

380 for _, filepath in ipairs(all_corrupted) do

381 os.remove(filepath)

382 end

383 end

385 return total_valid, completed_poems

392local function calculate_poem_similarities(poem_data, all_embeddings, output_file, sleep_duration)

393 -- FAIL-FAST: Validate input poem data

394 if not poem_data then

395 error(string.format(

396 "SIMILARITY CALCULATION FAILED: poem_data is nil\n" ..

397 " Context: Attempted to calculate similarities for nil poem\n" ..

398 " Output file: %s\n" ..

399 " Remedy: Check embedding loading - poem data may be missing from embeddings.json",

400 output_file or "nil"

401 ))

402 end

404 if not poem_data.embedding then

405 error(string.format(

406 "SIMILARITY CALCULATION FAILED: Missing embedding for poem %s (index %s)\n" ..

407 " Context: Poem exists but has no embedding vector\n" ..

408 " Output file: %s\n" ..

409 " Remedy: Regenerate embeddings with 'lua src/similarity-engine.lua -I' (option 1)",

410 tostring(poem_data.id or "nil"), tostring(poem_data.index or "nil"), output_file or "nil"

411 ))

412 end

414 if type(poem_data.embedding) ~= "table" or #poem_data.embedding == 0 then

415 error(string.format(

416 "SIMILARITY CALCULATION FAILED: Invalid embedding for poem %s (index %s)\n" ..

417 " Context: Embedding is not a valid vector (type=%s, length=%s)\n" ..

418 " Output file: %s\n" ..

419 " Remedy: Regenerate embeddings - this poem has corrupted embedding data",

420 tostring(poem_data.id or "nil"), tostring(poem_data.index or "nil"),

421 type(poem_data.embedding), tostring(#poem_data.embedding or 0), output_file or "nil"

422 ))

423 end

425 local similarities = {}

427 utils.log_info("Calculating similarities for poem " .. (poem_data.id or poem_data.index) .. " against " .. #all_embeddings .. " poems")

429 for j = 1, #all_embeddings do

430 local other_poem = all_embeddings[j]

432 -- FAIL-FAST: Validate each comparison target

433 if not other_poem.embedding then

434 error(string.format(

435 "SIMILARITY CALCULATION FAILED: Missing embedding for comparison poem %s (index %s)\n" ..

436 " Context: Source poem %s, comparison index %d of %d\n" ..

437 " Remedy: All poems must have embeddings. Regenerate embeddings.json",

438 tostring(other_poem.id or "nil"), tostring(other_poem.index or "nil"),

439 tostring(poem_data.id or poem_data.index), j, #all_embeddings

440 ))

441 end

443 -- Skip self-comparison

444 if poem_data.index ~= other_poem.index then

445 local similarity = cosine_similarity(poem_data.embedding, other_poem.embedding)

447 table.insert(similarities, {

448 id = other_poem.id,

449 index = other_poem.index,

450 similarity = similarity

451 })

452 end

453 end

455 -- Sort by similarity (highest first) - ALL similarities, not just top N

456 table.sort(similarities, function(a, b) return a.similarity > b.similarity end)

458 -- Create comprehensive similarity data for this poem

459 local poem_similarity_data = {

460 metadata = {

461 poem_id = poem_data.id,

462 poem_index = poem_data.index,

463 total_comparisons = #similarities,

464 calculated_at = os.date("%Y-%m-%d %H:%M:%S"),

465 algorithm = "cosine_similarity"

466 },

467 similarities = similarities

468 }

470 -- FAIL-FAST: File write must succeed

471 if not utils.write_json_file(output_file, poem_similarity_data) then

472 error(string.format(

473 "SIMILARITY CALCULATION FAILED: Could not write output file\n" ..

474 " File: %s\n" ..

475 " Poem: %s (index %s)\n" ..

476 " Comparisons: %d\n" ..

477 " Remedy: Check disk space, permissions, and path validity",

478 output_file, tostring(poem_data.id or "nil"), tostring(poem_data.index or "nil"),

479 #similarities

480 ))

481 end

483 -- Temperature control - sleep to prevent overheating

484 -- Uses native FFI sleep to allow Ctrl+C to work properly

485 if sleep_duration > 0 then

486 native_sleep(sleep_duration)

487 end

489 return true

496local function process_poem_batch(batch_poems, all_embeddings, output_dir, sleep_duration, thread_id)

497 local processed = 0

499 utils.log_info("Thread " .. thread_id .. ": Processing " .. #batch_poems .. " poems (fail-fast mode)")

501 for i, poem_data in ipairs(batch_poems) do

502 local output_file = get_poem_similarity_file(output_dir, poem_data.id, poem_data.index)

504 -- calculate_poem_similarities will error() on any failure

505 -- We wrap in pcall to add thread context to the error message, then re-raise

506 local success, err = pcall(function()

507 calculate_poem_similarities(poem_data, all_embeddings, output_file, sleep_duration)

508 end)

510 if not success then

511 -- Add thread context and re-raise the error

512 error(string.format(

513 "Thread %d FATAL ERROR at poem %d/%d:\n%s",

514 thread_id, i, #batch_poems, tostring(err)

515 ))

516 end

518 processed = processed + 1

519 utils.log_info("Thread " .. thread_id .. ": Completed poem " .. (poem_data.id or poem_data.index) .. " (" .. processed .. "/" .. #batch_poems .. ")")

520 end

522 utils.log_info("Thread " .. thread_id .. ": Completed batch - " .. processed .. " processed, 0 errors")

523 return processed, 0

533function M.calculate_similarity_matrix_parallel(embeddings_file, model_name, sleep_duration, force_regenerate, requested_threads)

534 sleep_duration = sleep_duration or 0.5

535 force_regenerate = force_regenerate or false

537 -- Determine thread count (use requested, or default to CPU count)

538 local cpu_count = get_cpu_count()

539 local thread_count = requested_threads or cpu_count

563 -- Load embeddings

564 utils.log_info("Loading embeddings from: " .. embeddings_file)

565 local embeddings_data = utils.read_json_file(embeddings_file)

566 if not embeddings_data or not embeddings_data.embeddings then

567 utils.log_error("Failed to load embeddings from " .. embeddings_file)

568 return false

569 end

571 -- Filter valid embeddings

572 -- Use poem_index as the unique identifier (Issue 8-019)

573 local valid_embeddings = {}

574 for key, item in pairs(embeddings_data.embeddings) do

575 if item.embedding and #item.embedding > 0 then

576 -- Prefer item.poem_index if available, fallback to storage key

577 local poem_index = item.poem_index or tonumber(key)

578 table.insert(valid_embeddings, {

579 index = poem_index,

580 id = item.id,

581 poem_index = poem_index,

582 embedding = item.embedding

583 })

584 end

585 end

587 utils.log_info("Found " .. #valid_embeddings .. " valid embeddings for similarity calculation")

589 -- Setup output directory

590 local output_dir = get_similarity_output_dir(model_name)

591 if not utils.directory_exists(output_dir) then

592 os.execute("mkdir -p '" .. output_dir .. "'")

593 utils.log_info("Created similarity output directory: " .. output_dir)

594 end

596 -- Check for existing work (resume capability) - use same thread count

597 local completed_count, completed_poems = count_completed_poems(output_dir, #valid_embeddings, thread_count)

599 if not force_regenerate and completed_count > 0 then

600 utils.log_info("Found " .. completed_count .. " existing similarity files")

602 if completed_count == #valid_embeddings then

603 utils.log_info("✅ All poem similarities already calculated!")

604 return true

605 else

606 utils.log_info("📄 Resuming from existing progress: " .. completed_count .. "/" .. #valid_embeddings .. " completed")

607 end

608 elseif force_regenerate then

609 utils.log_info("🗑️ Force regenerate enabled - clearing existing similarity files")

610 os.execute("rm -f '" .. output_dir .. "poem_*.json' 2>/dev/null")

611 completed_poems = {}

612 completed_count = 0

613 end

615 -- Filter out already completed poems

616 local remaining_poems = {}

617 for _, poem_data in ipairs(valid_embeddings) do

618 local filename = poem_data.id and ("poem_" .. poem_data.id) or ("poem_index_" .. poem_data.index)

619 if not completed_poems[filename] then

620 table.insert(remaining_poems, poem_data)

621 end

622 end

624 utils.log_info("Remaining poems to process: " .. #remaining_poems)

626 if #remaining_poems == 0 then

627 utils.log_info("✅ All poem similarities are up to date!")

628 return true

629 end

631 -- Limit thread count to remaining work

632 thread_count = math.min(thread_count, #remaining_poems)

634 utils.log_info("🧵 Using " .. thread_count .. " threads (detected " .. cpu_count .. " CPUs)")

635 utils.log_info("⏱️ Sleep duration per poem: " .. sleep_duration .. " seconds")

636

637 -- Divide work among threads

638 local poems_per_thread = math.ceil(#remaining_poems / thread_count)

639 local batches = {}

640

641 for t = 1, thread_count do

642 local start_idx = (t - 1) * poems_per_thread + 1

643 local end_idx = math.min(t * poems_per_thread, #remaining_poems)

644

645 if start_idx <= #remaining_poems then

646 local batch = {}

647 for i = start_idx, end_idx do

648 table.insert(batch, remaining_poems[i])

649 end

650 batches[t] = batch

651 utils.log_info("Thread " .. t .. ": " .. #batch .. " poems (indices " .. start_idx .. "-" .. end_idx .. ")")

652 end

653 end

654

655 -- Process batches using effil threading if available, otherwise sequential

656 utils.log_info("🚀 Starting similarity calculation...")

657 local start_time = os.time()

659 local total_processed = 0

660 local total_errors = 0

662 -- Process batches using effil multithreading (effil is required, verified at startup)

663 utils.log_info("🧵 Using effil multithreading with " .. thread_count .. " threads")

665 -- Track batch sizes for progress display

666 local batch_sizes = {}

667 local total_to_process = 0

668 for thread_id, batch in pairs(batches) do

669 batch_sizes[thread_id] = #batch

670 total_to_process = total_to_process + #batch

671 end

673 -- Create shared channel for progress updates from threads

674 -- Channel allows threads to send (thread_id, progress) messages to main thread

675 local progress_channel = effil.channel()

677 -- Create thread pool and tasks

678 local threads = {}

679 local tasks = {}

681 for thread_id, batch in pairs(batches) do

682 -- Create thread function with channel for progress reporting

683 local thread_func = effil.thread(function(batch_data, all_embeddings_data, output_dir, sleep_duration, thread_id, prog_channel)

684 -- Load required modules in thread context

685 package.path = package.path .. ';./libs/?.lua;./src/?.lua'

686 local utils = require('utils')

687 local dkjson = require('libs.dkjson')

688

689 -- Cosine similarity function

690 local function cosine_similarity(vec1, vec2)

691 if #vec1 ~= #vec2 then

692 error("Vector dimensions must match")

693 end

694

695 local dot_product = 0

696 local norm_a = 0

697 local norm_b = 0

698

699 for i = 1, #vec1 do

700 dot_product = dot_product + vec1[i] * vec2[i]

701 norm_a = norm_a + vec1[i] * vec1[i]

702 norm_b = norm_b + vec2[i] * vec2[i]

703 end

704

705 norm_a = math.sqrt(norm_a)

706 norm_b = math.sqrt(norm_b)

707

708 if norm_a == 0 or norm_b == 0 then

709 return 0

710 end

711

712 return dot_product / (norm_a * norm_b)

713 end

714

715 -- State codes for channel messages (negative = state, positive = progress count)

716 local STATE_RESTING = -1

717 local STATE_PROCESSING = -2

719 -- FAIL-FAST: Process poems in this thread with strict error checking

720 -- Any failure will immediately error() and stop the thread

721 local processed = 0

723 for poem_idx, poem_data in ipairs(batch_data) do

724 -- Signal that we're now processing

725 prog_channel:push(thread_id, STATE_PROCESSING)

727 -- FAIL-FAST: Validate poem data has embedding

728 if not poem_data.embedding then

729 error(string.format(

730 "Thread %d FAILED: Poem %s (index %s) has no embedding\n" ..

731 " Batch position: %d/%d\n" ..

732 " Remedy: Regenerate embeddings - this poem is missing embedding data",

733 thread_id, tostring(poem_data.id or "nil"), tostring(poem_data.index or "nil"),

734 poem_idx, #batch_data

735 ))

736 end

738 local similarities = {}

740 -- Issue 8-034: Calculate ONLY upper triangle (where other_index > this_index)

741 -- This provides 50% storage savings while maintaining symmetric lookup via access layer

742 for j = 1, #all_embeddings_data do

743 local other_poem = all_embeddings_data[j]

745 -- FAIL-FAST: Validate comparison target has embedding

746 if not other_poem.embedding then

747 error(string.format(

748 "Thread %d FAILED: Comparison poem %s (index %s) has no embedding\n" ..

749 " Source poem: %s (index %s)\n" ..

750 " Comparison index: %d/%d\n" ..

751 " Remedy: All poems must have embeddings. Regenerate embeddings.json",

752 thread_id, tostring(other_poem.id or "nil"), tostring(other_poem.index or "nil"),

753 tostring(poem_data.id or "nil"), tostring(poem_data.index or "nil"),

754 j, #all_embeddings_data

755 ))

756 end

758 -- Only calculate for higher indices (triangular upper)

759 if other_poem.index > poem_data.index then

760 local similarity = cosine_similarity(poem_data.embedding, other_poem.embedding)

762 table.insert(similarities, {

763 id = other_poem.id,

764 index = other_poem.index,

765 similarity = similarity

766 })

767 end

768 end

770 -- Sort by similarity (highest first)

771 table.sort(similarities, function(a, b) return a.similarity > b.similarity end)

773 -- Create similarity data (Issue 8-034: triangular format)

774 local poem_similarity_data = {

775 metadata = {

776 poem_id = poem_data.id,

777 poem_index = poem_data.index,

778 total_comparisons = #similarities,

779 format = "triangular_upper",

780 range = string.format("%d-%d", poem_data.index + 1, #all_embeddings_data),

781 calculated_at = os.date("%Y-%m-%d %H:%M:%S"),

782 algorithm = "cosine_similarity"

783 },

784 similarities = similarities

785 }

787 -- Atomic write: use temporary file + rename to prevent partial files

788 local filename = poem_data.id and ("poem_" .. poem_data.id .. ".json") or ("poem_index_" .. poem_data.index .. ".json")

789 local output_file = output_dir .. filename

790 local temp_file = output_dir .. filename .. ".tmp"

792 -- FAIL-FAST: JSON encoding must succeed

793 local json_string = dkjson.encode(poem_similarity_data, { indent = true })

794 if not json_string then

795 error(string.format(

796 "Thread %d FAILED: JSON encoding failed for poem %s\n" ..

797 " Output file: %s\n" ..

798 " Comparisons: %d\n" ..

799 " Remedy: Check poem data structure - dkjson.encode returned nil",

800 thread_id, tostring(poem_data.id or poem_data.index), output_file, #similarities

801 ))

802 end

804 -- FAIL-FAST: File write must succeed

805 local file = io.open(temp_file, "w")

806 if not file then

807 error(string.format(

808 "Thread %d FAILED: Could not open temp file for writing\n" ..

809 " Temp file: %s\n" ..

810 " Poem: %s\n" ..

811 " Remedy: Check disk space, permissions, and path validity",

812 thread_id, temp_file, tostring(poem_data.id or poem_data.index)

813 ))

814 end

816 file:write(json_string)

817 file:close()

819 -- FAIL-FAST: Atomic rename must succeed

820 local rename_success = os.rename(temp_file, output_file)

821 if not rename_success then

822 os.remove(temp_file) -- cleanup temp file

823 error(string.format(

824 "Thread %d FAILED: Could not rename temp file to final destination\n" ..

825 " Temp file: %s\n" ..

826 " Output file: %s\n" ..

827 " Poem: %s\n" ..

828 " Remedy: Check filesystem permissions and disk space",

829 thread_id, temp_file, output_file, tostring(poem_data.id or poem_data.index)

830 ))

831 end

833 processed = processed + 1

834 -- Send progress update through channel

835 prog_channel:push(thread_id, processed)

837 -- Temperature control: signal resting state before sleep

838 if sleep_duration > 0 then

839 prog_channel:push(thread_id, STATE_RESTING)

840 os.execute("sleep " .. sleep_duration)

841 end

842 end

844 return processed, 0 -- 0 errors because we fail-fast on any error

845 end)

846

847 -- Start thread (progress reported via channel)

848 local task = thread_func(batch, valid_embeddings, output_dir, sleep_duration, thread_id, progress_channel)

849 threads[thread_id] = task

850 tasks[thread_id] = thread_id

851 end

853 -- Per-thread progress tracking via channel messages

854 -- Each thread sends (thread_id, processed_count) after each poem

855 -- State codes: -1 = resting, -2 = processing, >=0 = progress count

856 local STATE_RESTING = -1

857 local STATE_PROCESSING = -2

858 local thread_progress = {}

859 local thread_state = {} -- "processing" or "resting"

860 for tid = 1, thread_count do

861 thread_progress[tid] = 0

862 thread_state[tid] = "processing" -- assume processing initially

863 end

864 local first_display = true

866 -- ANSI color codes for terminal display

867 local COLOR_GREEN = "\027[32m" -- processing

868 local COLOR_CYAN = "\027[36m" -- resting

869 local COLOR_RESET = "\027[0m"

871 local function drain_channel()

872 -- Read all available messages from channel (non-blocking)

873 -- Messages are either state updates (negative) or progress counts (>=0)

874 while true do

875 local tid, value = progress_channel:pop(0) -- timeout=0 for non-blocking

876 if tid == nil then break end

877 if value == STATE_RESTING then

878 thread_state[tid] = "resting"

879 elseif value == STATE_PROCESSING then

880 thread_state[tid] = "processing"

881 else

882 -- Progress count update

883 thread_progress[tid] = value

884 end

885 end

886 end

888 local function display_progress()

889 drain_channel()

891 -- Move cursor up to overwrite previous display

892 -- After initial blank lines: cursor is on line N+2, need to move up N+1

893 -- After subsequent displays: cursor is on summary line, need to move up N

894 -- Also add \r to go to column 1 (cursor up doesn't change column position)

895 if first_display then

896 io.write(string.format("\027[%dA\r", thread_count + 1))

897 first_display = false

898 else

899 io.write(string.format("\027[%dA\r", thread_count))

900 end

902 local total_done = 0

903 local total_size = 0

905 -- Print each thread's progress on its own line with color coding

906 -- Green = processing, Cyan = resting (with indicator)

907 for tid = 1, thread_count do

908 local done = thread_progress[tid] or 0

909 local size = batch_sizes[tid] or 0

910 local state = thread_state[tid] or "processing"

911 total_done = total_done + done

912 total_size = total_size + size

914 local pct = (done / math.max(size, 1)) * 100

915 local bar_width = 20

916 local filled = math.floor((done / math.max(size, 1)) * bar_width)

917 local bar = string.rep("#", filled) .. string.rep("-", bar_width - filled)

919 -- Choose color based on thread state

920 local color = (state == "resting") and COLOR_CYAN or COLOR_GREEN

921 local suffix = (state == "resting") and " (resting)" or ""

923 io.write(string.format("\r %sThread %2d: [%s] %4d/%4d (%5.1f%%)%s%s\027[K\n",

924 color, tid, bar, done, size, pct, suffix, COLOR_RESET))

925 end

927 -- Summary line (use ASCII characters for compatibility)

928 local elapsed = os.time() - start_time

929 local rate = total_done / math.max(elapsed, 1)

930 local eta = (total_size - total_done) / math.max(rate, 0.01)

931 local total_pct = (total_done / math.max(total_size, 1)) * 100

932 io.write(string.format("\r --- Total: %d/%d (%.1f%%) | %.2f/s | ETA: %ds\027[K",

933 total_done, total_size, total_pct, rate, math.floor(eta)))

934 io.flush()

935 end

937 -- Wait for all threads to complete with progress updates

938 utils.log_info("⏳ Waiting for " .. thread_count .. " threads to complete...")

939 utils.log_info("💡 Press Ctrl+C to gracefully stop (threads will finish current poem)")

940 -- Print initial blank lines for progress display area (thread_count + 1 for summary)

941 for _ = 1, thread_count + 1 do

942 print("")

943 end

945 -- Poll progress while threads are running (check for Ctrl+C interrupt)

946 local all_done = false

947 local was_interrupted = false

948 while not all_done do

949 display_progress()

951 -- Check for Ctrl+C interrupt

952 if is_interrupted() then

953 was_interrupted = true

954 -- Move below progress display before printing message

955 io.write("\n")

956 utils.log_warn("⚠️ Ctrl+C detected! Waiting for threads to finish current poem...")

957 break

958 end

960 all_done = true

961 for thread_id, task in pairs(threads) do

962 local status = task:status()

963 if status ~= "completed" and status ~= "failed" then

964 all_done = false

965 end

966 end

967 if not all_done then

968 native_sleep(0.5)

969 end

970 end

971 print("") -- newline after progress

973 -- If interrupted, wait briefly for threads to finish their current work

974 if was_interrupted then

975 utils.log_info("⏳ Giving threads 5 seconds to complete current work...")

976 native_sleep(5)

977 end

979 for thread_id, task in pairs(threads) do

980 local status, processed, errors = task:get(0) -- Non-blocking get with timeout=0

981 if status then

982 total_processed = total_processed + processed

983 total_errors = total_errors + errors

984 utils.log_info("Thread " .. thread_id .. " completed: " .. processed .. " processed, " .. errors .. " errors")

985 elseif was_interrupted then

986 -- Thread may still be running - that's ok, we'll count files later

987 utils.log_info("Thread " .. thread_id .. ": interrupted (partial results saved)")

988 else

989 utils.log_error("Thread " .. thread_id .. " failed: " .. tostring(processed))

990 total_errors = total_errors + 1

991 end

992 end

994 -- Reset interrupt flag for potential future runs

995 reset_interrupt()

996

997 local end_time = os.time()

998 local total_time = end_time - start_time

1000 -- Final verification

1001 local final_completed_count = count_completed_poems(output_dir, #valid_embeddings)

1003 if was_interrupted then

1004 utils.log_info("⏸️ Similarity calculation interrupted by user")

1005 else

1006 utils.log_info("🎉 Similarity calculation completed!")

1007 end

1008 utils.log_info("⏱️ Total time: " .. total_time .. " seconds")

1009 utils.log_info("📁 Output directory: " .. output_dir)

1010 utils.log_info("✅ Completed similarity files: " .. final_completed_count .. "/" .. #valid_embeddings)

1011 utils.log_info("📊 Processing results: " .. total_processed .. " processed, " .. total_errors .. " errors")

1013 if final_completed_count == #valid_embeddings then

1014 utils.log_info("🎯 All poem similarities successfully calculated!")

1015 return true

1016 else

1017 local missing = #valid_embeddings - final_completed_count

1018 if was_interrupted then

1019 utils.log_info("💡 Run again to resume - already-completed files will be skipped")

1020 else

1021 utils.log_warn("⚠️ " .. missing .. " poem similarities are missing")

1022 end

1023 return false

1024 end

1029function M.process_poem_batch_external(batch_poems, all_embeddings, output_dir, sleep_duration, thread_id)

1030 return process_poem_batch(batch_poems, all_embeddings, output_dir, sleep_duration, thread_id)

1035function M.main()

1036 if arg and arg[1] == "-I" then

1037 -- Load TUI menu library

1038 local menu_ok, menu = pcall(require, "menu")

1039 local tui_ok, tui = pcall(require, "tui")

1041 if not menu_ok or not tui_ok then

1042 -- Fallback to simple text menu if TUI not available

1043 utils.log_warn("TUI menu not available, using text mode")

1044 return M.main_text_mode()

1045 end

1047 -- Get CPU count for default thread value

1048 local cpu_count = get_cpu_count()

1050 -- Build menu configuration

1051 local config = {

1052 title = "Parallel Similarity Engine",

1053 subtitle = "Generate full similarity rankings for all poems",

1054 sections = {

1055 {

1056 id = "action",

1057 title = "Action",

1058 type = "single",

1059 items = {

1060 {

1061 id = "calculate",

1062 label = "Calculate similarity matrix (parallel)",

1063 type = "checkbox",

1064 value = "1",

1065 description = "Generate individual similarity files for all poems",

1066 shortcut = "c"

1067 },

1068 {

1069 id = "check_status",

1070 label = "Check calculation status",

1071 type = "checkbox",

1072 value = "0",

1073 description = "View progress of similarity file generation",

1074 shortcut = "s"

1075 }

1076 }

1077 },

1078 {

1079 id = "options",

1080 title = "Options",

1081 type = "multi",

1082 items = {

1083 {

1084 id = "force",

1085 label = "Force regenerate",

1086 type = "checkbox",

1087 value = "0",

1088 description = "Delete and recreate all existing similarity files",

1089 shortcut = "f"

1090 },

1091 {

1092 id = "sleep",

1093 label = "Sleep duration (sec)",

1094 type = "flag",

1095 value = "0.5",

1096 config = "6", -- Width of input field (separate from value)

1097 description = "Delay between poems for thermal management (e.g. 0.5)",

1098 flag = "--sleep"

1099 },

1100 {

1101 id = "threads",

1102 label = "Thread count",

1103 type = "flag",

1104 value = tostring(cpu_count),

1105 config = "3",

1106 description = "Number of parallel threads (detected " .. cpu_count .. " CPUs)",

1107 flag = "--threads"

1108 },

1109 {

1110 id = "model",

1111 label = "Embedding model",

1112 type = "text",

1113 value = "embeddinggemma:latest",

1114 description = "Model used for embeddings (stored in assets/embeddings/)"

1115 }

1116 }

1117 },

1118 {

1119 id = "run",

1120 title = "",

1121 type = "single",

1122 items = {

1123 {

1124 id = "run_action",

1125 label = "[Run]",

1126 type = "action",

1127 value = "",

1128 description = "Execute selected action with current options"

1129 }

1130 }

1131 }

1132 },

1133 -- Only enable some options when "calculate" is selected

1134 dependencies = {

1135 {

1136 item_id = "force",

1137 depends_on = "calculate",

1138 required_values = {"1"},

1139 invert = false,

1140 reason = "Only available for Calculate action"

1141 },

1142 {

1143 item_id = "sleep",

1144 depends_on = "calculate",

1145 required_values = {"1"},

1146 invert = false,

1147 reason = "Only available for Calculate action"

1148 }

1149 -- Note: threads is available for both actions (validation uses it too)

1150 }

1151 }

1153 -- Run TUI menu

1154 local action, values

1155 local success, err = pcall(function()

1156 menu.init(config)

1157 action, values = menu.run()

1158 menu.cleanup()

1159 end)

1161 if not success then

1162 pcall(menu.cleanup)

1163 utils.log_error("Menu error: " .. tostring(err))

1164 return false

1165 end

1167 -- Handle quit

1168 if action == "quit" then

1169 utils.log_info("Cancelled by user")

1170 return true

1171 end

1173 -- Execute selected action

1174 local thread_count = tonumber(values.threads) or cpu_count

1176 if values.calculate == "1" then

1177 local force = values.force == "1"

1178 local sleep_duration = tonumber(values.sleep) or 0.5

1179 local model = values.model or "embeddinggemma:latest"

1180 if model == "" then model = "embeddinggemma:latest" end

1182 -- Issue 10-054: read embeddings through the cache switch (RAM/disk).

1183 local embeddings_file = utils.embeddings_dir(model) .. "/embeddings.json"

1185 if not utils.file_exists(embeddings_file) then

1186 utils.log_error("Embeddings file not found: " .. embeddings_file)

1187 return false

1188 end

1190 return M.calculate_similarity_matrix_parallel(embeddings_file, model, sleep_duration, force, thread_count)

1192 elseif values.check_status == "1" then

1193 local model = values.model or "embeddinggemma:latest"

1194 if model == "" then model = "embeddinggemma:latest" end

1196 local output_dir = get_similarity_output_dir(model)

1197 utils.log_info("Checking " .. output_dir .. " ...")

1199 -- Quick count first (no validation)

1200 local quick_count_handle = io.popen("find '" .. output_dir .. "' -name 'poem_*.json' 2>/dev/null | wc -l")

1201 local quick_count = 0

1202 if quick_count_handle then

1203 quick_count = tonumber(quick_count_handle:read("*a")) or 0

1204 quick_count_handle:close()

1205 end

1206 utils.log_info("Found " .. quick_count .. " similarity files (validating...)")

1208 local completed_count = count_completed_poems(output_dir, 0, thread_count)

1210 utils.log_info("✅ Valid similarity files: " .. completed_count .. "/" .. quick_count)

1211 else

1212 utils.log_info("No action selected")

1213 end

1214 end

1216 return true

1222function M.main_text_mode()

1223 local cpu_count = get_cpu_count()

1225 utils.log_info("=== Parallel Similarity Engine (Text Mode) ===")

1226 print("1. Calculate similarity matrix (parallel)")

1227 print("2. Check similarity calculation status")

1228 print("q. Quit")

1229 io.write("Select option: ")

1230 io.flush()

1232 local choice = io.read()

1234 if choice == "q" or choice == "Q" then

1235 utils.log_info("Cancelled by user")

1236 return true

1237 elseif choice == "1" then

1238 io.write("Force regenerate existing files? (y/N): ")

1239 io.flush()

1240 local force = (io.read() or ""):lower() == "y"

1242 io.write("Sleep duration per poem (default 0.5s): ")

1243 io.flush()

1244 local sleep_input = io.read()

1245 local sleep_duration = tonumber(sleep_input) or 0.5

1247 io.write("Thread count (default " .. cpu_count .. " detected CPUs): ")

1248 io.flush()

1249 local thread_input = io.read()

1250 local thread_count = tonumber(thread_input) or cpu_count

1252 io.write("Embedding model (default: embeddinggemma:latest): ")

1253 io.flush()

1254 local model = io.read() or ""

1255 if model == "" then model = "embeddinggemma:latest" end

1257 -- Issue 10-054: read embeddings through the cache switch (RAM/disk).

1258 local embeddings_file = utils.embeddings_dir(model) .. "/embeddings.json"

1260 if not utils.file_exists(embeddings_file) then

1261 utils.log_error("Embeddings file not found: " .. embeddings_file)

1262 return false

1263 end

1265 return M.calculate_similarity_matrix_parallel(embeddings_file, model, sleep_duration, force, thread_count)

1267 elseif choice == "2" then

1268 io.write("Thread count for validation (default " .. cpu_count .. " detected CPUs): ")

1269 io.flush()

1270 local thread_input = io.read()

1271 local thread_count = tonumber(thread_input) or cpu_count

1273 io.write("Embedding model (default: embeddinggemma:latest): ")

1274 io.flush()

1275 local model = io.read() or ""

1276 if model == "" then model = "embeddinggemma:latest" end

1278 local output_dir = get_similarity_output_dir(model)

1279 utils.log_info("Checking " .. output_dir .. " ...")

1281 -- Quick count first (no validation)

1282 local quick_count_handle = io.popen("find '" .. output_dir .. "' -name 'poem_*.json' 2>/dev/null | wc -l")

1283 local quick_count = 0

1284 if quick_count_handle then

1285 quick_count = tonumber(quick_count_handle:read("*a")) or 0

1286 quick_count_handle:close()

1287 end

1288 utils.log_info("Found " .. quick_count .. " similarity files (validating...)")

1290 local completed_count = count_completed_poems(output_dir, 0, thread_count)

1292 utils.log_info("✅ Valid similarity files: " .. completed_count .. "/" .. quick_count)

1293 end

1295 return true