<commit_before>package client

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/troubling/hummingbird/common"
	"github.com/troubling/hummingbird/common/conf"
	"github.com/troubling/hummingbird/common/ring"
)

const PostQuorumTimeoutMs = 100

func mkquery(options map[string]string) string {
	query := ""
	for k, v := range options {
		query += url.QueryEscape(k) + "=" + url.QueryEscape(v) + "&"
	}
	if query != "" {
		return "?" + strings.TrimRight(query, "&")
	}
	return ""
}

func addUpdateHeaders(prefix string, headers http.Header, devices []*ring.Device, i, replicas int) {
	if i < len(devices) {
		host := ""
		device := ""
		for ; i < len(devices); i += replicas {
			host += fmt.Sprintf("%s:%d,", devices[i].Ip, devices[i].Port)
			device += devices[i].Device + ","
		}
		headers.Set(prefix+"-Host", strings.TrimRight(host, ","))
		headers.Set(prefix+"-Device", strings.TrimRight(device, ","))
	}
}

type ProxyDirectClient struct {
	policyList    conf.PolicyList
	client        *http.Client
	AccountRing   ring.Ring
	ContainerRing ring.Ring
}

func NewProxyDirectClient(policyList conf.PolicyList) (*ProxyDirectClient, error) {
	var xport http.RoundTripper = &http.Transport{
		DisableCompression: true,
		Dial: (&net.Dialer{
			Timeout:   10 * time.Second,
			KeepAlive: 5 * time.Second,
		}).Dial,
	}
	// Debug hook to auto-close responses and report on it. See debug.go
	// xport = &autoCloseResponses{transport: xport}
	c := &ProxyDirectClient{
		policyList: policyList,
		client: &http.Client{
			Transport: xport,
			Timeout:   120 * time.Minute,
		},
	}
	hashPathPrefix, hashPathSuffix, err := conf.GetHashPrefixAndSuffix()
	if err != nil {
		return nil, err
	}
	c.ContainerRing, err = ring.GetRing("container", hashPathPrefix, hashPathSuffix, 0)
	if err != nil {
		return nil, err
	}
	c.AccountRing, err = ring.GetRing("account", hashPathPrefix, hashPathSuffix, 0)
	if err != nil {
		return nil, err
	}
	return c, nil
}

func (c *ProxyDirectClient) quorumResponse(reqs ...*http.Request) *http.Response {
	// this is based on swift's best_response function.
	responses := make(chan *http.Response)
	cancel := make(chan struct{})
	defer close(cancel)
	for _, req := range reqs {
		go func(req *http.Request) {
			var entry *http.Response
			if resp, err := c.client.Do(req); err != nil {
				if resp != nil {
					resp.Body.Close()
				}
				entry = ResponseStub(http.StatusInternalServerError, err.Error())
			} else {
				entry = StubResponse(resp)
			}
			select {
			case responses <- entry:
			case <-cancel:
			}
		}(req)
	}
	quorum := int(math.Ceil(float64(len(reqs)) / 2.0))
	responseClasses := []int{0, 0, 0, 0, 0, 0}
	responseCount := 0
	var chosenResponse *http.Response
	for response := range responses {
		responseCount++
		class := response.StatusCode / 100
		if class <= 5 {
			responseClasses[class]++
			if responseClasses[class] >= quorum {
				chosenResponse = response
				break
			}
		}
	}
	// Give any pending requests *some* chance to finish. This will increase
	// the likelihood that a read immediately after a write will get the latest
	// information, as we'll force the caller to wait until all backends have
	// responded, with a timeout limit.
	timeout := time.After(PostQuorumTimeoutMs * time.Millisecond)
waiting:
	for responseCount < len(reqs) {
		select {
		case <-responses:
			responseCount++
		case <-timeout:
			break waiting
		}
	}
	return chosenResponse
}

func (c *ProxyDirectClient) firstResponse(reqs ...*http.Request) (resp *http.Response) {
	success := make(chan *http.Response)
	returned := make(chan struct{})
	defer close(returned)

	for _, req := range reqs {
		go func(r *http.Request) {
			response, err := c.client.Do(r)
			if err != nil {
				if response != nil {
					response.Body.Close()
				}
				response = nil
			}
			select {
			case success <- response:
			case <-returned:
				if response != nil {
					response.Body.Close()
				}
			}
		}(req)

		select {
		case resp = <-success:
			if resp != nil && (resp.StatusCode/100 == 2 || resp.StatusCode == http.StatusPreconditionFailed || resp.StatusCode == http.StatusNotModified || resp.StatusCode == http.StatusRequestedRangeNotSatisfiable) {
				resp = StubResponse(resp)
				resp.Header.Set("Accept-Ranges", "bytes")
				if etag := resp.Header.Get("Etag"); etag != "" {
					resp.Header.Set("Etag", strings.Trim(etag, "\""))
				}
				return resp
			} else if resp != nil {
				resp.Body.Close()
			}
		case <-time.After(time.Second):
		}
	}
	return ResponseStub(http.StatusNotFound, "")
}

type proxyClient struct {
	pdc *ProxyDirectClient
	mc  ring.MemcacheRing
	lc  map[string]*ContainerInfo
}<commit_msg>Made firstReponse 503 when appropriate (#225)<commit_after>package client

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/troubling/hummingbird/common"
	"github.com/troubling/hummingbird/common/conf"
	"github.com/troubling/hummingbird/common/ring"
)

const PostQuorumTimeoutMs = 100

func mkquery(options map[string]string) string {
	query := ""
	for k, v := range options {
		query += url.QueryEscape(k) + "=" + url.QueryEscape(v) + "&"
	}
	if query != "" {
		return "?" + strings.TrimRight(query, "&")
	}
	return ""
}

func addUpdateHeaders(prefix string, headers http.Header, devices []*ring.Device, i, replicas int) {
	if i < len(devices) {
		host := ""
		device := ""
		for ; i < len(devices); i += replicas {
			host += fmt.Sprintf("%s:%d,", devices[i].Ip, devices[i].Port)
			device += devices[i].Device + ","
		}
		headers.Set(prefix+"-Host", strings.TrimRight(host, ","))
		headers.Set(prefix+"-Device", strings.TrimRight(device, ","))
	}
}

type ProxyDirectClient struct {
	policyList    conf.PolicyList
	client        *http.Client
	AccountRing   ring.Ring
	ContainerRing ring.Ring
}

func NewProxyDirectClient(policyList conf.PolicyList) (*ProxyDirectClient, error) {
	var xport http.RoundTripper = &http.Transport{
		DisableCompression: true,
		Dial: (&net.Dialer{
			Timeout:   10 * time.Second,
			KeepAlive: 5 * time.Second,
		}).Dial,
	}
	// Debug hook to auto-close responses and report on it. See debug.go
	// xport = &autoCloseResponses{transport: xport}
	c := &ProxyDirectClient{
		policyList: policyList,
		client: &http.Client{
			Transport: xport,
			Timeout:   120 * time.Minute,
		},
	}
	hashPathPrefix, hashPathSuffix, err := conf.GetHashPrefixAndSuffix()
	if err != nil {
		return nil, err
	}
	c.ContainerRing, err = ring.GetRing("container", hashPathPrefix, hashPathSuffix, 0)
	if err != nil {
		return nil, err
	}
	c.AccountRing, err = ring.GetRing("account", hashPathPrefix, hashPathSuffix, 0)
	if err != nil {
		return nil, err
	}
	return c, nil
}

func (c *ProxyDirectClient) quorumResponse(reqs ...*http.Request) *http.Response {
	// this is based on swift's best_response function.
	responses := make(chan *http.Response)
	cancel := make(chan struct{})
	defer close(cancel)
	for _, req := range reqs {
		go func(req *http.Request) {
			var entry *http.Response
			if resp, err := c.client.Do(req); err != nil {
				if resp != nil {
					resp.Body.Close()
				}
				entry = ResponseStub(http.StatusInternalServerError, err.Error())
			} else {
				entry = StubResponse(resp)
			}
			select {
			case responses <- entry:
			case <-cancel:
			}
		}(req)
	}
	quorum := int(math.Ceil(float64(len(reqs)) / 2.0))
	responseClasses := []int{0, 0, 0, 0, 0, 0}
	responseCount := 0
	var chosenResponse *http.Response
	for response := range responses {
		responseCount++
		class := response.StatusCode / 100
		if class <= 5 {
			responseClasses[class]++
			if responseClasses[class] >= quorum {
				chosenResponse = response
				break
			}
		}
	}
	// Give any pending requests *some* chance to finish. This will increase
	// the likelihood that a read immediately after a write will get the latest
	// information, as we'll force the caller to wait until all backends have
	// responded, with a timeout limit.
	timeout := time.After(PostQuorumTimeoutMs * time.Millisecond)
waiting:
	for responseCount < len(reqs) {
		select {
		case <-responses:
			responseCount++
		case <-timeout:
			break waiting
		}
	}
	return chosenResponse
}

func (c *ProxyDirectClient) firstResponse(reqs ...*http.Request) (resp *http.Response) {
	success := make(chan *http.Response)
	returned := make(chan struct{})
	defer close(returned)

	internalErrors := 0
	for _, req := range reqs {
		go func(r *http.Request) {
			response, err := c.client.Do(r)
			if err != nil {
				if response != nil {
					response.Body.Close()
				}
				response = nil
			}
			select {
			case success <- response:
			case <-returned:
				if response != nil {
					response.Body.Close()
				}
			}
		}(req)

		select {
		case resp = <-success:
			if resp == nil || resp.StatusCode/100 == 5 {
				internalErrors++
			} else if resp.StatusCode/100 == 2 || resp.StatusCode == http.StatusPreconditionFailed || resp.StatusCode == http.StatusNotModified || resp.StatusCode == http.StatusRequestedRangeNotSatisfiable {
				resp = StubResponse(resp)
				resp.Header.Set("Accept-Ranges", "bytes")
				if etag := resp.Header.Get("Etag"); etag != "" {
					resp.Header.Set("Etag", strings.Trim(etag, "\""))
				}
				return resp
			} else if resp != nil {
				resp.Body.Close()
			}
		case <-time.After(time.Second):
		}
	}
	if internalErrors == len(reqs) {
		return ResponseStub(http.StatusServiceUnavailable, "")
	}
	return ResponseStub(http.StatusNotFound, "")
}

type proxyClient struct {
	pdc *ProxyDirectClient
	mc  ring.MemcacheRing
	lc  map[string]*ContainerInfo
}
{'fraction_non_alphanumeric': 0.14284232365145227, 'fraction_numerical': 0.0073651452282157675, 'mean_word_length': 3.5241670577193807, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '317001', 'n_tokens_mistral': 3705, 'n_tokens_neox': 3276, 'n_words': 809}