<commit_before>package core

import (
	"errors"
	"fmt"
	"github.com/MG-RAST/AWE/lib/conf"
	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/AWE/lib/user"
	"os"
	"time"
)

type ProxyMgr struct {
	CQMgr
}

func (qm *ProxyMgr) Lock()    {}
func (qm *ProxyMgr) Unlock()  {}
func (qm *ProxyMgr) RLock()   {}
func (qm *ProxyMgr) RUnlock() {}

func NewProxyMgr() *ProxyMgr {
	return &ProxyMgr{
		CQMgr: CQMgr{
			clientMap:    *NewClientMap(),
			workQueue:    NewWorkQueue(),
			suspendQueue: false,
			coReq:        make(chan CoReq),
			//coAck:        make(chan CoAck),
			feedback: make(chan Notice),
			coSem:    make(chan int, 1), //non-blocking buffered channel
		},
	}
}

func (qm *ProxyMgr) TaskHandle() {
	return
}

func (qm *ProxyMgr) ClientHandle() {
	for {
		select {
		case coReq := <-qm.coReq:
			logger.Debug(2, fmt.Sprintf("proxymgr: workunit checkout request received, Req=%v\n", coReq))
			var ack CoAck
			if qm.suspendQueue {
				// queue is suspended, return suspend error
				ack = CoAck{workunits: nil, err: errors.New(e.QueueSuspend)}
			} else {
				works, err := qm.popWorks(coReq)
				ack = CoAck{workunits: works, err: err}
			}
			//qm.coAck <- ack
			coReq.response <- ack
		case notice := <-qm.feedback:
			logger.Debug(2, fmt.Sprintf("proxymgr: workunit feedback received, workid=%s, status=%s, clientid=%s\n", notice.WorkId, notice.Status, notice.ClientId))
			if err := qm.handleWorkStatusChange(notice); err != nil {
				logger.Error("handleWorkStatusChange(): " + err.Error())
			}
		}
	}
}

func (qm *ProxyMgr) NoticeHandle() {
	//TODO copy code from ClientHandle, and/or reuse server code
	return
}

func (qm *ProxyMgr) SuspendQueue() {
	return
}

func (qm *ProxyMgr) ResumeQueue() {
	return
}

func (qm *ProxyMgr) QueueStatus() string {
	return ""
}

func (qm *ProxyMgr) GetQueue(name string) interface{} {
	return nil
}

func (qm *ProxyMgr) GetJsonStatus() (status map[string]map[string]int) {
	return status
}

func (qm *ProxyMgr) GetTextStatus() string {
	return ""
}

//---end of mgr methods

// workunit methods

//handle feedback from a client about the execution of a workunit
func (qm *ProxyMgr) handleWorkStatusChange(notice Notice) (err error) {
	//relay the notice to the server
	perf := new(WorkPerf)
	workid := notice.WorkId
	clientid := notice.ClientId
	if client, ok := qm.GetClient(clientid, true); ok {
		//delete(client.Current_work, workid)
		client.LockNamed("ProxyMgr/handleWorkStatusChange A")
		client.Current_work_delete(workid, false)
		if client.Current_work_length(false) == 0 {
			client.Status = CLIENT_STAT_ACTIVE_IDLE
		}
		qm.AddClient(client, true)
		client.Unlock()
	}
	if work, ok := qm.workQueue.Get(workid); ok {<commit_msg>minor debug change<commit_after>package core

import (
	"errors"
	"fmt"
	"github.com/MG-RAST/AWE/lib/conf"
	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/AWE/lib/user"
	"os"
	"time"
)

type ProxyMgr struct {
	CQMgr
}

func (qm *ProxyMgr) Lock()    {}
func (qm *ProxyMgr) Unlock()  {}
func (qm *ProxyMgr) RLock()   {}
func (qm *ProxyMgr) RUnlock() {}

func NewProxyMgr() *ProxyMgr {
	return &ProxyMgr{
		CQMgr: CQMgr{
			clientMap:    *NewClientMap(),
			workQueue:    NewWorkQueue(),
			suspendQueue: false,
			coReq:        make(chan CoReq),
			//coAck:        make(chan CoAck),
			feedback: make(chan Notice),
			coSem:    make(chan int, 1), //non-blocking buffered channel
		},
	}
}

func (qm *ProxyMgr) TaskHandle() {
	return
}

func (qm *ProxyMgr) ClientHandle() {
	for {
		select {
		case coReq := <-qm.coReq:
			logger.Debug(2, fmt.Sprintf("proxymgr: workunit checkout request received, Req=%v\n", coReq))
			var ack CoAck
			if qm.suspendQueue {
				// queue is suspended, return suspend error
				ack = CoAck{workunits: nil, err: errors.New(e.QueueSuspend)}
			} else {
				works, err := qm.popWorks(coReq)
				ack = CoAck{workunits: works, err: err}
			}
			//qm.coAck <- ack
			coReq.response <- ack
		case notice := <-qm.feedback:
			logger.Debug(2, fmt.Sprintf("proxymgr: workunit feedback received, workid=%s, status=%s, clientid=%s\n", notice.WorkId, notice.Status, notice.ClientId))
			if err := qm.handleWorkStatusChange(notice); err != nil {
				logger.Error("handleWorkStatusChange(): " + err.Error())
			}
		}
	}
}

func (qm *ProxyMgr) NoticeHandle() {
	//TODO copy code from ClientHandle, and/or reuse server code
	return
}

func (qm *ProxyMgr) SuspendQueue() {
	return
}

func (qm *ProxyMgr) ResumeQueue() {
	return
}

func (qm *ProxyMgr) QueueStatus() string {
	return ""
}

func (qm *ProxyMgr) GetQueue(name string) interface{} {
	return nil
}

func (qm *ProxyMgr) GetJsonStatus() (status map[string]map[string]int) {
	return status
}

func (qm *ProxyMgr) GetTextStatus() string {
	return ""
}

//---end of mgr methods

// workunit methods

//handle feedback from a client about the execution of a workunit
func (qm *ProxyMgr) handleWorkStatusChange(notice Notice) (err error) {
	//relay the notice to the server
	perf := new(WorkPerf)
	workid := notice.WorkId
	clientid := notice.ClientId
	if client, ok := qm.GetClient(clientid, true); ok {
		//delete(client.Current_work, workid)
		client.LockNamed("ProxyMgr/handleWorkStatusChange A2")
		client.Current_work_delete(workid, false)
		if client.Current_work_length(false) == 0 {
			client.Status = CLIENT_STAT_ACTIVE_IDLE
		}
		qm.AddClient(client, true)
		client.Unlock()
	}
	if work, ok := qm.workQueue.Get(workid); ok {
{'fraction_non_alphanumeric': 0.14531754574811626, 'fraction_numerical': 0.0016146393972012918, 'mean_word_length': 3.7772065124250216, 'pattern_counts': {'":': 0, '<': 11, '<?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': '4782867', 'n_tokens_mistral': 2243, 'n_tokens_neox': 2133, 'n_words': 496}