<commit_before>package main

import (
	"fmt"
	"io"
	"log"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/inominate/apicache"
)

var apiClient apicache.Client

// worker tracking
var activeWorkerCount, workerCount int32
var workCount []int32

type apiReq struct {
	apiReq  *apicache.Request
	apiResp *apicache.Response

	apiErr apicache.APIError

	expires time.Time

	worker   int
	httpCode int
	err      error
	respChan chan apiReq
}

// Channel for sending jobs to workers
var workChan chan apiReq

func APIReq(url string, params map[string]string) (*apicache.Response, error) {
	var errorStr string

	if atomic.LoadInt32(&workerCount) <= 0 {
		panic("No workers!")
	}

	// Build the request
	apireq := apicache.NewRequest(url)
	for k, v := range params {
		switch k {
		case "force":
			if v != "" {
				apireq.Force = true
			}
		default:
			apireq.Set(k, v)
		}
	}

	workerID := "C"
	// Don't send it to a worker if we can just yank it fromm the cache
	apiResp, err := apireq.GetCached()
	// 221's are the bane of my existence.
	if err != nil || apireq.Force || apiResp.Error.ErrorCode == 221 {
		for i := 0; i < conf.Retries; i++ {
			respChan := make(chan apiReq)
			req := apiReq{apiReq: apireq, respChan: respChan}
			workChan <- req

			resp := <-respChan
			close(respChan)

			apiResp = resp.apiResp
			err = resp.err
			workerID = fmt.Sprintf("%d", resp.worker)

			// Attempt to recover from server issues, invalidate flag means we
			// believe this is not a server failure.
			// 418 is the tempban code
			// 500/900 are panic codes
			if err == nil || apiResp.Invalidate || apiResp.HTTPCode == 418 || apiResp.HTTPCode == 500 || apiResp.HTTPCode == 900 {
				break
			}
			time.Sleep(2 * time.Second)
			apireq.Force = true
		}
	}
	var singleDebug bool
	if apiResp.Error.ErrorCode == 221 {
		log.Printf("Failed to recover from error 221.")
		singleDebug = true
	}

	// This is similar to the request log, but knows more about where it came from.
	if debug || singleDebug {
		if apiResp.Error.ErrorCode != 0 {
			errorStr = fmt.Sprintf(" Error %d: %s", apiResp.Error.ErrorCode, apiResp.Error.ErrorText)
		}
		logParams := ""
		var paramVal string
		for k, _ := range params {
			if conf.Logging.CensorLog && strings.ToLower(k) == "vcode" && len(params[k]) > 8 {
				paramVal = params[k][0:8] + "..."
			} else {
				paramVal = params[k]
			}
			logParams = fmt.Sprintf("%s&%s=%s", logParams, k, paramVal)
		}
		if logParams != "" {
			logParams = "?" + logParams[1:]
		}

		debugLog.Printf("w%s: %s%s HTTP: %d Expires: %s%s", workerID, url, logParams, apiResp.HTTPCode, apiResp.Expires.Format("2006-01-02 15:04:05"), errorStr)
	}
	return apiResp, err
}

func worker(reqChan chan apiReq, workerID int) {
	atomic.AddInt32(&workerCount, 1)
	var err error

	var eErr error
	var rErr error

	for req := range reqChan {
		var errStr string

		atomic.AddInt32(&activeWorkerCount, 1)

		// Run both of the error limiters simultaneously rather than in
		// sequence. Still need both before we continue.
		errorLimiter := make(chan error)
		rpsLimiter := make(chan error)
		go func() {
			err := errorRateLimiter.Start(30 * time.Second)
			errorLimiter <- err
		}()
		go func() {
			err := rateLimiter.Start(30 * time.Second)
			rpsLimiter <- err
		}()
		eErr = <-errorLimiter
		rErr = <-rpsLimiter
		err = nil

		// Check the error limiter for timeouts
		if eErr != nil {
			err = eErr
			errStr = "error throttling"

			// If the rate limiter didn't timeout be sure to signal it that we
			// didn't do anything.
			if rErr == nil {
				rateLimiter.Finish(true)
			}
		}<commit_msg>better fix for worker error bug<commit_after>package main

import (
	"fmt"
	"io"
	"log"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/inominate/apicache"
)

var apiClient apicache.Client

// worker tracking
var activeWorkerCount, workerCount int32
var workCount []int32

type apiReq struct {
	apiReq  *apicache.Request
	apiResp *apicache.Response

	apiErr apicache.APIError

	expires time.Time

	worker   int
	httpCode int
	err      error
	respChan chan apiReq
}

// Channel for sending jobs to workers
var workChan chan apiReq

func APIReq(url string, params map[string]string) (*apicache.Response, error) {
	var errorStr string

	if atomic.LoadInt32(&workerCount) <= 0 {
		panic("No workers!")
	}

	// Build the request
	apireq := apicache.NewRequest(url)
	for k, v := range params {
		switch k {
		case "force":
			if v != "" {
				apireq.Force = true
			}
		default:
			apireq.Set(k, v)
		}
	}

	workerID := "C"
	// Don't send it to a worker if we can just yank it fromm the cache
	apiResp, err := apireq.GetCached()
	// 221's are the bane of my existence.
	if err != nil || apireq.Force || apiResp.Error.ErrorCode == 221 {
		for i := 0; i < conf.Retries; i++ {
			respChan := make(chan apiReq)
			req := apiReq{apiReq: apireq, respChan: respChan}
			workChan <- req

			resp := <-respChan
			close(respChan)

			apiResp = resp.apiResp
			err = resp.err
			workerID = fmt.Sprintf("%d", resp.worker)

			// Attempt to recover from server issues, invalidate flag means we
			// believe this is not a server failure.
			// 418 is the tempban code
			// 500/900 are panic codes
			if err == nil || apiResp.Invalidate || apiResp.HTTPCode == 418 || apiResp.HTTPCode == 500 || apiResp.HTTPCode == 900 {
				break
			}
			time.Sleep(2 * time.Second)
			apireq.Force = true
		}
	}
	var singleDebug bool
	if apiResp.Error.ErrorCode == 221 {
		log.Printf("Failed to recover from error 221.")
		singleDebug = true
	}

	// This is similar to the request log, but knows more about where it came from.
	if debug || singleDebug {
		if apiResp.Error.ErrorCode != 0 {
			errorStr = fmt.Sprintf(" Error %d: %s", apiResp.Error.ErrorCode, apiResp.Error.ErrorText)
		}
		logParams := ""
		var paramVal string
		for k, _ := range params {
			if conf.Logging.CensorLog && strings.ToLower(k) == "vcode" && len(params[k]) > 8 {
				paramVal = params[k][0:8] + "..."
			} else {
				paramVal = params[k]
			}
			logParams = fmt.Sprintf("%s&%s=%s", logParams, k, paramVal)
		}
		if logParams != "" {
			logParams = "?" + logParams[1:]
		}

		debugLog.Printf("w%s: %s%s HTTP: %d Expires: %s%s", workerID, url, logParams, apiResp.HTTPCode, apiResp.Expires.Format("2006-01-02 15:04:05"), errorStr)
	}
	return apiResp, err
}

func worker(reqChan chan apiReq, workerID int) {
	atomic.AddInt32(&workerCount, 1)

	for req := range reqChan {
		var err, eErr, rErr error
		var errStr string

		atomic.AddInt32(&activeWorkerCount, 1)

		// Run both of the error limiters simultaneously rather than in
		// sequence. Still need both before we continue.
		errorLimiter := make(chan error)
		rpsLimiter := make(chan error)
		go func() {
			err := errorRateLimiter.Start(30 * time.Second)
			errorLimiter <- err
		}()
		go func() {
			err := rateLimiter.Start(30 * time.Second)
			rpsLimiter <- err
		}()
		eErr = <-errorLimiter
		rErr = <-rpsLimiter

		// Check the error limiter for timeouts
		if eErr != nil {
			err = eErr
			errStr = "error throttling"

			// If the rate limiter didn't timeout be sure to signal it that we
			// didn't do anything.
			if rErr == nil {
				rateLimiter.Finish(true)
			}
		}
{'fraction_non_alphanumeric': 0.12489557226399331, 'fraction_numerical': 0.01893622946254525, 'mean_word_length': 3.106918238993711, 'pattern_counts': {'":': 2, '<': 19, '<?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': '164281', 'n_tokens_mistral': 2955, 'n_tokens_neox': 2517, 'n_words': 755}