chore: determine docker runtime from executor class definition

Impacts: sdk, facilitator, validator, miner

Author: emnoor-reef

compute_horde/compute_horde/fv_protocol/facilitator_requests.py

@@ -61,7 +61,7 @@ class V2JobRequest(SignedRequest, BaseModel, extra="forbid"):
     docker_image: str
     args: list[str]
     env: dict[str, str]
-    use_gpu: bool
+    use_gpu: bool = True  # DEPRECATED: this field has no effect
     volume: Volume | None = None
     output_upload: OutputUpload | None = None
     artifacts_dir: str | None = None

compute_horde_sdk/src/compute_horde_core/signature.py

@@ -57,7 +57,7 @@ class SignedFields(BaseModel):
     docker_image: str
     args: list[str]
     env: dict[str, str]
-    use_gpu: bool
+    use_gpu: bool  # DEPRECATED: this field has no effect
     artifacts_dir: str
     on_trusted_miner: bool
     download_time_limit: int

compute_horde_sdk/src/compute_horde_sdk/_internal/sdk.py

@@ -528,7 +528,7 @@ async def create_job(self, job_spec: ComputeHordeJobSpec, on_trusted_miner: bool
             "job_namespace": job_spec.job_namespace,
             "args": job_spec.args or [],  # type: ignore
             "env": job_spec.env or {},  # type: ignore
-            "use_gpu": True,
+            "use_gpu": True,  # DEPRECATED: this field has no effect
             "artifacts_dir": job_spec.artifacts_dir,
             "on_trusted_miner": on_trusted_miner,
             "download_time_limit": job_spec.download_time_limit_sec,

executor/app/src/compute_horde_executor/executor/job_driver.py

@@ -3,6 +3,7 @@
 
 import packaging.version
 import sentry_sdk
+from compute_horde.executor_class import EXECUTOR_CLASS
 from compute_horde.job_errors import HordeError, JobError
 from compute_horde.protocol_consts import (
     HordeFailureReason,
@@ -151,8 +152,9 @@ async def _startup_stage(self) -> V0InitialJobRequest:
         self._enter_stage(JobStage.EXECUTOR_STARTUP)
         if not settings.DEBUG_NO_GPU_MODE:
             self.specs = await get_machine_specs()
-        await self.run_security_checks_or_fail()
         initial_job_request = await self.miner_client.initial_msg
+        has_gpu = EXECUTOR_CLASS[initial_job_request.executor_class].has_gpu
+        await self.run_security_checks_or_fail(has_gpu)
         await self.runner.prepare_initial(initial_job_request)
         await self.miner_client.send_executor_ready()
         if initial_job_request.streaming_details is not None:
@@ -189,9 +191,9 @@ async def _upload_stage(self):
         job_result.specs = self.specs
         await self.miner_client.send_result(job_result)
 
-    async def run_security_checks_or_fail(self):
+    async def run_security_checks_or_fail(self, has_gpu: bool):
         await self.run_cve_2022_0492_check_or_fail()
-        if not settings.DEBUG_NO_GPU_MODE:
+        if has_gpu and not settings.DEBUG_NO_GPU_MODE:
             await self.run_nvidia_toolkit_version_check_or_fail()
 
     async def run_cve_2022_0492_check_or_fail(self):

facilitator/app/src/project/core/models.py

@@ -151,7 +151,7 @@ class Job(ExportModelOperationsMixin("job"), models.Model):
         help_text="arguments passed to the script or docker image",
     )
     env = models.JSONField(blank=True, default=dict, help_text="environment variables for the job")
-    use_gpu = models.BooleanField(default=False, help_text="Whether to use GPU for the job")
+    use_gpu = models.BooleanField(default=False, help_text="DEPRECATED: Whether to use GPU for the job")
     target_validator_hotkey = models.TextField(help_text="target validator")
     volumes = SchemaField(schema=list[MuliVolumeAllowedVolume], blank=True, default=list)
     uploads = SchemaField(schema=list[SingleFileUpload], blank=True, default=list)

facilitator/app/src/project/core/tests/test_api.py

@@ -113,7 +113,6 @@ def check_docker_job(job_result):
     assert job_result["docker_image"] == "hello-world"
     assert job_result["args"] == ["my", "args"]
     assert job_result["env"] == {"MY_ENV": "my value"}
-    assert job_result["use_gpu"] is True
     assert set(job_result.keys()) & generated_fields == generated_fields
 
 
@@ -168,7 +167,6 @@ def test_docker_job_viewset_create(api_client, user, connected_validator, mock_s
     assert job.docker_image == "hello-world"
     assert job.args == ["my", "args"]
     assert job.env == {"MY_ENV": "my value"}
-    assert job.use_gpu is True
     assert job.user == user
 
 

validator/app/src/compute_horde_validator/validator/management/commands/debug_run_organic_job.py

@@ -65,7 +65,6 @@ def add_arguments(self, parser):
         parser.add_argument(
             "--cmd_args", default="", help="arguments passed to the script or docker image"
         )
-        parser.add_argument("--use_gpu", action="store_true", help="use gpu for job execution")
 
         parser.add_argument(
             "--download_time_limit", default=10, type=int, help="download time limit in seconds"
@@ -120,7 +119,6 @@ def handle(self, *args, **options):
             docker_image=options["docker_image"],
             args=shlex.split(options["cmd_args"]),
             env={},
-            use_gpu=options["use_gpu"],
             download_time_limit=options["download_time_limit"],
             execution_time_limit=options["execution_time_limit"],
             upload_time_limit=options["upload_time_limit"],

validator/app/src/compute_horde_validator/validator/organic_jobs/miner_driver.py

@@ -6,6 +6,7 @@
 import sentry_sdk
 from asgiref.sync import sync_to_async
 from channels.layers import get_channel_layer
+from compute_horde.executor_class import EXECUTOR_CLASS
 from compute_horde.fv_protocol.facilitator_requests import OrganicJobRequest
 from compute_horde.fv_protocol.validator_requests import (
     HordeFailureDetails,
@@ -299,11 +300,14 @@ async def streaming_ready_callback(msg: V0StreamingJobReadyRequest) -> None:
     miner_client.notify_streaming_readiness = streaming_ready_callback  # type: ignore[method-assign]
     # TODO: remove method assignment above and properly handle notify_* cases
 
+    executor_class = ExecutorClass(job_request.executor_class)
+    has_gpu = EXECUTOR_CLASS[executor_class].has_gpu
+
     job_details = OrganicJobDetails(
         job_uuid=str(job.job_uuid),
-        executor_class=ExecutorClass(job_request.executor_class),
+        executor_class=executor_class,
         docker_image=job_request.docker_image,
-        docker_run_options_preset="nvidia_all" if job_request.use_gpu else "none",
+        docker_run_options_preset="nvidia_all" if has_gpu else "none",
         docker_run_cmd=job_request.get_args(),
         env=job_request.env,
         total_job_timeout=OrganicJobDetails.total_job_timeout,

validator/app/src/compute_horde_validator/validator/routing/tests/test_job_routing.py

@@ -28,7 +28,6 @@
     docker_image="doesntmatter",
     args=[],
     env={},
-    use_gpu=False,
     download_time_limit=1,
     execution_time_limit=1,
     streaming_start_time_limit=1,

validator/app/src/compute_horde_validator/validator/routing/tests/test_job_routing_incident.py

@@ -53,7 +53,6 @@ class MinerScenario:
     docker_image="doesntmatter",
     args=[],
     env={},
-    use_gpu=False,
     download_time_limit=1,
     execution_time_limit=1,
     streaming_start_time_limit=1,
@@ -140,7 +139,6 @@ async def _report_incidents(miner_hotkey: str, incidents: int, executor_class):
                 docker_image=JOB_REQUEST.docker_image,
                 args=list(JOB_REQUEST.args),
                 env=dict(JOB_REQUEST.env),
-                use_gpu=JOB_REQUEST.use_gpu,
                 download_time_limit=JOB_REQUEST.download_time_limit,
                 execution_time_limit=JOB_REQUEST.execution_time_limit,
                 streaming_start_time_limit=JOB_REQUEST.streaming_start_time_limit,
@@ -348,7 +346,6 @@ async def fake_get_expected_miner_executor_count(**_kwargs):
         docker_image="ubuntu:latest",
         args=[],
         env={},
-        use_gpu=False,
         download_time_limit=1,
         execution_time_limit=1,
         streaming_start_time_limit=1,