<commit_before>/*
Package couchbase provides a smart client for go.

Usage:

 client, err := couchbase.Connect("http://myserver:8091/")
 handleError(err)
 pool, err := client.GetPool("default")
 handleError(err)
 bucket, err := pool.GetBucket("MyAwesomeBucket")
 handleError(err)
 ...

or a shortcut for the bucket directly

 bucket, err := couchbase.GetBucket("http://myserver:8091/", "default", "default")

in any case, you can specify authentication credentials using
standard URL userinfo syntax:

 b, err := couchbase.GetBucket("http://bucketname:bucketpass@myserver:8091/",
         "default", "bucket")
*/
package couchbase

import (
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"

	"github.com/couchbase/gomemcached"
	"github.com/couchbase/gomemcached/client" // package name is 'memcached'
	"github.com/couchbase/goutils/logging"
)

// Mutation Token
type MutationToken struct {
	VBid  uint16 // vbucket id
	Guard uint64 // vbuuid
	Value uint64 // sequence number
}

// Maximum number of times to retry a chunk of a bulk get on error.
var MaxBulkRetries = 5000
var backOffDuration time.Duration = 100 * time.Millisecond
var MaxBackOffRetries = 25 // exponentail backOff result in over 30sec (25*13*0.1s)

// If this is set to a nonzero duration, Do() and ViewCustom() will log a warning if the call
// takes longer than that.
var SlowServerCallWarningThreshold time.Duration

func slowLog(startTime time.Time, format string, args ...interface{}) {
	if elapsed := time.Now().Sub(startTime); elapsed > SlowServerCallWarningThreshold {
		pc, _, _, _ := runtime.Caller(2)
		caller := runtime.FuncForPC(pc).Name()
		logging.Infof("go-couchbase: "+format+" in "+caller+" took "+elapsed.String(), args...)
	}
}

// Return true if error is KEY_ENOENT. Required by cbq-engine
func IsKeyEExistsError(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && res.Status == gomemcached.KEY_EEXISTS {
		return true
	}

	return false
}

// Return true if error is KEY_ENOENT. Required by cbq-engine
func IsKeyNoEntError(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && res.Status == gomemcached.KEY_ENOENT {
		return true
	}

	return false
}

// Return true if error suggests a bucket refresh is required. Required by cbq-engine
func IsRefreshRequired(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && (res.Status == gomemcached.NO_BUCKET || res.Status == gomemcached.NOT_MY_VBUCKET) {
		return true
	}

	return false
}

// Return true if a collection is not known. Required by cbq-engine
func IsUnknownCollection(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && (res.Status == gomemcached.UNKNOWN_COLLECTION) {
		return true
	}

	return false
}

// ClientOpCallback is called for each invocation of Do.
var ClientOpCallback func(opname, k string, start time.Time, err error)

// Do executes a function on a memcached connection to the node owning key "k"
//
// Note that this automatically handles transient errors by replaying
// your function on a "not-my-vbucket" error, so don't assume
// your command will only be executed only once.
func (b *Bucket) Do(k string, f func(mc *memcached.Client, vb uint16) error) (err error) {
	return b.Do2(k, f, true)
}

func (b *Bucket) Do2(k string, f func(mc *memcached.Client, vb uint16) error, deadline bool) (err error) {
	var lastError error

	if SlowServerCallWarningThreshold > 0 {
		defer slowLog(time.Now(), "call to Do(%q)", k)
	}

	vb := b.VBHash(k)
	maxTries := len(b.Nodes()) * 2
	for i := 0; i < maxTries; i++ {
		conn, pool, err := b.getConnectionToVBucket(vb)
		if err != nil {
			if (err == errNoPool || isConnError(err)) && backOff(i, maxTries, backOffDuration, true) {
				b.Refresh()
				continue
			}
			return err
		}

		if deadline && DefaultTimeout > 0 {
			conn.SetDeadline(getDeadline(noDeadline, DefaultTimeout))
		} else {
			conn.SetDeadline(noDeadline)
		}
		lastError = f(conn, uint16(vb))

		retry := false
		discard := isOutOfBoundsError(lastError) || IsReadTimeOutError(lastError)

		// MB-30967 / MB-31001 implement back off for transient errors
		if resp, ok := lastError.(*gomemcached.MCResponse); ok {
			switch resp.Status {
			case gomemcached.NOT_MY_VBUCKET:
				b.Refresh()
				// MB-28842: in case of NMVB, check if the node is still part of the map
				// and ditch the connection if it isn't.
				discard = b.checkVBmap(pool.Node())
				retry = true
			case gomemcached.NOT_SUPPORTED:
				discard = true
				retry = true
			case gomemcached.ENOMEM:
				fallthrough
			case gomemcached.TMPFAIL, gomemcached.EBUSY:
				retry = backOff(i, maxTries, backOffDuration, true)
			}
		} else if lastError != nil && isConnError(lastError) && backOff(i, maxTries, backOffDuration, true) {
			retry = true
		}

		if discard {
			pool.Discard(conn)
		} else {
			pool.Return(conn)
		}

		if !retry {
			return lastError
		}
	}

	if resp, ok := lastError.(*gomemcached.MCResponse); ok {
		return fmt.Errorf("unable to complete action after %v attemps: %v", maxTries, resp.Status)
	} else {
		return fmt.Errorf("unable to complete action after %v attemps: %v", maxTries, lastError)
	}
}

type GatheredStats struct {
	Server string
	Stats  map[string]string
	Err    error
}

func getStatsParallel(sn string, b *Bucket, offset int, which string,
	ch chan<- GatheredStats) {
	pool := b.getConnPool(offset)
	var gatheredStats GatheredStats

	conn, err := pool.Get()
	defer func() {
		pool.Return(conn)
		ch <- gatheredStats
	}()

	if err != nil {<commit_msg>MB-44328 clarify memcached status in retry error<commit_after>/*
Package couchbase provides a smart client for go.

Usage:

 client, err := couchbase.Connect("http://myserver:8091/")
 handleError(err)
 pool, err := client.GetPool("default")
 handleError(err)
 bucket, err := pool.GetBucket("MyAwesomeBucket")
 handleError(err)
 ...

or a shortcut for the bucket directly

 bucket, err := couchbase.GetBucket("http://myserver:8091/", "default", "default")

in any case, you can specify authentication credentials using
standard URL userinfo syntax:

 b, err := couchbase.GetBucket("http://bucketname:bucketpass@myserver:8091/",
         "default", "bucket")
*/
package couchbase

import (
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"

	"github.com/couchbase/gomemcached"
	"github.com/couchbase/gomemcached/client" // package name is 'memcached'
	"github.com/couchbase/goutils/logging"
)

// Mutation Token
type MutationToken struct {
	VBid  uint16 // vbucket id
	Guard uint64 // vbuuid
	Value uint64 // sequence number
}

// Maximum number of times to retry a chunk of a bulk get on error.
var MaxBulkRetries = 5000
var backOffDuration time.Duration = 100 * time.Millisecond
var MaxBackOffRetries = 25 // exponentail backOff result in over 30sec (25*13*0.1s)

// If this is set to a nonzero duration, Do() and ViewCustom() will log a warning if the call
// takes longer than that.
var SlowServerCallWarningThreshold time.Duration

func slowLog(startTime time.Time, format string, args ...interface{}) {
	if elapsed := time.Now().Sub(startTime); elapsed > SlowServerCallWarningThreshold {
		pc, _, _, _ := runtime.Caller(2)
		caller := runtime.FuncForPC(pc).Name()
		logging.Infof("go-couchbase: "+format+" in "+caller+" took "+elapsed.String(), args...)
	}
}

// Return true if error is KEY_EEXISTS. Required by cbq-engine
func IsKeyEExistsError(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && res.Status == gomemcached.KEY_EEXISTS {
		return true
	}

	return false
}

// Return true if error is KEY_ENOENT. Required by cbq-engine
func IsKeyNoEntError(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && res.Status == gomemcached.KEY_ENOENT {
		return true
	}

	return false
}

// Return true if error suggests a bucket refresh is required. Required by cbq-engine
func IsRefreshRequired(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && (res.Status == gomemcached.NO_BUCKET || res.Status == gomemcached.NOT_MY_VBUCKET) {
		return true
	}

	return false
}

// Return true if a collection is not known. Required by cbq-engine
func IsUnknownCollection(err error) bool {

	res, ok := err.(*gomemcached.MCResponse)
	if ok && (res.Status == gomemcached.UNKNOWN_COLLECTION) {
		return true
	}

	return false
}

// ClientOpCallback is called for each invocation of Do.
var ClientOpCallback func(opname, k string, start time.Time, err error)

// Do executes a function on a memcached connection to the node owning key "k"
//
// Note that this automatically handles transient errors by replaying
// your function on a "not-my-vbucket" error, so don't assume
// your command will only be executed only once.
func (b *Bucket) Do(k string, f func(mc *memcached.Client, vb uint16) error) (err error) {
	return b.Do2(k, f, true)
}

func (b *Bucket) Do2(k string, f func(mc *memcached.Client, vb uint16) error, deadline bool) (err error) {
	var lastError error

	if SlowServerCallWarningThreshold > 0 {
		defer slowLog(time.Now(), "call to Do(%q)", k)
	}

	vb := b.VBHash(k)
	maxTries := len(b.Nodes()) * 2
	for i := 0; i < maxTries; i++ {
		conn, pool, err := b.getConnectionToVBucket(vb)
		if err != nil {
			if (err == errNoPool || isConnError(err)) && backOff(i, maxTries, backOffDuration, true) {
				b.Refresh()
				continue
			}
			return err
		}

		if deadline && DefaultTimeout > 0 {
			conn.SetDeadline(getDeadline(noDeadline, DefaultTimeout))
		} else {
			conn.SetDeadline(noDeadline)
		}
		lastError = f(conn, uint16(vb))

		retry := false
		discard := isOutOfBoundsError(lastError) || IsReadTimeOutError(lastError)

		// MB-30967 / MB-31001 implement back off for transient errors
		if resp, ok := lastError.(*gomemcached.MCResponse); ok {
			switch resp.Status {
			case gomemcached.NOT_MY_VBUCKET:
				b.Refresh()
				// MB-28842: in case of NMVB, check if the node is still part of the map
				// and ditch the connection if it isn't.
				discard = b.checkVBmap(pool.Node())
				retry = true
			case gomemcached.NOT_SUPPORTED:
				discard = true
				retry = true
			case gomemcached.ENOMEM:
				fallthrough
			case gomemcached.TMPFAIL, gomemcached.EBUSY:
				retry = backOff(i, maxTries, backOffDuration, true)
			}
		} else if lastError != nil && isConnError(lastError) && backOff(i, maxTries, backOffDuration, true) {
			retry = true
		}

		if discard {
			pool.Discard(conn)
		} else {
			pool.Return(conn)
		}

		if !retry {
			return lastError
		}
	}

	if resp, ok := lastError.(*gomemcached.MCResponse); ok {
		err := gomemcached.StatusNames[resp.Status]
		if err == "" {
			err = fmt.Sprintf("KV status %v", resp.Status)
		}
		return fmt.Errorf("unable to complete action after %v attemps: %v", maxTries, err)
	} else {
		return fmt.Errorf("unable to complete action after %v attemps: %v", maxTries, lastError)
	}
}

type GatheredStats struct {
	Server string
	Stats  map[string]string
	Err    error
}

func getStatsParallel(sn string, b *Bucket, offset int, which string,
	ch chan<- GatheredStats) {
	pool := b.getConnPool(offset)
	var gatheredStats GatheredStats

	conn, err := pool.Get()
	defer func() {
		pool.Return(conn)
		ch <- gatheredStats
	}()

	if err != nil {
{'fraction_non_alphanumeric': 0.11995049942543975, 'fraction_numerical': 0.01157959869177053, 'mean_word_length': 3.8412494651262303, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '2668542', 'n_tokens_mistral': 4126, 'n_tokens_neox': 3809, 'n_words': 1149}