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
}
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 ( " " , 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
CollectActiveObjectsAndTags collects all objects / tags that are not deleted or eligible for garbage collection