<commit_before>package containerizedengine

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"sync"
	"time"

	"github.com/containerd/containerd/content"
	"github.com/containerd/containerd/errdefs"
	"github.com/containerd/containerd/remotes"
	"github.com/docker/docker/pkg/jsonmessage"
	digest "github.com/opencontainers/go-digest"
	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
	"github.com/sirupsen/logrus"
)

func showProgress(ctx context.Context, ongoing *jobs, cs content.Store, out io.WriteCloser) {
	var (
		ticker   = time.NewTicker(100 * time.Millisecond)
		start    = time.Now()
		enc      = json.NewEncoder(out)
		statuses = map[string]statusInfo{}
		done     bool
	)
	defer ticker.Stop()

outer:
	for {
		select {
		case <-ticker.C:

			resolved := "resolved"
			if !ongoing.isResolved() {
				resolved = "resolving"
			}
			statuses[ongoing.name] = statusInfo{
				Ref:    ongoing.name,
				Status: resolved,
			}
			keys := []string{ongoing.name}

			activeSeen := map[string]struct{}{}
			if !done {
				active, err := cs.ListStatuses(ctx, "")
				if err != nil {
					logrus.Debugf("active check failed: %s", err)
					continue
				}
				// update status of active entries!
				for _, active := range active {
					statuses[active.Ref] = statusInfo{
						Ref:       active.Ref,
						Status:    "downloading",
						Offset:    active.Offset,
						Total:     active.Total,
						StartedAt: active.StartedAt,
						UpdatedAt: active.UpdatedAt,
					}
					activeSeen[active.Ref] = struct{}{}
				}
			}

			err := updateNonActive(ctx, ongoing, cs, statuses, keys, activeSeen, &done, start)
			if err != nil {
				continue outer
			}

			var ordered []statusInfo
			for _, key := range keys {
				ordered = append(ordered, statuses[key])
			}

			for _, si := range ordered {
				jm := si.JSONMessage()
				err := enc.Encode(jm)
				if err != nil {
					logrus.Debugf("failed to encode progress message: %s", err)
				}
			}

			if done {
				out.Close()
				return
			}
		case <-ctx.Done():
			done = true // allow ui to update once more
		}
	}
}

func updateNonActive(ctx context.Context, ongoing *jobs, cs content.Store, statuses map[string]statusInfo, keys []string, activeSeen map[string]struct{}, done *bool, start time.Time) error {

	for _, j := range ongoing.jobs() {
		key := remotes.MakeRefKey(ctx, j)
		keys = append(keys, key)<commit_msg>Fix progress reporting for containerd pulls<commit_after>package containerizedengine

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"sync"
	"time"

	"github.com/containerd/containerd/content"
	"github.com/containerd/containerd/errdefs"
	"github.com/containerd/containerd/remotes"
	"github.com/docker/docker/pkg/jsonmessage"
	digest "github.com/opencontainers/go-digest"
	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
	"github.com/sirupsen/logrus"
)

func showProgress(ctx context.Context, ongoing *jobs, cs content.Store, out io.WriteCloser) {
	var (
		ticker   = time.NewTicker(100 * time.Millisecond)
		start    = time.Now()
		enc      = json.NewEncoder(out)
		statuses = map[string]statusInfo{}
		done     bool
	)
	defer ticker.Stop()

outer:
	for {
		select {
		case <-ticker.C:

			resolved := "resolved"
			if !ongoing.isResolved() {
				resolved = "resolving"
			}
			statuses[ongoing.name] = statusInfo{
				Ref:    ongoing.name,
				Status: resolved,
			}
			keys := []string{ongoing.name}

			activeSeen := map[string]struct{}{}
			if !done {
				active, err := cs.ListStatuses(ctx, "")
				if err != nil {
					logrus.Debugf("active check failed: %s", err)
					continue
				}
				// update status of active entries!
				for _, active := range active {
					statuses[active.Ref] = statusInfo{
						Ref:       active.Ref,
						Status:    "downloading",
						Offset:    active.Offset,
						Total:     active.Total,
						StartedAt: active.StartedAt,
						UpdatedAt: active.UpdatedAt,
					}
					activeSeen[active.Ref] = struct{}{}
				}
			}

			err := updateNonActive(ctx, ongoing, cs, statuses, &keys, activeSeen, &done, start)
			if err != nil {
				continue outer
			}

			var ordered []statusInfo
			for _, key := range keys {
				ordered = append(ordered, statuses[key])
			}

			for _, si := range ordered {
				jm := si.JSONMessage()
				err := enc.Encode(jm)
				if err != nil {
					logrus.Debugf("failed to encode progress message: %s", err)
				}
			}

			if done {
				out.Close()
				return
			}
		case <-ctx.Done():
			done = true // allow ui to update once more
		}
	}
}

func updateNonActive(ctx context.Context, ongoing *jobs, cs content.Store, statuses map[string]statusInfo, keys *[]string, activeSeen map[string]struct{}, done *bool, start time.Time) error {

	for _, j := range ongoing.jobs() {
		key := remotes.MakeRefKey(ctx, j)
		*keys = append(*keys, key)
{'fraction_non_alphanumeric': 0.13496677740863788, 'fraction_numerical': 0.0016611295681063123, 'mean_word_length': 2.98099173553719, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6947061', 'n_tokens_mistral': 1998, 'n_tokens_neox': 1644, 'n_words': 376}