def resources_for_task()

in jetstream/diagnostics/resource_profiling_plugin.py [0:0]

• 13 lines of code
• 3 McCabe index (conditional complexity)

    def resources_for_task(self, worker_address: str):
        """The worker finished its previous task.
        Return its resource usage and then reset it.
        """
        with self._lock:
            mem_result = self._worker_memory[worker_address]
            if not mem_result:
                mem_result = [0]
            del self._worker_memory[worker_address]

            cpu_result = self._worker_cpu[worker_address]
            if not cpu_result:
                cpu_result = [0]
            del self._worker_cpu[worker_address]

            return ResourceUsage(
                worker_address=worker_address, memory_usage=mem_result, cpu_usage=cpu_result
            )