def do_select()

in src/dubbo/cluster/loadbalances.py [0:0]

• 17 lines of code
• 6 McCabe index (conditional complexity)

    def do_select(self, invokers: list[Invoker], invocation: Invocation) -> Optional[Invoker]:
        # get the CPU usage
        cpu_usages = self._monitor.get_cpu_usage(invokers)
        # Select the caller with the lowest CPU usage, 0 means CPU usage is unknown.
        available_invokers = []
        unknown_invokers = []

        for invoker, cpu_usage in cpu_usages.items():
            if cpu_usage == 0:
                unknown_invokers.append((cpu_usage, invoker))
            else:
                available_invokers.append((cpu_usage, invoker))

        if available_invokers:
            # sort and select the invoker with the lowest CPU usage
            available_invokers.sort(key=lambda x: x[0])
            return available_invokers[0][1]
        elif unknown_invokers:
            # get the invoker with unknown CPU usage randomly
            randint = random.randint(0, len(unknown_invokers) - 1)
            return unknown_invokers[randint][1]
        else:
            return None