Serverside search.

This commit is contained in:
2024-04-10 13:17:58 +02:00
parent 1a14b9b520
commit 41eb0f9245
6 changed files with 161860 additions and 2834 deletions
+41 -22
View File
@@ -2,15 +2,19 @@ from queue import Queue, Empty
from threading import Thread
q = Queue()
#MAXTHREADS = 39
MAXTHREADS = 4
DATABASE_PATH = r'C:\\Database\\chunks_1\\'
#TODO: TO w klasie jako parametry wybierane czy robimy pelen run czy testowy
#MAXTHREADS = 3
MAXTHREADS = 1
#DATABASE_PATH = r'C:\\Database\\chunks_1\\'
DATABASE_PATH = r'C:\\Database\\chunks\\'
ENCODING = "utf-8"
CHUNk = "_chunk.txt"
_sentinel = object()
result_list = []
#TODO: Doisy wchodzą jako lista krotek (doi, False) - po znalezieniu zaznaczamy na True i potem już nie porównujemy
#TODO: lista krotek - jeśli wszystkie mają True dajemy terma
#TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
def producer(out_q, control_q, filename):
with open(DATABASE_PATH + filename, "r+", encoding=ENCODING) as operated_file:
@@ -26,55 +30,70 @@ def producer(out_q, control_q, filename):
print("TERM signal received")
control_q.put(check)
break
print("Worker finished")
print(f"Worker finished: {filename}")
out_q.put(_sentinel)
#IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK
#TODO: Change doi to tuple list and only put term when all doi are found => (doi, True)
def consumer(in_q, control_q, doi):
def consumer(in_q, control_q, doi, live_results, logger):
sentinels = 0
for item in doi:
result_list.append({"DOI" : item[0], "exists" : False, "data" : item[1]})
while True:
#if in_q.empty():
# print("Queue empty!")
#if in_q.full():
# print("Queue full!")
done_check = True
data = in_q.get()
if data is _sentinel:
print("Worker finished")
logger.info("Worker finished")
sentinels += 1
else:
for item in result_list:
if item["DOI"] in data:
print(data)
print("HIT")
logger.info(data)
logger.info("HIT")
item["exists"] = True
live_results.append(item)
done_check = done_check and item["exists"]
if done_check:
control_q.put(_sentinel)
if sentinels >= MAXTHREADS:
print("All workers finished")
logger.info("All workers finished")
break
def search_for_doi(doi):
def search_for_doi(doi, live_results, logger):
threads = []
work_q = Queue(maxsize=100)
work_q = Queue(maxsize=200000)
control_q = Queue()
t_cons = Thread(target = consumer, args = (work_q, control_q, doi))
t_cons = Thread(target = consumer, args = (work_q, control_q, doi, live_results, logger))
logger.info("Consumer thread created")
threads.append(t_cons)
for i in range(0,MAXTHREADS):
filename = "test" + str(i) + CHUNk
filename = str(i) + CHUNk
#print(filename)
#filename = "test" + str(i) + CHUNk
filename = "40" + CHUNk
#filename = str(i) + CHUNk
logger.info(f"Creating worker thread no: {i}")
threads.append(Thread(target = producer, args = (work_q, control_q, filename)))
for worker in threads:
worker.start()
for worker in threads:
worker.join()
return result_list
if __name__ == "__main__":
print(search_for_doi([
import logging
logger = logging.getLogger()
logger.setLevel("DEBUG")
h1 = logging.StreamHandler()
logger.addHandler(h1)
logger.info("TEST RUN")
live_result = []
logger.info(search_for_doi([
("10.1002/9781118786352.wbieg0998.pub2","DATA"),
("10.1002/j.2050-0416.2002.tb00563.x","DATA"),
("10.1111/j.1365-2958.1994.tb00448.x","DATA"),
@@ -83,5 +102,5 @@ if __name__ == "__main__":
("10.15803/ijnc.7.2_419","DATA"),
("10.1051/0004-6361/201321596e","DATA"),
("10.2307/40835941","DATA"),
]))
], live_result, logger))
logger.info(live_result)