// registry.go
//
// The Registry provides a 'threadsafe' interface to various global information stores.
//
// The registry dispatch thread is forbidden from performing any work that is likely to block.
// Result channels must be buffered with enough space for the full set of results.

package main

import (
	o "orchestra"
	"sort"
	"time"
	"container/list"
)

// Request Types
const (
	requestAddClient = iota
	requestGetClient
	requestDeleteClient
	requestSyncClients

	requestAddJob
	requestGetJob
	requestAddJobResult
	requestGetJobResult
	requestGetJobResultNames
	requestDisqualifyPlayer
	requestReviewJobStatus

	requestWriteJobUpdate
	requestWriteJobAll

	requestQueueSize		= 10

	jobLingerTime			= int64(30e9)
)

type registryRequest struct {
	operation		int
	id			uint64
	hostname		string
	hostlist		[]string
	job			*JobRequest
	tresp			*TaskResponse
	responseChannel		chan *registryResponse
}

type registryResponse struct {
	success			bool
	info			*ClientInfo
	tresp			*TaskResponse
	names			[]string
	jobs			[]*JobRequest
}

var (
	chanRegistryRequest	= make(chan *registryRequest, requestQueueSize)
 	clientList 		= make(map[string]*ClientInfo)
	jobRegister 		= make(map[uint64]*JobRequest)
	expiryChan		<-chan int64
	expiryJobid 		uint64
	expiryList		*list.List

	expiryLoopFudge 	int64 = 10e6; /* 10 ms should be enough fudgefactor */
)

func init() {
	expiryList = list.New()
}

func regInternalAdd(hostname string) {
	o.Warn("Registry: New Host \"%s\"", hostname)
	clientList[hostname] = NewClientInfo()
	// do this initialisation here since it'll help unmask sequencing errors
	clientList[hostname].pendingTasks = make(map[uint64]*TaskRequest)
	clientList[hostname].Player = hostname
}

func regInternalDel(hostname string) {
	o.Warn("Registry: Deleting Host \"%s\"", hostname)
	/* remove it from the registry */
	clientList[hostname] = nil, false
}

func regInternalExpireJob(jobid uint64) {
	job, exists := jobRegister[jobid]
	if exists {
		if job.State.Finished() {
			jobRegister[jobid] = nil, false
		} else {
			o.Assert("Tried to expire incomplete job.")
		}
	}
}

func regInternalFindNextExpiry() {
	if expiryChan != nil {
		o.Assert("Attempted to Find Next Expiry avenue with expiry timer active.")
	}
	// if there's nothing to expire, do nothing.
	if expiryList.Len() == 0 {
		return
	}

	for expiryChan == nil && expiryList.Len() > 0 {
		jobif := expiryList.Remove(expiryList.Front())
		req, ok := jobif.(*JobRequest)
		if !ok {
			o.Assert("item in expiryList not a *JobRequest")
		}
		if (time.Nanoseconds() + expiryLoopFudge) > req.expirytime {
			regInternalExpireJob(req.Id)
		} else {
			expiryChan = time.After(req.expirytime - time.Nanoseconds())
			expiryJobid = req.Id
		}
	}
}

func regInternalMarkJobForExpiry(job *JobRequest) {
	job.expirytime = time.Nanoseconds() + jobLingerTime
	expiryList.PushBack(job)
	// if there is no job pending expiry, feed it into the delay loop
	if expiryChan == nil {
		regInternalFindNextExpiry()
	}
}

var registryHandlers = map[int] func(*registryRequest, *registryResponse) {
requestAddClient:	regintAddClient,
requestGetClient:	regintGetClient,
requestDeleteClient:	regintDeleteClient,
requestSyncClients:	regintSyncClients,
requestAddJob:		regintAddJob,
requestGetJob:		regintGetJob,
requestAddJobResult:	regintAddJobResult,
requestGetJobResult:	regintGetJobResult,
requestGetJobResultNames:	regintGetJobResultNames,
requestDisqualifyPlayer:	regintDisqualifyPlayer,
requestReviewJobStatus:	regintReviewJobStatus,
requestWriteJobUpdate:	regintWriteJobUpdate,
requestWriteJobAll:	regintWriteJobAll,
}

func manageRegistry() {
	for {
		select {
		case req := <-chanRegistryRequest:
			resp := new(registryResponse)
			// by default, we failed.
			resp.success = false
			// find the operation
			handler, exists := registryHandlers[req.operation]
			if exists {
				handler(req, resp)
			}
			if req.responseChannel != nil {
				req.responseChannel <- resp
			}
		case <-expiryChan:
			o.Debug("job%d: Expiring Job Record", expiryJobid)
			regInternalExpireJob(expiryJobid)
			expiryChan = nil
			regInternalFindNextExpiry()
		}
	}
}

func StartRegistry() {
	go manageRegistry()
}

func newRequest(wants_response bool) (req *registryRequest) {
	req = new(registryRequest)
	if wants_response {
		req.responseChannel = make(chan *registryResponse, 1)
	}

	return req
}
	
func ClientAdd(hostname string) (success bool) {
	r := newRequest(true)
	r.operation = requestAddClient
	r.hostname = hostname
	chanRegistryRequest <- r
	resp := <- r.responseChannel
	
	return resp.success
}

func regintAddClient(req *registryRequest, resp *registryResponse) {
	_, exists := clientList[req.hostname]
	if exists {
		resp.success = false
	} else {
		regInternalAdd(req.hostname)
		resp.success = true
	}
}

func ClientDelete(hostname string) (success bool) {
	r := newRequest(true)
	r.operation = requestDeleteClient
	r.hostname = hostname
	chanRegistryRequest <- r
	resp := <- r.responseChannel
	
	return resp.success
}

func regintDeleteClient(req *registryRequest, resp *registryResponse) {
	_, exists := clientList[req.hostname]
	if exists {
		resp.success = true
		regInternalDel(req.hostname)
	} else {
		resp.success = false
	}
}

func ClientGet(hostname string) (info *ClientInfo) {
	r := newRequest(true)
	r.operation = requestGetClient
	r.hostname = hostname
	chanRegistryRequest <- r
	resp := <- r.responseChannel
	if resp.success {
		return resp.info
	}
	return nil
}

func regintGetClient(req *registryRequest, resp *registryResponse) {
	clinfo, exists := clientList[req.hostname]
	if exists {
		resp.success = true
		resp.info = clinfo
	} else {
		resp.success = false
	}
}

func ClientUpdateKnown(hostnames []string) {
	/* this is an asynchronous, we feed it into the registry 
	 * and it'll look after itself.
	*/
	r := newRequest(false)
	r.operation = requestSyncClients
	r.hostlist = hostnames
	chanRegistryRequest <- r
}

func regintSyncClients(req *registryRequest, resp *registryResponse) {
	// we need to make sure the registered clients matches the
	// hostlist we're given.
	//
	// First, we transform the array into a map
	newhosts := make(map[string]bool)
	for k,_ := range req.hostlist {
		newhosts[req.hostlist[k]] = true
	}
	// now, scan the current list, checking to see if they exist.
	// Remove them from the newhosts map if they do exist.
	for k,_ := range clientList {
		_, exists := newhosts[k]
		if exists {
			// remove it from the newhosts map
			newhosts[k] = false, false
		} else {
			regInternalDel(k)
		}
	}
	// now that we're finished, we should only have new clients in
	// the newhosts list left.
	for k,_ := range newhosts {
		regInternalAdd(k)
	}
	// and we're done.
}

// Add a Job to the registry.  Return true if successful, returns
// false if the job is lacking critical information (such as a JobId)
// and can't be registered.
func JobAdd(job *JobRequest) bool {
	rr := newRequest(true)
	rr.operation = requestAddJob
	rr.job = job

	chanRegistryRequest <- rr
	resp := <- rr.responseChannel 
	return resp.success
}

func regintAddJob(req *registryRequest, resp *registryResponse) {
	if nil == req.job {
		return
	}
	// ensure that the players are sorted!
	sort.Strings(req.job.Players)
	// update the state
	req.job.updateState()
	// and register the job
	_, overwrite := jobRegister[req.job.Id]
	if !overwrite {
		jobRegister[req.job.Id] = req.job
		// force a queue update.
		req.job.UpdateInSpool()
		if req.job.State.Finished() {
			regInternalMarkJobForExpiry(req.job)
		}
		resp.success = true
	}
}

// Get a Job from the registry.  Returns the job if successful,
// returns nil if the job couldn't be found.
func JobGet(id uint64) *JobRequest {
	rr := newRequest(true)
	rr.operation = requestGetJob
	rr.id = id

	chanRegistryRequest <- rr
	resp := <- rr.responseChannel
	if resp.jobs == nil {
		return nil
	}
	return resp.jobs[0]
}

func regintGetJob(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.id]
	resp.success = exists
	if exists {
		resp.jobs = make([]*JobRequest, 1)
		resp.jobs[0] = job
	} else {
		o.Warn("Received Request for job%d which is not in memory", req.id)
		go regintGetJobDeferred(req.id, req.responseChannel)
		// mask out the responseChannel so the deferred handler can use it.
		req.responseChannel = nil
	}
}

func regintGetJobDeferred(jobid uint64, responseChannel chan<- *registryResponse) {
	resp := new(registryResponse)
	resp.success = false
	defer func (resp *registryResponse, rChan chan<- *registryResponse) {
		rChan <- resp;
	}(resp, responseChannel)

	req, err := LoadFromFinished(jobid)
	if err != nil {
		o.Warn("Couldn't load job%d from disk.  Doesn't exist?", jobid)
		return
	}
	// fix up the state, and stuff it back into the system
	RestoreJobState(req)
	resp.jobs = make([]*JobRequest, 1)
	resp.jobs[0] = req
	resp.success = true
}

// Attach a result to a Job in the Registry
//
// This exists in order to prevent nasty concurrency problems
// when trying to put results back onto the job.  Reading a job is far
// less of a problem than writing to it.
func JobAddResult(playername string, task *TaskResponse) bool {
	rr := newRequest(true)
	rr.operation = requestAddJobResult
	rr.tresp = task
	rr.hostname = playername
	chanRegistryRequest <- rr
	resp := <- rr.responseChannel
	return resp.success
}

func regintAddJobResult(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.tresp.id]
	resp.success = exists
	if exists {
		job.Results[req.hostname] = req.tresp
		// force a queue update.
		job.UpdateInSpool()
	}
}

// Get a result from the registry
func JobGetResult(id uint64, playername string) (tresp *TaskResponse) {
	rr := newRequest(true)
	rr.operation = requestGetJobResult
	rr.id = id
	rr.hostname = playername
	chanRegistryRequest <- rr
	resp := <- rr.responseChannel
	return resp.tresp
}

func regintGetJobResult(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.id]
	if exists {
		result, exists := job.Results[req.hostname]
		resp.success = exists
		if exists {
			resp.tresp = result
		}
	} else {
		resp.success = false
	}
}

// Get a list of names we have results for against a given job.
func JobGetResultNames(id uint64) (names []string) {
	rr := newRequest(true)
	rr.operation = requestGetJobResultNames
	rr.id = id

	chanRegistryRequest <- rr
	resp := <- rr.responseChannel 
	return resp.names
}

func regintGetJobResultNames(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.id]
	resp.success = exists
	if exists {
		resp.names = make([]string, len(job.Results))
		idx := 0
		for k, _ := range job.Results {
			resp.names[idx] = k
			idx++
		}
	}
}

//  Disqualify a player from servicing a job
func JobDisqualifyPlayer(id uint64, playername string) bool {
	rr := newRequest(true)
	rr.operation = requestDisqualifyPlayer
	rr.id = id
	rr.hostname = playername

	chanRegistryRequest <- rr
	resp := <- rr.responseChannel

	return resp.success
}

func regintDisqualifyPlayer(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.id]
	if exists {
		idx := sort.Search(len(job.Players), func(idx int) bool { return job.Players[idx] >= req.hostname })
		if (job.Players[idx] == req.hostname) {
			resp.success = true
			newplayers := make([]string, len(job.Players)-1)
			copy(newplayers[0:idx], job.Players[0:idx])
			copy(newplayers[idx:len(job.Players)-1], job.Players[idx+1:len(job.Players)])
			job.Players = newplayers
			job.updateState()
			// force a queue update.
			job.UpdateInSpool()
		} else {
			resp.success = false
		}
	} else {
		resp.success = false
	}
}

func JobReviewState(id uint64) bool {
	rr := newRequest(true)
	rr.operation = requestReviewJobStatus
	rr.id = id

	chanRegistryRequest <- rr
	resp := <- rr.responseChannel

	return resp.success
}

func regintReviewJobStatus(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.id]
	resp.success = exists
	if exists {
		job.updateState()
		// force a queue update.
		job.UpdateInSpool()
	}
}

func JobWriteUpdate(id uint64) {
	rr := newRequest(false)
	rr.operation = requestWriteJobUpdate
	rr.id = id
	chanRegistryRequest <- rr
}

func regintWriteJobUpdate(req *registryRequest, resp *registryResponse) {
	job, exists := jobRegister[req.id]
	resp.success = exists
	if exists {
		job.UpdateInSpool()
	}
}

func JobWriteAll() bool {
	rr := newRequest(true)
	rr.operation = requestWriteJobAll

	chanRegistryRequest <- rr
	resp := <-rr.responseChannel

	return resp.success
}

func regintWriteJobAll(req *registryRequest, resp *registryResponse) {
	for _, job := range jobRegister {
		job.UpdateInSpool()
	}
	resp.success = true
}

// Ugh.
func (job *JobRequest) updateState() {
	if job.Results == nil {
		o.Assert("job.Results nil for jobid %d", job.Id)
		return
	}
	was_finished := job.State.Finished()
	switch job.Scope {
	case SCOPE_ONEOF:
		// look for a success (any success) in the responses
		var success bool = false
		for host, res := range job.Results {
			if res == nil {
				o.Debug("nil result for %s?", host)
				continue
			}
			if res.State == RESP_FINISHED {
				success = true
				break
			}
		}
		// update the job state based upon these findings
		if success {
			job.State = JOB_SUCCESSFUL
		} else {
			if len(job.Players) < 1 {
				job.State = JOB_FAILED
			} else {
				job.State = JOB_PENDING
			}
		}
	case SCOPE_ALLOF:
		var success int = 0
		var failed  int = 0
		
		for pidx := range job.Players {
			p := job.Players[pidx]
			resp, exists := job.Results[p]
			if exists {
				if resp.DidFail() {
					failed++
				} else if resp.State == RESP_FINISHED {
					success++
				}
			}
		}
		if (success + failed) < len(job.Players) {
			job.State = JOB_PENDING
		} else if success == len(job.Players) {
			job.State = JOB_SUCCESSFUL
		} else if failed == len(job.Players) {
			job.State = JOB_FAILED
		} else {
			job.State = JOB_FAILED_PARTIAL
		}
	}
	if !was_finished && job.State.Finished() {
		o.Debug("job%d: Finished - Setting Expiry Time", job.Id)
		regInternalMarkJobForExpiry(job)
	}
}