Merge pull request #20 from liubin/fix/make-num-thread-configurable

make thread-num for nydusd should be configurable

cmd/containerd-nydus-grpc/pkg/command/flags.go

@@ -47,6 +47,7 @@ type Args struct {
 	DisableCacheManager  bool
 	LogToStdout          bool
 	EnableNydusOverlayFS bool
+	NydusdThreadNum      int
 }
 
 type Flags struct {
@@ -180,6 +181,11 @@ func buildFlags(args *Args) []cli.Flag {
 			Usage:       "whether to enable nydus-overlayfs to mount",
 			Destination: &args.EnableNydusOverlayFS,
 		},
+		&cli.IntFlag{
+			Name:        "nydusd-thread-num",
+			Usage:       "Nydusd daemon thread-num, default will be set to the number of CPUs",
+			Destination: &args.NydusdThreadNum,
+		},
 	}
 }
 
@@ -233,6 +239,7 @@ func Validate(args *Args, cfg *config.Config) error {
 	cfg.EnableStargz = args.EnableStargz
 	cfg.DisableCacheManager = args.DisableCacheManager
 	cfg.EnableNydusOverlayFS = args.EnableNydusOverlayFS
+	cfg.NydusdThreadNum = args.NydusdThreadNum
 
 	d, err := time.ParseDuration(args.GCPeriod)
 	if err != nil {

config/config.go

@@ -50,6 +50,7 @@ type Config struct {
 	LogToStdout          bool          `toml:"log_to_stdout"`
 	DisableCacheManager  bool          `toml:"disable_cache_manager"`
 	EnableNydusOverlayFS bool          `toml:"enable_nydus_overlayfs"`
+	NydusdThreadNum      int           `toml:"nydusd_thread_num"`
 }
 
 func (c *Config) FillupWithDefaults() error {

pkg/daemon/config.go

@@ -131,3 +131,10 @@ func WithAPISock(apiSock string) NewDaemonOpt {
 		return nil
 	}
 }
+
+func WithNydusdThreadNum(nydusdThreadNum int) NewDaemonOpt {
+	return func(d *Daemon) error {
+		d.nydusdThreadNum = nydusdThreadNum
+		return nil
+	}
+}

pkg/daemon/daemon.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
 
 	"github.com/containerd/nydus-snapshotter/config"
 	"github.com/containerd/nydus-snapshotter/pkg/nydussdk"
@@ -39,6 +40,7 @@ type Daemon struct {
 	apiSock          *string
 	RootMountPoint   *string
 	CustomMountPoint *string
+	nydusdThreadNum  int
 }
 
 func (d *Daemon) SharedMountPoint() string {
@@ -67,6 +69,17 @@ func (d *Daemon) ConfigFile() string {
 	return filepath.Join(d.ConfigDir, "config.json")
 }
 
+// NydusdThreadNum returns `nydusd-thread-num` for nydusd if set,
+// otherwise will return the number of CPUs as default.
+func (d *Daemon) NydusdThreadNum() string {
+	if d.nydusdThreadNum > 0 {
+		return strconv.Itoa(d.nydusdThreadNum)
+	}
+	// if nydusd-thread-num is not set, return empty string
+	// to let manager don't set thread-num option.
+	return ""
+}
+
 func (d *Daemon) APISock() string {
 	if d.apiSock != nil {
 		return *d.apiSock

pkg/filesystem/nydus/config.go

@@ -135,3 +135,10 @@ func WithLogToStdout(logToStdout bool) NewFSOpt {
 		return nil
 	}
 }
+
+func WithNydusdThreadNum(nydusdThreadNum int) NewFSOpt {
+	return func(d *filesystem) error {
+		d.nydusdThreadNum = nydusdThreadNum
+		return nil
+	}
+}

pkg/filesystem/nydus/fs.go

@@ -45,6 +45,7 @@ type filesystem struct {
 	logLevel         string
 	logDir           string
 	logToStdout      bool
+	nydusdThreadNum  int
 }
 
 // NewFileSystem initialize Filesystem instance
@@ -79,6 +80,7 @@ func (fs *filesystem) newSharedDaemon() (*daemon.Daemon, error) {
 		daemon.WithRootMountPoint(filepath.Join(fs.RootDir, "mnt")),
 		daemon.WithLogLevel(fs.logLevel),
 		daemon.WithLogToStdout(fs.logToStdout),
+		daemon.WithNydusdThreadNum(fs.nydusdThreadNum),
 		modeOpt,
 	)
 	if err != nil {
@@ -361,6 +363,7 @@ func (fs *filesystem) createNewDaemon(snapshotID string, imageID string) (*daemo
 		daemon.WithLogLevel(fs.logLevel),
 		daemon.WithLogToStdout(fs.logToStdout),
 		daemon.WithCustomMountPoint(customMountPoint),
+		daemon.WithNydusdThreadNum(fs.nydusdThreadNum),
 	); err != nil {
 		return nil, err
 	}
@@ -397,6 +400,7 @@ func (fs *filesystem) createSharedDaemon(snapshotID string, imageID string) (*da
 		daemon.WithImageID(imageID),
 		daemon.WithLogLevel(fs.logLevel),
 		daemon.WithLogToStdout(fs.logToStdout),
+		daemon.WithNydusdThreadNum(fs.nydusdThreadNum),
 	); err != nil {
 		return nil, err
 	}

pkg/filesystem/stargz/config.go

@@ -97,3 +97,10 @@ func WithLogToStdout(logToStdout bool) NewFSOpt {
 }
 
 type NewFSOpt func(d *filesystem) error
+
+func WithNydusdThreadNum(nydusdThreadNum int) NewFSOpt {
+	return func(d *filesystem) error {
+		d.nydusdThreadNum = nydusdThreadNum
+		return nil
+	}
+}

pkg/filesystem/stargz/fs.go

@@ -41,6 +41,7 @@ type filesystem struct {
 	logLevel              string
 	logDir                string
 	logToStdout           bool
+	nydusdThreadNum       int
 }
 
 func NewFileSystem(ctx context.Context, opt ...NewFSOpt) (fs.FileSystem, error) {
@@ -149,6 +150,7 @@ func (f *filesystem) createNewDaemon(snapshotID string, imageID string) (*daemon
 		daemon.WithImageID(imageID),
 		daemon.WithLogLevel(f.logLevel),
 		daemon.WithLogToStdout(f.logToStdout),
+		daemon.WithNydusdThreadNum(f.nydusdThreadNum),
 	)
 	if err != nil {
 		return nil, err

pkg/process/manager.go

@@ -134,8 +134,12 @@ func (m *Manager) buildStartCommand(d *daemon.Daemon) (*exec.Cmd, error) {
 	args := []string{
 		"--apisock", d.APISock(),
 		"--log-level", d.LogLevel,
-		"--thread-num", "10",
 	}
+	nydusdThreadNum := d.NydusdThreadNum()
+	if nydusdThreadNum != "" {
+		args = append(args, "--thread-num", nydusdThreadNum)
+	}
+
 	if !d.LogToStdout {
 		args = append(args, "--log-file", d.LogFile())
 	}

snapshot/snapshot.go

@@ -113,6 +113,7 @@ func NewSnapshotter(ctx context.Context, cfg *config.Config) (snapshots.Snapshot
 		nydus.WithLogLevel(cfg.LogLevel),
 		nydus.WithLogDir(cfg.LogDir),
 		nydus.WithLogToStdout(cfg.LogToStdout),
+		nydus.WithNydusdThreadNum(cfg.NydusdThreadNum),
 	}
 
 	if !cfg.DisableCacheManager {
@@ -149,6 +150,7 @@ func NewSnapshotter(ctx context.Context, cfg *config.Config) (snapshots.Snapshot
 				stargz.WithLogLevel(cfg.LogLevel),
 				stargz.WithLogDir(cfg.LogDir),
 				stargz.WithLogToStdout(cfg.LogToStdout),
+				stargz.WithNydusdThreadNum(cfg.NydusdThreadNum),
 			)
 			if err != nil {
 				return nil, errors.Wrap(err, "failed to initialize stargz filesystem")