change default to main thread
This commit is contained in:
parent
3828385fd1
commit
a15c7061b4
100
reflex/app.py
100
reflex/app.py
@ -17,6 +17,7 @@ import sys
|
|||||||
import traceback
|
import traceback
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from timeit import default_timer as timer
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
from typing import (
|
from typing import (
|
||||||
TYPE_CHECKING,
|
TYPE_CHECKING,
|
||||||
@ -73,7 +74,7 @@ from reflex.components.core.client_side_routing import (
|
|||||||
from reflex.components.core.sticky import sticky
|
from reflex.components.core.sticky import sticky
|
||||||
from reflex.components.core.upload import Upload, get_upload_dir
|
from reflex.components.core.upload import Upload, get_upload_dir
|
||||||
from reflex.components.radix import themes
|
from reflex.components.radix import themes
|
||||||
from reflex.config import environment, get_config
|
from reflex.config import ExecutorType, environment, get_config
|
||||||
from reflex.event import (
|
from reflex.event import (
|
||||||
_EVENT_FIELDS,
|
_EVENT_FIELDS,
|
||||||
Event,
|
Event,
|
||||||
@ -1062,10 +1063,23 @@ class App(MiddlewareMixin, LifespanMixin):
|
|||||||
app_wrappers[(1, "ToasterProvider")] = toast_provider
|
app_wrappers[(1, "ToasterProvider")] = toast_provider
|
||||||
|
|
||||||
with console.timing("Evaluate Pages (Frontend)"):
|
with console.timing("Evaluate Pages (Frontend)"):
|
||||||
|
performance_metrics: list[tuple[str, float]] = []
|
||||||
for route in self._unevaluated_pages:
|
for route in self._unevaluated_pages:
|
||||||
console.debug(f"Evaluating page: {route}")
|
console.debug(f"Evaluating page: {route}")
|
||||||
|
start = timer()
|
||||||
self._compile_page(route, save_page=should_compile)
|
self._compile_page(route, save_page=should_compile)
|
||||||
|
end = timer()
|
||||||
|
performance_metrics.append((route, end - start))
|
||||||
progress.advance(task)
|
progress.advance(task)
|
||||||
|
console.debug(
|
||||||
|
"Slowest pages:\n"
|
||||||
|
+ "\n".join(
|
||||||
|
f"{route}: {time * 1000:.1f}ms"
|
||||||
|
for route, time in sorted(
|
||||||
|
performance_metrics, key=lambda x: x[1], reverse=True
|
||||||
|
)[:10]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
# Add the optional endpoints (_upload)
|
# Add the optional endpoints (_upload)
|
||||||
self._add_optional_endpoints()
|
self._add_optional_endpoints()
|
||||||
@ -1157,51 +1171,69 @@ class App(MiddlewareMixin, LifespanMixin):
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
executor_type = environment.REFLEX_COMPILE_EXECUTOR.get()
|
||||||
|
|
||||||
# Use a forking process pool, if possible. Much faster, especially for large sites.
|
# Use a forking process pool, if possible. Much faster, especially for large sites.
|
||||||
# Fallback to ThreadPoolExecutor as something that will always work.
|
# Fallback to ThreadPoolExecutor as something that will always work.
|
||||||
executor = None
|
if executor_type is None:
|
||||||
if environment.REFLEX_COMPILE_USE_MAIN_THREAD.get():
|
reflex_compile_processes = environment.REFLEX_COMPILE_PROCESSES.get()
|
||||||
T = TypeVar("T")
|
reflex_compile_threads = environment.REFLEX_COMPILE_THREADS.get()
|
||||||
|
if (
|
||||||
|
platform.system() not in ("Linux", "Darwin")
|
||||||
|
and reflex_compile_processes is not None
|
||||||
|
):
|
||||||
|
console.warn("Multiprocessing is only supported on Linux and MacOS.")
|
||||||
|
|
||||||
class MainThreadExecutor:
|
if (
|
||||||
def __enter__(self):
|
platform.system() in ("Linux", "Darwin")
|
||||||
return self
|
and reflex_compile_processes is not None
|
||||||
|
):
|
||||||
|
executor_type = ExecutorType.PROCESS
|
||||||
|
elif reflex_compile_threads is not None:
|
||||||
|
executor_type = ExecutorType.THREAD
|
||||||
|
else:
|
||||||
|
executor_type = ExecutorType.MAIN_THREAD
|
||||||
|
|
||||||
def __exit__(self, *args):
|
match executor_type:
|
||||||
pass
|
case ExecutorType.PROCESS:
|
||||||
|
executor = concurrent.futures.ProcessPoolExecutor(
|
||||||
|
max_workers=environment.REFLEX_COMPILE_PROCESSES.get(),
|
||||||
|
mp_context=multiprocessing.get_context("fork"),
|
||||||
|
)
|
||||||
|
case ExecutorType.THREAD:
|
||||||
|
executor = concurrent.futures.ThreadPoolExecutor(
|
||||||
|
max_workers=environment.REFLEX_COMPILE_THREADS.get()
|
||||||
|
)
|
||||||
|
case ExecutorType.MAIN_THREAD:
|
||||||
|
T = TypeVar("T")
|
||||||
|
|
||||||
def submit(
|
class MainThreadExecutor:
|
||||||
self, fn: Callable[..., T], *args, **kwargs
|
def __enter__(self):
|
||||||
) -> concurrent.futures.Future[T]:
|
return self
|
||||||
future_job = concurrent.futures.Future()
|
|
||||||
future_job.set_result(fn(*args, **kwargs))
|
|
||||||
return future_job
|
|
||||||
|
|
||||||
executor = MainThreadExecutor()
|
def __exit__(self, *args):
|
||||||
elif (
|
pass
|
||||||
platform.system() in ("Linux", "Darwin")
|
|
||||||
and (number_of_processes := environment.REFLEX_COMPILE_PROCESSES.get())
|
def submit(
|
||||||
is not None
|
self, fn: Callable[..., T], *args, **kwargs
|
||||||
):
|
) -> concurrent.futures.Future[T]:
|
||||||
executor = concurrent.futures.ProcessPoolExecutor(
|
future_job = concurrent.futures.Future()
|
||||||
max_workers=number_of_processes or None,
|
future_job.set_result(fn(*args, **kwargs))
|
||||||
mp_context=multiprocessing.get_context("fork"),
|
return future_job
|
||||||
)
|
|
||||||
else:
|
executor = MainThreadExecutor()
|
||||||
executor = concurrent.futures.ThreadPoolExecutor(
|
|
||||||
max_workers=environment.REFLEX_COMPILE_THREADS.get() or None
|
|
||||||
)
|
|
||||||
|
|
||||||
for route, component in zip(self._pages, page_components, strict=True):
|
for route, component in zip(self._pages, page_components, strict=True):
|
||||||
ExecutorSafeFunctions.COMPONENTS[route] = component
|
ExecutorSafeFunctions.COMPONENTS[route] = component
|
||||||
|
|
||||||
ExecutorSafeFunctions.STATE = self._state
|
ExecutorSafeFunctions.STATE = self._state
|
||||||
|
|
||||||
with executor:
|
with console.timing("Compile to Javascript"), executor as executor:
|
||||||
result_futures: list[concurrent.futures.Future[tuple[str, str]]] = []
|
result_futures: list[concurrent.futures.Future[tuple[str, str]]] = []
|
||||||
|
|
||||||
def _submit_work(fn: Callable[..., tuple[str, str]], *args, **kwargs):
|
def _submit_work(fn: Callable[..., tuple[str, str]], *args, **kwargs):
|
||||||
f = executor.submit(fn, *args, **kwargs)
|
f = executor.submit(fn, *args, **kwargs)
|
||||||
|
f.add_done_callback(lambda _: progress.advance(task))
|
||||||
result_futures.append(f)
|
result_futures.append(f)
|
||||||
|
|
||||||
# Compile the pre-compiled pages.
|
# Compile the pre-compiled pages.
|
||||||
@ -1227,10 +1259,10 @@ class App(MiddlewareMixin, LifespanMixin):
|
|||||||
_submit_work(compiler.remove_tailwind_from_postcss)
|
_submit_work(compiler.remove_tailwind_from_postcss)
|
||||||
|
|
||||||
# Wait for all compilation tasks to complete.
|
# Wait for all compilation tasks to complete.
|
||||||
with console.timing("Compile to Javascript"):
|
compile_results.extend(
|
||||||
for future in concurrent.futures.as_completed(result_futures):
|
future.result()
|
||||||
compile_results.append(future.result())
|
for future in concurrent.futures.as_completed(result_futures)
|
||||||
progress.advance(task)
|
)
|
||||||
|
|
||||||
app_root = self._app_root(app_wrappers=app_wrappers)
|
app_root = self._app_root(app_wrappers=app_wrappers)
|
||||||
|
|
||||||
|
@ -481,6 +481,14 @@ class PerformanceMode(enum.Enum):
|
|||||||
OFF = "off"
|
OFF = "off"
|
||||||
|
|
||||||
|
|
||||||
|
class ExecutorType(enum.Enum):
|
||||||
|
"""Executor for compiling the frontend."""
|
||||||
|
|
||||||
|
THREAD = "thread"
|
||||||
|
PROCESS = "process"
|
||||||
|
MAIN_THREAD = "main_thread"
|
||||||
|
|
||||||
|
|
||||||
class EnvironmentVariables:
|
class EnvironmentVariables:
|
||||||
"""Environment variables class to instantiate environment variables."""
|
"""Environment variables class to instantiate environment variables."""
|
||||||
|
|
||||||
@ -522,8 +530,7 @@ class EnvironmentVariables:
|
|||||||
Path(constants.Dirs.UPLOADED_FILES)
|
Path(constants.Dirs.UPLOADED_FILES)
|
||||||
)
|
)
|
||||||
|
|
||||||
# Whether to use the main thread for compiling the frontend. Takes precedence over other compile options.
|
REFLEX_COMPILE_EXECUTOR: EnvVar[Optional[ExecutorType]] = env_var(None)
|
||||||
REFLEX_COMPILE_USE_MAIN_THREAD: EnvVar[bool] = env_var(False)
|
|
||||||
|
|
||||||
# Whether to use separate processes to compile the frontend and how many. If not set, defaults to thread executor.
|
# Whether to use separate processes to compile the frontend and how many. If not set, defaults to thread executor.
|
||||||
REFLEX_COMPILE_PROCESSES: EnvVar[Optional[int]] = env_var(None)
|
REFLEX_COMPILE_PROCESSES: EnvVar[Optional[int]] = env_var(None)
|
||||||
|
Loading…
Reference in New Issue
Block a user