⚡️ Speed up function get_cache_service by 190% in PR #10768 (fix-agent-flow-lfx)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #10768

If you approve this dependent PR, these changes will be merged into the original PR branch fix-agent-flow-lfx.

This PR will be automatically closed if the original PR is merged.

---

📄 190% (1.90x) speedup for get_cache_service in src/backend/base/langflow/services/deps.py

⏱️ Runtime : 133 microseconds → 45.9 microseconds (best of 134 runs)

📝 Explanation and details

The optimized code achieves a 190% speedup by eliminating redundant operations through strategic caching of expensive resources at module level.

Key Optimizations:

1. Service Manager Caching: The original code imported and called get_service_manager() on every invocation (120 times in the profile). The optimization caches this in _service_manager global variable, reducing it to a one-time cost.

2. Factory Registration Tracking: Added _factories_registered flag to prevent redundant calls to register_factories(). The original code checked are_factories_registered() 120 times and called the expensive register_factories() method unnecessarily.

3. Factory Instance Caching: The CacheServiceFactory() instantiation is cached in _cache_service_factory_instance, eliminating repeated object creation (was called 120 times, now just once).

Performance Impact Analysis:

From the line profiler results, the most expensive operations were:

• service_manager.get(): ~27ms in both versions (unchanged, as expected)
• Import and manager setup: Reduced from ~0.36ms to ~0.05ms per call
• Factory instantiation in get_cache_service(): Moved from per-call to one-time cost

The optimizations are particularly effective for repeated calls - evident from the test cases calling these functions 50-100 times. The caching strategy transforms O(n) redundant work into O(1) amortized cost.

Workload Benefits:

• High-frequency service access: Applications making many cache service requests will see significant gains
• Service initialization hotpaths: Reduces startup overhead when services are frequently accessed during application bootstrap
• Memory efficiency: Prevents creation of redundant factory instances

The optimizations maintain identical behavior and thread safety while dramatically improving performance for repeated invocations.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 117 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

from __future__ import annotations

import sys
import types

# imports
import pytest
from langflow.services.deps import get_cache_service

# --- End: test scaffolding and helpers ---

# --- Begin: Unit tests for get_cache_service ---

# 1. Basic Test Cases

def test_returns_cache_service_instance_when_not_registered():
    """
    Scenario: No factories registered yet.
    Expectation: get_cache_service returns the instance from the default CacheServiceFactory.
    """
    # The DummyServiceManager starts with _registered = False, so register_factories will be called
    codeflash_output = get_cache_service(); result = codeflash_output

def test_returns_cache_service_instance_when_already_registered():
    """
    Scenario: Factories are already registered.
    Expectation: get_cache_service returns the instance from the service manager without registering again.
    """
    # Register factories first
    from lfx.services.manager import get_service_manager
    mgr = get_service_manager()
    mgr.register_factories({"cache_service": "already_registered_cache_service"})
    # Now call get_cache_service
    codeflash_output = get_cache_service(); result = codeflash_output

def test_returns_factory_instance_if_no_registered_and_no_factory():
    """
    Scenario: No factories registered, but default factory returns a value.
    Expectation: get_cache_service returns what the default factory returns.
    """
    # Patch the DummyServiceManager to return nothing for get_factories
    import langflow.services.manager
    class ServiceManager:
        @staticmethod
        def get_factories():
            return {}
    langflow.services.manager.ServiceManager = ServiceManager
    # Should fall back to the default factory
    codeflash_output = get_cache_service(); result = codeflash_output

# 2. Edge Test Cases

def test_factory_returns_none(monkeypatch):
    """
    Scenario: The default CacheServiceFactory returns None.
    Expectation: get_cache_service returns None.
    """
    # Patch CacheServiceFactory to return None
    class NoneFactory:
        def __call__(self):
            return None
    sys.modules["langflow.services.cache.factory"].CacheServiceFactory = NoneFactory
    # Patch get_factories to return empty so default is used
    import langflow.services.manager
    class ServiceManager:
        @staticmethod
        def get_factories():
            return {}
    langflow.services.manager.ServiceManager = ServiceManager
    codeflash_output = get_cache_service(); result = codeflash_output

def test_service_manager_get_returns_none(monkeypatch):
    """
    Scenario: The service manager's get method returns None.
    Expectation: get_cache_service returns None.
    """
    # Patch DummyServiceManager.get to always return None
    from lfx.services.manager import get_service_manager
    mgr = get_service_manager()
    def always_none(service_type, default):
        return None
    mgr.get = always_none
    codeflash_output = get_cache_service(); result = codeflash_output



def test_many_factories_registered():
    """
    Scenario: Large number of factories registered.
    Expectation: get_cache_service still returns the correct cache service.
    """
    from lfx.services.manager import get_service_manager

    # Register 999 unrelated services and one cache service
    many_factories = {f"service_{i}": f"instance_{i}" for i in range(999)}
    many_factories["cache_service"] = "big_cache_service"
    mgr = get_service_manager()
    mgr.register_factories(many_factories)
    codeflash_output = get_cache_service(); result = codeflash_output

def test_get_cache_service_multiple_calls_consistency():
    """
    Scenario: Call get_cache_service repeatedly.
    Expectation: Always returns the same cache service instance.
    """
    from lfx.services.manager import get_service_manager
    mgr = get_service_manager()
    mgr.register_factories({"cache_service": "persistent_cache_service"})
    results = [get_cache_service() for _ in range(50)]

def test_performance_with_large_number_of_calls():
    """
    Scenario: Call get_cache_service many times with large number of registered services.
    Expectation: No performance degradation, all return correct instance.
    """
    from lfx.services.manager import get_service_manager
    many_factories = {f"service_{i}": f"instance_{i}" for i in range(999)}
    many_factories["cache_service"] = "large_scale_cache_service"
    mgr = get_service_manager()
    mgr.register_factories(many_factories)
    for _ in range(100):
        codeflash_output = get_cache_service(); result = codeflash_output
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import sys
import types
# --- Function to test (from langflow/services/deps.py) ---
from typing import Any

# imports
import pytest
from langflow.services.deps import get_cache_service


# Minimal stubs to support the test environment, since we can't import actual langflow code.
class ServiceType:
    CACHE_SERVICE = "CACHE_SERVICE"
    OTHER_SERVICE = "OTHER_SERVICE"

class CacheService:
    def __init__(self, name="sync"):
        self.name = name
    def get(self, key): return f"got-{key}"
    def set(self, key, value): return f"set-{key}-{value}"

class AsyncBaseCacheService:
    def __init__(self, name="async"):
        self.name = name
    async def aget(self, key): return f"agot-{key}"
    async def aset(self, key, value): return f"aset-{key}-{value}"

class CacheServiceFactory:
    def __init__(self, kind="sync"):
        self.kind = kind
    def __call__(self):
        if self.kind == "sync":
            return CacheService()
        else:
            return AsyncBaseCacheService()

class ServiceManager:
    _factories_registered = False
    _factories = {ServiceType.CACHE_SERVICE: CacheServiceFactory()}
    _services = {}

    @classmethod
    def get_factories(cls):
        return cls._factories

    def are_factories_registered(self):
        return self._factories_registered

    def register_factories(self, factories):
        self._factories_registered = True
        self._factories = factories

    def get(self, service_type, default=None):
        if service_type in self._services:
            return self._services[service_type]
        if service_type in self._factories:
            self._services[service_type] = self._factories[service_type]()
            return self._services[service_type]
        if default is not None:
            return default()
        raise ValueError("Service not found")

# Global singleton for service manager (simulating import)
_service_manager_instance = ServiceManager()
from langflow.services.deps import get_cache_service

# --- Unit tests ---

# 1. Basic Test Cases

def test_get_cache_service_returns_cache_service_instance():
    """
    Test that get_cache_service returns an instance of CacheService or AsyncBaseCacheService.
    """
    codeflash_output = get_cache_service(); result = codeflash_output

def test_get_cache_service_returns_same_instance_on_subsequent_calls():
    """
    Test that get_cache_service returns the same instance on multiple calls (singleton behavior).
    """
    codeflash_output = get_cache_service(); a = codeflash_output
    codeflash_output = get_cache_service(); b = codeflash_output


def test_get_cache_service_when_factories_already_registered(monkeypatch):
    """
    Test that get_cache_service works when factories are already registered.
    """
    # Register factories first
    _service_manager_instance.register_factories(ServiceManager.get_factories())
    # Should not re-register, but still work
    codeflash_output = get_cache_service(); result = codeflash_output

def test_get_cache_service_with_custom_factory(monkeypatch):
    """
    Test that get_cache_service can use a custom factory if provided.
    """
    # Patch the factory to return an async cache service
    class CustomFactory:
        def __call__(self):
            return AsyncBaseCacheService("custom-async")
    _service_manager_instance._factories[ServiceType.CACHE_SERVICE] = CustomFactory()
    # Remove existing service to force re-creation
    _service_manager_instance._services.pop(ServiceType.CACHE_SERVICE, None)
    codeflash_output = get_cache_service(); result = codeflash_output


def test_get_cache_service_is_isolated_from_other_services():
    """
    Test that get_cache_service does not interfere with other service types.
    """
    # Add a dummy service for OTHER_SERVICE
    class DummyService: pass
    _service_manager_instance._factories[ServiceType.OTHER_SERVICE] = lambda: DummyService()
    _service_manager_instance._services[ServiceType.OTHER_SERVICE] = DummyService()
    # Call get_cache_service and ensure OTHER_SERVICE is unaffected
    codeflash_output = get_cache_service(); cache = codeflash_output

def test_get_cache_service_with_unusual_factory(monkeypatch):
    """
    Test that get_cache_service works if the factory returns a nonstandard object.
    """
    class WeirdCache:
        pass
    class WeirdFactory:
        def __call__(self):
            return WeirdCache()
    _service_manager_instance._factories[ServiceType.CACHE_SERVICE] = WeirdFactory()
    _service_manager_instance._services.pop(ServiceType.CACHE_SERVICE, None)
    codeflash_output = get_cache_service(); result = codeflash_output
    # Restore default
    _service_manager_instance._factories[ServiceType.CACHE_SERVICE] = CacheServiceFactory()
    _service_manager_instance._services.pop(ServiceType.CACHE_SERVICE, None)

# 3. Large Scale Test Cases


def test_get_cache_service_with_large_number_of_services():
    """
    Test that get_cache_service works when the service manager manages many services.
    """
    # Add a bunch of dummy services
    for i in range(500):
        stype = f"DUMMY_{i}"
        _service_manager_instance._factories[stype] = (lambda n=i: lambda: f"service-{n}")()
    # Now call get_cache_service and check it's unaffected
    codeflash_output = get_cache_service(); cache = codeflash_output
    # Clean up
    for i in range(500):
        stype = f"DUMMY_{i}"
        _service_manager_instance._factories.pop(stype, None)
        _service_manager_instance._services.pop(stype, None)

def test_get_cache_service_factory_returns_different_types():
    """
    Test that get_cache_service can handle a factory that alternates between sync and async cache services.
    """
    class AlternatingFactory:
        def __init__(self):
            self.toggle = True
        def __call__(self):
            if self.toggle:
                self.toggle = False
                return CacheService("alt-sync")
            else:
                self.toggle = True
                return AsyncBaseCacheService("alt-async")
    _service_manager_instance._factories[ServiceType.CACHE_SERVICE] = AlternatingFactory()
    _service_manager_instance._services.pop(ServiceType.CACHE_SERVICE, None)
    # First call: sync
    codeflash_output = get_cache_service(); result1 = codeflash_output
    # Remove and call again: async
    _service_manager_instance._services.pop(ServiceType.CACHE_SERVICE, None)
    codeflash_output = get_cache_service(); result2 = codeflash_output
    # Restore default
    _service_manager_instance._factories[ServiceType.CACHE_SERVICE] = CacheServiceFactory()
    _service_manager_instance._services.pop(ServiceType.CACHE_SERVICE, None)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-pr10768-2025-12-10T19.10.20 and push.

[coderabbitai]

Important

Review skipped

Bot user detected.

To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov]

Codecov Report

❌ Patch coverage is 85.71429% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 33.05%. Comparing base (936ae87) to head (5014832).

Files with missing lines	Patch %	Lines
src/backend/base/langflow/services/deps.py	85.71%	2 Missing ⚠️

❌ Your project status has failed because the head coverage (40.00%) is below the target coverage (60.00%). You can increase the head coverage or adjust the target coverage.

Additional details and impacted files

@@                 Coverage Diff                 @@
##           fix-agent-flow-lfx   #10958   +/-   ##
===================================================
  Coverage               33.05%   33.05%           
===================================================
  Files                    1368     1368           
  Lines                   63832    63839    +7     
  Branches                 9395     9395           
===================================================
+ Hits                    21099    21102    +3     
- Misses                  41687    41690    +3     
- Partials                 1046     1047    +1     

Flag	Coverage Δ
backend	52.79% <85.71%> (+<0.01%)	⬆️
lfx	40.00% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
src/backend/base/langflow/services/deps.py	83.56% <85.71%> (+0.22%)	⬆️

... and 4 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[codeflash-ai]

This PR has been automatically closed because the original PR #10768 by ogabrielluiz was closed.