<commit_before>// Copyright 2017 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package downloader

import (
	"archive/tar"
	"bytes"
	"context"
	"encoding/json"
	stderrors "errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	humanize "github.com/dustin/go-humanize"
	"golang.org/x/sync/errgroup"

	"go.chromium.org/luci/client/internal/common"
	"go.chromium.org/luci/common/clock"
	"go.chromium.org/luci/common/data/caching/cache"
	"go.chromium.org/luci/common/data/stringset"
	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/isolated"
	"go.chromium.org/luci/common/isolatedclient"
	"go.chromium.org/luci/common/logging"
	"go.chromium.org/luci/common/retry"
	"go.chromium.org/luci/common/retry/transient"
)

// Downloader is a high level interface to an isolatedclient.Client.
//
// Downloader provides functionality to download full isolated trees.
type Downloader struct {
	// Immutable variables.
	ctx    context.Context
	cancel func()

	c         *isolatedclient.Client
	rootHash  isolated.HexDigest
	outputDir string
	options   Options

	// Mutable variables.
	mu       sync.Mutex
	started  bool
	finished bool

	err errors.MultiError

	interval time.Duration
	isoMap   map[isolated.HexDigest]*isolated.Isolated

	fileStats       FileStats
	lastFileStatsCb time.Time

	// dirCache is a cache of known existing directories which is extended
	// and read from by ensureDir.
	muCache  sync.RWMutex
	dirCache stringset.Set

	// pool is a goroutine priority pool which manages jobs to download
	// isolated trees and files.
	pool *common.GoroutinePriorityPool
}

// Options are some optional bits you can pass to New.
type Options struct {
	// FileCallback allows you to set a callback function that will be called with
	// every file name and metadata which is extracted to disk by the Downloader.
	//
	// This callback should execute quickly (e.g. push to channel, append to list),
	// as it will partially block the process of the download.
	//
	// Tarball archives behave a bit differently. The callback will be called for
	// individual files in the tarball, but the 'Digest' field will be empty. The
	// Size and Mode fields will be populated, however. The callback will ALSO be
	// called for the tarfile as a whole (but the tarfile will not actually exist
	// on disk).
	FileCallback func(string, *isolated.File)

	// FileStatsCallback is a callback function that will be called at intervals
	// with relevant statistics (see MaxFileStatsInterval).
	//
	// This callback should execute quickly (e.g. push to channel, append to list,
	// etc.) as it will partially block the process of the download.  However,
	// since it's only called once every "interval" amount of time, being a bit
	// slow here (e.g. doing console IO) isn't the worst.
	//
	// To allow this callback to actuate meaningfully for small downloads, this
	// will be called more frequently at the beginning of the download, and will
	// taper off to MaxFileStatsInterval.
	FileStatsCallback func(FileStats, time.Duration)

	// MaxFileStatsInterval changes the maximum interval that the
	// FileStatsCallback will be called at.
	//
	// At the beginning of the download the interval is 100ms, but it will ramp up
	// to the provided maxInterval. If you specify a MaxFileStatsInterval
	// smaller than 100ms, there will be no ramp up, just a fixed interval at the
	// one you specify here.
	//
	// Default: 5 seconds
	MaxFileStatsInterval time.Duration

	// MaxConcurrentJobs is the number of parallel worker goroutines the
	// downloader will have.
	//
	// Default: 8
	MaxConcurrentJobs int

	// Cache is used to save/load isolated items to/from cache.
	Cache cache.Cache
}

// normalizePathSeparator returns path having os native path separator.
func normalizePathSeparator(p string) string {
	if filepath.Separator == '/' {
		return strings.Replace(p, `\`, string(filepath.Separator), -1)
	}
	return strings.Replace(p, "/", string(filepath.Separator), -1)
}

// New returns a Downloader instance, good to download one isolated.
//
// ctx will be used for logging and clock.
//
// The Client, hash and outputDir must be specified.
//
// If options is nil, this will use defaults as described in the Options struct.
func New(ctx context.Context, c *isolatedclient.Client, hash isolated.HexDigest,
	outputDir string, options *Options) *Downloader {

	var opt Options
	if options != nil {
		opt = *options
	}
	if opt.MaxConcurrentJobs == 0 {
		opt.MaxConcurrentJobs = 8
	}
	if opt.MaxFileStatsInterval == 0 {
		opt.MaxFileStatsInterval = time.Second * 5
	}
<commit_msg>client: set smaller default max concurrent jobs in 32bit client<commit_after>// Copyright 2017 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package downloader

import (
	"archive/tar"
	"bytes"
	"context"
	"encoding/json"
	stderrors "errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strings"
	"sync"
	"time"

	humanize "github.com/dustin/go-humanize"
	"golang.org/x/sync/errgroup"

	"go.chromium.org/luci/client/internal/common"
	"go.chromium.org/luci/common/clock"
	"go.chromium.org/luci/common/data/caching/cache"
	"go.chromium.org/luci/common/data/stringset"
	"go.chromium.org/luci/common/errors"
	"go.chromium.org/luci/common/isolated"
	"go.chromium.org/luci/common/isolatedclient"
	"go.chromium.org/luci/common/logging"
	"go.chromium.org/luci/common/retry"
	"go.chromium.org/luci/common/retry/transient"
)

// Downloader is a high level interface to an isolatedclient.Client.
//
// Downloader provides functionality to download full isolated trees.
type Downloader struct {
	// Immutable variables.
	ctx    context.Context
	cancel func()

	c         *isolatedclient.Client
	rootHash  isolated.HexDigest
	outputDir string
	options   Options

	// Mutable variables.
	mu       sync.Mutex
	started  bool
	finished bool

	err errors.MultiError

	interval time.Duration
	isoMap   map[isolated.HexDigest]*isolated.Isolated

	fileStats       FileStats
	lastFileStatsCb time.Time

	// dirCache is a cache of known existing directories which is extended
	// and read from by ensureDir.
	muCache  sync.RWMutex
	dirCache stringset.Set

	// pool is a goroutine priority pool which manages jobs to download
	// isolated trees and files.
	pool *common.GoroutinePriorityPool
}

// Options are some optional bits you can pass to New.
type Options struct {
	// FileCallback allows you to set a callback function that will be called with
	// every file name and metadata which is extracted to disk by the Downloader.
	//
	// This callback should execute quickly (e.g. push to channel, append to list),
	// as it will partially block the process of the download.
	//
	// Tarball archives behave a bit differently. The callback will be called for
	// individual files in the tarball, but the 'Digest' field will be empty. The
	// Size and Mode fields will be populated, however. The callback will ALSO be
	// called for the tarfile as a whole (but the tarfile will not actually exist
	// on disk).
	FileCallback func(string, *isolated.File)

	// FileStatsCallback is a callback function that will be called at intervals
	// with relevant statistics (see MaxFileStatsInterval).
	//
	// This callback should execute quickly (e.g. push to channel, append to list,
	// etc.) as it will partially block the process of the download.  However,
	// since it's only called once every "interval" amount of time, being a bit
	// slow here (e.g. doing console IO) isn't the worst.
	//
	// To allow this callback to actuate meaningfully for small downloads, this
	// will be called more frequently at the beginning of the download, and will
	// taper off to MaxFileStatsInterval.
	FileStatsCallback func(FileStats, time.Duration)

	// MaxFileStatsInterval changes the maximum interval that the
	// FileStatsCallback will be called at.
	//
	// At the beginning of the download the interval is 100ms, but it will ramp up
	// to the provided maxInterval. If you specify a MaxFileStatsInterval
	// smaller than 100ms, there will be no ramp up, just a fixed interval at the
	// one you specify here.
	//
	// Default: 5 seconds
	MaxFileStatsInterval time.Duration

	// MaxConcurrentJobs is the number of parallel worker goroutines the
	// downloader will have.
	//
	// Default: 8
	MaxConcurrentJobs int

	// Cache is used to save/load isolated items to/from cache.
	Cache cache.Cache
}

// normalizePathSeparator returns path having os native path separator.
func normalizePathSeparator(p string) string {
	if filepath.Separator == '/' {
		return strings.Replace(p, `\`, string(filepath.Separator), -1)
	}
	return strings.Replace(p, "/", string(filepath.Separator), -1)
}

// New returns a Downloader instance, good to download one isolated.
//
// ctx will be used for logging and clock.
//
// The Client, hash and outputDir must be specified.
//
// If options is nil, this will use defaults as described in the Options struct.
func New(ctx context.Context, c *isolatedclient.Client, hash isolated.HexDigest,
	outputDir string, options *Options) *Downloader {

	var opt Options
	if options != nil {
		opt = *options
	}
	if opt.MaxConcurrentJobs == 0 {
		if runtime.GOARCH == "386" {
			// set lower value to prevent memory exceeded error.
			opt.MaxConcurrentJobs = 4
		} else {
			opt.MaxConcurrentJobs = 8
		}

	}
	if opt.MaxFileStatsInterval == 0 {
		opt.MaxFileStatsInterval = time.Second * 5
	}

{'fraction_non_alphanumeric': 0.08943634969325154, 'fraction_numerical': 0.004792944785276074, 'mean_word_length': 3.4892426850258174, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2990280', 'n_tokens_mistral': 3208, 'n_tokens_neox': 3077, 'n_words': 1212}