vicilliar
commented
1 year ago Models endpoint after the bug occurred:
{'models': [{'model_name': 'hf/all_datasets_v4_MiniLM-L6', 'model_device': 'cpu'}, {'model_name': 'hf/all_datasets_v4_MiniLM-L6', 'model_device': 'cuda'}, {'model_name': 'ViT-L/14', 'model_device': 'cpu'}, {'model_name': 'ViT-L/14', 'model_device': 'cuda'}, {'model_name': 'hf/e5-base', 'model_device': 'cuda'}]}
Describe the bug A 500 error is received a few minutes through trying to index the simplewiki dataset with the
hf/e5-basemodel with multiple client threads.Error Messages From the client terminal:
From the marqo terminal:
To Reproduce Steps to reproduce the behavior:
Run a marqo instance (0.0.21 tag)
Run a managed OpenSearch instance
Run a Sagemaker client
Run the following multithreaded indexing script
try: mq.create_index(index_name, settings_dict=index_settings) print(f"Successfully created index {index_name}!")
except Exception as e: print(f"ERROR CREATING INDEX: {e}") print(f"Index {index_name} already exists! Skipping creation.")
print("Confirming index settings are: ") pprint.pprint(mq.index(index_name).get_settings())
#####################################################
STEP 3. indexing with marqo
#####################################################
TODO: track add docs throughput
Add documents in parallel
def send_docs(docs, error_queue): try: responses = mq.index(index_name).add_documents(docs, device=args.device, client_batch_size=10, non_tensor_fields=["docDate", "domain", "url"]) except Exception as e: error_queue.put(e)
t0 = time.time()
THREAD_COUNT = 7 split_size = math.ceil(len(data)/THREAD_COUNT) splits = [data[i: i + split_size] for i in range(0, len(data), split_size)] allocation_dict = { i: splits[i] for i in range(THREAD_COUNT) }
error_queue = queue.Queue() threads = [threading.Thread(target=senddocs, args=(split, errorqueue)) for split in splits] for thread in threads: thread.start() for thread in threads: thread.join()