src/server/pps/server/api_server.go
func CollectActiveObjectsAndTags(ctx context.Context, pachClient *client.APIClient, repoInfos []*pfs.RepoInfo, pipelineInfos []*pps.PipelineInfo, memoryAllowance int, storageRoot string) (*ActiveStat, error) {
	if memoryAllowance == 0 {
		memoryAllowance = defaultGCMemory
	}
	result := &ActiveStat{
		// Each bloom filter gets half the memory allowance, times 8 to convert
		// from bytes to bits.
		Objects: bloom.New(uint(memoryAllowance*8/2), 10),
		Tags:    bloom.New(uint(memoryAllowance*8/2), 10),
	}
	var activeObjectsMu sync.Mutex
	// A helper function for adding active objects in a thread-safe way
	addActiveObjects := func(objects ...*pfs.Object) {
		activeObjectsMu.Lock()
		defer activeObjectsMu.Unlock()
		for _, object := range objects {
			if object != nil {
				result.NObjects++
				result.Objects.AddString(object.Hash)
			}
		}
	}
	// A helper function for adding objects that are actually hash trees,
	// which in turn contain active objects.
	addActiveTree := func(object *pfs.Object) error {
		if object == nil {
			return nil
		}
		addActiveObjects(object)
		tree, err := hashtree.GetHashTreeObject(pachClient, storageRoot, object)
		if err != nil {
			return err
		}
		return tree.Walk("/", func(path string, node *hashtree.NodeProto) error {
			if node.FileNode != nil {
				addActiveObjects(node.FileNode.Objects...)
			}
			return nil
		})
	}

	// Get all commit trees
	limiter := limit.New(100)
	var eg errgroup.Group
	for _, repo := range repoInfos {
		repo := repo
		client, err := pachClient.ListCommitStream(ctx, &pfs.ListCommitRequest{
			Repo: repo.Repo,
		})
		if err != nil {
			return nil, err
		}
		for {
			ci, err := client.Recv()
			if err == io.EOF {
				break
			} else if err != nil {
				return nil, grpcutil.ScrubGRPC(err)
			}
			limiter.Acquire()
			eg.Go(func() error {
				defer limiter.Release()
				// (bryce) This needs some notion of active blockrefs since these trees do not use objects
				addActiveObjects(ci.Trees...)
				addActiveObjects(ci.Datums)
				return addActiveTree(ci.Tree)
			})
		}
	}
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	eg = errgroup.Group{}
	for _, pipelineInfo := range pipelineInfos {
		tags, err := pachClient.ObjectAPIClient.ListTags(pachClient.Ctx(), &pfs.ListTagsRequest{
			Prefix:        client.DatumTagPrefix(pipelineInfo.Salt),
			IncludeObject: true,
		})
		if err != nil {
			return nil, fmt.Errorf("error listing tagged objects: %v", err)
		}

		for resp, err := tags.Recv(); err != io.EOF; resp, err = tags.Recv() {
			resp := resp
			if err != nil {
				return nil, err
			}
			result.Tags.AddString(resp.Tag.Name)
			result.NTags++
			limiter.Acquire()
			eg.Go(func() error {
				defer limiter.Release()
				// (bryce) Same as above
				addActiveObjects(resp.Object)
				return nil
			})
		}
	}
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	return result, nil
}
func CollectActiveObjectsAndTags(ctx context.Context, pachClient *client.APIClient, repoInfos []*pfs.RepoInfo, pipelineInfos []*pps.PipelineInfo, memoryAllowance int, storageRoot string) (*ActiveStat, error) {
	if memoryAllowance == 0 {
		memoryAllowance = defaultGCMemory
	}
	result := &ActiveStat{
		// Each bloom filter gets half the memory allowance, times 8 to convert
		// from bytes to bits.
		Objects: bloom.New(uint(memoryAllowance*8/2), 10),
		Tags:    bloom.New(uint(memoryAllowance*8/2), 10),
	}
	var activeObjectsMu sync.Mutex
	// A helper function for adding active objects in a thread-safe way
	addActiveObjects := func(objects ...*pfs.Object) {
		activeObjectsMu.Lock()
		defer activeObjectsMu.Unlock()
		for _, object := range objects {
			if object != nil {
				result.NObjects++
				result.Objects.AddString(object.Hash)
			}
		}
	}
	// A helper function for adding objects that are actually hash trees,
	// which in turn contain active objects.
	addActiveTree := func(object *pfs.Object) error {
		if object == nil {
			return nil
		}
		addActiveObjects(object)
		tree, err := hashtree.GetHashTreeObject(pachClient, storageRoot, object)
		if err != nil {
			return err
		}
		return tree.Walk("/", func(path string, node *hashtree.NodeProto) error {
			if node.FileNode != nil {
				addActiveObjects(node.FileNode.Objects...)
			}
			return nil
		})
	}

	// Get all commit trees
	limiter := limit.New(100)
	var eg errgroup.Group
	for _, repo := range repoInfos {
		repo := repo
		client, err := pachClient.ListCommitStream(ctx, &pfs.ListCommitRequest{
			Repo: repo.Repo,
		})
		if err != nil {
			return nil, err
		}
		for {
			ci, err := client.Recv()
			if err == io.EOF {
				break
			} else if err != nil {
				return nil, grpcutil.ScrubGRPC(err)
			}
			limiter.Acquire()
			eg.Go(func() error {
				defer limiter.Release()
				// (bryce) This needs some notion of active blockrefs since these trees do not use objects
				addActiveObjects(ci.Trees...)
				addActiveObjects(ci.Datums)
				return addActiveTree(ci.Tree)
			})
		}
	}
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	eg = errgroup.Group{}
	for _, pipelineInfo := range pipelineInfos {
		tags, err := pachClient.ObjectAPIClient.ListTags(pachClient.Ctx(), &pfs.ListTagsRequest{
			Prefix:        client.DatumTagPrefix(pipelineInfo.Salt),
			IncludeObject: true,
		})
		if err != nil {
			return nil, fmt.Errorf("error listing tagged objects: %v", err)
		}

		for resp, err := tags.Recv(); err != io.EOF; resp, err = tags.Recv() {
			resp := resp
			if err != nil {
				return nil, err
			}
			result.Tags.AddString(resp.Tag.Name)
			result.NTags++
			limiter.Acquire()
			eg.Go(func() error {
				defer limiter.Release()
				// (bryce) Same as above
				addActiveObjects(resp.Object)
				return nil
			})
		}
	}
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	return result, nil
}
// CollectActiveObjectsAndTags collects all objects/tags that are not deleted
// or eligible for garbage collection