highperfocused/danswer
1
0
Fork 0
mirror of https://github.com/danswer-ai/danswer.git synced 2025-10-10 05:05:34 +02:00
Code Issues Packages Projects Releases Wiki Activity

saved files

Browse Source
This commit is contained in:
Richard Kuo (Danswer)
2025-01-13 10:46:20 -08:00
parent f5bdf9d2c9
commit be3cfdd4a6
3 changed files with 1 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
backend/onyx/background/celery/apps/heavy.py
View File

@@ -1,4 +1,3 @@
import multiprocessing
from typing import Any from typing import Any
from celery import Celery from celery import Celery
@@ -57,24 +56,6 @@ def on_celeryd_init(sender: Any = None, conf: Any = None, **kwargs: Any) -> None
def on_worker_init(sender: Any, **kwargs: Any) -> None: def on_worker_init(sender: Any, **kwargs: Any) -> None:
logger.info("worker_init signal received.") logger.info("worker_init signal received.")
all_start_methods: list[str] = multiprocessing.get_all_start_methods()
logger.info(f"Multiprocessing all start methods: {all_start_methods}")
try:
multiprocessing.set_start_method("spawn") # fork is unsafe, set to spawn
except Exception:
logger.info("multiprocessing.set_start_method exceptioned.")
try:
multiprocessing.set_start_method(
"spawn", force=True
) # fork is unsafe, set to spawn
except Exception:
logger.info("multiprocessing.set_start_method force=True exceptioned.")
logger.info(
f"Multiprocessing selected start method: {multiprocessing.get_start_method()}"
)
SqlEngine.set_app_name(POSTGRES_CELERY_WORKER_HEAVY_APP_NAME) SqlEngine.set_app_name(POSTGRES_CELERY_WORKER_HEAVY_APP_NAME)
SqlEngine.init_engine(pool_size=4, max_overflow=12) SqlEngine.init_engine(pool_size=4, max_overflow=12)

13
backend/onyx/background/celery/apps/indexing.py
View File

@@ -1,4 +1,3 @@
import multiprocessing
from typing import Any from typing import Any
from celery import Celery from celery import Celery
@@ -58,21 +57,11 @@ def on_celeryd_init(sender: Any = None, conf: Any = None, **kwargs: Any) -> None
def on_worker_init(sender: Any, **kwargs: Any) -> None: def on_worker_init(sender: Any, **kwargs: Any) -> None:
logger.info("worker_init signal received.") logger.info("worker_init signal received.")
all_start_methods: list[str] = multiprocessing.get_all_start_methods()
logger.info(f"Multiprocessing all start methods: {all_start_methods}")
multiprocessing.set_start_method("spawn") # fork is unsafe, set to spawn
logger.info(
f"Multiprocessing selected start method: {multiprocessing.get_start_method()}"
)
SqlEngine.set_app_name(POSTGRES_CELERY_WORKER_INDEXING_APP_NAME) SqlEngine.set_app_name(POSTGRES_CELERY_WORKER_INDEXING_APP_NAME)
# rkuo: been seeing transient connection exceptions here, so upping the connection count # rkuo: been seeing transient connection exceptions here, so upping the connection count
# from just concurrency/concurrency to concurrency/concurrency*2 # from just concurrency/concurrency to concurrency/concurrency*2
SqlEngine.init_engine( SqlEngine.init_engine(pool_size=sender.concurrency, max_overflow=8)
pool_size=sender.concurrency, max_overflow=sender.concurrency * 2
)
app_base.wait_for_redis(sender, **kwargs) app_base.wait_for_redis(sender, **kwargs)
app_base.wait_for_db(sender, **kwargs) app_base.wait_for_db(sender, **kwargs)

9
backend/onyx/background/celery/apps/primary.py
View File

@@ -1,5 +1,4 @@
import logging import logging
import multiprocessing
from typing import Any from typing import Any
from typing import cast from typing import cast
@@ -81,14 +80,6 @@ def on_celeryd_init(sender: Any = None, conf: Any = None, **kwargs: Any) -> None
def on_worker_init(sender: Any, **kwargs: Any) -> None: def on_worker_init(sender: Any, **kwargs: Any) -> None:
logger.info("worker_init signal received.") logger.info("worker_init signal received.")
all_start_methods: list[str] = multiprocessing.get_all_start_methods()
logger.info(f"Multiprocessing all start methods: {all_start_methods}")
multiprocessing.set_start_method("spawn") # fork is unsafe, set to spawn
logger.info(
f"Multiprocessing selected start method: {multiprocessing.get_start_method()}"
)
SqlEngine.set_app_name(POSTGRES_CELERY_WORKER_PRIMARY_APP_NAME) SqlEngine.set_app_name(POSTGRES_CELERY_WORKER_PRIMARY_APP_NAME)
SqlEngine.init_engine(pool_size=8, max_overflow=0) SqlEngine.init_engine(pool_size=8, max_overflow=0)