46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
import sys
|
|
import os
|
|
import threading
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.abspath('gateway'))
|
|
import pdfengine
|
|
|
|
def worker(thread_id: int, iterations: int):
|
|
pdf_files = ['utf-8.pdf']
|
|
for i in range(iterations):
|
|
for pdf_file in pdf_files:
|
|
try:
|
|
filepath = os.path.abspath(f"corpus/fonts/{pdf_file}")
|
|
doc = pdfengine.PdfDocument.load_from_file(filepath)
|
|
page = doc.get_page(0)
|
|
|
|
fonts = doc.get_fonts(0, -1)
|
|
|
|
glyphs = page.extract_text_with_bounds()
|
|
|
|
except Exception as e:
|
|
print(f"[Thread {thread_id}] Error: {e}")
|
|
|
|
def run_concurrency_test(num_threads: int, iterations: int):
|
|
print(f"Starting concurrency test with {num_threads} threads, {iterations} iterations each...")
|
|
start_time = time.time()
|
|
|
|
threads = []
|
|
for i in range(num_threads):
|
|
t = threading.Thread(target=worker, args=(i, iterations))
|
|
threads.append(t)
|
|
t.start()
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
elapsed = time.time() - start_time
|
|
print(f"Concurrency test completed in {elapsed:.2f} seconds.")
|
|
print("No crashes or deadlocks detected.")
|
|
|
|
if __name__ == "__main__":
|
|
run_concurrency_test(10, 20)
|
|
|
|
run_concurrency_test(50, 10)
|