diff --git a/cli/command_benchmark_compression.go b/cli/command_benchmark_compression.go index 93a72333d..4ecc597a2 100644 --- a/cli/command_benchmark_compression.go +++ b/cli/command_benchmark_compression.go @@ -27,7 +27,7 @@ type commandBenchmarkCompression struct { byAllocated bool verifyStable bool optionPrint bool - parallel int + parallel uint deprecated bool operations string algorithms string @@ -41,7 +41,7 @@ func (c *commandBenchmarkCompression) setup(svc appServices, parent commandParen cmd.Flag("data-file", "Use data from the given file").Required().ExistingFileVar(&c.dataFile) cmd.Flag("by-size", "Sort results by size").BoolVar(&c.bySize) cmd.Flag("by-alloc", "Sort results by allocated bytes").BoolVar(&c.byAllocated) - cmd.Flag("parallel", "Number of parallel goroutines").Default("1").IntVar(&c.parallel) + cmd.Flag("parallel", "Number of parallel goroutines").Default("1").UintVar(&c.parallel) cmd.Flag("operations", "Operations").Default("both").EnumVar(&c.operations, "compress", "decompress", "both") cmd.Flag("verify-stable", "Verify that compression is stable").BoolVar(&c.verifyStable) cmd.Flag("print-options", "Print out options usable for repository creation").BoolVar(&c.optionPrint) @@ -212,7 +212,7 @@ func (c *commandBenchmarkCompression) runCompression(ctx context.Context, data [ return compressedSize } - outputBuffers := makeOutputBuffers(c.parallel, defaultCompressedDataByMethod) + outputBuffers := makeOutputBuffers(parallelismAsInt(c.parallel), defaultCompressedDataByMethod) tt := timetrack.Start() @@ -281,7 +281,7 @@ func (c *commandBenchmarkCompression) runDecompression(ctx context.Context, data return uint64(compressedInput.Length()) } - outputBuffers := makeOutputBuffers(c.parallel, defaultCompressedDataByMethod) + outputBuffers := makeOutputBuffers(parallelismAsInt(c.parallel), defaultCompressedDataByMethod) tt := timetrack.Start() diff --git a/cli/command_blob_gc.go b/cli/command_blob_gc.go index 56865a005..b41f04874 100644 --- a/cli/command_blob_gc.go +++ b/cli/command_blob_gc.go @@ -12,7 +12,7 @@ type commandBlobGC struct { delete string - parallel int + parallel uint prefix string safety maintenance.SafetyParameters @@ -22,7 +22,7 @@ type commandBlobGC struct { func (c *commandBlobGC) setup(svc appServices, parent commandParent) { cmd := parent.Command("gc", "Garbage-collect unused blobs").Hidden() cmd.Flag("delete", "Whether to delete unused blobs").StringVar(&c.delete) - cmd.Flag("parallel", "Number of parallel blob scans").Default("16").IntVar(&c.parallel) + cmd.Flag("parallel", "Number of parallel blob scans").Default("16").UintVar(&c.parallel) cmd.Flag("prefix", "Only GC blobs with given prefix").StringVar(&c.prefix) safetyFlagVar(cmd, &c.safety) cmd.Action(svc.directRepositoryWriteAction(c.run)) @@ -35,7 +35,7 @@ func (c *commandBlobGC) run(ctx context.Context, rep repo.DirectRepositoryWriter opts := maintenance.DeleteUnreferencedPacksOptions{ DryRun: c.delete != "yes", - Parallel: c.parallel, + Parallel: parallelismAsInt(c.parallel), Prefix: blob.ID(c.prefix), } diff --git a/cli/command_cache_sync.go b/cli/command_cache_sync.go index b714cdd5d..1ccbdf586 100644 --- a/cli/command_cache_sync.go +++ b/cli/command_cache_sync.go @@ -12,12 +12,12 @@ ) type commandCacheSync struct { - parallel int + parallel uint } func (c *commandCacheSync) setup(svc appServices, parent commandParent) { cmd := parent.Command("sync", "Synchronizes the metadata cache with blobs in storage") - cmd.Flag("parallel", "Fetch parallelism").Default("16").IntVar(&c.parallel) + cmd.Flag("parallel", "Fetch parallelism").Default("16").UintVar(&c.parallel) cmd.Action(svc.directRepositoryWriteAction(c.run)) } diff --git a/cli/command_content_verify.go b/cli/command_content_verify.go index 44a5d32e3..be328efe9 100644 --- a/cli/command_content_verify.go +++ b/cli/command_content_verify.go @@ -14,7 +14,7 @@ ) type commandContentVerify struct { - contentVerifyParallel int + contentVerifyParallel uint contentVerifyFull bool contentVerifyIncludeDeleted bool contentVerifyPercent float64 @@ -26,7 +26,7 @@ type commandContentVerify struct { func (c *commandContentVerify) setup(svc appServices, parent commandParent) { cmd := parent.Command("verify", "Verify that each content is backed by a valid blob") - cmd.Flag("parallel", "Parallelism").Default("16").IntVar(&c.contentVerifyParallel) + cmd.Flag("parallel", "Parallelism").Default("16").UintVar(&c.contentVerifyParallel) cmd.Flag("full", "Full verification (including download)").BoolVar(&c.contentVerifyFull) cmd.Flag("include-deleted", "Include deleted contents").BoolVar(&c.contentVerifyIncludeDeleted) cmd.Flag("download-percent", "Download a percentage of files [0.0 .. 100.0]").Float64Var(&c.contentVerifyPercent) @@ -70,7 +70,7 @@ func (c *commandContentVerify) run(ctx context.Context, rep repo.DirectRepositor ContentIDRange: c.contentRange.contentIDRange(), ContentReadPercentage: c.contentVerifyPercent, IncludeDeletedContents: c.contentVerifyIncludeDeleted, - ContentIterateParallelism: c.contentVerifyParallel, + ContentIterateParallelism: parallelismAsInt(c.contentVerifyParallel), ProgressCallbackInterval: 1, ProgressCallback: func(vps content.VerifyProgressStats) { diff --git a/cli/command_index_inspect.go b/cli/command_index_inspect.go index 48edd5550..8f2f2a887 100644 --- a/cli/command_index_inspect.go +++ b/cli/command_index_inspect.go @@ -21,7 +21,7 @@ type commandIndexInspect struct { blobIDs []string contentIDs []string - parallel int + parallel uint out textOutput } @@ -31,7 +31,7 @@ func (c *commandIndexInspect) setup(svc appServices, parent commandParent) { cmd.Flag("all", "Inspect all index blobs in the repository, including inactive").BoolVar(&c.all) cmd.Flag("active", "Inspect all active index blobs").BoolVar(&c.active) cmd.Flag("content-id", "Inspect all active index blobs").StringsVar(&c.contentIDs) - cmd.Flag("parallel", "Parallelism").Default("8").IntVar(&c.parallel) + cmd.Flag("parallel", "Parallelism").Default("8").UintVar(&c.parallel) cmd.Arg("blobs", "Names of index blobs to inspect").StringsVar(&c.blobIDs) cmd.Action(svc.directRepositoryReadAction(c.run)) diff --git a/cli/command_index_recover.go b/cli/command_index_recover.go index 6ffde86b1..4aa398920 100644 --- a/cli/command_index_recover.go +++ b/cli/command_index_recover.go @@ -20,7 +20,7 @@ type commandIndexRecover struct { blobPrefixes []string commit bool ignoreErrors bool - parallel int + parallel uint deleteIndexes bool svc appServices @@ -30,7 +30,7 @@ func (c *commandIndexRecover) setup(svc appServices, parent commandParent) { cmd := parent.Command("recover", "Recover indexes from pack blobs") cmd.Flag("blob-prefixes", "Prefixes of pack blobs to recover from (default=all packs)").StringsVar(&c.blobPrefixes) cmd.Flag("blobs", "Names of pack blobs to recover from (default=all packs)").StringsVar(&c.blobIDs) - cmd.Flag("parallel", "Recover parallelism").Default("1").IntVar(&c.parallel) + cmd.Flag("parallel", "Recover parallelism").Default("1").UintVar(&c.parallel) cmd.Flag("ignore-errors", "Ignore errors when recovering").BoolVar(&c.ignoreErrors) cmd.Flag("delete-indexes", "Delete all indexes before recovering").BoolVar(&c.deleteIndexes) cmd.Flag("commit", "Commit recovered content").BoolVar(&c.commit) diff --git a/cli/command_repository_sync.go b/cli/command_repository_sync.go index 4b10df194..e6358591f 100644 --- a/cli/command_repository_sync.go +++ b/cli/command_repository_sync.go @@ -27,7 +27,7 @@ type commandRepositorySyncTo struct { repositorySyncUpdate bool repositorySyncDelete bool repositorySyncDryRun bool - repositorySyncParallelism int + repositorySyncParallelism uint repositorySyncDestinationMustExist bool repositorySyncTimes bool @@ -43,7 +43,7 @@ func (c *commandRepositorySyncTo) setup(svc advancedAppServices, parent commandP cmd.Flag("update", "Whether to update blobs present in destination and source if the source is newer.").Default(trueStr).BoolVar(&c.repositorySyncUpdate) cmd.Flag("delete", "Whether to delete blobs present in destination but not source.").BoolVar(&c.repositorySyncDelete) cmd.Flag("dry-run", "Do not perform copying.").Short('n').BoolVar(&c.repositorySyncDryRun) - cmd.Flag("parallel", "Copy parallelism.").Default("1").IntVar(&c.repositorySyncParallelism) + cmd.Flag("parallel", "Copy parallelism.").Default("1").UintVar(&c.repositorySyncParallelism) cmd.Flag("must-exist", "Fail if destination does not have repository format blob.").BoolVar(&c.repositorySyncDestinationMustExist) cmd.Flag("times", "Synchronize blob times if supported.").BoolVar(&c.repositorySyncTimes) diff --git a/cli/command_restore.go b/cli/command_restore.go index a1469857f..780ab7311 100644 --- a/cli/command_restore.go +++ b/cli/command_restore.go @@ -116,7 +116,7 @@ type commandRestore struct { restoreWriteSparseFiles bool restoreConsistentAttributes bool restoreMode string - restoreParallel int + restoreParallel uint restoreIgnorePermissionErrors bool restoreWriteFilesAtomically bool restoreSkipTimes bool @@ -147,7 +147,7 @@ func (c *commandRestore) setup(svc appServices, parent commandParent) { cmd.Flag("write-sparse-files", "When doing a restore, attempt to write files sparsely-allocating the minimum amount of disk space needed.").Default(falseStr).BoolVar(&c.restoreWriteSparseFiles) cmd.Flag("consistent-attributes", "When multiple snapshots match, fail if they have inconsistent attributes").Envar(svc.EnvName("KOPIA_RESTORE_CONSISTENT_ATTRIBUTES")).BoolVar(&c.restoreConsistentAttributes) cmd.Flag("mode", "Override restore mode").Default(restoreModeAuto).EnumVar(&c.restoreMode, restoreModeAuto, restoreModeLocal, restoreModeZip, restoreModeZipNoCompress, restoreModeTar, restoreModeTgz) - cmd.Flag("parallel", "Restore parallelism (1=disable)").Default("8").IntVar(&c.restoreParallel) + cmd.Flag("parallel", "Restore parallelism (1=disable)").Default("8").UintVar(&c.restoreParallel) cmd.Flag("skip-owners", "Skip owners during restore").BoolVar(&c.restoreSkipOwners) cmd.Flag("skip-permissions", "Skip permissions during restore").BoolVar(&c.restoreSkipPermissions) cmd.Flag("skip-times", "Skip times during restore").BoolVar(&c.restoreSkipTimes) @@ -441,7 +441,7 @@ func (c *commandRestore) run(ctx context.Context, rep repo.Repository) error { } st, err := restore.Entry(ctx, rep, output, rootEntry, restore.Options{ - Parallel: c.restoreParallel, + Parallel: parallelismAsInt(c.restoreParallel), Incremental: c.restoreIncremental, DeleteExtra: c.restoreDeleteExtra, IgnoreErrors: c.restoreIgnoreErrors, diff --git a/cli/command_snapshot_create.go b/cli/command_snapshot_create.go index 6787463e6..b1abe8c3c 100644 --- a/cli/command_snapshot_create.go +++ b/cli/command_snapshot_create.go @@ -33,7 +33,7 @@ type commandSnapshotCreate struct { snapshotCreateCheckpointInterval time.Duration snapshotCreateFailFast bool snapshotCreateForceHash float64 - snapshotCreateParallelUploads int + snapshotCreateParallelUploads uint snapshotCreateStartTime string snapshotCreateEndTime string snapshotCreateForceEnableActions bool @@ -67,7 +67,7 @@ func (c *commandSnapshotCreate) setup(svc appServices, parent commandParent) { cmd.Flag("description", "Free-form snapshot description.").StringVar(&c.snapshotCreateDescription) cmd.Flag("fail-fast", "Fail fast when creating snapshot.").Envar(svc.EnvName("KOPIA_SNAPSHOT_FAIL_FAST")).BoolVar(&c.snapshotCreateFailFast) cmd.Flag("force-hash", "Force hashing of source files for a given percentage of files [0.0 .. 100.0]").Default("0").Float64Var(&c.snapshotCreateForceHash) - cmd.Flag("parallel", "Upload N files in parallel").PlaceHolder("N").Default("0").IntVar(&c.snapshotCreateParallelUploads) + cmd.Flag("parallel", "Upload N files in parallel").PlaceHolder("N").Default("0").UintVar(&c.snapshotCreateParallelUploads) cmd.Flag("start-time", "Override snapshot start timestamp.").StringVar(&c.snapshotCreateStartTime) cmd.Flag("end-time", "Override snapshot end timestamp.").StringVar(&c.snapshotCreateEndTime) cmd.Flag("force-enable-actions", "Enable snapshot actions even if globally disabled on this client").Hidden().BoolVar(&c.snapshotCreateForceEnableActions) @@ -264,7 +264,7 @@ func (c *commandSnapshotCreate) setupUploader(rep repo.RepositoryWriter) *upload c.svc.onTerminate(u.Cancel) u.ForceHashPercentage = c.snapshotCreateForceHash - u.ParallelUploads = c.snapshotCreateParallelUploads + u.ParallelUploads = parallelismAsInt(c.snapshotCreateParallelUploads) u.FailFast = c.snapshotCreateFailFast u.Progress = c.svc.getProgress() diff --git a/cli/command_snapshot_fix.go b/cli/command_snapshot_fix.go index f276d9dda..86c18d965 100644 --- a/cli/command_snapshot_fix.go +++ b/cli/command_snapshot_fix.go @@ -30,7 +30,7 @@ type commonRewriteSnapshots struct { manifestIDs []string sources []string commit bool - parallel int + parallel uint invalidDirHandling string } @@ -47,7 +47,7 @@ func (c *commonRewriteSnapshots) setup(svc appServices, cmd *kingpin.CmdClause) cmd.Flag("manifest-id", "Manifest IDs").StringsVar(&c.manifestIDs) cmd.Flag("source", "Source to target (username@hostname:/path)").StringsVar(&c.sources) cmd.Flag("commit", "Update snapshot manifests").BoolVar(&c.commit) - cmd.Flag("parallel", "Parallelism").IntVar(&c.parallel) + cmd.Flag("parallel", "Parallelism").UintVar(&c.parallel) cmd.Flag("invalid-directory-handling", "Handling of invalid directories").Default(invalidEntryStub).EnumVar(&c.invalidDirHandling, invalidEntryFail, invalidEntryStub, invalidEntryKeep) } @@ -66,7 +66,7 @@ func failedEntryCallback(rep repo.RepositoryWriter, enumVal string) snapshotfs.R func (c *commonRewriteSnapshots) rewriteMatchingSnapshots(ctx context.Context, rep repo.RepositoryWriter, rewrite snapshotfs.RewriteDirEntryCallback) error { rw, err := snapshotfs.NewDirRewriter(ctx, rep, snapshotfs.DirRewriterOptions{ - Parallel: c.parallel, + Parallel: parallelismAsInt(c.parallel), RewriteEntry: rewrite, OnDirectoryReadFailure: failedEntryCallback(rep, c.invalidDirHandling), }) diff --git a/cli/command_snapshot_migrate.go b/cli/command_snapshot_migrate.go index 8b310aa5c..4049c7b58 100644 --- a/cli/command_snapshot_migrate.go +++ b/cli/command_snapshot_migrate.go @@ -22,7 +22,7 @@ type commandSnapshotMigrate struct { migratePolicies bool migrateOverwritePolicies bool migrateLatestOnly bool - migrateParallel int + migrateParallel uint applyIgnoreRules bool svc advancedAppServices @@ -37,7 +37,7 @@ func (c *commandSnapshotMigrate) setup(svc advancedAppServices, parent commandPa cmd.Flag("policies", "Migrate policies too").Default(trueStr).BoolVar(&c.migratePolicies) cmd.Flag("overwrite-policies", "Overwrite policies").BoolVar(&c.migrateOverwritePolicies) cmd.Flag("latest-only", "Only migrate the latest snapshot").BoolVar(&c.migrateLatestOnly) - cmd.Flag("parallel", "Number of sources to migrate in parallel").Default("1").IntVar(&c.migrateParallel) + cmd.Flag("parallel", "Number of sources to migrate in parallel").Default("1").UintVar(&c.migrateParallel) cmd.Flag("apply-ignore-rules", "When migrating also apply current ignore rules").BoolVar(&c.applyIgnoreRules) cmd.Action(svc.repositoryWriterActionWithMaintenance(c.run)) diff --git a/cli/command_snapshot_verify.go b/cli/command_snapshot_verify.go index b8b3f3820..2122c52f5 100644 --- a/cli/command_snapshot_verify.go +++ b/cli/command_snapshot_verify.go @@ -22,7 +22,7 @@ type commandSnapshotVerify struct { verifyCommandSnapshotIDs []string verifyCommandAllSources bool verifyCommandSources []string - verifyCommandParallel int + verifyCommandParallel uint verifyCommandFilesPercent float64 fileQueueLength int @@ -42,7 +42,7 @@ func (c *commandSnapshotVerify) setup(svc appServices, parent commandParent) { cmd.Flag("file-id", "File object IDs to verify").StringsVar(&c.verifyCommandFileObjectIDs) cmd.Flag("all-sources", "Verify all snapshots (DEPRECATED)").Hidden().BoolVar(&c.verifyCommandAllSources) cmd.Flag("sources", "Verify the provided sources").StringsVar(&c.verifyCommandSources) - cmd.Flag("parallel", "Parallelization").Default("8").IntVar(&c.verifyCommandParallel) + cmd.Flag("parallel", "Parallelization").Default("8").UintVar(&c.verifyCommandParallel) cmd.Flag("file-queue-length", "Queue length for file verification").Default("20000").IntVar(&c.fileQueueLength) cmd.Flag("file-parallelism", "Parallelism for file verification").IntVar(&c.fileParallelism) cmd.Flag("verify-files-percent", "Randomly verify a percentage of files by downloading them [0.0 .. 100.0]").Default("0").Float64Var(&c.verifyCommandFilesPercent) diff --git a/cli/flag_parallel.go b/cli/flag_parallel.go new file mode 100644 index 000000000..177e07dc1 --- /dev/null +++ b/cli/flag_parallel.go @@ -0,0 +1,13 @@ +package cli + +import "math" + +// parallelismAsInt converts --parallel-style flag values from +// uint to int and clamps returned values at MaxInt32. +func parallelismAsInt(v uint) int { + if v > math.MaxInt32 { + return math.MaxInt32 + } + + return int(v) +} diff --git a/cli/flag_parallel_internal_test.go b/cli/flag_parallel_internal_test.go new file mode 100644 index 000000000..49e9a0e16 --- /dev/null +++ b/cli/flag_parallel_internal_test.go @@ -0,0 +1,31 @@ +package cli + +import ( + "math" + "testing" +) + +// TestParallelismAsInt verifies the uint->int conversion used by the +// --parallel flags to prevent large values from wrapping to a negative int. +func TestParallelismAsInt(t *testing.T) { + cases := []struct { + name string + input uint + want int + }{ + {name: "zero", input: 0, want: 0}, + {name: "one", input: 1, want: 1}, + {name: "typical", input: 16, want: 16}, + {name: "maxint32", input: math.MaxInt32, want: math.MaxInt32}, + {name: "above maxint32 clamps", input: math.MaxInt32 + 1, want: math.MaxInt32}, + {name: "maxuint does not wrap negative", input: math.MaxUint, want: math.MaxInt32}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := parallelismAsInt(tc.input); got != tc.want { + t.Fatalf("parallelismAsInt(%d) = %d, want %d", tc.input, got, tc.want) + } + }) + } +} diff --git a/cli/flag_parallel_test.go b/cli/flag_parallel_test.go new file mode 100644 index 000000000..f36abc3e1 --- /dev/null +++ b/cli/flag_parallel_test.go @@ -0,0 +1,33 @@ +package cli_test + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/kopia/kopia/tests/testenv" +) + +// TestNegativeParallelRejected verifies that a negative --parallel value is +// rejected at flag-parse time and returns an error identifying the bad value +// rather than panicking once the command runs. +func TestNegativeParallelRejected(t *testing.T) { + t.Parallel() + + env := testenv.NewCLITest(t, testenv.RepoFormatNotImportant, testenv.NewInProcRunner(t)) + + for _, args := range [][]string{ + {"snapshot", "migrate", "--parallel=-1", "--all"}, + {"content", "verify", "--parallel=-1"}, + } { + _, _, err := env.Run(t, true, args...) + // check rejection message so the test fails if + // the flag ever stops validating, otherwise a + // bare "command failed" check would pass + // since these commands fail regardless given that + // no repository is connected. + require.ErrorContains(t, err, "invalid syntax", + "'kopia %v' should fail parsing the negative value", strings.Join(args, " ")) + } +}