Make I/O and prefetch worker threads configurable

cfg/rippled-example.cfg

@@ -779,6 +779,14 @@
 #   number of processor threads plus 2 for networked nodes. Nodes running in
 #   stand alone mode default to 1 worker.
 #
+# [io_workers]
+#
+#   Configures the number of threads for processing raw inbound and outbound IO.
+#
+# [prefetch_workers]
+#
+#   Configures the number of threads for performing nodestore prefetching.
+#
 #
 #
 # [network_id]

src/ripple/app/main/Application.cpp

@@ -246,6 +246,10 @@ public:
 #if RIPPLE_SINGLE_IO_SERVICE_THREAD
         return 1;
 #else
+
+        if (config.IO_WORKERS > 0)
+            return config.IO_WORKERS;
+
         auto const cores = std::thread::hardware_concurrency();
 
         // Use a single thread when running on under-provisioned systems
@@ -342,7 +346,8 @@ public:
               m_collectorManager->collector(),
               logs_->journal("Resource")))
 
-        , m_nodeStore(m_shaMapStore->makeNodeStore(4))
+        , m_nodeStore(m_shaMapStore->makeNodeStore(
+              config_->PREFETCH_WORKERS > 0 ? config_->PREFETCH_WORKERS : 4))
 
         , nodeFamily_(*this, *m_collectorManager)
 

src/ripple/core/Config.h

@@ -214,8 +214,11 @@ public:
     // Amendment majority time
     std::chrono::seconds AMENDMENT_MAJORITY_TIME = defaultAmendmentMajorityTime;
 
-    // Thread pool configuration
-    int WORKERS = 0;
+    // Thread pool configuration (0 = choose for me)
+    int WORKERS = 0;           // jobqueue thread count. default: upto 6
+    int IO_WORKERS = 0;        // io svc thread count. default: 2
+    int PREFETCH_WORKERS = 0;  // prefetch thread count. default: 4
+
     // Can only be set in code, specifically unit tests
     bool FORCE_MULTI_THREAD = false;
 

src/ripple/core/ConfigSections.h

@@ -96,6 +96,8 @@ struct ConfigSection
 #define SECTION_VALIDATOR_TOKEN "validator_token"
 #define SECTION_VETO_AMENDMENTS "veto_amendments"
 #define SECTION_WORKERS "workers"
+#define SECTION_IO_WORKERS "io_workers"
+#define SECTION_PREFETCH_WORKERS "prefetch_workers"
 #define SECTION_LEDGER_REPLAY "ledger_replay"
 #define SECTION_BETA_RPC_API "beta_rpc_api"
 #define SECTION_SWEEP_INTERVAL "sweep_interval"

src/ripple/core/impl/Config.cpp

@@ -655,6 +655,26 @@ Config::loadFromString(std::string const& fileContents)
                 ": must be between 1 and 1024 inclusive.");
     }
 
+    if (getSingleSection(secConfig, SECTION_IO_WORKERS, strTemp, j_))
+    {
+        IO_WORKERS = beast::lexicalCastThrow<int>(strTemp);
+
+        if (IO_WORKERS < 1 || IO_WORKERS > 1024)
+            Throw<std::runtime_error>(
+                "Invalid " SECTION_IO_WORKERS
+                ": must be between 1 and 1024 inclusive.");
+    }
+
+    if (getSingleSection(secConfig, SECTION_PREFETCH_WORKERS, strTemp, j_))
+    {
+        PREFETCH_WORKERS = beast::lexicalCastThrow<int>(strTemp);
+
+        if (PREFETCH_WORKERS < 1 || PREFETCH_WORKERS > 1024)
+            Throw<std::runtime_error>(
+                "Invalid " SECTION_PREFETCH_WORKERS
+                ": must be between 1 and 1024 inclusive.");
+    }
+
     if (getSingleSection(secConfig, SECTION_COMPRESSION, strTemp, j_))
         COMPRESSION = beast::lexicalCastThrow<bool>(strTemp);