Small fixes for grouped benchmark run

Author: pgmpablo157321

loadgen/benchmark/repro.cpp

@@ -33,6 +33,9 @@ class QSL : public mlperf::QuerySampleLibrary {
   const std::string& Name() override { return mName; }
   size_t TotalSampleCount() override { return 1000000; }
   size_t PerformanceSampleCount() override { return TotalSampleCount(); }
+  size_t GroupSize(size_t i) override { return 1; }
+  size_t GroupOf(size_t i) override { return i; }
+  size_t NumberOfGroups() override { return TotalSampleCount(); }
   void LoadSamplesToRam(const std::vector<mlperf::QuerySampleIndex>&) override {
   }
   void UnloadSamplesFromRam(

loadgen/bindings/python_api.cc

@@ -113,6 +113,7 @@ class QuerySampleLibraryTrampoline : public QuerySampleLibrary {
   size_t TotalSampleCount() override { return total_sample_count_; }
   size_t PerformanceSampleCount() override { return performance_sample_count_; }
   size_t GroupSize(size_t i) override { return 1; }
+  size_t GroupOf(size_t i) override { return i; }
   size_t NumberOfGroups() override { return total_sample_count_; }
 
   void LoadSamplesToRam(const std::vector<QuerySampleIndex>& samples) override {
@@ -157,7 +158,7 @@ class GroupedQuerySampleLibraryTrampoline : public QuerySampleLibrary {
       for(ssize_t i = 0; i < group_sizes.shape()[0]; i++){
         group_sizes_.push_back(ptr[i]);
         total_sample_count_ += ptr[i];
-        for(ssize_t j = 0; j < ptr[i]; j++){
+        for(size_t j = 0; j < ptr[i]; j++){
           group_idx_.push_back(i);
         }
       }
@@ -330,7 +331,7 @@ void StartTestWithLogSettings(uintptr_t sut, uintptr_t qsl,
                     audit_config_filename);
 }
 
-void StartTestWithGroupedTest(
+void StartTestWithGroupedQSL(
   uintptr_t sut, uintptr_t qsl, mlperf::TestSettings test_settings,
                const std::string& audit_config_filename){
   pybind11::gil_scoped_release gil_releaser;
@@ -551,7 +552,7 @@ PYBIND11_MODULE(mlperf_loadgen, m) {
         "Destroy the object created by ConstructQDL.");
 
   m.def("ConstructGroupedQSL", &py::ConstructGroupedQSL,
-        "Construct the query sample library.");
+        "Construct grouped query sample library.");
   m.def("DestroyGroupedQSL", &py::DestroyQSL,
         "Destroy the object created by ConstructGroupedQSL.");
 
@@ -577,6 +578,9 @@ PYBIND11_MODULE(mlperf_loadgen, m) {
         "IssueQuery calls have finished.",
         pybind11::arg("responses"),
         pybind11::arg("response_cb") = ResponseCallback{});
+  m.def("StartTestWithGroupedQSL", &py::StartTestWithGroupedQSL,
+        "Run tests on a SUT created by ConstructSUT() and a QSL created by"
+        "ConstructGroupedQSL");
 }
 
 }  // namespace py

loadgen/demos/py_demo_grouped_qsl.py

@@ -0,0 +1,82 @@
+# Copyright 2019 The MLPerf Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =============================================================================
+
+"""Python demo showing how to use the MLPerf Inference load generator bindings.
+"""
+
+from __future__ import print_function
+
+import threading
+import time
+
+import numpy as np
+from absl import app
+import mlperf_loadgen
+
+
+def load_samples_to_ram(query_samples):
+    del query_samples
+    return
+
+
+def unload_samples_from_ram(query_samples):
+    del query_samples
+    return
+
+
+def process_query_async(query_samples):
+    time.sleep(0.001)
+    responses = []
+    for s in query_samples:
+        print(s.index)
+        responses.append(mlperf_loadgen.QuerySampleResponse(s.id, 0, 0))
+    mlperf_loadgen.QuerySamplesComplete(responses)
+
+
+def issue_query(query_samples):
+    threading.Thread(target=process_query_async, args=[query_samples]).start()
+
+
+def flush_queries():
+    pass
+
+
+def main(argv):
+    del argv
+    settings = mlperf_loadgen.TestSettings()
+    settings.scenario = mlperf_loadgen.TestScenario.Server
+    settings.mode = mlperf_loadgen.TestMode.PerformanceOnly
+    settings.server_target_qps = 100
+    settings.server_target_latency_ns = 100000000
+    settings.min_query_count = 100
+    settings.min_duration_ms = 10000
+    settings.server_constant_gen = True
+    settings.use_grouped_qsl = True
+
+    sut = mlperf_loadgen.ConstructSUT(issue_query, flush_queries)
+    qsl = mlperf_loadgen.ConstructGroupedQSL(
+        np.array([16 for _ in range(64)] + [32 for _ in range(32)], dtype=np.uint64), 16*128, load_samples_to_ram, unload_samples_from_ram
+    )
+
+    # qsl = mlperf_loadgen.ConstructQSL(
+    #     1024, 128, load_samples_to_ram, unload_samples_from_ram
+    # )
+    mlperf_loadgen.StartTestWithGroupedQSL(sut, qsl, settings, "")
+    mlperf_loadgen.DestroyQSL(qsl)
+    mlperf_loadgen.DestroySUT(sut)
+
+
+if __name__ == "__main__":
+    app.run(main)

loadgen/loadgen.cc

@@ -208,7 +208,7 @@ auto ScheduleDistribution<TestScenario::Server>(double qps) {
 }
 
 auto ScheduleConstantDistribution(double qps){
-  return [dist = std::uniform_real_distribution<>(1.0 / qps)](auto& gen) mutable {
+  return [dist = std::uniform_real_distribution<>(1.0 / qps, 1.0 / qps)](auto& gen) mutable {
     return std::chrono::duration_cast<std::chrono::nanoseconds>(
         std::chrono::duration<double>(dist(gen)));
   };
@@ -353,7 +353,7 @@ std::vector<QueryMetadata> GenerateQueries(
   std::vector<size_t> groups;
   std::vector<size_t> groups_first;
   size_t number_of_groups = 0;
-  size_t g, group_size;
+  size_t g = 0, group_size = 1;
 
   if (settings.use_grouped_qsl) {
     size_t current_idx = 0;
@@ -362,10 +362,9 @@ std::vector<QueryMetadata> GenerateQueries(
       size_t current_group = qsl->GroupOf(loaded_samples[current_idx]);
       groups.push_back(current_group);
       groups_first.push_back(current_idx);
-      current_idx += qsl->GroupSize(loaded_samples[current_idx]);
+      current_idx += qsl->GroupSize(current_group);
       number_of_groups++;
     }
-    
   }
 
   auto grouped_sample_distribution = SampleDistribution<mode>(
@@ -419,7 +418,7 @@ std::vector<QueryMetadata> GenerateQueries(
       }
     } else if (settings.use_grouped_qsl) {
       g = grouped_sample_distribution(sample_rng);
-      group_size = qsl->GroupSize(loaded_samples[groups_first[g]]);
+      group_size = qsl->GroupSize(qsl->GroupOf(groups_first[g]));
     } else {
       for (auto& s : samples) {
         s = loaded_samples[settings.performance_issue_unique
@@ -430,23 +429,24 @@ std::vector<QueryMetadata> GenerateQueries(
                                : sample_distribution(sample_rng)];
       }
     }
+    prev_timestamp = timestamp;
     if (!settings.use_grouped_qsl) {
       queries.emplace_back(samples, timestamp, response_delegate, sequence_gen);
     } else {
       for (size_t i = 0; i < group_size; i++){
         samples[0] = loaded_samples[groups_first[g]+i];
         queries.emplace_back(samples, timestamp, response_delegate, sequence_gen);
+        timestamp += schedule_constant_distribution(schedule_rng);
       }
+      prev_timestamp = timestamp - schedule_constant_distribution(schedule_rng);
     }
-    prev_timestamp = timestamp;
-    if (settings.server_constant_gen && (scenario == TestScenario::Server)){
-      if(!settings.use_grouped_qsl){
+
+    if (!settings.use_grouped_qsl){
+      if (settings.server_constant_gen && (scenario == TestScenario::Server)){
         timestamp += schedule_constant_distribution(schedule_rng);
       } else {
-        timestamp += group_size * schedule_constant_distribution(schedule_rng);
+        timestamp += schedule_distribution(schedule_rng);
       }
-    } else {
-      timestamp += schedule_distribution(schedule_rng);
     }
     // In equal_issue mode, the min_queries will be bumped up by a multiple of
     // the dataset size if the test time has not met the threshold.
@@ -455,6 +455,7 @@ std::vector<QueryMetadata> GenerateQueries(
         (scenario != TestScenario::Offline)) {
       min_queries += loaded_samples.size();
     }
+    
   }
 
   // See if we need to create a "remainder" query for offline+accuracy to