feat: support progressed output in blob push and blob get

cmd/oras/internal/display/console/testutils/testutils.go

@@ -29,30 +29,30 @@ import (
 )
 
 // NewPty creates a new pty pair for testing, caller is responsible for closing
-// the returned slave if err is not nil.
+// the returned device file if err is not nil.
 func NewPty() (console.Console, *os.File, error) {
-	pty, slavePath, err := console.NewPty()
+	pty, devicePath, err := console.NewPty()
 	if err != nil {
 		return nil, nil, err
 	}
-	slave, err := os.OpenFile(slavePath, os.O_RDWR, 0)
+	device, err := os.OpenFile(devicePath, os.O_RDWR, 0)
 	if err != nil {
 		return nil, nil, err
 	}
-	return pty, slave, nil
+	return pty, device, nil
 }
 
 // MatchPty checks that the output matches the expected strings in specified
 // order.
-func MatchPty(pty console.Console, slave *os.File, expected ...string) error {
+func MatchPty(pty console.Console, device *os.File, expected ...string) error {
 	var wg sync.WaitGroup
 	wg.Add(1)
 	var buffer bytes.Buffer
 	go func() {
 		defer wg.Done()
 		_, _ = io.Copy(&buffer, pty)
 	}()
-	slave.Close()
+	device.Close()
 	wg.Wait()
 
 	return OrderedMatch(buffer.String(), expected...)

cmd/oras/internal/display/progress/humanize.go → cmd/oras/internal/display/progress/humanize/bytes.go

@@ -13,31 +13,40 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package progress
+package humanize
 
 import (
+	"fmt"
 	"math"
 )
 
-var (
-	units = []string{"B", "kB", "MB", "GB", "TB"}
-	base  = 1024.0
-)
+const base = 1024.0
+
+var units = []string{"B", "kB", "MB", "GB", "TB"}
 
-type bytes struct {
-	size float64
-	unit string
+type Bytes struct {
+	Size float64
+	Unit string
 }
 
 // ToBytes converts size in bytes to human readable format.
-func ToBytes(sizeInBytes int64) bytes {
+func ToBytes(sizeInBytes int64) Bytes {
 	f := float64(sizeInBytes)
 	if f < base {
-		return bytes{f, units[0]}
+		return Bytes{f, units[0]}
+	}
+	e := int(math.Floor(math.Log(f) / math.Log(base)))
+	if e >= len(units) {
+		// only support up to TB
+		e = len(units) - 1
 	}
-	e := math.Floor(math.Log(f) / math.Log(base))
-	p := f / math.Pow(base, e)
-	return bytes{RoundTo(p), units[int(e)]}
+	p := f / math.Pow(base, float64(e))
+	return Bytes{RoundTo(p), units[e]}
+}
+
+// String returns the string representation of Bytes.
+func (b Bytes) String() string {
+	return fmt.Sprintf("%v %2s", b.Size, b.Unit)
 }
 
 // RoundTo makes length of the size string to less than or equal to 4.

cmd/oras/internal/display/progress/humanize_test.go → cmd/oras/internal/display/progress/humanize/bytes_test.go

@@ -13,7 +13,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package progress
+package humanize
 
 import (
 	"reflect"
@@ -50,17 +50,17 @@ func TestToBytes(t *testing.T) {
 	tests := []struct {
 		name string
 		args args
-		want bytes
+		want Bytes
 	}{
-		{"0 bytes", args{0}, bytes{0, "B"}},
-		{"1023 bytes", args{1023}, bytes{1023, "B"}},
-		{"1 kB", args{1024}, bytes{1, "kB"}},
-		{"1.5 kB", args{1024 + 512}, bytes{1.5, "kB"}},
-		{"12.5 kB", args{1024 * 12.5}, bytes{12.5, "kB"}},
-		{"512.5 kB", args{1024 * 512.5}, bytes{513, "kB"}},
-		{"1 MB", args{1024 * 1024}, bytes{1, "MB"}},
-		{"1 GB", args{1024 * 1024 * 1024}, bytes{1, "GB"}},
-		{"1 TB", args{1024 * 1024 * 1024 * 1024}, bytes{1, "TB"}},
+		{"0 bytes", args{0}, Bytes{0, "B"}},
+		{"1023 bytes", args{1023}, Bytes{1023, "B"}},
+		{"1 kB", args{1024}, Bytes{1, "kB"}},
+		{"1.5 kB", args{1024 + 512}, Bytes{1.5, "kB"}},
+		{"12.5 kB", args{1024 * 12.5}, Bytes{12.5, "kB"}},
+		{"512.5 kB", args{1024 * 512.5}, Bytes{513, "kB"}},
+		{"1 MB", args{1024 * 1024}, Bytes{1, "MB"}},
+		{"1 GB", args{1024 * 1024 * 1024}, Bytes{1, "GB"}},
+		{"1 TB", args{1024 * 1024 * 1024 * 1024}, Bytes{1, "TB"}},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

cmd/oras/internal/display/progress/manager.go

@@ -24,22 +24,23 @@ import (
 	"oras.land/oras/cmd/oras/internal/display/console"
 )
 
-// BufferSize is the size of the status channel buffer.
-const BufferSize = 20
+const (
+	// BufferSize is the size of the status channel buffer.
+	BufferSize       = 1
+	bufFlushDuration = 200 * time.Millisecond
+)
 
 var errManagerStopped = errors.New("progress output manage has already been stopped")
 
 // Status is print message channel
-type Status chan<- *status
+type Status chan *status
 
 // Manager is progress view master
 type Manager interface {
 	Add() (Status, error)
 	Close() error
 }
 
-const bufFlushDuration = 200 * time.Millisecond
-
 type manager struct {
 	status       []*status
 	statusLock   sync.RWMutex
@@ -73,7 +74,6 @@ func (m *manager) start() {
 		for {
 			select {
 			case <-m.renderDone:
-				m.updating.Wait()
 				m.render()
 				close(m.renderClosed)
 				return
@@ -136,9 +136,11 @@ func (m *manager) Close() error {
 	if m.closed() {
 		return errManagerStopped
 	}
-	// 1. stop periodic rendering
+	// 1. wait for update to stop
+	m.updating.Wait()
+	// 2. stop periodic rendering
 	close(m.renderDone)
-	// 2. wait for the render stop
+	// 3. wait for the render stop
 	<-m.renderClosed
 	return nil
 }

cmd/oras/internal/display/progress/status.go

@@ -17,18 +17,20 @@ package progress
 
 import (
 	"fmt"
+	"math"
 	"strings"
 	"sync"
 	"time"
 	"unicode/utf8"
 
 	"github.com/morikuni/aec"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"oras.land/oras/cmd/oras/internal/display/progress/humanize"
 )
 
 const (
 	barLength    = 20
-	speedLength  = 9    // speed_size(4) + space(1) + speed_unit(4)
+	speedLength  = 7    // speed_size(4) + space(1) + speed_unit(2)
 	zeroDuration = "0s" // default zero value of time.Duration.String()
 	zeroStatus   = "loading status..."
 	zeroDigest   = "  └─ loading digest..."
@@ -40,7 +42,7 @@ type status struct {
 	prompt         string
 	descriptor     ocispec.Descriptor
 	offset         int64
-	total          bytes
+	total          humanize.Bytes
 	lastOffset     int64
 	lastRenderTime time.Time
 
@@ -113,11 +115,11 @@ func (s *status) String(width int) (string, string) {
 	var offset string
 	switch percent {
 	case 1: // 100%, show exact size
-		offset = fmt.Sprint(s.total.size)
+		offset = fmt.Sprint(s.total.Size)
 	default: // 0% ~ 99%, show 2-digit precision
-		offset = fmt.Sprintf("%.2f", RoundTo(s.total.size*percent))
+		offset = fmt.Sprintf("%.2f", humanize.RoundTo(s.total.Size*percent))
 	}
-	right := fmt.Sprintf(" %s/%v %s %6.2f%% %6s", offset, s.total.size, s.total.unit, percent*100, s.durationString())
+	right := fmt.Sprintf(" %s/%s %6.2f%% %6s", offset, s.total, percent*100, s.durationString())
 	lenRight := utf8.RuneCountInString(right)
 
 	var left string
@@ -126,10 +128,9 @@ func (s *status) String(width int) (string, string) {
 		lenBar := int(percent * barLength)
 		bar := fmt.Sprintf("[%s%s]", aec.Inverse.Apply(strings.Repeat(" ", lenBar)), strings.Repeat(".", barLength-lenBar))
 		speed := s.calculateSpeed()
-		speedStr := fmt.Sprintf("%v %2s/s", speed.size, speed.unit)
-		left = fmt.Sprintf("%c %s(%*s) %s %s", s.mark.symbol(), bar, speedLength, speedStr, s.prompt, name)
-		// bar + wrapper(2) + space(1) + speed + wrapper(2) = len(bar) + len(speed) + 5
-		lenLeft = barLength + speedLength + 5
+		left = fmt.Sprintf("%c %s(%*s/s) %s %s", s.mark.symbol(), bar, speedLength, speed, s.prompt, name)
+		// bar + wrapper(2) + space(1) + speed + "/s"(2) + wrapper(2) = len(bar) + len(speed) + 7
+		lenLeft = barLength + speedLength + 7
 	} else {
 		left = fmt.Sprintf("√ %s %s", s.prompt, name)
 	}
@@ -147,15 +148,16 @@ func (s *status) String(width int) (string, string) {
 
 // calculateSpeed calculates the speed of the progress and update last status.
 // caller must hold the lock.
-func (s *status) calculateSpeed() bytes {
+func (s *status) calculateSpeed() humanize.Bytes {
 	now := time.Now()
 	secondsTaken := now.Sub(s.lastRenderTime).Seconds()
+	secondsTaken = math.Max(secondsTaken, float64(bufFlushDuration.Milliseconds())/1000)
 	bytes := float64(s.offset - s.lastOffset)
 
 	s.lastOffset = s.offset
 	s.lastRenderTime = now
 
-	return ToBytes(int64(bytes / secondsTaken))
+	return humanize.ToBytes(int64(bytes / secondsTaken))
 }
 
 // durationString returns a viewable TTY string of the status with duration.
@@ -190,7 +192,7 @@ func (s *status) Update(n *status) {
 	if n.offset >= 0 {
 		s.offset = n.offset
 		if n.descriptor.Size != s.descriptor.Size {
-			s.total = ToBytes(n.descriptor.Size)
+			s.total = humanize.ToBytes(n.descriptor.Size)
 		}
 		s.descriptor = n.descriptor
 	}

cmd/oras/internal/display/progress/status_test.go

@@ -22,6 +22,7 @@ import (
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras/cmd/oras/internal/display/console"
 	"oras.land/oras/cmd/oras/internal/display/console/testutils"
+	"oras.land/oras/cmd/oras/internal/display/progress/humanize"
 )
 
 func Test_status_String(t *testing.T) {
@@ -41,16 +42,16 @@ func Test_status_String(t *testing.T) {
 		},
 		startTime: time.Now().Add(-time.Minute),
 		offset:    0,
-		total:     ToBytes(2),
+		total:     humanize.ToBytes(2),
 	})
 	// full name
 	statusStr, digestStr := s.String(120)
-	if err := testutils.OrderedMatch(statusStr+digestStr, " [\x1b[7m\x1b[0m....................]", s.prompt, s.descriptor.MediaType, "0.00/2 B", "0.00%", s.descriptor.Digest.String()); err != nil {
+	if err := testutils.OrderedMatch(statusStr+digestStr, " [\x1b[7m\x1b[0m....................]", s.prompt, s.descriptor.MediaType, "0.00/2  B", "0.00%", s.descriptor.Digest.String()); err != nil {
 		t.Error(err)
 	}
 	// partial name
 	statusStr, digestStr = s.String(console.MinWidth)
-	if err := testutils.OrderedMatch(statusStr+digestStr, " [\x1b[7m\x1b[0m....................]", s.prompt, "application/vn.", "0.00/2 B", "0.00%", s.descriptor.Digest.String()); err != nil {
+	if err := testutils.OrderedMatch(statusStr+digestStr, " [\x1b[7m\x1b[0m....................]", s.prompt, "application/v.", "0.00/2  B", "0.00%", s.descriptor.Digest.String()); err != nil {
 		t.Error(err)
 	}
 
@@ -61,7 +62,7 @@ func Test_status_String(t *testing.T) {
 		descriptor: s.descriptor,
 	})
 	statusStr, digestStr = s.String(120)
-	if err := testutils.OrderedMatch(statusStr+digestStr, "√", s.prompt, s.descriptor.MediaType, "2/2 B", "100.00%", s.descriptor.Digest.String()); err != nil {
+	if err := testutils.OrderedMatch(statusStr+digestStr, "√", s.prompt, s.descriptor.MediaType, "2/2  B", "100.00%", s.descriptor.Digest.String()); err != nil {
 		t.Error(err)
 	}
 }

cmd/oras/internal/display/track/reader.go

@@ -92,12 +92,13 @@ func (r *reader) Read(p []byte) (int, error) {
 		if r.offset != r.descriptor.Size {
 			return n, io.ErrUnexpectedEOF
 		}
-		r.status <- progress.NewStatus(r.actionPrompt, r.descriptor, r.offset)
 	}
-
-	if len(r.status) < progress.BufferSize {
-		// intermediate progress might be ignored if buffer is full
-		r.status <- progress.NewStatus(r.actionPrompt, r.descriptor, r.offset)
+	for {
+		select {
+		case r.status <- progress.NewStatus(r.actionPrompt, r.descriptor, r.offset):
+			// purge the channel until successfully pushed
+			return n, err
+		case <-r.status:
+		}
 	}
-	return n, err
 }

cmd/oras/internal/option/common.go

@@ -39,7 +39,7 @@ type Common struct {
 func (opts *Common) ApplyFlags(fs *pflag.FlagSet) {
 	fs.BoolVarP(&opts.Debug, "debug", "d", false, "debug mode")
 	fs.BoolVarP(&opts.Verbose, "verbose", "v", false, "verbose output")
-	fs.BoolVarP(&opts.noTTY, "no-tty", "", false, "[Preview] avoid using stdout as a terminal")
+	fs.BoolVarP(&opts.noTTY, "no-tty", "", false, "[Preview] do not show progress output")
 }
 
 // WithContext returns a new FieldLogger and an associated Context derived from ctx.

cmd/oras/internal/option/common_unix_test.go

@@ -24,21 +24,21 @@ import (
 )
 
 func TestCommon_parseTTY(t *testing.T) {
-	_, slave, err := testutils.NewPty()
+	_, device, err := testutils.NewPty()
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer slave.Close()
+	defer device.Close()
 	var opts Common
 
 	// TTY output
-	if err := opts.parseTTY(slave); err != nil {
+	if err := opts.parseTTY(device); err != nil {
 		t.Errorf("unexpected error with TTY output: %v", err)
 	}
 
 	// --debug
 	opts.Debug = true
-	if err := opts.parseTTY(slave); err == nil {
+	if err := opts.parseTTY(device); err == nil {
 		t.Error("expected error when debug is set with TTY output")
 	}
 }