func (export *progressMeter) LogProgress(label string, definitions int, files int) {
	export.mutex.Lock()
	defer export.mutex.Unlock()
	if export.label != label {
		export.Log()
		export.definitions = 0
		export.files = 0
		export.resetAt = time.Now()
	}
	export.label = label
	export.definitions += definitions
	export.files += files
}
LogProgress() accumulates totals until a new label is supplied, when it will write a log message
package codegen
import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"
	"github.com/Azure/k8s-infra/hack/generator/pkg/astmodel"
	"github.com/pkg/errors"
	kerrors "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/klog/v2"
)
func exportPackages(outputPath string) PipelineStage {
	description := fmt.Sprintf("Export packages to %q", outputPath)
	return MakePipelineStage(
		"exportPackages",
		description,
		func(ctx context.Context, types astmodel.Types) (astmodel.Types, error) {
			packages, err := CreatePackagesForDefinitions(types)
			if err != nil {
				return nil, errors.Wrapf(err, "failed to assign generated definitions to packages")
			}
			err = writeFiles(ctx, packages, outputPath)
			if err != nil {
				return nil, errors.Wrapf(err, "unable to write files into %q", outputPath)
			}
			return types, nil
		})
}
func CreatePackagesForDefinitions(definitions astmodel.Types) (map[astmodel.PackageReference]*astmodel.PackageDefinition, error) {
	genVersion := combinedVersion()
	packages := make(map[astmodel.PackageReference]*astmodel.PackageDefinition)
	for _, def := range definitions {
		defName := def.Name()
		pkgRef, ok := defName.PackageReference.AsLocalPackage()
		if !ok {
			klog.Errorf("Definition %v from external package %v skipped", defName.Name(), defName.PackageReference)
			continue
		}
		groupName := pkgRef.Group()
		pkgName := pkgRef.PackageName()
		if pkg, ok := packages[pkgRef]; ok {
			pkg.AddDefinition(def)
		} else {
			pkg = astmodel.NewPackageDefinition(groupName, pkgName, genVersion)
			pkg.AddDefinition(def)
			packages[pkgRef] = pkg
		}
	}
	return packages, nil
}
func writeFiles(ctx context.Context, packages map[astmodel.PackageReference]*astmodel.PackageDefinition, outputPath string) error {
	var pkgs []*astmodel.PackageDefinition
	for _, pkg := range packages {
		pkgs = append(pkgs, pkg)
	}
	sort.Slice(pkgs, func(i int, j int) bool {
		iPkg := pkgs[i]
		jPkg := pkgs[j]
		return iPkg.GroupName < jPkg.GroupName ||
			(iPkg.GroupName == jPkg.GroupName && iPkg.PackageName < jPkg.PackageName)
	})
	klog.V(0).Infof("Writing %d packages into %q", len(pkgs), outputPath)
	globalProgress := newProgressMeter()
	groupProgress := newProgressMeter()
	var wg sync.WaitGroup
	pkgQueue := make(chan *astmodel.PackageDefinition, 100)
	errs := make(chan error, 10)  
	for c := 0; c < 8; c++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for pkg := range pkgQueue {
				if ctx.Err() != nil {  
					return
				}
				outputDir := filepath.Join(outputPath, pkg.GroupName, pkg.PackageName)
				if _, err := os.Stat(outputDir); os.IsNotExist(err) {
					klog.V(5).Infof("Creating directory %q\n", outputDir)
					err = os.MkdirAll(outputDir, 0700)
					if err != nil {
						select {  
						case errs <- errors.Wrapf(err, "unable to create directory %q", outputDir):
						default:
						}
						return
					}
				}
				count, err := pkg.EmitDefinitions(outputDir, packages)
				if err != nil {
					select {  
					case errs <- errors.Wrapf(err, "error writing definitions into %q", outputDir):
					default:
					}
					return
				} else {
					globalProgress.LogProgress("", pkg.DefinitionCount(), count)
					groupProgress.LogProgress(pkg.GroupName, pkg.DefinitionCount(), count)
				}
			}
		}()
	}
	for _, pkg := range pkgs {
		pkgQueue <- pkg
	}
	close(pkgQueue)
	wg.Wait()
	close(errs)
	var totalErrs []error
	for err := range errs {
		totalErrs = append(totalErrs, err)
	}
	err := kerrors.NewAggregate(totalErrs)
	if err != nil {
		return err
	}
	globalProgress.mutex.Lock()
	defer globalProgress.mutex.Unlock()
	globalProgress.Log()
	return nil
}
func newProgressMeter() *progressMeter {
	return &progressMeter{
		resetAt: time.Now(),
	}
}
type progressMeter struct {
	label       string
	definitions int
	files       int
	resetAt     time.Time
	mutex sync.Mutex
}
func (export *progressMeter) Log() {
	started := export.resetAt
	export.resetAt = time.Now()
	if export.definitions == 0 && export.files == 0 {
		return
	}
	elapsed := time.Since(started).Round(time.Millisecond)
	if export.label != "" {
		klog.V(2).Infof("Wrote %d files containing %d definitions for %v in %v", export.files, export.definitions, export.label, elapsed)
	} else {
		klog.V(2).Infof("Wrote %d files containing %d definitions in %v", export.files, export.definitions, time.Since(started))
	}
	export.resetAt = time.Now()
}