<commit_before>package s3gof3r

import (
	"bytes"
	"crypto/md5"
	"fmt"
	"hash"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"net/url"
	"sync"
	"syscall"
)

type getter struct {
	url    url.URL
	client *http.Client
	b      *Bucket
	bufsz  int64
	err    error
	wg     sync.WaitGroup

	cur_chunk_id   int
	cur_chunk      *chunk
	content_length int64
	chunk_total    int
	read_ch        chan *chunk
	get_ch         chan *chunk

	//	get   chan *bytes.Buffer
	//	give  chan *bytes.Buffer
	bp *bp

	q_wait map[int]*chunk

	concurrency int
	nTry        int
	closed      bool
	c           *Config

	md5 hash.Hash
}

type chunk struct {
	id     int
	header http.Header
	start  int64
	size   int64
	b      *bytes.Buffer
	len    int64
}

func newGetter(p_url url.URL, c *Config, b *Bucket) (io.ReadCloser, http.Header, error) {
	// initialize getter
	g := new(getter)
	g.url = p_url
	g.bufsz = c.PartSize
	g.get_ch = make(chan *chunk)
	g.read_ch = make(chan *chunk)
	g.nTry = c.NTry
	g.q_wait = make(map[int]*chunk)
	g.b = b
	g.c = c
	g.client = c.Client
	g.md5 = md5.New()

	// get content length
	resp, err := g.retryRequest("HEAD", p_url.String(), nil)
	if err != nil {
		return nil, nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return nil, nil, newRespError(resp)
	}
	g.content_length = resp.ContentLength
	g.chunk_total = int((g.content_length + g.bufsz - 1) / g.bufsz) // round up, integer division
	g.concurrency = min(c.Concurrency, g.chunk_total)
	log.Println("chunk total: ", g.chunk_total)
	log.Println("content length : ", g.content_length)
	log.Println("concurrency: ", g.concurrency)

	g.bp = NewBufferPool(g.bufsz)

	for i := 0; i < g.concurrency; i++ {
		go g.worker()
	}
	go g.init_chunks()
	return g, resp.Header, nil
}

func (g *getter) retryRequest(method, urlStr string, body io.ReadSeeker) (resp *http.Response, err error) {
	for i := 0; i < g.nTry; i++ {
		var req *http.Request
		req, err = http.NewRequest(method, urlStr, body)
		if err != nil {
			return
		}
		g.b.Sign(req)
		resp, err = g.client.Do(req)
		if err == nil {
			return
		}
		log.Println(err)
		if body != nil {
			body.Seek(0, 0)
		}
	}
	return
}

func (g *getter) init_chunks() {
	id := 0
	for i := int64(0); i < g.content_length; {
		size := min64(g.bufsz, g.content_length-i)
		c := &chunk{
			id: id,
			header: http.Header{
				"Range": {fmt.Sprintf("bytes=%d-%d",
					i, i+size-1)},
			},<commit_msg>Don't blow up memory on download.<commit_after>package s3gof3r

import (
	"bytes"
	"crypto/md5"
	"fmt"
	"hash"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"net/url"
	"sync"
	"syscall"
	"time"
)

type getter struct {
	url    url.URL
	client *http.Client
	b      *Bucket
	bufsz  int64
	err    error
	wg     sync.WaitGroup

	cur_chunk_id   int
	cur_chunk      *chunk
	content_length int64
	chunk_total    int
	read_ch        chan *chunk
	get_ch         chan *chunk

	//	get   chan *bytes.Buffer
	//	give  chan *bytes.Buffer
	bp *bp

	q_wait map[int]*chunk

	concurrency int
	nTry        int
	closed      bool
	c           *Config

	md5 hash.Hash
}

type chunk struct {
	id     int
	header http.Header
	start  int64
	size   int64
	b      *bytes.Buffer
	len    int64
}

func newGetter(p_url url.URL, c *Config, b *Bucket) (io.ReadCloser, http.Header, error) {
	// initialize getter
	g := new(getter)
	g.url = p_url
	g.bufsz = c.PartSize
	g.get_ch = make(chan *chunk)
	g.read_ch = make(chan *chunk)
	g.nTry = c.NTry
	g.q_wait = make(map[int]*chunk)
	g.b = b
	g.c = c
	g.client = c.Client
	g.md5 = md5.New()

	// get content length
	resp, err := g.retryRequest("HEAD", p_url.String(), nil)
	if err != nil {
		return nil, nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return nil, nil, newRespError(resp)
	}
	g.content_length = resp.ContentLength
	g.chunk_total = int((g.content_length + g.bufsz - 1) / g.bufsz) // round up, integer division
	g.concurrency = min(c.Concurrency, g.chunk_total)
	log.Println("chunk total: ", g.chunk_total)
	log.Println("content length : ", g.content_length)
	log.Println("concurrency: ", g.concurrency)

	g.bp = NewBufferPool(g.bufsz)

	for i := 0; i < g.concurrency; i++ {
		go g.worker()
	}
	go g.init_chunks()
	return g, resp.Header, nil
}

func (g *getter) retryRequest(method, urlStr string, body io.ReadSeeker) (resp *http.Response, err error) {
	for i := 0; i < g.nTry; i++ {
		var req *http.Request
		req, err = http.NewRequest(method, urlStr, body)
		if err != nil {
			return
		}
		g.b.Sign(req)
		resp, err = g.client.Do(req)
		if err == nil {
			return
		}
		log.Println(err)
		if body != nil {
			body.Seek(0, 0)
		}
	}
	return
}

func (g *getter) init_chunks() {
	id := 0
	for i := int64(0); i < g.content_length; {
		for len(g.q_wait) > g.concurrency {
			time.Sleep(500 * time.Millisecond)
		}
		size := min64(g.bufsz, g.content_length-i)
		c := &chunk{
			id: id,
			header: http.Header{
				"Range": {fmt.Sprintf("bytes=%d-%d",
					i, i+size-1)},
			},
{'fraction_non_alphanumeric': 0.14285714285714285, 'fraction_numerical': 0.013227513227513227, 'mean_word_length': 2.8249027237354087, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6130130', 'n_tokens_mistral': 2173, 'n_tokens_neox': 2034, 'n_words': 501}