Spaces:
Runtime error
Runtime error
File size: 2,223 Bytes
d5b7ee9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | #!/usr/bin/env python
"""Test script to verify FinBERT loads correctly with multithreading."""
import sys
import threading
import time
def load_finbert_in_thread(thread_id: int):
"""Load FinBERT in a thread to test the workaround."""
print(f"[Thread {thread_id}] Starting FinBERT load...")
from trading_cli.sentiment.finbert import FinBERTAnalyzer
analyzer = FinBERTAnalyzer.get_instance()
def progress_callback(msg: str):
print(f"[Thread {thread_id}] Progress: {msg}")
success = analyzer.load(progress_callback=progress_callback)
if success:
print(f"[Thread {thread_id}] ✓ FinBERT loaded successfully!")
# Test inference
result = analyzer.analyze_batch(["Test headline for sentiment analysis"])
print(f"[Thread {thread_id}] Test result: {result}")
else:
print(f"[Thread {thread_id}] ✗ FinBERT failed to load: {analyzer.load_error}")
return success
def main():
print("=" * 60)
print("Testing FinBERT multithreaded loading with fds_to_keep workaround")
print("=" * 60)
# Try loading in multiple threads to trigger the issue
threads = []
results = []
for i in range(3):
t = threading.Thread(target=lambda idx=i: results.append(load_finbert_in_thread(idx)))
threads.append(t)
t.start()
time.sleep(0.5) # Small delay between thread starts
# Wait for all threads to complete
for t in threads:
t.join()
print("\n" + "=" * 60)
print("Test Results:")
print("=" * 60)
# The singleton should only load once
if len(results) > 0:
print(f"✓ At least one thread attempted loading")
if any(results):
print(f"✓ FinBERT loaded successfully in multithreaded context")
print("\n✅ TEST PASSED - fds_to_keep workaround is working!")
return 0
else:
print(f"✗ All threads failed to load FinBERT")
print("\n❌ TEST FAILED - workaround did not resolve the issue")
return 1
else:
print("✗ No threads completed")
return 1
if __name__ == "__main__":
sys.exit(main())
|