Ffw simulation command

dev/.env.docker-compose

@@ -23,4 +23,13 @@ DJANGO_PHOLD_MPI_PROCESSES=4
 CODE_EXAMPLE_DIR=/opt/codes
 
 # Directory containing FFW build
-FFW_BUILD_DIR=/opt/FFW
+DJANGO_FFW_BUILD_DIR=/opt/FFW/build/local
+
+# FFW binary
+DJANGO_FFW_BINARY_PATH=/opt/FFW/build/local/src/model-net-fluid-flow-wan-trace-traffic
+
+# FFW Config path
+DJANGO_FFW_CONFIG_PATH=/opt/FFW/build/local/doc/example/fluid-flow-wan-trace-traffic-pause.yaml
+
+# Directory for FFW output files
+DJANGO_FFW_OUTPUT_DIR=/tmp/ffw_output

net_maestro/core/management/commands/run_ffw.py

@@ -0,0 +1,166 @@
+"""RUN FFW simulation management command.
+
+Execute Fluid-Flow WAN simulation with specified parameters and ingest results.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from django.conf import settings
+import djclick as click
+
+from net_maestro.core.constants import RunStatus, SimulationType
+from net_maestro.core.models import Run
+from net_maestro.core.tasks.simulation import run_ffw_simulation
+
+
+def _create_or_update_run(
+    name: str,
+    description: str | None,
+    run_id: int | None,
+) -> Run:
+    """Create a new Run object or update an existing one's metadata.
+
+    Args:
+        name: Run identifier
+        description: Run description
+        run_id: Existing run ID to update, or None to create new
+
+    Returns:
+        The Run object
+
+    Raises:
+        Run.DoesNotExist: If run_id is provided but no matching Run exists
+    """
+    if run_id:
+        run = Run.objects.get(id=run_id)
+        run.name = name
+        run.description = description or ""
+        run.save()
+    else:
+        run = Run.objects.create(
+            name=name,
+            description=description or "",
+            status=RunStatus.PENDING,
+            simulation_type=SimulationType.FFW,
+        )
+    return run
+
+
+# TODO: consider adding the stats path as an option
+# get help text for new options
+@click.command()
+@click.option(
+    "-n",
+    "--name",
+    "name",
+    type=str,
+    help="Run identifier.",
+    required=True,
+)
+@click.option(
+    "--np",
+    "np",
+    type=int,
+    default=1,
+    help="Number of MPI processes.",
+)
+@click.option(
+    "--binary-path",
+    "binary_path",
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    help="Path to FFW binary.",
+)
+@click.option(
+    "--sync",
+    "sync",
+    type=int,
+    default=1,
+    help=(
+        "Synchronization protocol (1=sequential, 2=conservative, 3=optimistic, "
+        "4=optimistic-debug, 5=optimistic-realtime, 6=reverse-check)."
+    ),
+)
+@click.option(
+    "--model-stats",
+    "model_stats",
+    type=int,
+    default=4,
+)
+@click.option(
+    "--num-gvt",
+    "num_gvt",
+    type=int,
+    default=1,
+)
+@click.option(
+    "--rt-interval",
+    "rt_interval",
+    type=int,
+    default=1,
+)
+@click.option("--vt-interval", "vt_interval", type=int, default=1e8)
+@click.option("--vt-samp-end", "vt_samp_end", type=int, default=1.7e10)
+@click.option(
+    "--config-path",
+    "config_path",
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    help="Path to FFW configuration file.",
+)
+@click.option(
+    "--working-dir",
+    type=click.Path(exists=True, path_type=Path),
+)
+@click.option(
+    "--description",
+    "description",
+    type=str,
+    help="Description of the run.",
+)
+@click.option(
+    "--run-id",
+    "run_id",
+    type=int,
+    default=None,
+    help="Existing run ID to update instead of creating a new run.",
+)
+def run_ffw(  # noqa: PLR0913
+    name: str,
+    np: int,
+    binary_path: Path,
+    sync: int,
+    model_stats: int,
+    num_gvt: int,
+    rt_interval: int,
+    vt_interval: int,
+    vt_samp_end: int,
+    config_path: Path,
+    working_dir: Path,
+    description: str | None = None,
+    run_id: int | None = None,
+) -> None:
+    working_dir = (
+        Path(working_dir) if working_dir else Path(getattr(settings, "FFW_BUILD_PATH", "."))
+    )
+    # Logs dir needs to exist for the model to run
+    logs_dir = working_dir / "doc" / "example" / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+
+    # Create or update Run object
+    run = _create_or_update_run(name, description, run_id)
+    run_ffw_simulation.apply(
+        kwargs={
+            "run_id": run.id,
+            "np": np,
+            "sync": sync,
+            "model_stats": model_stats,
+            "num_gvt": num_gvt,
+            "rt_interval": rt_interval,
+            "vt_interval": vt_interval,
+            "vt_samp_end": vt_samp_end,
+            "config_path": str(config_path) if config_path else None,
+            "working_dir": str(working_dir),
+            "binary_path": str(binary_path) if binary_path else None,
+        }
+    )

net_maestro/core/services/ffw.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import logging
+import subprocess
+
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+
+def execute_ffw_model(  # noqa: PLR0913
+    *,
+    np: int,
+    sync: int,
+    model_stats: int,
+    num_gvt: int,
+    rt_interval: int,
+    vt_interval: int,
+    vt_samp_end: int,
+    config_path: str,
+    working_dir: str,
+    binary_path: str,
+) -> int:
+    stats_output = settings.FFW_OUTPUT_DIR
+    cmd = [
+        "mpirun",
+        "-np",
+        str(np),
+        binary_path,
+        f"--sync={sync}",
+        f"--model-stats={model_stats}",
+        f"--num-gvt={num_gvt}",
+        f"--rt-interval={rt_interval}",
+        f"--vt-interval={vt_interval}",
+        f"--vt-samp-end={vt_samp_end}",
+        f"--stats-path={stats_output}",
+        "--",
+        config_path,
+    ]
+    try:
+        result = subprocess.run(cmd, cwd=working_dir, check=True, capture_output=True, text=True)  # noqa: S603
+        logger.info("FFW simulation stdout: %s", result.stdout)
+    except subprocess.CalledProcessError as e:
+        logger.exception("FFW simulation failed with error code %s", e.returncode)
+        logger.info("FFW simulation stderr: %s", result.stderr)
+    return result.returncode

net_maestro/core/tasks/__init__.py

@@ -2,6 +2,12 @@
 
 from .events import run_event_task
 from .models import run_model_task
-from .simulation import run_phold_simulation, run_simulation_task
+from .simulation import run_ffw_simulation, run_phold_simulation, run_simulation_task
 
-__all__ = ["run_event_task", "run_model_task", "run_phold_simulation", "run_simulation_task"]
+__all__ = [
+    "run_event_task",
+    "run_ffw_simulation",
+    "run_model_task",
+    "run_phold_simulation",
+    "run_simulation_task",
+]

net_maestro/core/tasks/simulation.py

@@ -18,6 +18,7 @@
 from net_maestro.core.models.simulation_pe_record import SimulationPeRecord
 from net_maestro.core.parsers.ross_binary_file import RecordType
 from net_maestro.core.parsers.ross_binary_file import ROSSFile as SimulationFileParser
+from net_maestro.core.services.ffw import execute_ffw_model
 
 logger = logging.getLogger(__name__)
 
@@ -219,3 +220,51 @@ def run_phold_simulation(  # noqa: PLR0913
         # Update status to FAILED
         run.status = RunStatus.FAILED
         run.save()
+
+
+@shared_task
+def run_ffw_simulation(  # noqa: PLR0913
+    *,
+    run_id: int,
+    np: int = 1,
+    sync: int = 1,
+    model_stats: int = 3,
+    num_gvt: int = 1,
+    rt_interval: int = 1,
+    vt_interval: int = 1e8,
+    vt_samp_end: int = 1.7e10,
+    working_dir: str = "",
+    binary_path: str = "",
+    config_path: str = "",
+) -> None:
+    """Execute FFW simulation with given parameters and update run status."""
+    run = Run.objects.get(pk=run_id)
+    run.status = RunStatus.RUNNING
+    run.save()
+
+    # Use example defaults if values not provided
+    ffw_working_dir = working_dir or getattr(settings, "FFW_BUILD_PATH", "")
+    ffw_binary_path = binary_path or getattr(settings, "FFW_BINARY_PATH", "")
+    ffw_config_path = config_path or getattr(settings, "FFW_CONFIG_PATH", "")
+
+    try:
+        execute_ffw_model(
+            np=np,
+            sync=sync,
+            model_stats=model_stats,
+            num_gvt=num_gvt,
+            rt_interval=rt_interval,
+            vt_interval=vt_interval,
+            vt_samp_end=vt_samp_end,
+            config_path=ffw_config_path,
+            working_dir=ffw_working_dir,
+            binary_path=ffw_binary_path,
+        )
+        run.status = RunStatus.COMPLETED
+        run.save()
+        logger.info("FFW simulation finished for run %s", run_id)
+
+    except Exception:
+        run.status = RunStatus.FAILED
+        run.save()
+        logger.exception("FFW simulation failed for run %s", run_id)

net_maestro/settings/development.py

@@ -65,3 +65,9 @@
 PHOLD_BINARY_PATH = env.str("DJANGO_PHOLD_BINARY_PATH", default="")
 PHOLD_OUTPUT_DIR = env.str("DJANGO_PHOLD_OUTPUT_DIR", default="/tmp/phold_output")  # noqa: S108
 PHOLD_MPI_PROCESSES = env.int("DJANGO_PHOLD_MPI_PROCESSES", default=2)
+
+# FFW simulation configuration
+FFW_BUILD_PATH = env.str("DJANGO_FFW_BUILD_DIR", default="")
+FFW_BINARY_PATH = env.str("DJANGO_FFW_BINARY_PATH", default="")
+FFW_CONFIG_PATH = env.str("DJANGO_FFW_CONFIG_PATH", default="")
+FFW_OUTPUT_DIR = env.str("DJANGO_FFW_OUTPUT_DIR", default="/tmp/ffw_output")  # noqa: S108