DSperce integration

.devcontainer/devcontainer.json

@@ -33,7 +33,7 @@
     }
   },
   // Install ezkl cli, done here instead of the Dockerfile to test new versions without rebuilding the image.
-  "onCreateCommand": "curl https://raw.githubusercontent.com/zkonduit/ezkl/main/install_ezkl_cli.sh | bash -s v19.0.7",
+  "onCreateCommand": "curl https://raw.githubusercontent.com/zkonduit/ezkl/main/install_ezkl_cli.sh | bash -s v22.2.1",
   "postCreateCommand": "uv tool install bittensor-cli",
   "remoteEnv": {
     "PATH": "${containerEnv:PATH}:/home/vscode/.ezkl"

neurons/_validator/api/__init__.py

@@ -23,6 +23,7 @@
 import bittensor as bt
 from _validator.models.poc_rpc_request import ProofOfComputationRPCRequest
 from _validator.models.pow_rpc_request import ProofOfWeightsRPCRequest
+from _validator.models.base_rpc_request import QueuedRequestDataModel
 import hashlib
 from constants import (
     MAX_SIGNATURE_LIFESPAN,
@@ -70,9 +71,10 @@ def _should_rate_limit(ip: str):
 class ValidatorAPI:
     def __init__(self, config: ValidatorConfig):
         self.config = config
-        self.external_requests_queue: list[
-            ProofOfWeightsRPCRequest | ProofOfComputationRPCRequest
-        ] = []
+        # a Queue of requests to be sent to miners
+        # consists of "real world requests" ProofOfWeightsRPCRequest and ProofOfComputationRPCRequest
+        # and a Request with one slice of a DSperse model (DSlice)
+        self.stacked_requests_queue: list[QueuedRequestDataModel] = []
         self.ws_manager = WebSocketManager()
         self.recent_requests: dict[str, int] = {}
         self.validator_keys_cache = ValidatorKeysCache(config)
@@ -282,7 +284,7 @@ async def handle_proof_of_weights(
                 return InvalidParams(str(e))
 
             self.pending_requests[external_request.hash] = asyncio.Event()
-            self.external_requests_queue.insert(0, external_request)
+            self.stacked_requests_queue.insert(0, external_request)
             bt.logging.success(
                 f"External request with hash {external_request.hash} added to queue"
             )
@@ -341,7 +343,7 @@ async def handle_proof_of_computation(
                 return InvalidParams(str(e))
 
             self.pending_requests[external_request.hash] = asyncio.Event()
-            self.external_requests_queue.insert(0, external_request)
+            self.stacked_requests_queue.insert(0, external_request)
             bt.logging.success(
                 f"External request with hash {external_request.hash} added to queue"
             )

neurons/_validator/core/dsperse_manager.py

@@ -0,0 +1,85 @@
+import random
+import os
+import uuid
+import tempfile
+import json
+from pathlib import Path
+
+from bittensor import logging
+from dsperse.src.slice.utils.converter import Converter
+from dsperse.src.run.runner import Runner
+
+from constants import DSPERSE_SLICES_FILE_NAME
+from deployment_layer.circuit_store import circuit_store
+from execution_layer.circuit import CircuitType, Circuit
+from _validator.models.dslice_request import DSliceQueuedProofRequest
+from _validator.models.request_type import RequestType
+from _validator.api import ValidatorAPI
+
+
+class DSperseManager:
+    def __init__(self, api: ValidatorAPI):
+        self.api = api
+        self.circuits: list[Circuit] = [
+            circuit
+            for circuit in circuit_store.circuits
+            if circuit.metadata.type == CircuitType.DSPERSE_PROOF_GENERATION
+        ]
+        self.runs = {}
+
+    def generate_dslice_requests(self) -> list:
+        """
+        Generate DSlice requests for DSperse models.
+        Each DSlice request corresponds to one slice of a DSperse model.
+        """
+        if self.api.stacked_requests_queue or not self.circuits:
+            # there are already requests stacked, do not generate new DSlice requests
+            return []
+
+        circuit = random.choice(self.circuits)
+        run_uid = str(uuid.uuid4())
+        logging.info(
+            f"Generating DSlice requests for circuit {circuit.metadata.name}... Run UID: {run_uid}"
+        )
+
+        # TODO: ...
+        self.run_dsperse(circuit, run_uid)
+        dslice_requests = []
+        # Logic to create DSlice requests goes here
+        return dslice_requests
+
+    def extract_slices(self, circuit: Circuit) -> Path:
+        """
+        Extract slices from a DSperse circuit file.
+        Returns the path to the folder containing the extracted slices.
+        """
+        dsperse_file_path = (
+            Path(circuit.paths.external_base_path) / DSPERSE_SLICES_FILE_NAME
+        )
+        dsperse_slices_folder = dsperse_file_path.with_suffix("")
+        # TODO: Check is already converted
+        Converter.convert(
+            path=dsperse_file_path,
+            output_type="dirs",
+            output_path=dsperse_slices_folder,
+        )
+        return dsperse_slices_folder
+
+    def run_dsperse(self, circuit: Circuit, run_uid: str) -> None:
+        slices_path = self.extract_slices(circuit)
+
+        # Create temporary folder for run metadata
+        run_metadata_path = Path(tempfile.mkdtemp(prefix=f"dsperse_run_{run_uid}_"))
+        save_metadata_path = run_metadata_path / "metadata.json"
+        logging.info(f"Running DSperse model. Run metadata path: {run_metadata_path}")
+
+        # Generate benchmarking input JSON
+        input_json_path = run_metadata_path / "input.json"
+        with open(input_json_path, "w") as f:
+            json.dump(circuit.input_handler(RequestType.BENCHMARK).generate(), f)
+
+        # init runner and run the sliced model
+        runner = Runner(
+            run_metadata_path=run_metadata_path, save_metadata_path=save_metadata_path
+        )
+        results = runner.run(input_json_path=input_json_path, slice_path=slices_path)

neurons/_validator/core/request_pipeline.py

@@ -34,25 +34,6 @@ def __init__(
         self.api = api
         self.hash_guard = HashGuard()
 
-    def prepare_requests(self, filtered_uids) -> list[Request]:
-        """
-        Prepare requests for the current validation step.
-        This includes both regular benchmark requests and any external requests.
-
-        Args:
-            filtered_uids (list): List of UIDs to send requests to.
-
-        Returns:
-            list[Request]: List of prepared requests.
-        """
-        if len(filtered_uids) == 0:
-            bt.logging.error("No UIDs to query")
-            return []
-
-        if self.api.external_requests_queue:
-            return self._prepare_real_world_requests(filtered_uids)
-        return self._prepare_benchmark_requests(filtered_uids)
-
     def _check_and_create_request(
         self,
         uid: int,
@@ -91,13 +72,14 @@ def _check_and_create_request(
             circuit=circuit,
             request_type=request_type,
             # 'inputs' are used for verification later on validator side:
+            #   I suppose `RWR` passed here to prevent new data generation
             inputs=GenericInput(RequestType.RWR, input_data),
             request_hash=request_hash,
             save=save,
         )
 
     def _prepare_real_world_requests(self, filtered_uids: list[int]) -> list[Request]:
-        external_request = self.api.external_requests_queue.pop()
+        external_request = self.api.stacked_requests_queue.pop()
         requests = []
 
         for uid in filtered_uids:
@@ -207,6 +189,9 @@ def get_request_data(
                 QueryZkProof(query_input=inputs, model_id=circuit.id, query_output=""),
                 False,
             )
+        elif circuit.metadata.type == CircuitType.DSPERSE_PROOF_GENERATION:
+            # TODO: Handle DSPERSE_PROOF_GENERATION request data preparation
+            pass
 
         return (
             ProofOfWeightsDataModel(
@@ -230,7 +215,7 @@ def prepare_single_request(self, uid: int) -> Request | None:
         Returns:
             Request | None: The prepared request, or None if preparation failed.
         """
-        if self.api.external_requests_queue:
+        if self.api.stacked_requests_queue:
             requests = self._prepare_real_world_requests([uid])
         else:
             requests = self._prepare_benchmark_requests([uid])

neurons/_validator/core/validator_loop.py

@@ -19,6 +19,7 @@
 from _validator.competitions.competition import Competition
 from _validator.config import ValidatorConfig
 from _validator.core.capacity_manager import CapacityManager
+from _validator.core.dsperse_manager import DSperseManager
 from _validator.core.prometheus import (
     log_error,
     log_queue_metrics,
@@ -120,6 +121,7 @@ def __init__(self, config: ValidatorConfig):
         self.request_pipeline = RequestPipeline(
             self.config, self.score_manager, self.api
         )
+        self.dsperse_manager = DSperseManager(self.api)
 
         self.request_queue = asyncio.Queue()
         self.active_tasks: dict[int, asyncio.Task] = {}
@@ -342,6 +344,9 @@ async def maintain_request_pool(self):
                 slots_available = self.current_concurrency - len(self.active_tasks)
 
                 if slots_available > 0:
+                    # TODO: some conditions to trigger dsperse requests generation?
+                    self.dsperse_manager.generate_dslice_requests()
+
                     available_uids = [
                         uid
                         for uid in self.queryable_uids

neurons/_validator/models/base_rpc_request.py

@@ -3,7 +3,12 @@
 from _validator.utils.api import hash_inputs
 
 
-class RealWorldRequest(BaseModel):
+class QueuedRequestDataModel(BaseModel):
+    """
+    Base model for requests that are stacked in the validator's queue and waiting to be sent to miners.
+    At the moment, that's a Real World Request (RWR) or a Request with one slice of a DSperse model (DSlice).
+    """
+
     circuit: Circuit
     inputs: dict
 

neurons/_validator/models/dslice_request.py

@@ -0,0 +1,12 @@
+from _validator.models.base_rpc_request import QueuedRequestDataModel
+from pydantic import Field
+
+
+class DSliceQueuedProofRequest(QueuedRequestDataModel):
+    """
+    Request for a DSperse slice.
+    """
+
+    slice_num: str = Field(..., description="Num of the DSperse slice")
+    run_uid: str = Field(..., description="UID of the DSperse run")
+    outputs: dict = Field(..., description="Outputs of the DSperse slice")

neurons/_validator/models/poc_rpc_request.py

@@ -1,10 +1,10 @@
-from _validator.models.base_rpc_request import RealWorldRequest
+from _validator.models.base_rpc_request import QueuedRequestDataModel
 from pydantic import Field
 from deployment_layer.circuit_store import circuit_store
 from execution_layer.circuit import CircuitType
 
 
-class ProofOfComputationRPCRequest(RealWorldRequest):
+class ProofOfComputationRPCRequest(QueuedRequestDataModel):
     """
     Request for the Proof of Computation RPC method.
     """

neurons/_validator/models/pow_rpc_request.py

@@ -1,10 +1,10 @@
 from __future__ import annotations
-from _validator.models.base_rpc_request import RealWorldRequest
+from _validator.models.base_rpc_request import QueuedRequestDataModel
 from pydantic import Field
 from deployment_layer.circuit_store import circuit_store
 
 
-class ProofOfWeightsRPCRequest(RealWorldRequest):
+class ProofOfWeightsRPCRequest(QueuedRequestDataModel):
     """
     Request for the Proof of Weights RPC method.
     """

neurons/_validator/models/request_type.py

@@ -2,14 +2,31 @@
 
 
 class RequestType(Enum):
+    """
+    Enumeration of different types of requests that the validator can send to miner.
+     - BENCHMARK: Requests with generated input data for benchmarking purposes.
+       In case of empty RWR queue the validator generates some input data to keep miners busy.
+     - RWR: Real World Requests with actual input data for real-world inference.
+       Validator collects such requests from external users and stacks them in a queue to be sent to miners.
+     - DSLICE: That's a tricky one. We use DSperse app for slicing large models into smaller parts.
+       And some requests involve sliced model and each slice is sent as a separate request to the miner.
+       That each slice request is of type DSLICE.
+       At the moment we just stack DSlices to the same RWR queue, and behave as normal RWR requests.
+       That means DSLICE request is a part of RWR or BENCHMARK request.
+       XXX: Yeah, not very elegant, but we'll improve it later, I promise.
+    """
+
     BENCHMARK = "benchmark_request"
     RWR = "real_world_request"
+    DSLICE = "dslice_request"
 
     def __str__(self) -> str:
         if self == RequestType.BENCHMARK:
             return "Benchmark"
         elif self == RequestType.RWR:
             return "Real World Request"
+        elif self == RequestType.DSPERSE:
+            return "DSperse Request (one slice)"
         else:
             raise ValueError(f"Unknown request type: {self}")
 

neurons/constants.py

@@ -173,6 +173,8 @@ class Roles:
 ONE_YEAR = ONE_DAY * 365
 # Temporary folder for storing proof files
 TEMP_FOLDER = "/tmp/subnet-2"
+# DSperse slices file name
+DSPERSE_SLICES_FILE_NAME = "slices.dsperse"
 
 # Queue size limits
 MAX_POW_QUEUE_SIZE = 1024