mirror of
https://github.com/frappe/gunicorn.git
synced 2026-07-01 10:11:30 +08:00
8559854b4f
Allow dirty apps to specify how many workers should load them, enabling significant memory savings for heavy applications like ML models. - Add `workers` class attribute to DirtyApp (None = all workers) - Add `parse_dirty_app_spec()` to parse "module:Class:N" format - Add `DirtyNoWorkersAvailableError` for app-specific error handling - Update DirtyArbiter with per-app worker tracking and routing - Maintain backward compatibility when no dirty_apps configured Example: 8 workers x 10GB model = 80GB RAM needed With workers=2: 2 x 10GB = 20GB RAM (75% savings) Configuration formats: - Class attribute: `workers = 2` on DirtyApp subclass - Config format: `module:class:N` (e.g., `myapp.ml:HugeModel:2`)
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
#
|
|
# This file is part of gunicorn released under the MIT license.
|
|
# See the NOTICE for more information.
|
|
|
|
"""Tests for dirty errors module."""
|
|
|
|
import pytest
|
|
|
|
from gunicorn.dirty.errors import (
|
|
DirtyError,
|
|
DirtyNoWorkersAvailableError,
|
|
)
|
|
|
|
|
|
class TestDirtyNoWorkersAvailableError:
|
|
"""Tests for DirtyNoWorkersAvailableError exception."""
|
|
|
|
def test_error_contains_app_path(self):
|
|
"""Error includes the app_path."""
|
|
error = DirtyNoWorkersAvailableError("myapp:Model")
|
|
assert error.app_path == "myapp:Model"
|
|
assert "myapp:Model" in str(error)
|
|
assert "No workers available" in str(error)
|
|
|
|
def test_error_with_custom_message(self):
|
|
"""Error can have a custom message."""
|
|
error = DirtyNoWorkersAvailableError(
|
|
"myapp:Model",
|
|
message="Custom: no workers for heavy model"
|
|
)
|
|
assert error.app_path == "myapp:Model"
|
|
assert "Custom: no workers" in str(error)
|
|
|
|
def test_error_serialization_roundtrip(self):
|
|
"""Error survives to_dict/from_dict cycle."""
|
|
original = DirtyNoWorkersAvailableError("myapp.ml:HugeModel")
|
|
|
|
# Serialize
|
|
data = original.to_dict()
|
|
assert data["error_type"] == "DirtyNoWorkersAvailableError"
|
|
assert data["details"]["app_path"] == "myapp.ml:HugeModel"
|
|
|
|
# Deserialize
|
|
restored = DirtyError.from_dict(data)
|
|
assert isinstance(restored, DirtyNoWorkersAvailableError)
|
|
assert restored.app_path == "myapp.ml:HugeModel"
|
|
assert "No workers available" in str(restored)
|
|
|
|
def test_error_is_dirty_error_subclass(self):
|
|
"""DirtyNoWorkersAvailableError is a DirtyError subclass."""
|
|
error = DirtyNoWorkersAvailableError("app:Class")
|
|
assert isinstance(error, DirtyError)
|
|
|
|
def test_web_app_can_catch_specific_error(self):
|
|
"""Web app can catch DirtyNoWorkersAvailableError specifically."""
|
|
def simulate_execute():
|
|
raise DirtyNoWorkersAvailableError("myapp:HeavyModel")
|
|
|
|
# Catch specific error
|
|
try:
|
|
simulate_execute()
|
|
assert False, "Should have raised"
|
|
except DirtyNoWorkersAvailableError as e:
|
|
assert e.app_path == "myapp:HeavyModel"
|
|
|
|
def test_can_catch_as_base_error(self):
|
|
"""Can catch DirtyNoWorkersAvailableError as DirtyError."""
|
|
def simulate_execute():
|
|
raise DirtyNoWorkersAvailableError("myapp:Model")
|
|
|
|
try:
|
|
simulate_execute()
|
|
assert False, "Should have raised"
|
|
except DirtyError as e:
|
|
# Should catch it as the base class
|
|
assert hasattr(e, "app_path")
|