func do(appEnvObj interface{}) error {
	appEnv := appEnvObj.(*appEnv)
	logrus.Register()
	address := appEnv.PachydermPfsd1Port
	if address == "" {
		address = appEnv.PfsAddress
	} else {
		address = strings.Replace(address, "tcp://", "", -1)
	}
	return dockervolume.Serve(
		dockervolume.NewVolumeDriverHandler(
			newVolumeDriver(
				func() (fuse.Mounter, error) {
					clientConn, err := grpc.Dial(address, grpc.WithInsecure())
					if err != nil {
						return nil, err
					}
					return fuse.NewMounter(
						pfs.NewApiClient(
							clientConn,
						),
					), nil
				},
				appEnv.BaseMountpoint,
			),
			dockervolume.VolumeDriverHandlerOptions{},
		),
		dockervolume.ProtocolUnix,
		volumeDriverName,
		volumeDriverGroup,
	)
}

func testMountBig(t *testing.T, apiClient pfs.APIClient, cluster Cluster) {
	repoName := "testMountBigRepo"

	err := pfsutil.CreateRepo(apiClient, repoName)
	require.NoError(t, err)

	directory := "/compile/testMount"
	mounter := fuse.NewMounter("localhost", apiClient)
	ready := make(chan bool)
	go func() {
		err = mounter.Mount(directory, 0, 1, ready)
		require.NoError(t, err)
	}()
	<-ready

	_, err = os.Stat(filepath.Join(directory, repoName))
	require.NoError(t, err)

	_, err = os.Stat(filepath.Join(directory, repoName, "scratch"))
	require.NoError(t, err)

	commit, err := pfsutil.StartCommit(apiClient, repoName, "scratch")
	require.NoError(t, err)
	require.NotNil(t, commit)
	newCommitID := commit.Id

	bigValue := make([]byte, 1024*1024*300)
	for i := 0; i < 1024*1024*300; i++ {
		bigValue[i] = 'a'
	}

	wg := sync.WaitGroup{}
	for j := 0; j < 5; j++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			err := ioutil.WriteFile(filepath.Join(directory, repoName, newCommitID, fmt.Sprintf("big%d", j)), bigValue, 0666)
			require.NoError(t, err)
		}(j)
	}
	wg.Wait()

	err = pfsutil.FinishCommit(apiClient, repoName, newCommitID)
	require.NoError(t, err)

	wg = sync.WaitGroup{}
	for j := 0; j < 5; j++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			data, err := ioutil.ReadFile(filepath.Join(directory, repoName, newCommitID, fmt.Sprintf("big%d", j)))
			require.NoError(t, err)
			require.Equal(t, bigValue, data)
		}(j)
	}
	wg.Wait()

	err = mounter.Unmount(directory)
	require.NoError(t, err)
}

func TestMountBig(t *testing.T) {
	t.Skip()
	t.Parallel()
	apiClient := getPfsClient(t)
	repoName := uniqueString("testMountBigRepo")

	err := pfsutil.CreateRepo(apiClient, repoName)
	require.NoError(t, err)

	directory := "/compile/testMount"
	mounter := fuse.NewMounter("localhost", apiClient)
	ready := make(chan bool)
	go func() {
		err = mounter.Mount(directory, &pfs.Shard{Number: 0, Modulus: 1}, nil, ready)
		require.NoError(t, err)
	}()
	<-ready

	_, err = os.Stat(filepath.Join(directory, repoName))
	require.NoError(t, err)

	commit, err := pfsutil.StartCommit(apiClient, repoName, "")
	require.NoError(t, err)
	require.NotNil(t, commit)
	newCommitID := commit.Id

	bigValue := make([]byte, 1024*1024*300)
	for i := 0; i < 1024*1024*300; i++ {
		bigValue[i] = 'a'
	}

	wg := sync.WaitGroup{}
	for j := 0; j < 5; j++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			err := ioutil.WriteFile(filepath.Join(directory, repoName, newCommitID, fmt.Sprintf("big%d", j)), bigValue, 0666)
			require.NoError(t, err)
		}(j)
	}
	wg.Wait()

	err = pfsutil.FinishCommit(apiClient, repoName, newCommitID)
	require.NoError(t, err)

	wg = sync.WaitGroup{}
	for j := 0; j < 5; j++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			data, err := ioutil.ReadFile(filepath.Join(directory, repoName, newCommitID, fmt.Sprintf("big%d", j)))
			require.NoError(t, err)
			require.Equal(t, bigValue, data)
		}(j)
	}
	wg.Wait()

	err = mounter.Unmount(directory)
	require.NoError(t, err)
}

func testMountBig(t *testing.T, apiClient pfs.ApiClient, internalAPIClient pfs.InternalApiClient, cluster Cluster) {
	repositoryName := TestRepositoryName()

	err := pfsutil.InitRepository(apiClient, repositoryName)
	require.NoError(t, err)

	directory := "/compile/testMount"
	mounter := fuse.NewMounter(apiClient)
	err = mounter.Mount(repositoryName, "", directory, 0, 1)
	require.NoError(t, err)

	_, err = os.Stat(filepath.Join(directory, "scratch"))
	require.NoError(t, err)

	commit, err := pfsutil.Branch(apiClient, repositoryName, "scratch")
	require.NoError(t, err)
	require.NotNil(t, commit)
	newCommitID := commit.Id

	bigValue := make([]byte, 1024*1024*300)
	for i := 0; i < 1024*1024*300; i++ {
		bigValue[i] = 'a'
	}

	wg := sync.WaitGroup{}
	for j := 0; j < 5; j++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			err := ioutil.WriteFile(filepath.Join(directory, newCommitID, fmt.Sprintf("big%d", j)), bigValue, 0666)
			require.NoError(t, err)
		}(j)
	}
	wg.Wait()

	err = pfsutil.Write(apiClient, repositoryName, newCommitID)
	require.NoError(t, err)

	wg = sync.WaitGroup{}
	for j := 0; j < 5; j++ {
		wg.Add(1)
		go func(j int) {
			defer wg.Done()
			data, err := ioutil.ReadFile(filepath.Join(directory, newCommitID, fmt.Sprintf("big%d", j)))
			require.NoError(t, err)
			require.Equal(t, bigValue, data)
		}(j)
	}
	wg.Wait()

	err = mounter.Unmount(directory)
	require.NoError(t, err)
	err = mounter.Wait(directory)
	require.NoError(t, err)
}

func BenchmarkFuse(b *testing.B) {
	apiClient := getPfsClient(b)
	repoName := uniqueString("benchMountRepo")

	if err := pfsutil.CreateRepo(apiClient, repoName); err != nil {
		b.Error(err)
	}

	directory := "/compile/benchMount"
	mounter := fuse.NewMounter("localhost", apiClient)
	ready := make(chan bool)
	go func() {
		err := mounter.Mount(directory, &pfs.Shard{Number: 0, Modulus: 1}, nil, ready)
		require.NoError(b, err)
	}()
	<-ready

	defer func() {
		if err := mounter.Unmount(directory); err != nil {
			b.Error(err)
		}
	}()

	bigValue := make([]byte, 1024*1024)
	for i := 0; i < 1024*1024; i++ {
		bigValue[i] = 'a'
	}

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		commit, err := pfsutil.StartCommit(apiClient, repoName, "")
		if err != nil {
			b.Error(err)
		}
		if commit == nil {
			b.Error("nil branch")
		}
		newCommitID := commit.Id
		var wg sync.WaitGroup
		for j := 0; j < 1024; j++ {
			wg.Add(1)
			go func(j int) {
				defer wg.Done()
				if err = ioutil.WriteFile(filepath.Join(directory, repoName, newCommitID, fmt.Sprintf("big%d", j)), bigValue, 0666); err != nil {
					b.Error(err)
				}
			}(j)
		}
		wg.Wait()
		if err := pfsutil.FinishCommit(apiClient, repoName, newCommitID); err != nil {
			b.Error(err)
		}
	}
}

func benchMount(b *testing.B, apiClient pfs.ApiClient) {
	repositoryName := "benchMountRepo"

	if err := pfsutil.CreateRepo(apiClient, repositoryName); err != nil {
		b.Error(err)
	}

	directory := "/compile/benchMount"
	mounter := fuse.NewMounter(apiClient)
	if err := mounter.Mount(repositoryName, "", directory, 0, 1); err != nil {
		b.Error(err)
	}

	defer func() {
		if err := mounter.Unmount(directory); err != nil {
			b.Error(err)
		}
		if err := mounter.Wait(directory); err != nil {
			b.Error(err)
		}
	}()

	bigValue := make([]byte, 1024*1024)
	for i := 0; i < 1024*1024; i++ {
		bigValue[i] = 'a'
	}

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		commit, err := pfsutil.StartCommit(apiClient, repositoryName, "scratch")
		if err != nil {
			b.Error(err)
		}
		if commit == nil {
			b.Error("nil branch")
		}
		newCommitID := commit.Id
		var wg sync.WaitGroup
		for j := 0; j < 1024; j++ {
			wg.Add(1)
			go func(j int) {
				defer wg.Done()
				if err = ioutil.WriteFile(filepath.Join(directory, newCommitID, fmt.Sprintf("big%d", j)), bigValue, 0666); err != nil {
					b.Error(err)
				}
			}(j)
		}
		wg.Wait()
		if err := pfsutil.FinishCommit(apiClient, repositoryName, newCommitID); err != nil {
			b.Error(err)
		}
	}
}

//.........这里部分代码省略.........
			}
			return writer.Flush()
		},
	}.ToCobraCommand()
	listChange.Flags().IntVarP(&shard, "shard", "s", 0, "shard to read from")
	listChange.Flags().IntVarP(&modulus, "modulus", "m", 1, "modulus of the shards")

	inspectServer := cobramainutil.Command{
		Use:     "inspect-server server-id",
		Short:   "Inspect a server.",
		Long:    "Inspect a server.",
		NumArgs: 1,
		Run: func(cmd *cobra.Command, args []string) error {
			serverInfo, err := pfsutil.InspectServer(apiClient, args[0])
			if err != nil {
				return err
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintServerInfoHeader(writer)
			pretty.PrintServerInfo(writer, serverInfo)
			return writer.Flush()
		},
	}.ToCobraCommand()

	listServer := cobramainutil.Command{
		Use:     "list-server",
		Short:   "Return all servers in the cluster.",
		Long:    "Return all servers in the cluster.",
		NumArgs: 0,
		Run: func(cmd *cobra.Command, args []string) error {
			serverInfos, err := pfsutil.ListServer(apiClient)
			if err != nil {
				return err
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintServerInfoHeader(writer)
			for _, serverInfo := range serverInfos {
				pretty.PrintServerInfo(writer, serverInfo)
			}
			return writer.Flush()
		},
	}.ToCobraCommand()

	mount := cobramainutil.Command{
		Use:        "mount mountpoint repo-name [commit-id]",
		Short:      "Mount a repo as a local file system.",
		Long:       "Mount a repo as a local file system.",
		MinNumArgs: 2,
		MaxNumArgs: 3,
		Run: func(cmd *cobra.Command, args []string) error {
			mountPoint := args[0]
			repo := args[1]
			commitID := ""
			if len(args) == 3 {
				commitID = args[2]
			}
			mounter := fuse.NewMounter(apiClient)
			if err := mounter.Mount(repo, commitID, mountPoint, uint64(shard), uint64(modulus)); err != nil {
				return err
			}
			return mounter.Wait(mountPoint)
		},
	}.ToCobraCommand()
	mount.Flags().IntVarP(&shard, "shard", "s", 0, "shard to read from")
	mount.Flags().IntVarP(&modulus, "modulus", "m", 1, "modulus of the shards")

	rootCmd := &cobra.Command{
		Use: "pfs",
		Long: `Access the PFS API.

Note that this CLI is experimental and does not even check for common errors.
The environment variable PFS_ADDRESS controls what server the CLI connects to, the default is 0.0.0.0:650.`,
	}

	rootCmd.AddCommand(protoclient.NewVersionCommand(clientConn, pachyderm.Version, nil))
	rootCmd.AddCommand(createRepo)
	rootCmd.AddCommand(inspectRepo)
	rootCmd.AddCommand(listRepo)
	rootCmd.AddCommand(deleteRepo)
	rootCmd.AddCommand(startCommit)
	rootCmd.AddCommand(finishCommit)
	rootCmd.AddCommand(inspectCommit)
	rootCmd.AddCommand(listCommit)
	rootCmd.AddCommand(deleteCommit)
	rootCmd.AddCommand(mkdir)
	rootCmd.AddCommand(putBlock)
	rootCmd.AddCommand(getBlock)
	rootCmd.AddCommand(inspectBlock)
	rootCmd.AddCommand(listBlock)
	rootCmd.AddCommand(putFile)
	rootCmd.AddCommand(getFile)
	rootCmd.AddCommand(inspectFile)
	rootCmd.AddCommand(listFile)
	rootCmd.AddCommand(deleteFile)
	rootCmd.AddCommand(listChange)
	rootCmd.AddCommand(inspectServer)
	rootCmd.AddCommand(listServer)
	rootCmd.AddCommand(mount)
	return rootCmd.Execute()
}

func TestWriteAndRead(t *testing.T) {
	t.Parallel()

	// don't leave goroutines running
	var wg sync.WaitGroup
	defer wg.Wait()

	tmp, err := ioutil.TempDir("", "pachyderm-test-")
	if err != nil {
		t.Fatalf("tempdir: %v", err)
	}
	defer func() {
		if err := os.RemoveAll(tmp); err != nil {
			t.Errorf("cannot remove tempdir: %v", err)
		}
	}()

	// closed on successful termination
	quit := make(chan struct{})
	defer close(quit)
	listener, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("cannot listen: %v", err)
	}
	defer func() {
		_ = listener.Close()
	}()

	// TODO try to share more of this setup code with various main
	// functions
	localAddress := listener.Addr().String()
	srv := grpc.NewServer()
	const (
		numShards = 1
	)
	sharder := shard.NewLocalSharder(localAddress, numShards)
	hasher := pfs.NewHasher(numShards, 1)
	router := shard.NewRouter(
		sharder,
		grpcutil.NewDialer(
			grpc.WithInsecure(),
		),
		localAddress,
	)

	blockDir := filepath.Join(tmp, "blocks")
	blockServer, err := server.NewLocalBlockAPIServer(blockDir)
	if err != nil {
		t.Fatalf("NewLocalBlockAPIServer: %v", err)
	}
	pfs.RegisterBlockAPIServer(srv, blockServer)

	driver, err := drive.NewDriver(localAddress)
	if err != nil {
		t.Fatalf("NewDriver: %v", err)
	}

	apiServer := server.NewAPIServer(
		hasher,
		router,
	)
	pfs.RegisterAPIServer(srv, apiServer)

	internalAPIServer := server.NewInternalAPIServer(
		hasher,
		router,
		driver,
	)
	pfs.RegisterInternalAPIServer(srv, internalAPIServer)

	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := srv.Serve(listener); err != nil {
			select {
			case <-quit:
				// orderly shutdown
				return
			default:
				t.Errorf("grpc serve: %v", err)
			}
		}
	}()

	clientConn, err := grpc.Dial(localAddress, grpc.WithInsecure())
	if err != nil {
		t.Fatalf("grpc dial: %v", err)
	}
	apiClient := pfs.NewAPIClient(clientConn)
	mounter := fuse.NewMounter(localAddress, apiClient)

	mountpoint := filepath.Join(tmp, "mnt")
	if err := os.Mkdir(mountpoint, 0700); err != nil {
		t.Fatalf("mkdir mountpoint: %v", err)
	}

	ready := make(chan bool)
	wg.Add(1)
	go func() {
		defer wg.Done()
//.........这里部分代码省略.........

func TestCommitFinishedReadDir(t *testing.T) {
	t.Parallel()

	// don't leave goroutines running
	var wg sync.WaitGroup
	defer wg.Wait()

	tmp, err := ioutil.TempDir("", "pachyderm-test-")
	if err != nil {
		t.Fatalf("tempdir: %v", err)
	}
	defer func() {
		if err := os.RemoveAll(tmp); err != nil {
			t.Errorf("cannot remove tempdir: %v", err)
		}
	}()

	// closed on successful termination
	quit := make(chan struct{})
	defer close(quit)
	listener, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("cannot listen: %v", err)
	}
	defer func() {
		_ = listener.Close()
	}()

	// TODO try to share more of this setup code with various main
	// functions
	localAddress := listener.Addr().String()
	srv := grpc.NewServer()
	const (
		numShards = 1
	)
	sharder := shard.NewLocalSharder(localAddress, numShards)
	hasher := pfs.NewHasher(numShards, 1)
	router := shard.NewRouter(
		sharder,
		grpcutil.NewDialer(
			grpc.WithInsecure(),
		),
		localAddress,
	)

	blockDir := filepath.Join(tmp, "blocks")
	blockServer, err := server.NewLocalBlockAPIServer(blockDir)
	if err != nil {
		t.Fatalf("NewLocalBlockAPIServer: %v", err)
	}
	pfs.RegisterBlockAPIServer(srv, blockServer)

	driver, err := drive.NewDriver(localAddress)
	if err != nil {
		t.Fatalf("NewDriver: %v", err)
	}

	apiServer := server.NewAPIServer(
		hasher,
		router,
	)
	pfs.RegisterAPIServer(srv, apiServer)

	internalAPIServer := server.NewInternalAPIServer(
		hasher,
		router,
		driver,
	)
	pfs.RegisterInternalAPIServer(srv, internalAPIServer)

	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := srv.Serve(listener); err != nil {
			select {
			case <-quit:
				// orderly shutdown
				return
			default:
				t.Errorf("grpc serve: %v", err)
			}
		}
	}()

	clientConn, err := grpc.Dial(localAddress, grpc.WithInsecure())
	if err != nil {
		t.Fatalf("grpc dial: %v", err)
	}
	apiClient := pfs.NewAPIClient(clientConn)
	mounter := fuse.NewMounter(localAddress, apiClient)

	mountpoint := filepath.Join(tmp, "mnt")
	if err := os.Mkdir(mountpoint, 0700); err != nil {
		t.Fatalf("mkdir mountpoint: %v", err)
	}

	ready := make(chan bool)
	wg.Add(1)
	go func() {
		defer wg.Done()
//.........这里部分代码省略.........

func TestRootReadDir(t *testing.T) {
	t.Parallel()

	// don't leave goroutines running
	var wg sync.WaitGroup
	defer wg.Wait()

	tmp, err := ioutil.TempDir("", "pachyderm-test-")
	if err != nil {
		t.Fatalf("tempdir: %v", err)
	}
	defer func() {
		if err := os.RemoveAll(tmp); err != nil {
			t.Errorf("cannot remove tempdir: %v", err)
		}
	}()

	// closed on successful termination
	quit := make(chan struct{})
	defer close(quit)
	listener, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("cannot listen: %v", err)
	}
	defer func() {
		_ = listener.Close()
	}()

	// TODO try to share more of this setup code with various main
	// functions
	localAddress := listener.Addr().String()
	srv := grpc.NewServer()
	const (
		numShards = 1
	)
	sharder := shard.NewLocalSharder(localAddress, numShards)
	hasher := pfs.NewHasher(numShards, 1)
	router := shard.NewRouter(
		sharder,
		grpcutil.NewDialer(
			grpc.WithInsecure(),
		),
		localAddress,
	)

	blockDir := filepath.Join(tmp, "blocks")
	blockServer, err := server.NewLocalBlockAPIServer(blockDir)
	if err != nil {
		t.Fatalf("NewLocalBlockAPIServer: %v", err)
	}
	pfs.RegisterBlockAPIServer(srv, blockServer)

	driver, err := drive.NewDriver(localAddress)
	if err != nil {
		t.Fatalf("NewDriver: %v", err)
	}

	apiServer := server.NewAPIServer(
		hasher,
		router,
	)
	pfs.RegisterAPIServer(srv, apiServer)

	internalAPIServer := server.NewInternalAPIServer(
		hasher,
		router,
		driver,
	)
	pfs.RegisterInternalAPIServer(srv, internalAPIServer)

	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := srv.Serve(listener); err != nil {
			select {
			case <-quit:
				// orderly shutdown
				return
			default:
				t.Errorf("grpc serve: %v", err)
			}
		}
	}()

	clientConn, err := grpc.Dial(localAddress, grpc.WithInsecure())
	if err != nil {
		t.Fatalf("grpc dial: %v", err)
	}
	apiClient := pfs.NewAPIClient(clientConn)
	mounter := fuse.NewMounter(localAddress, apiClient)

	mountpoint := filepath.Join(tmp, "mnt")
	if err := os.Mkdir(mountpoint, 0700); err != nil {
		t.Fatalf("mkdir mountpoint: %v", err)
	}

	ready := make(chan bool)
	wg.Add(1)
	go func() {
		defer wg.Done()
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		Long:  "Return info about a file.",
		Run: pkgcobra.RunFixedArgs(3, func(args []string) error {
			apiClient, err := getAPIClient(address)
			if err != nil {
				return err
			}
			fileInfo, err := pfsutil.InspectFile(apiClient, args[0], args[1], args[2], "", shard())
			if err != nil {
				return err
			}
			if fileInfo == nil {
				return fmt.Errorf("file %s not found", args[2])
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintFileInfoHeader(writer)
			pretty.PrintFileInfo(writer, fileInfo)
			return writer.Flush()
		}),
	}
	addShardFlags(inspectFile)

	listFile := &cobra.Command{
		Use:   "list-file repo-name commit-id path/to/dir",
		Short: "Return the files in a directory.",
		Long:  "Return the files in a directory.",
		Run: pkgcobra.RunBoundedArgs(pkgcobra.Bounds{Min: 2, Max: 3}, func(args []string) error {
			apiClient, err := getAPIClient(address)
			if err != nil {
				return err
			}
			var path string
			if len(args) == 3 {
				path = args[2]
			}
			fileInfos, err := pfsutil.ListFile(apiClient, args[0], args[1], path, "", shard())
			if err != nil {
				return err
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintFileInfoHeader(writer)
			for _, fileInfo := range fileInfos {
				pretty.PrintFileInfo(writer, fileInfo)
			}
			return writer.Flush()
		}),
	}
	addShardFlags(listFile)

	deleteFile := &cobra.Command{
		Use:   "delete-file repo-name commit-id path/to/file",
		Short: "Delete a file.",
		Long:  "Delete a file.",
		Run: pkgcobra.RunFixedArgs(2, func(args []string) error {
			apiClient, err := getAPIClient(address)
			if err != nil {
				return err
			}
			return pfsutil.DeleteFile(apiClient, args[0], args[1], args[2])
		}),
	}

	var mountPoint string
	mount := &cobra.Command{
		Use:   "mount [repo/commit:alias...]",
		Short: "Mount pfs locally.",
		Long:  "Mount pfs locally.",
		Run: pkgcobra.Run(func(args []string) error {
			//lion.SetLevel(lion.LevelDebug)
			apiClient, err := getAPIClient(address)
			if err != nil {
				return err
			}
			mounter := fuse.NewMounter(address, apiClient)
			return mounter.Mount(mountPoint, shard(), parseCommitMounts(args), nil)
		}),
	}
	mount.Flags().StringVarP(&mountPoint, "mount-point", "p", "/pfs", "root of mounted filesystem")
	addShardFlags(mount)

	var result []*cobra.Command
	result = append(result, repo)
	result = append(result, createRepo)
	result = append(result, inspectRepo)
	result = append(result, listRepo)
	result = append(result, deleteRepo)
	result = append(result, commit)
	result = append(result, startCommit)
	result = append(result, finishCommit)
	result = append(result, inspectCommit)
	result = append(result, listCommit)
	result = append(result, deleteCommit)
	result = append(result, file)
	result = append(result, putFile)
	result = append(result, getFile)
	result = append(result, inspectFile)
	result = append(result, listFile)
	result = append(result, deleteFile)
	result = append(result, mount)
	return result
}

func do(appEnvObj interface{}) error {
	lion.SetLevel(lion.LevelDebug)
	appEnv := appEnvObj.(*appEnv)
	rootCmd := &cobra.Command{
		Use:   os.Args[0] + " job-id",
		Short: `Pachyderm job-shim, coordinates with ppsd to create an output commit and run user work.`,
		Long:  `Pachyderm job-shim, coordinates with ppsd to create an output commit and run user work.`,
		Run: func(cmd *cobra.Command, args []string) {
			client, err := getPachClient(appEnv)
			if err != nil {
				errorAndExit(err.Error())
			}
			response, err := client.StartJob(
				context.Background(),
				&pps.StartJobRequest{
					Job: &pps.Job{
						Id: args[0],
					}})
			if err != nil {
				fmt.Fprintf(os.Stderr, "%s\n", err.Error())
				os.Exit(0)
			}

			mounter := fuse.NewMounter(appEnv.PachydermAddress, client)
			ready := make(chan bool)
			go func() {
				if err := mounter.Mount(
					"/pfs",
					nil,
					response.CommitMounts,
					ready,
				); err != nil {
					errorAndExit(err.Error())
				}
			}()
			<-ready
			defer func() {
				if err := mounter.Unmount("/pfs"); err != nil {
					errorAndExit(err.Error())
				}
			}()
			var readers []io.Reader
			for _, line := range response.Transform.Stdin {
				readers = append(readers, strings.NewReader(line+"\n"))
			}
			io := pkgexec.IO{
				Stdin:  io.MultiReader(readers...),
				Stdout: os.Stdout,
				Stderr: os.Stderr,
			}
			success := true
			if err := pkgexec.RunIO(io, response.Transform.Cmd...); err != nil {
				fmt.Fprintf(os.Stderr, "%s\n", err.Error())
				success = false
			}
			if _, err := client.FinishJob(
				context.Background(),
				&pps.FinishJobRequest{
					Job: &pps.Job{
						Id: args[0],
					},
					Index:   response.Index,
					Success: success,
				},
			); err != nil {
				errorAndExit(err.Error())
			}
		},
	}

	return rootCmd.Execute()
}

//.........这里部分代码省略.........
		NumArgs: 3,
		Run: func(cmd *cobra.Command, args []string) error {
			listFilesResponse, err := pfsutil.ListFiles(apiClient, args[0], args[1], args[2], uint64(shard), uint64(modulus))
			if err != nil {
				return err
			}
			for _, fileInfo := range listFilesResponse.FileInfo {
				fmt.Printf("%+v\n", fileInfo)
			}
			return nil
		},
	}.ToCobraCommand()
	lsCmd.Flags().IntVarP(&shard, "shard", "s", 0, "shard to read from")
	lsCmd.Flags().IntVarP(&modulus, "modulus", "m", 1, "modulus of the shards")

	branchCmd := cobramainutil.Command{
		Use:     "branch repository-name commit-id",
		Long:    "Branch a commit. commit-id must be a readable commit.",
		NumArgs: 2,
		Run: func(cmd *cobra.Command, args []string) error {
			branchResponse, err := pfsutil.Branch(apiClient, args[0], args[1])
			if err != nil {
				return err
			}
			fmt.Println(branchResponse.Commit.Id)
			return nil
		},
	}.ToCobraCommand()

	commitCmd := cobramainutil.Command{
		Use:     "commit repository-name branch-id",
		Long:    "Commit a branch. branch-id must be a writeable commit.",
		NumArgs: 2,
		Run: func(cmd *cobra.Command, args []string) error {
			return pfsutil.Commit(apiClient, args[0], args[1])
		},
	}.ToCobraCommand()

	commitInfoCmd := cobramainutil.Command{
		Use:     "commit-info repository-name commit-id",
		Long:    "Get info for a commit.",
		NumArgs: 2,
		Run: func(cmd *cobra.Command, args []string) error {
			commitInfoResponse, err := pfsutil.GetCommitInfo(apiClient, args[0], args[1])
			if err != nil {
				return err
			}
			fmt.Printf("%+v\n", commitInfoResponse.CommitInfo)
			return nil
		},
	}.ToCobraCommand()

	listCommitsCmd := cobramainutil.Command{
		Use:     "list-commits repository-name",
		Long:    "List commits on the repository.",
		NumArgs: 1,
		Run: func(cmd *cobra.Command, args []string) error {
			listCommitsResponse, err := pfsutil.ListCommits(apiClient, args[0])
			if err != nil {
				return err
			}
			for _, commitInfo := range listCommitsResponse.CommitInfo {
				fmt.Printf("%+v\n", commitInfo)
			}
			return nil
		},
	}.ToCobraCommand()

	mountCmd := cobramainutil.Command{
		Use:     "mount repository-name",
		Long:    "Mount a repository as a local file system.",
		NumArgs: 1,
		Run: func(cmd *cobra.Command, args []string) error {
			return fuse.NewMounter().Mount(apiClient, args[0], args[0], uint64(shard), uint64(modulus))
		},
	}.ToCobraCommand()
	mountCmd.Flags().IntVarP(&shard, "shard", "s", 0, "shard to read from")
	mountCmd.Flags().IntVarP(&modulus, "modulus", "m", 1, "modulus of the shards")

	rootCmd := &cobra.Command{
		Use: "pfs",
		Long: `Access the PFS API.

Note that this CLI is experimental and does not even check for common errors.
The environment variable PFS_ADDRESS controls what server the CLI connects to, the default is 0.0.0.0:650.`,
	}

	rootCmd.AddCommand(cobramainutil.NewVersionCommand(clientConn, pachyderm.Version))
	rootCmd.AddCommand(initCmd)
	rootCmd.AddCommand(mkdirCmd)
	rootCmd.AddCommand(putCmd)
	rootCmd.AddCommand(getCmd)
	rootCmd.AddCommand(lsCmd)
	rootCmd.AddCommand(branchCmd)
	rootCmd.AddCommand(commitCmd)
	rootCmd.AddCommand(commitInfoCmd)
	rootCmd.AddCommand(listCommitsCmd)
	rootCmd.AddCommand(mountCmd)
	return rootCmd.Execute()
}

func testMount(t *testing.T, apiClient pfs.ApiClient, internalAPIClient pfs.InternalApiClient) {
	repositoryName := TestRepositoryName()

	err := pfsutil.InitRepository(apiClient, repositoryName, false)
	require.NoError(t, err)

	directory := "/compile/testMount"
	mounter := fuse.NewMounter()
	go func() {
		err = mounter.Mount(apiClient, repositoryName, directory, 0, 1)
		require.NoError(t, err)
	}()
	mounter.Ready()

	_, err = os.Stat(filepath.Join(directory, "scratch"))
	require.NoError(t, err)

	branchResponse, err := pfsutil.Branch(apiClient, repositoryName, "scratch")
	require.NoError(t, err)
	require.NotNil(t, branchResponse)
	newCommitID := branchResponse.Commit.Id

	_, err = os.Stat(filepath.Join(directory, newCommitID))
	require.NoError(t, err)

	err = ioutil.WriteFile(filepath.Join(directory, newCommitID, "foo"), []byte("foo"), 0666)
	require.NoError(t, err)

	_, err = pfsutil.PutFile(apiClient, repositoryName, newCommitID, "bar", 0, strings.NewReader("bar"))
	require.NoError(t, err)

	bigValue := make([]byte, 1024*1024)
	for i := 0; i < 1024*1024; i++ {
		bigValue[i] = 'a'
	}

	err = ioutil.WriteFile(filepath.Join(directory, newCommitID, "big1"), bigValue, 0666)
	require.NoError(t, err)

	_, err = pfsutil.PutFile(apiClient, repositoryName, newCommitID, "big2", 0, bytes.NewReader(bigValue))
	require.NoError(t, err)

	err = pfsutil.Commit(apiClient, repositoryName, newCommitID)
	require.NoError(t, err)

	fInfo, err := os.Stat(filepath.Join(directory, newCommitID, "foo"))
	require.NoError(t, err)
	require.Equal(t, int64(3), fInfo.Size())

	data, err := ioutil.ReadFile(filepath.Join(directory, newCommitID, "foo"))
	require.NoError(t, err)
	require.Equal(t, "foo", string(data))

	data, err = ioutil.ReadFile(filepath.Join(directory, newCommitID, "bar"))
	require.NoError(t, err)
	require.Equal(t, "bar", string(data))

	data, err = ioutil.ReadFile(filepath.Join(directory, newCommitID, "big1"))
	require.NoError(t, err)
	require.Equal(t, bigValue, data)

	data, err = ioutil.ReadFile(filepath.Join(directory, newCommitID, "big2"))
	require.NoError(t, err)
	require.Equal(t, bigValue, data)
}

func RunTest(
	t *testing.T,
	f func(*testing.T, pfs.APIClient, pps.JobAPIClient, pps.PipelineAPIClient),
) {
	pfstesting.RunTest(
		t,
		func(t *testing.T, pfsAPIClient pfs.APIClient, _ pfstesting.Cluster) {
			pfsMountDir, err := ioutil.TempDir("", "")
			require.NoError(t, err)
			mounter := fuse.NewMounter("localhost", pfsAPIClient)
			ready := make(chan bool)
			go func() {
				require.NoError(t, mounter.Mount(pfsMountDir, &pfs.Shard{Number: 0, Modulus: 1}, nil, ready))
			}()
			<-ready
			persistservertesting.RunTestWithRethinkAPIServer(
				t,
				func(t *testing.T, persistAPIServer persist.APIServer) {
					persistAPIClient := persist.NewLocalAPIClient(persistAPIServer)
					var pipelineAPIServer pipelineserver.APIServer
					prototest.RunT(
						t,
						testNumServers,
						func(servers map[string]*grpc.Server) {
							jobAPIServer := jobserver.NewAPIServer(
								pfsAPIClient,
								persistAPIClient,
								nil,
							)
							jobAPIClient := pps.NewLocalJobAPIClient(jobAPIServer)
							pipelineAPIServer = pipelineserver.NewAPIServer(
								pfsAPIClient,
								jobAPIClient,
								persistAPIClient,
							)
							for _, server := range servers {
								pps.RegisterJobAPIServer(server, jobAPIServer)
								pps.RegisterPipelineAPIServer(server, pipelineAPIServer)
							}
						},
						func(t *testing.T, clientConns map[string]*grpc.ClientConn) {
							pipelineAPIServer.Start()
							var clientConn *grpc.ClientConn
							for _, c := range clientConns {
								clientConn = c
								break
							}
							f(
								t,
								pfsAPIClient,
								pps.NewJobAPIClient(
									clientConn,
								),
								pps.NewPipelineAPIClient(
									clientConn,
								),
							)
						},
						func() {
							_ = mounter.Unmount(pfsMountDir)
						},
					)
				},
			)
		},
	)
}

//.........这里部分代码省略.........
				return err
			}
			changeInfos, err := pfsutil.ListChange(apiClient, args[0], args[1], args[2], shard())
			if err != nil {
				return err
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintChangeHeader(writer)
			for _, changeInfo := range changeInfos {
				pretty.PrintChange(writer, changeInfo)
			}
			return writer.Flush()
		}),
	}
	listChange.Flags().IntVarP(&number, "shard", "s", 0, "shard to read from")
	listChange.Flags().IntVarP(&modulus, "modulus", "m", 1, "modulus of the shards")

	inspectServer := &cobra.Command{
		Use:   "inspect-server server-id",
		Short: "Inspect a server.",
		Long:  "Inspect a server.",
		Run: pkgcobra.RunFixedArgs(1, func(args []string) error {
			clusterAPIClient, err := getClusterAPIClient(address)
			if err != nil {
				return err
			}
			serverInfo, err := pfsutil.InspectServer(clusterAPIClient, args[0])
			if err != nil {
				return err
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintServerInfoHeader(writer)
			pretty.PrintServerInfo(writer, serverInfo)
			return writer.Flush()
		}),
	}

	listServer := &cobra.Command{
		Use:   "list-server",
		Short: "Return all servers in the cluster.",
		Long:  "Return all servers in the cluster.",
		Run: pkgcobra.RunFixedArgs(0, func(args []string) error {
			clusterAPIClient, err := getClusterAPIClient(address)
			if err != nil {
				return err
			}
			serverInfos, err := pfsutil.ListServer(clusterAPIClient)
			if err != nil {
				return err
			}
			writer := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
			pretty.PrintServerInfoHeader(writer)
			for _, serverInfo := range serverInfos {
				pretty.PrintServerInfo(writer, serverInfo)
			}
			return writer.Flush()
		}),
	}

	var mountPoint string
	mount := &cobra.Command{
		Use:   "mount [repo/commit:alias...]",
		Short: "Mount pfs locally.",
		Long:  "Mount pfs locally.",
		Run: pkgcobra.Run(func(args []string) error {
			protolog.SetLevel(protolog.Level_LEVEL_DEBUG)
			apiClient, err := getAPIClient(address)
			if err != nil {
				return err
			}
			mounter := fuse.NewMounter(address, apiClient)
			return mounter.Mount(mountPoint, shard(), parseCommitMounts(args), nil)
		}),
	}
	mount.Flags().IntVarP(&number, "shard", "s", 0, "shard to read from")
	mount.Flags().IntVarP(&modulus, "modulus", "m", 1, "modulus of the shards")
	mount.Flags().StringVarP(&mountPoint, "mount-point", "p", "/pfs", "root of mounted filesystem")

	var result []*cobra.Command
	result = append(result, createRepo)
	result = append(result, inspectRepo)
	result = append(result, listRepo)
	result = append(result, deleteRepo)
	result = append(result, startCommit)
	result = append(result, finishCommit)
	result = append(result, inspectCommit)
	result = append(result, listCommit)
	result = append(result, deleteCommit)
	result = append(result, mkdir)
	result = append(result, putFile)
	result = append(result, getFile)
	result = append(result, inspectFile)
	result = append(result, listFile)
	result = append(result, deleteFile)
	result = append(result, listChange)
	result = append(result, inspectServer)
	result = append(result, listServer)
	result = append(result, mount)
	return result, nil
}

func do(appEnvObj interface{}) error {
	protolog.SetLevel(protolog.Level_LEVEL_DEBUG)
	appEnv := appEnvObj.(*appEnv)
	rootCmd := &cobra.Command{
		Use:   os.Args[0] + " job-id",
		Short: `Pachyderm job-shim, coordinates with ppsd to create an output commit and run user work.`,
		Long:  `Pachyderm job-shim, coordinates with ppsd to create an output commit and run user work.`,
		Run: func(cmd *cobra.Command, args []string) {
			pfsAPIClient, err := getPfsAPIClient(getPfsdAddress(appEnv))
			if err != nil {
				errorAndExit(err.Error())
			}

			ppsAPIClient, err := getPpsAPIClient(getPpsdAddress(appEnv))
			if err != nil {
				errorAndExit(err.Error())
			}

			response, err := ppsAPIClient.StartJob(
				context.Background(),
				&pps.StartJobRequest{
					Job: &pps.Job{
						Id: args[0],
					}})
			if err != nil {
				fmt.Fprintf(os.Stderr, "%s\n", err.Error())
				os.Exit(0)
			}

			mounter := fuse.NewMounter(getPfsdAddress(appEnv), pfsAPIClient)
			ready := make(chan bool)
			go func() {
				if err := mounter.Mount(
					"/pfs",
					response.CommitMounts,
					ready,
				); err != nil {
					errorAndExit(err.Error())
				}
			}()
			<-ready
			defer func() {
				if err := mounter.Unmount("/pfs"); err != nil {
					errorAndExit(err.Error())
				}
			}()
			io := pkgexec.IO{
				Stdin:  strings.NewReader(response.Transform.Stdin),
				Stdout: os.Stdout,
				Stderr: os.Stderr,
			}
			success := true
			if err := pkgexec.RunIO(io, response.Transform.Cmd...); err != nil {
				fmt.Fprintf(os.Stderr, "%s\n", err.Error())
				success = false
			}
			if _, err := ppsAPIClient.FinishJob(
				context.Background(),
				&pps.FinishJobRequest{
					Job: &pps.Job{
						Id: args[0],
					},
					Index:   response.Index,
					Success: success,
				},
			); err != nil {
				errorAndExit(err.Error())
			}
		},
	}

	return rootCmd.Execute()
}

func testMount(t *testing.T, apiClient pfs.ApiClient, internalAPIClient pfs.InternalApiClient, cluster Cluster) {
	repositoryName := "testMountRepo"

	err := pfsutil.CreateRepo(apiClient, repositoryName)
	require.NoError(t, err)

	directory := "/compile/testMount"
	mounter := fuse.NewMounter(apiClient)
	err = mounter.Mount(repositoryName, directory, "", 0, 1)
	require.NoError(t, err)

	_, err = os.Stat(filepath.Join(directory, "scratch"))
	require.NoError(t, err)

	commit, err := pfsutil.StartCommit(apiClient, repositoryName, "scratch")
	require.NoError(t, err)
	require.NotNil(t, commit)
	newCommitID := commit.Id

	_, err = os.Stat(filepath.Join(directory, newCommitID))
	require.NoError(t, err)

	err = ioutil.WriteFile(filepath.Join(directory, newCommitID, "foo"), []byte("foo"), 0666)
	require.NoError(t, err)

	_, err = pfsutil.PutFile(apiClient, repositoryName, newCommitID, "bar", 0, strings.NewReader("bar"))
	require.NoError(t, err)

	bigValue := make([]byte, 1024*1024)
	for i := 0; i < 1024*1024; i++ {
		bigValue[i] = 'a'
	}

	err = ioutil.WriteFile(filepath.Join(directory, newCommitID, "big1"), bigValue, 0666)
	require.NoError(t, err)

	_, err = pfsutil.PutFile(apiClient, repositoryName, newCommitID, "big2", 0, bytes.NewReader(bigValue))
	require.NoError(t, err)

	err = pfsutil.FinishCommit(apiClient, repositoryName, newCommitID)
	require.NoError(t, err)

	fInfo, err := os.Stat(filepath.Join(directory, newCommitID, "foo"))
	require.NoError(t, err)
	require.Equal(t, int64(3), fInfo.Size())

	data, err := ioutil.ReadFile(filepath.Join(directory, newCommitID, "foo"))
	require.NoError(t, err)
	require.Equal(t, "foo", string(data))

	data, err = ioutil.ReadFile(filepath.Join(directory, newCommitID, "bar"))
	require.NoError(t, err)
	require.Equal(t, "bar", string(data))

	data, err = ioutil.ReadFile(filepath.Join(directory, newCommitID, "big1"))
	require.NoError(t, err)
	require.Equal(t, bigValue, data)

	data, err = ioutil.ReadFile(filepath.Join(directory, newCommitID, "big2"))
	require.NoError(t, err)
	require.Equal(t, bigValue, data)

	err = mounter.Unmount(directory)
	require.NoError(t, err)
	err = mounter.Wait(directory)
	require.NoError(t, err)
}