DAOS-14408 common: enable NDCTL for DCPM

src/control/cmd/dmg/auto_test.go

@@ -566,6 +566,8 @@ engines:
  provider: ofi+verbs
  fabric_iface: ib0
  fabric_iface_port: 31416
+ env_vars:
+ - ABT_THREAD_STACKSIZE=18432
  pinned_numa_node: 0
 - targets: 6
  nr_xs_helpers: 0
@@ -583,6 +585,8 @@ engines:
  provider: ofi+verbs
  fabric_iface: ib1
  fabric_iface_port: 32416
+ env_vars:
+ - ABT_THREAD_STACKSIZE=18432
  pinned_numa_node: 1
 disable_vfio: false
 disable_vmd: false

src/control/server/config/server_test.go

@@ -129,8 +129,14 @@ func mockConfigFromFile(t *testing.T, path string) (*Server, error) {
  t.Helper()
  c := DefaultServer()
  c.Path = path
+ err := c.Load()
+ if err == nil {
+ for i := 0; i < len(c.Engines); i++ {
+ c.Engines[i] = c.Engines[i].WithSDSForRam().WithStackSizeForDCPM()
+ }
+ }
 
- return c, c.Load()
+ return c, err
 }
 
 func TestServerConfig_MarshalUnmarshal(t *testing.T) {

src/control/server/engine/config.go

@@ -9,6 +9,7 @@ package engine
 import (
  "fmt"
  "os"
+ "strconv"
  "strings"
 
  "github.com/pkg/errors"
@@ -28,6 +29,8 @@ const (
  envLogMasks = "D_LOG_MASK"
  envLogDbgStreams = "DD_MASK"
  envLogSubsystems = "DD_SUBSYS"
+
+ MIN_ABT_THREAD_STACKSIZE_FOR_DCPM = 18432
 )
 
 // FabricConfig encapsulates networking fabric configuration.
@@ -343,6 +346,41 @@ func (c *Config) Validate() error {
  return errors.Wrap(err, "validate engine log subsystems")
  }
 
+ // Ensure proper environment variables for PMDK w/ NDCTL enabled.
+ pmemobj_conf_str, pmemobj_conf_err := c.GetEnvVar("PMEMOBJ_CONF")
+ if c.Storage.Tiers[0].Class == storage.ClassDcpm {
+ // Ensure 18KiB ABT stack size for an engine with DCPM storage class.
+ stacksize_str, err := c.GetEnvVar("ABT_THREAD_STACKSIZE")
+ if err != nil {
+ fmt.Printf("env_var ABT_THREAD_STACKSIZE set to %d for `dcpm` storage class\n",
+ MIN_ABT_THREAD_STACKSIZE_FOR_DCPM)
+ c.EnvVars = append(c.EnvVars, fmt.Sprintf("ABT_THREAD_STACKSIZE=%d",
+ MIN_ABT_THREAD_STACKSIZE_FOR_DCPM))
+ } else {
+ stacksize_value, err := strconv.Atoi(stacksize_str)
+ if err != nil || stacksize_value < MIN_ABT_THREAD_STACKSIZE_FOR_DCPM {
+ return errors.New(fmt.Sprintf("env_var ABT_THREAD_STACKSIZE should be >= %d for `dcpm` storage class",
+ MIN_ABT_THREAD_STACKSIZE_FOR_DCPM))
+ }
+ }
+ // Ensure default handling of shutdown state (SDS) for DCPM storage class.
+ if pmemobj_conf_err != nil && strings.Contains(pmemobj_conf_str, "sds.at_create") {
+ return errors.New("env_var PMEMOBJ_CONF should NOT be set to sds.at_create=? for non-`dcpm` storage class")
+ }
+ } else {
+ // Disable shutdown state (SDS) (part of RAS) for RAM-based simulated SCM.
+ // RAM doesn't support this feature and trying to use
+ // it will fail the create/open operations.
+ if pmemobj_conf_err != nil {
+ fmt.Printf("env_var PMEMOBJ_CONF set to sds.at_create=0 for non-`dcpm` storage class\n")
+ c.EnvVars = append(c.EnvVars, "PMEMOBJ_CONF=sds.at_create=0")
+ } else if strings.Contains(pmemobj_conf_str, "sds.at_create") {
+ if strings.Contains(pmemobj_conf_str, "sds.at_create=1") {
+ return errors.New("env_var PMEMOBJ_CONF should be set to sds.at_create=0 for non-`dcpm` storage class")
+ }
+ }
+ }
+
  return nil
 }
 
@@ -480,7 +518,7 @@ func (c *Config) WithSystemName(name string) *Config {
 func (c *Config) WithStorage(cfgs ...*storage.TierConfig) *Config {
  c.Storage.Tiers = storage.TierConfigs{}
  c.AppendStorage(cfgs...)
- return c
+ return c.WithStackSizeForDCPM().WithSDSForRam()
 }
 
 // AppendStorage appends the given storage tier configurations to
@@ -690,3 +728,17 @@ func (c *Config) WithStorageIndex(i uint32) *Config {
  c.Storage.EngineIdx = uint(i)
  return c
 }
+
+func (c *Config) WithStackSizeForDCPM() *Config {
+ if len(c.Storage.Tiers) > 0 && c.Storage.Tiers[0].Class == storage.ClassDcpm {
+ c.EnvVars = append(c.EnvVars, fmt.Sprintf("ABT_THREAD_STACKSIZE=%d", MIN_ABT_THREAD_STACKSIZE_FOR_DCPM))
+ }
+ return c
+}
+
+func (c *Config) WithSDSForRam() *Config {
+ if len(c.Storage.Tiers) > 0 && c.Storage.Tiers[0].Class != storage.ClassDcpm {
+ c.EnvVars = append(c.EnvVars, "PMEMOBJ_CONF=sds.at_create=0")
+ }
+ return c
+}

src/control/server/engine/config_test.go

@@ -177,6 +177,8 @@ func TestConfig_Constructed(t *testing.T) {
  t.Fatal(err)
  }
 
+ fromDisk = fromDisk.WithSDSForRam().WithStackSizeForDCPM()
+
  if diff := cmp.Diff(fromDisk, constructed, defConfigCmpOpts...); diff != "" {
  t.Fatalf("(-want, +got):\n%s", diff)
  }
@@ -730,6 +732,7 @@ func TestConfig_ToCmdVals(t *testing.T) {
  "D_LOG_MASK=" + logMask,
  "CRT_TIMEOUT=" + strconv.FormatUint(uint64(crtTimeout), 10),
  "FI_OFI_RXM_USE_SRX=0",
+ "PMEMOBJ_CONF=sds.at_create=0",
  }
 
  gotArgs, err := cfg.CmdLineArgs()

src/control/server/engine/exec_test.go

@@ -184,6 +184,7 @@ func TestRunnerNormalExit(t *testing.T) {
  "CRT_TIMEOUT=30",
  "D_INTERFACE=qib0",
  "D_LOG_MASK=DEBUG,MGMT=DEBUG,RPC=ERR,MEM=ERR",
+ "PMEMOBJ_CONF=sds.at_create=0",
  allowedUserEnv + "=" + allowedUserVal,
  }
  sort.Strings(env)

src/mgmt/srv_target.c

@@ -1180,7 +1180,7 @@ ds_mgmt_hdlr_tgt_create(crt_rpc_t *tc_req)
  /* A zero size accommodates the existing file */
  vpa.vpa_scm_size = 0;
  vpa.vpa_nvme_size = tc_in->tc_nvme_size / dss_tgt_nr;
- rc = dss_thread_collective(tgt_vos_create_one, &vpa, 0);
+ rc = dss_thread_collective(tgt_vos_create_one, &vpa, DSS_ULT_DEEP_STACK);
  if (rc) {
  D_ERROR(DF_UUID": thread collective tgt_vos_create_one failed, "DF_RC"\n",
  DP_UUID(tc_in->tc_pool_uuid), DP_RC(rc));