Merge pull request #1364 from sporksmith/sched-options

sporksmith · web-flow · commit 52b1ed146aa1 · 2021-05-18T11:03:04.000-05:00
Update scheduler options

* The Experimental option `max-concurrency` is now the General option `parallelism`.
* The default scheduler policy is now Host.
* The General option `workers` is now the Experimental option `worker-threads`, and no longer has the `-w` shortcut. Users should use `parallelism` instead. If not set explicitly, the number of worker threads will be set to the number of hosts in the simulation.
* "0" for "parallelism" is now rejected instead of being silently changed to 1.
diff --git a/docs/3.1-Shadow-Config.md b/docs/3.1-Shadow-Config.md
@@ -27,16 +27,15 @@ hosts:
 - [`general.data_directory`](#generaldata_directory)
 - [`general.heartbeat_interval`](#generalheartbeat_interval)
 - [`general.log_level`](#generallog_level)
+- [`general.parallelism`](#generalparallelism)
 - [`general.seed`](#generalseed)
 - [`general.stop_time`](#generalstop_time)
 - [`general.template_directory`](#generaltemplate_directory)
-- [`general.workers`](#generalworkers)
 - [`topology`](#topology)
 - [`experimental`](#experimental)
 - [`experimental.interface_buffer`](#experimentalinterface_buffer)
 - [`experimental.interface_qdisc`](#experimentalinterface_qdisc)
 - [`experimental.interpose_method`](#experimentalinterpose_method)
-- [`experimental.max_concurrency`](#experimentalmax_concurrency)
 - [`experimental.preload_spin_max`](#experimentalpreload_spin_max)
 - [`experimental.runahead`](#experimentalrunahead)
 - [`experimental.scheduler_policy`](#experimentalscheduler_policy)
@@ -52,6 +51,7 @@ hosts:
 - [`experimental.use_sched_fifo`](#experimentaluse_sched_fifo)
 - [`experimental.use_shim_syscall_handler`](#experimentaluse_shim_syscall_handler)
 - [`experimental.use_syscall_counters`](#experimentaluse_syscall_counters)
+- [`experimental.worker_threads`](#experimentalworker_threads)
 - [`host_defaults`](#host_defaults)
 - [`host_defaults.city_code_hint`](#host_defaultscity_code_hint)
 - [`host_defaults.country_code_hint`](#host_defaultscountry_code_hint)
@@ -106,6 +106,14 @@ Type: "error" OR "warning" OR "info" OR "debug" OR "trace"
 
 Log level of output written on stdout. If Shadow was built in release mode, then messages at level 'trace' will always be dropped.
 
+#### `general.parallelism`
+
+Default: null  
+Type: Integer OR null
+
+How many parallel threads to use to run the simulation. Optimal performance is
+usually obtained with `nproc`, or sometimes `nproc/2` with hyperthreading.
+
 #### `general.seed`
 
 Default: 1  
@@ -127,13 +135,6 @@ Type: String OR null
 
 Path to recursively copy during startup and use as the data-directory.
 
-#### `general.workers`
-
-Default: 0  
-Type: Integer
-
-Run concurrently with N worker threads.
-
 #### `topology`
 
 *Required*  
@@ -181,13 +182,6 @@ Type: "ptrace" OR "preload" OR "hybrid"
 
 Which interposition method to use.
 
-#### `experimental.max_concurrency`
-
-Default: null  
-Type: Integer OR null
-
-Maximum number of workers to allow to run at once.
-
 #### `experimental.preload_spin_max`
 
 Default: 0  
@@ -204,7 +198,7 @@ If set, overrides the automatically calculated minimum time workers may run ahea
 
 #### `experimental.scheduler_policy`
 
-Default: "steal"  
+Default: "host"  
 Type: "host" OR "steal" OR "thread" OR "threadxthread" OR "threadxhost"
 
 The event scheduler's policy for thread synchronization.
@@ -293,6 +287,19 @@ Type: Bool
 
 Count the number of occurrences for individual syscalls.
 
+#### `experimental.worker_threads`
+
+Default: # of hosts in the simulation  
+Type: Integer
+
+Create N worker threads. Note though, that `general.parallelism` of them will be
+allowed to run simultaneously. If unset, will create a thread for each simulated
+Host. This is to work around limitations in ptrace, and may change in the
+future.
+
+"0" is a valid value, and will cause the simulation to be run directly on the
+main Shadow thread, but this functionality may be removed in the future.
+
 #### `host_defaults`
 
 Default options for all hosts. These options can also be overridden for each host individually.
diff --git a/src/main/bindings/c/bindings.h b/src/main/bindings/c/bindings.h
@@ -118,7 +118,7 @@ bool config_getUseShimSyscallHandler(const struct ConfigOptions *config);
 
 int32_t config_getPreloadSpinMax(const struct ConfigOptions *config);
 
-int32_t config_getMaxConcurrency(const struct ConfigOptions *config);
+uint32_t config_getParallelism(const struct ConfigOptions *config);
 
 SimulationTime config_getStopTime(const struct ConfigOptions *config);
 
diff --git a/src/main/core/scheduler/scheduler.c b/src/main/core/scheduler/scheduler.c
@@ -26,8 +26,8 @@
 #include "main/utility/utility.h"
 #include "support/logger/logger.h"
 
-static int _maxConcurrency = -1;
-ADD_CONFIG_HANDLER(config_getMaxConcurrency, _maxConcurrency)
+static int _parallelism;
+ADD_CONFIG_HANDLER(config_getParallelism, _parallelism)
 
 struct _Scheduler {
     // Unowned back-pointer.
@@ -130,9 +130,8 @@ Scheduler* scheduler_new(Manager* manager, SchedulerPolicyType policyType,
     // Unowned back-pointer
     scheduler->manager = manager;
 
-    scheduler->workerPool =
-        workerpool_new(manager, scheduler, /*nThreads=*/nWorkers,
-                       /*nConcurrent=*/_maxConcurrency);
+    scheduler->workerPool = workerpool_new(manager, scheduler, /*nThreads=*/nWorkers,
+                                           /*nParallel=*/_parallelism);
 
     scheduler->endTime = endTime;
     scheduler->currentRound.endTime = scheduler->endTime;// default to one single round
@@ -160,12 +159,12 @@ Scheduler* scheduler_new(Manager* manager, SchedulerPolicyType policyType,
             break;
         }
         case SP_PARALLEL_HOST_STEAL: {
-            if (nWorkers > _maxConcurrency) {
+            if (nWorkers > _parallelism) {
                 // Proceeding will cause the scheduler to deadlock, since the
                 // work stealing scheduler threads spin-wait for each-other to
                 // finish.
                 utility_panic("Host stealing scheduler is incompatible with "
-                              "--workers > --max-concurrency");
+                              "--workers > --parallelism");
                 abort();
             }
             scheduler->policy = schedulerpolicyhoststeal_new();
diff --git a/src/main/core/support/configuration.rs b/src/main/core/support/configuration.rs
@@ -8,6 +8,8 @@ use merge::Merge;
 use once_cell::sync::Lazy;
 use schemars::{schema_for, JsonSchema};
 use serde::{Deserialize, Serialize};
+use std::convert::TryInto;
+use std::num::NonZeroU32;
 
 use super::simulation_time::{SIMTIME_ONE_NANOSECOND, SIMTIME_ONE_SECOND};
 use super::units::{self, Unit};
@@ -133,11 +135,13 @@ pub struct GeneralOptions {
     #[serde(default = "default_some_1")]
     seed: Option<u32>,
 
-    /// Run concurrently with N worker threads
-    #[clap(long, short = 'w', value_name = "N")]
-    #[clap(about = GENERAL_HELP.get("workers").unwrap())]
-    #[serde(default = "default_some_0")]
-    workers: Option<u32>,
+    /// How many parallel threads to use to run the simulation. Optimal
+    /// performance is usually obtained with `nproc`, or sometimes `nproc/2`
+    /// with hyperthreading.
+    #[clap(long, short = 'p', value_name = "cores")]
+    #[clap(about = GENERAL_HELP.get("parallelism").unwrap())]
+    #[serde(default = "default_some_nz_1")]
+    parallelism: Option<NonZeroU32>,
 
     #[clap(long, value_name = "seconds")]
     #[clap(about = GENERAL_HELP.get("bootstrap_end_time").unwrap())]
@@ -221,11 +225,6 @@ pub struct ExperimentalOptions {
     #[clap(about = EXP_HELP.get("preload_spin_max").unwrap())]
     preload_spin_max: Option<i32>,
 
-    /// Maximum number of workers to allow to run at once
-    #[clap(long, value_name = "workers")]
-    #[clap(about = EXP_HELP.get("max_concurrency").unwrap())]
-    max_concurrency: Option<i32>,
-
     /// Use the MemoryManager. It can be useful to disable for debugging, but will hurt performance in
     /// most cases
     #[clap(long, value_name = "bool")]
@@ -286,6 +285,16 @@ pub struct ExperimentalOptions {
     #[clap(long, value_name = "mode")]
     #[clap(about = EXP_HELP.get("interface_qdisc").unwrap())]
     interface_qdisc: Option<QDiscMode>,
+
+    /// Create N worker threads. Note though, that `--parallelism` of them will
+    /// be allowed to run simultaneously. If unset, will create a thread for
+    /// each simulated Host. This is to work around limitations in ptrace, and
+    /// may change in the future. "0" is a valid value, and will cause the
+    /// simulation to be run directly on the main Shadow thread, but this
+    /// functionality may be removed in the future.
+    #[clap(long, value_name = "N")]
+    #[clap(about = EXP_HELP.get("worker_threads").unwrap())]
+    worker_threads: Option<u32>,
 }
 
 impl ExperimentalOptions {
@@ -305,19 +314,19 @@ impl Default for ExperimentalOptions {
             use_syscall_counters: Some(false),
             use_object_counters: Some(true),
             preload_spin_max: Some(0),
-            max_concurrency: None,
             use_memory_manager: Some(true),
             use_shim_syscall_handler: Some(true),
             use_cpu_pinning: Some(true),
             interpose_method: Some(InterposeMethod::Ptrace),
             runahead: None,
-            scheduler_policy: Some(SchedulerPolicy::Steal),
+            scheduler_policy: Some(SchedulerPolicy::Host),
             socket_send_buffer: Some(units::Bytes::new(131_072, units::SiPrefixUpper::Base)),
             socket_send_autotune: Some(true),
             socket_recv_buffer: Some(units::Bytes::new(174_760, units::SiPrefixUpper::Base)),
             socket_recv_autotune: Some(true),
             interface_buffer: Some(units::Bytes::new(1_024_000, units::SiPrefixUpper::Base)),
             interface_qdisc: Some(QDiscMode::Fifo),
+            worker_threads: None,
         }
     }
 }
@@ -640,16 +649,16 @@ fn default_some_time_0() -> Option<units::Time<units::TimePrefixUpper>> {
     Some(units::Time::new(0, units::TimePrefixUpper::Sec))
 }
 
-/// Helper function for serde default `Some(0)` values.
-fn default_some_0() -> Option<u32> {
-    Some(0)
-}
-
 /// Helper function for serde default `Some(1)` values.
 fn default_some_1() -> Option<u32> {
     Some(1)
 }
 
+/// Helper function for serde default `Some(1)` values.
+fn default_some_nz_1() -> Option<NonZeroU32> {
+    Some(std::num::NonZeroU32::new(1).unwrap())
+}
+
 /// Helper function for serde default `Some(0)` values.
 fn default_some_time_1() -> Option<units::Time<units::TimePrefixUpper>> {
     Some(units::Time::new(1, units::TimePrefixUpper::Sec))
@@ -1106,13 +1115,10 @@ mod export {
     }
 
     #[no_mangle]
-    pub extern "C" fn config_getMaxConcurrency(config: *const ConfigOptions) -> i32 {
+    pub extern "C" fn config_getParallelism(config: *const ConfigOptions) -> NonZeroU32 {
         assert!(!config.is_null());
         let config = unsafe { &*config };
-        match config.experimental.max_concurrency {
-            Some(x) => x,
-            None => -1,
-        }
+        config.general.parallelism.unwrap()
     }
 
     #[no_mangle]
@@ -1149,7 +1155,13 @@ mod export {
     pub extern "C" fn config_getWorkers(config: *const ConfigOptions) -> libc::c_uint {
         assert!(!config.is_null());
         let config = unsafe { &*config };
-        config.general.workers.unwrap()
+        match &config.experimental.worker_threads {
+            Some(w) => *w,
+            None => {
+                // By default use 1 worker per host.
+                config.hosts.len().try_into().unwrap()
+            }
+        }
     }
 
     #[no_mangle]
diff --git a/src/main/core/worker.c b/src/main/core/worker.c
@@ -134,17 +134,12 @@ static Worker* _worker_new(WorkerPool*, int);
 static void _worker_free(Worker*);
 
 WorkerPool* workerpool_new(Manager* manager, Scheduler* scheduler, int nWorkers,
-                           int nConcurrent) {
-    int nLogicalProcessors = 0;
-    if (nWorkers == 0 || nConcurrent == 0) {
-        // With no concurrency, we still use a single logical processor.
-        nLogicalProcessors = 1;
-    } else if (nConcurrent < 0 || nConcurrent > nWorkers) {
-        // Never makes sense to use more logical processors than workers.
-        nLogicalProcessors = nWorkers;
-    } else {
-        nLogicalProcessors = nConcurrent;
-    }
+                           int nParallel) {
+    // Should have been ensured earlier by `config_getParallelism`.
+    utility_assert(nParallel >= 1);
+
+    // Never makes sense to use more logical processors than workers.
+    int nLogicalProcessors = MIN(nParallel, nWorkers);
 
     WorkerPool* pool = g_new(WorkerPool, 1);
     *pool = (WorkerPool){
diff --git a/src/test/determinism/CMakeLists.txt b/src/test/determinism/CMakeLists.txt
@@ -11,13 +11,13 @@ foreach(METHOD ptrace hybrid)
         BASENAME determinism1a
         METHODS ${METHOD}
         SHADOW_CONFIG ${CMAKE_CURRENT_SOURCE_DIR}/determinism1.test.shadow.config.yaml
-        ARGS --use-cpu-pinning true --workers 2
+        ARGS --use-cpu-pinning true --parallelism 2
         PROPERTIES RUN_SERIAL TRUE)
     add_shadow_tests(
         BASENAME determinism1b
         METHODS ${METHOD}
         SHADOW_CONFIG ${CMAKE_CURRENT_SOURCE_DIR}/determinism1.test.shadow.config.yaml
-        ARGS --use-cpu-pinning true --workers 2
+        ARGS --use-cpu-pinning true --parallelism 2
         PROPERTIES RUN_SERIAL TRUE)
     add_test(
         NAME determinism1-compare-shadow-${METHOD}
@@ -39,13 +39,13 @@ foreach(METHOD ptrace hybrid)
         BASENAME determinism2a
         METHODS ${METHOD}
         SHADOW_CONFIG ${CMAKE_CURRENT_SOURCE_DIR}/determinism2.test.shadow.config.yaml
-        ARGS --use-cpu-pinning true --workers 2
+        ARGS --use-cpu-pinning true --parallelism 2
         PROPERTIES RUN_SERIAL TRUE)
     add_shadow_tests(
         BASENAME determinism2b
         METHODS ${METHOD}
         SHADOW_CONFIG ${CMAKE_CURRENT_SOURCE_DIR}/determinism2.test.shadow.config.yaml
-        ARGS --use-cpu-pinning true --workers 2
+        ARGS --use-cpu-pinning true --parallelism 2
         PROPERTIES RUN_SERIAL TRUE)
     ## now compare the output
     ## TODO enable this test and fix the remaining determinism issue
diff --git a/src/test/phold/CMakeLists.txt b/src/test/phold/CMakeLists.txt
@@ -27,13 +27,13 @@ add_shadow_tests(
     BASENAME phold-parallel
     METHODS hybrid ptrace
     LOGLEVEL info
-    ARGS --use-cpu-pinning true --workers 2
+    ARGS --use-cpu-pinning true --parallelism 2
     PROPERTIES RUN_SERIAL TRUE)
 add_shadow_tests(
     BASENAME phold-parallel
     METHODS preload
     LOGLEVEL info
-    ARGS --use-cpu-pinning true --workers 2
+    ARGS --use-cpu-pinning true --parallelism 2
     PROPERTIES RUN_SERIAL TRUE
     CONFIGURATIONS ilibc)
 
diff --git a/src/test/tor/minimal/CMakeLists.txt b/src/test/tor/minimal/CMakeLists.txt
@@ -25,7 +25,7 @@ add_shadow_tests(BASENAME tor-minimal
                  LOGLEVEL info
                  ARGS
                    --use-cpu-pinning true
-                   --workers 2
+                   --parallelism 2
                    --template-directory "shadow.data.template"
                  POST_CMD "${CMAKE_CURRENT_SOURCE_DIR}/verify.sh"
                  PROPERTIES
