<commit_before>package worker

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/MG-RAST/AWE/lib/conf"
	"github.com/MG-RAST/AWE/lib/core"
	e "github.com/MG-RAST/AWE/lib/errors"
	"github.com/MG-RAST/AWE/lib/logger"
	"github.com/MG-RAST/AWE/lib/logger/event"
	"github.com/MG-RAST/golib/httpclient"
	"io/ioutil"
	"os"
	"strings"
	"time"
)

type WorkResponse struct {
	Code int            `bson:"status" json:"status"`
	Data *core.Workunit `bson:"data" json:"data"`
	Errs []string       `bson:"error" json:"error"`
}

type TokenResponse struct {
	Code int      `bson:"status" json:"status"`
	Data string   `bson:"data" json:"data"`
	Errs []string `bson:"error" json:"error"`
}

func workStealer(control chan int) {
	//fmt.Printf("workStealer launched, client=%s\n", core.Self.Id)
	logger.Debug(0, fmt.Sprintf("workStealer launched, client=%s\n", core.Self.Id))
	defer fmt.Printf("workStealer exiting...\n")
	retry := 0
	for {
		if core.Service == "proxy" {
			<-core.ProxyWorkChan
		}
		wu, err := CheckoutWorkunitRemote()
		if err != nil {
			if err.Error() == e.QueueEmpty || err.Error() == e.NoEligibleWorkunitFound {
				//normal, do nothing
			} else if err.Error() == e.ClientNotFound {
				//server may be restarted, waiting for the hearbeater goroutine to try re-register
				ReRegisterWithSelf(conf.SERVER_URL)
			} else if err.Error() == e.ClientSuspended {
				logger.Error("client suspended, waiting for repair or resume request...")
				//TODO: send out email notice that this client has problem and been suspended
				time.Sleep(2 * time.Minute)
			} else if err.Error() == e.ClientDeleted {
				fmt.Printf("client deleted, exiting...\n")
				os.Exit(1) // TODO is there a better way of exiting ? E.g. in regard of the logger who wants to flush....
			} else {
				//something is wrong, server may be down

				logger.Error(fmt.Sprintf("error in checking out workunit: %s, retry=%d", err.Error(), retry))
				retry += 1
			}
			//if retry == 12 {
			//	fmt.Printf("failed to checkout workunits for 12 times, exiting...\n")
			//	logger.Error("failed to checkout workunits for 12 times, exiting...")
			//	os.Exit(1) // TODO fix !
			//}
			if core.Service != "proxy" { //proxy: event driven, client: timer driven
				if retry <= 10 {
					time.Sleep(10 * time.Second)
				} else {
					time.Sleep(30 * time.Second)
				}
			}
			continue
		} else {
			retry = 0
		}
		logger.Debug(1, "workStealer: checked out workunit, id="+wu.Id)
		//log event about work checktout (WC)
		logger.Event(event.WORK_CHECKOUT, "workid="+wu.Id)
		core.Self.Total_checkout += 1
		core.Self.Current_work[wu.Id] = true
		workmap[wu.Id] = ID_WORKSTEALER

		//hand the work to the next step handler: dataMover
		workstat := core.NewWorkPerf(wu.Id)
		workstat.Checkout = time.Now().Unix()
		rawWork := &mediumwork{
			workunit: wu,
			perfstat: workstat,
		}
		fromStealer <- rawWork // sends to dataMover

		//if worker overlap is inhibited, wait until deliverer finishes processing the workunit
		if conf.WORKER_OVERLAP == false && core.Service != "proxy" {
			chanPermit <- true
		}
	}
	control <- ID_WORKSTEALER //we are ending
}

func CheckoutWorkunitRemote() (workunit *core.Workunit, err error) {
	response := new(WorkResponse)
	targeturl := fmt.Sprintf("%s/work?client=%s", conf.SERVER_URL, core.Self.Id)
	//res, err := http.Get(fmt.Sprintf("%s/work?client=%s", conf.SERVER_URL, core.Self.Id))
	var headers httpclient.Header
	if conf.CLIENT_GROUP_TOKEN != "" {
		headers = httpclient.Header{
			"Authorization": []string{"CG_TOKEN " + conf.CLIENT_GROUP_TOKEN},
		}
	}
	res, err := httpclient.Get(targeturl, headers, nil, nil)
	logger.Debug(3, fmt.Sprintf("client %s sent a checkout request to %s", core.Self.Id, conf.SERVER_URL))
	if err != nil {
		fmt.Printf("err=%s\n", err.Error())
		return<commit_msg>No timeout desired when checking out workunit.<commit_after>package worker

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/MG-RAST/AWE/lib/conf"
	"github.com/MG-RAST/AWE/lib/core"
	e "github.com/MG-RAST/AWE/lib/errors"
	"github.com/MG-RAST/AWE/lib/logger"
	"github.com/MG-RAST/AWE/lib/logger/event"
	"github.com/MG-RAST/golib/httpclient"
	"io/ioutil"
	"os"
	"strings"
	"time"
)

type WorkResponse struct {
	Code int            `bson:"status" json:"status"`
	Data *core.Workunit `bson:"data" json:"data"`
	Errs []string       `bson:"error" json:"error"`
}

type TokenResponse struct {
	Code int      `bson:"status" json:"status"`
	Data string   `bson:"data" json:"data"`
	Errs []string `bson:"error" json:"error"`
}

func workStealer(control chan int) {
	//fmt.Printf("workStealer launched, client=%s\n", core.Self.Id)
	logger.Debug(0, fmt.Sprintf("workStealer launched, client=%s\n", core.Self.Id))
	defer fmt.Printf("workStealer exiting...\n")
	retry := 0
	for {
		if core.Service == "proxy" {
			<-core.ProxyWorkChan
		}
		wu, err := CheckoutWorkunitRemote()
		if err != nil {
			if err.Error() == e.QueueEmpty || err.Error() == e.NoEligibleWorkunitFound {
				//normal, do nothing
			} else if err.Error() == e.ClientNotFound {
				//server may be restarted, waiting for the hearbeater goroutine to try re-register
				ReRegisterWithSelf(conf.SERVER_URL)
			} else if err.Error() == e.ClientSuspended {
				logger.Error("client suspended, waiting for repair or resume request...")
				//TODO: send out email notice that this client has problem and been suspended
				time.Sleep(2 * time.Minute)
			} else if err.Error() == e.ClientDeleted {
				fmt.Printf("client deleted, exiting...\n")
				os.Exit(1) // TODO is there a better way of exiting ? E.g. in regard of the logger who wants to flush....
			} else {
				//something is wrong, server may be down

				logger.Error(fmt.Sprintf("error in checking out workunit: %s, retry=%d", err.Error(), retry))
				retry += 1
			}
			//if retry == 12 {
			//	fmt.Printf("failed to checkout workunits for 12 times, exiting...\n")
			//	logger.Error("failed to checkout workunits for 12 times, exiting...")
			//	os.Exit(1) // TODO fix !
			//}
			if core.Service != "proxy" { //proxy: event driven, client: timer driven
				if retry <= 10 {
					time.Sleep(10 * time.Second)
				} else {
					time.Sleep(30 * time.Second)
				}
			}
			continue
		} else {
			retry = 0
		}
		logger.Debug(1, "workStealer: checked out workunit, id="+wu.Id)
		//log event about work checktout (WC)
		logger.Event(event.WORK_CHECKOUT, "workid="+wu.Id)
		core.Self.Total_checkout += 1
		core.Self.Current_work[wu.Id] = true
		workmap[wu.Id] = ID_WORKSTEALER

		//hand the work to the next step handler: dataMover
		workstat := core.NewWorkPerf(wu.Id)
		workstat.Checkout = time.Now().Unix()
		rawWork := &mediumwork{
			workunit: wu,
			perfstat: workstat,
		}
		fromStealer <- rawWork // sends to dataMover

		//if worker overlap is inhibited, wait until deliverer finishes processing the workunit
		if conf.WORKER_OVERLAP == false && core.Service != "proxy" {
			chanPermit <- true
		}
	}
	control <- ID_WORKSTEALER //we are ending
}

func CheckoutWorkunitRemote() (workunit *core.Workunit, err error) {
	response := new(WorkResponse)
	targeturl := fmt.Sprintf("%s/work?client=%s", conf.SERVER_URL, core.Self.Id)
	//res, err := http.Get(fmt.Sprintf("%s/work?client=%s", conf.SERVER_URL, core.Self.Id))
	var headers httpclient.Header
	if conf.CLIENT_GROUP_TOKEN != "" {
		headers = httpclient.Header{
			"Authorization": []string{"CG_TOKEN " + conf.CLIENT_GROUP_TOKEN},
		}
	}
	res, err := httpclient.DoTimeout("GET", targeturl, headers, nil, nil, time.Second*0)
	logger.Debug(3, fmt.Sprintf("client %s sent a checkout request to %s", core.Self.Id, conf.SERVER_URL))
	if err != nil {
		fmt.Printf("err=%s\n", err.Error())
		return
{'fraction_non_alphanumeric': 0.15299334811529933, 'fraction_numerical': 0.005869310029998695, 'mean_word_length': 3.4607329842931938, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 3, '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': '7117569', 'n_tokens_mistral': 2983, 'n_tokens_neox': 2688, 'n_words': 675}