Fixed CPU time limit calculation (#4909)

golem/task/helpers.py

@@ -14,6 +14,8 @@ def calculate_subtask_payment(
     # price_per_hour is
     # computation_time is expressed in seconds
     """
+    This is equivalent to: math.ceil(price_per_hour * computation_time // 3600)
+
     Don't use math.ceil (this is general advice, not specific to the case here)
     >>> math.ceil(10 ** 18 / 6)
     166666666666666656
@@ -33,7 +35,13 @@ def calculate_max_usage(budget: int, price_per_hour: int) -> int:
     :param price_per_hour: [ GNT wei / hour ]
     :return: [ seconds ]
     """
-    # budget is expressed in GNT wei
-    # price_per_hour is expressed in GNT wei / hour
+    """
+    This is equivalent to: math.ceil(budget * 3600 // price_per_hour)
 
+    Don't use math.ceil (this is general advice, not specific to the case here)
+    >>> math.ceil(10 ** 18 / 6)
+    166666666666666656
+    >>> (10 ** 18 + 5) // 6
+    166666666666666667
+    """
     return (budget * 3600 + price_per_hour - 1) // price_per_hour

golem/task/taskcomputer.py

@@ -86,7 +86,11 @@ def dir_manager(self) -> DirManager:
         # FIXME: This shouldn't be part of the public interface probably
         return self._old_computer.dir_manager
 
-    def task_given(self, ctd: ComputeTaskDef) -> None:
+    def task_given(
+            self,
+            ctd: ComputeTaskDef,
+            cpu_time_limit: Optional[int] = None
+    ) -> None:
         assert not self._new_computer.has_assigned_task()
         assert not self._old_computer.has_assigned_task()
 
@@ -95,7 +99,7 @@ def task_given(self, ctd: ComputeTaskDef) -> None:
         if task_header.environment_prerequisites is not None:
             self._new_computer.task_given(task_header, ctd)
         else:
-            self._old_computer.task_given(ctd)
+            self._old_computer.task_given(ctd, cpu_time_limit)
 
     def has_assigned_task(self) -> bool:
         return self._new_computer.has_assigned_task() \
@@ -475,9 +479,16 @@ def __init__(
         self.support_direct_computation = False
         self.finished_cb = finished_cb
 
-    def task_given(self, ctd: ComputeTaskDef) -> None:
+        self.cpu_limit: Optional[int] = None
+
+    def task_given(
+            self,
+            ctd: ComputeTaskDef,
+            cpu_time_limit: Optional[int] = None
+    ) -> None:
         assert self.assigned_subtask is None
         self.assigned_subtask = ctd
+        self.cpu_limit = cpu_time_limit
         ProviderTimer.start()
 
     def has_assigned_task(self) -> bool:
@@ -667,14 +678,13 @@ def start_computation(self) -> None:  # pylint: disable=too-many-locals
             return
 
         deadline = min(task_header.deadline, subtask_deadline)
-        cpu_limit = task_header.subtask_budget
         task_timeout = deadline_to_timeout(deadline)
 
         unique_str = str(uuid.uuid4())
 
         logger.info("Starting computation of subtask %r (task: %r, deadline: "
-                    "%r, docker images: %r)", subtask_id, task_id, deadline,
-                    docker_images)
+                    "%r, docker images: %r, cpu limit: %r)", subtask_id,
+                    task_id, deadline, docker_images, self.cpu_limit)
 
         with self.dir_lock:
             resource_dir = self.dir_manager.get_task_resource_dir(task_id)
@@ -690,7 +700,12 @@ def start_computation(self) -> None:  # pylint: disable=too-many-locals
             dir_mapping = DockerTaskThread.generate_dir_mapping(resource_dir,
                                                                 temp_dir)
             tt: TaskThread = DockerTaskThread(
-                docker_images, extra_data, dir_mapping, task_timeout, cpu_limit)
+                docker_images,
+                extra_data,
+                dir_mapping,
+                task_timeout,
+                self.cpu_limit
+            )
         elif self.support_direct_computation:
             tt = PyTaskThread(extra_data, resource_dir, temp_dir,
                               task_timeout)

golem/task/taskserver.py

@@ -72,6 +72,7 @@
 )
 from golem.resource.resourcemanager import ResourceManager
 from golem.rpc import utils as rpc_utils
+from golem.task import helpers as task_helpers
 from golem.task import timer
 from golem.task.acl import get_acl, setup_acl, AclRule, _DenyAcl as DenyAcl
 from golem.task.exceptions import ComputationInProgress
@@ -495,8 +496,13 @@ def task_given(
             logger.error("Trying to assign a task, when it's already assigned")
             return False
 
-        self.task_computer.task_given(msg.compute_task_def)
-        if msg.want_to_compute_task.task_header.environment_prerequisites:
+        task_header: dt_tasks.TaskHeader = msg.want_to_compute_task.task_header
+
+        cpu_time_limit = task_helpers.calculate_max_usage(
+            task_header.subtask_budget, msg.want_to_compute_task.price)
+        self.task_computer.task_given(msg.compute_task_def, cpu_time_limit)
+
+        if task_header.environment_prerequisites:
             subtask_inputs_dir = self.task_computer.get_subtask_inputs_dir()
             resources_options = msg.resources_options or dict()
             client_options = self.resource_manager.build_client_options(

requirements.txt

@@ -38,7 +38,7 @@ eth-utils==1.0.3
 ethereum==1.6.1
 eventlet==0.24.1
 fs==2.4.4
-Golem-Messages==3.14.0
+Golem-Messages==3.14.1
 Golem-Smart-Contracts-Interface==1.10.3
 Golem-Task-Api==0.24.0
 greenlet==0.4.15

requirements_to-freeze.txt

@@ -18,7 +18,7 @@ docker==3.5.0
 enforce==0.3.4
 eth-utils==1.0.3
 ethereum==1.6.1
-Golem-Messages==3.14.0
+Golem-Messages==3.14.1
 Golem-Smart-Contracts-Interface==1.10.3
 Golem-Task-Api==0.24.0
 html2text==2018.1.9

tests/golem/task/test_taskcomputeradapter.py

@@ -89,7 +89,7 @@ def test_old_task_ok(self):
         }
         self.adapter.task_given(ctd)
         self.new_computer.task_given.assert_not_called()
-        self.old_computer.task_given.assert_called_once_with(ctd)
+        self.old_computer.task_given.assert_called_once_with(ctd, None)
 
 
 class TestHasAssignedTask(TaskComputerAdapterTestBase):

tests/golem/task/test_taskserver.py

@@ -48,6 +48,7 @@
 from golem.task import tasksession
 from golem.task.acl import DenyReason as AclDenyReason, AclRule
 from golem.task.benchmarkmanager import BenchmarkManager
+from golem.task import helpers as task_helpers
 from golem.task.result.resultmanager import EncryptedResultPackageManager
 from golem.task.server import concent as server_concent
 from golem.task.taskarchiver import TaskArchiver
@@ -1370,15 +1371,18 @@ class TestTaskGiven(TaskServerTestBase):
     def test_ok(
             self, logger_mock, dispatcher_mock, update_requestor_assigned_sum,
             request_resource):
-
         self.ts.task_computer.has_assigned_task.return_value = False
         ttc = msg_factories.tasks.TaskToComputeFactory()
 
+        task_header: dt_tasks.TaskHeader = ttc.want_to_compute_task.task_header
+        max_cpu_usage: int = task_helpers.calculate_max_usage(
+            task_header.subtask_budget, ttc.want_to_compute_task.price)
+
         result = self.ts.task_given(ttc)
         self.assertEqual(result, True)
 
         self.ts.task_computer.task_given.assert_called_once_with(
-            ttc.compute_task_def
+            ttc.compute_task_def, max_cpu_usage
         )
         request_resource.assert_called_once_with(
             ttc.task_id,