fix a bug

Author: linjianma

recommendation/dlrm_v3/accuracy.py

@@ -23,8 +23,8 @@
 
 import numpy as np
 import torch
-from generative_recommenders.dlrm_v3.configs import get_hstu_configs
-from generative_recommenders.dlrm_v3.utils import MetricsLogger
+from configs import get_hstu_configs
+from utils import MetricsLogger
 
 logger: logging.Logger = logging.getLogger("main")
 

recommendation/dlrm_v3/main.py

@@ -19,9 +19,9 @@
 
 import argparse
 import array
-import json
 import logging
 import random
+import threading
 
 logging.basicConfig(level=logging.INFO)
 import math
@@ -53,7 +53,6 @@
     SUPPORTED_DATASETS,
 )
 
-
 logger: logging.Logger = logging.getLogger("main")
 
 torch.multiprocessing.set_start_method("spawn", force=True)
@@ -99,16 +98,16 @@ def get_args():  # pyre-ignore [3]
         "--find-peak-performance", default=False, help="Whether to find peak performance in the benchmark"
     )
     parser.add_argument(
-        "--dataset-path-prefix", default="", help="Prefix to the dataset path. Example: /home/username/"
+        "--dataset-path-prefix", default=f"/home/{os.getlogin()}/", help="Prefix to the dataset path. Example: /home/username/"
     )
     parser.add_argument(
-        "--warmup-ratio", default=0.1, help="The ratio of the dataset used to warmup SUT"
+        "--warmup-ratio", default=0.3, help="The ratio of the dataset used to warmup SUT"
     )
     parser.add_argument(
         "--num-queries", default=500000, help="Number of queries to run in the benchmark"
     )
     parser.add_argument(
-        "--target-qps", default=1500, help="Benchmark target QPS. Needs to be tuned for different implementations to balance latency and throughput"
+        "--target-qps", default=1000, help="Benchmark target QPS. Needs to be tuned for different implementations to balance latency and throughput"
     )
     parser.add_argument(
         "--numpy-rand-seed", default=123, help="Numpy random seed"
@@ -332,6 +331,7 @@ def __init__(
             get_num_queries(input_queries, self.total_requests) // self.total_requests
         )
         self.repeat: int = 0
+        self._lock = threading.Lock()
 
     def get_num_requests(self, warmup_ratio: float) -> List[int]:
         return [
@@ -359,6 +359,7 @@ def init_sut(self) -> None:
         self.ts = self.start_ts
         self.ds.set_ts(self.start_ts)
         self.cnt = 0
+        self.repeat = 0
 
     def load_query_samples(self, query_ids: List[Optional[int]]) -> None:
         length = len(query_ids)
@@ -382,25 +383,27 @@ def unload_query_samples(self, sample_list: List[int]) -> None:
     def get_samples(self, id_list: List[int]) -> Samples:
         batch_size: int = len(id_list)
         ts_idx: int = 0
-        while self.num_requests_cumsum[ts_idx] <= self.cnt:
-            ts_idx += 1
-        offset: int = 0 if ts_idx == 0 else self.num_requests_cumsum[ts_idx - 1]
+        with self._lock:
+            current_cnt: int = self.cnt
+            while self.num_requests_cumsum[ts_idx] <= current_cnt:
+                ts_idx += 1
+            offset: int = 0 if ts_idx == 0 else self.num_requests_cumsum[ts_idx - 1]
+            self.repeat += 1
+            if self.repeat == self.num_repeats:
+                self.repeat = 0
+                self.cnt += batch_size
         output: Samples = self.ds.get_samples_with_ts(
-            self.run_order[ts_idx][self.cnt - offset : self.cnt + batch_size - offset],
+            self.run_order[ts_idx][current_cnt - offset : current_cnt + batch_size - offset],
             ts_idx + self.start_ts,
         )
-        self.repeat += 1
-        if self.repeat == self.num_repeats:
-            self.repeat = 0
-            self.cnt += batch_size
         return output
 
     def get_item_count(self) -> int:
         return self.total_requests
 
 
 def run(
-    dataset: str = "debug",
+    dataset: str = "sampled-streaming-100b",
     model_path: str = "",
     scenario_name: str = "Server",
     batchsize: int = 16,

recommendation/dlrm_v3/model_family.py

@@ -306,10 +306,10 @@ def load(self, model_path: str) -> None:
             processes.append(p)
 
     def distributed_setup(self, rank: int, world_size: int, model_path: str) -> None:
-        nprocs_per_rank = 16
-        start_core: int = nprocs_per_rank * rank
-        cores: set[int] = set([start_core + i for i in range(nprocs_per_rank)])
-        os.sched_setaffinity(0, cores)
+        # nprocs_per_rank = 16
+        # start_core: int = nprocs_per_rank * rank + 128
+        # cores: set[int] = set([start_core + i for i in range(nprocs_per_rank)])
+        # os.sched_setaffinity(0, cores)
         set_is_inference(is_inference=not self.compute_eval)
         model = get_hstu_model(
             table_config=self.table_config,
@@ -366,6 +366,9 @@ def distributed_setup(self, rank: int, world_size: int, model_path: str) -> None
                         max_num_candidates=max_num_candidates,
                         num_candidates=num_candidates,
                     )
+                    # mt_target_preds = torch.empty(1, 2048 * 20).to(device="cpu")
+                    # mt_target_labels = None
+                    # mt_target_weights = None
                     assert mt_target_preds is not None
                     mt_target_preds = mt_target_preds.detach().to(device="cpu")
                     if mt_target_labels is not None:

recommendation/dlrm_v3/user.conf

@@ -1,6 +1,6 @@
 # Please set these fields depending on the performance of your system to
 # override default LoadGen settings.
-*.SingleStream.target_latency = 150
-*.MultiStream.target_latency = 150
-*.Server.target_latency = 150
-*.Server.min_duration = 20000
+*.SingleStream.target_latency = 100
+*.MultiStream.target_latency = 100
+*.Server.target_latency = 100
+# *.Server.min_duration = 20000