<commit_before>package herd

import (
	"fmt"
	"github.com/Symantec/Dominator/lib/constants"
	"github.com/Symantec/Dominator/lib/hash"
	subproto "github.com/Symantec/Dominator/proto/sub"
	"net/rpc"
	"strings"
	"time"
)

func (sub *Sub) tryMakeBusy() bool {
	sub.busyMutex.Lock()
	defer sub.busyMutex.Unlock()
	if sub.busy {
		return false
	}
	sub.busy = true
	return true
}

func (sub *Sub) makeUnbusy() {
	sub.busyMutex.Lock()
	defer sub.busyMutex.Unlock()
	sub.busy = false
}

func (sub *Sub) connectAndPoll() {
	sub.status = statusConnecting
	hostname := strings.SplitN(sub.hostname, "*", 2)[0]
	var err error
	connection, err := rpc.DialHTTP("tcp",
		fmt.Sprintf("%s:%d", hostname, constants.SubPortNumber))
	if err != nil {
		sub.status = statusFailedToConnect
		return
	}
	defer connection.Close()
	sub.status = statusWaitingToPoll
	sub.herd.pollSemaphore <- true
	sub.status = statusPolling
	sub.poll(connection)
	<-sub.herd.pollSemaphore
}

func (sub *Sub) poll(connection *rpc.Client) {
	var request subproto.PollRequest
	request.HaveGeneration = sub.generationCount
	var reply subproto.PollResponse
	err := connection.Call("Subd.Poll", request, &reply)
	if err != nil {
		sub.status = statusFailedToPoll
		fmt.Printf("Error calling\t%s\n", err)
		return
	}
	sub.lastSuccessfulPoll = time.Now()
	if reply.GenerationCount == 0 {
		sub.fileSystem = nil
		sub.generationCount = 0
	}
	if fs := reply.FileSystem; fs != nil {
		fs.RebuildPointers()
		sub.fileSystem = fs
		sub.generationCount = reply.GenerationCount
		// TODO(rgooch): Remove debugging output.
		fmt.Printf("Polled: %s, GenerationCount=%d\n",
			sub.hostname, reply.GenerationCount)
	}
	if sub.fileSystem == nil {
		sub.status = statusSubNotReady
		return
	}
	if reply.FetchInProgress {
		sub.status = statusFetching
		return
	}
	if reply.UpdateInProgress {
		sub.status = statusUpdating
		return
	}
	if sub.generationCountAtChangeStart == sub.generationCount {
		sub.status = statusWaitingForNextPoll
		return
	}
	if idle, status := sub.fetchMissingObjects(connection,
		sub.requiredImage); !idle {
		sub.status = status
		return
	}
	if idle, status := sub.sendUpdate(connection); !idle {
		sub.status = status
		return
	}
	if idle, status := sub.fetchMissingObjects(connection,
		sub.plannedImage); !idle {
		sub.status = status
		return
	}
	sub.status = statusSynced
}

// Returns true if all required objects are available.
func (sub *Sub) fetchMissingObjects(connection *rpc.Client, imageName string) (
	bool, uint) {
	image := sub.herd.getImage(imageName)
	if image == nil {
		return false, statusImageNotReady
	}
	missingObjects := make(map[hash.Hash]bool)
	for _, inode := range image.FileSystem.RegularInodeTable {
		if inode.Size > 0 {
			missingObjects[inode.Hash] = true
		}
	}
	for _, hash := range sub.fileSystem.ObjectCache {
		delete(missingObjects, hash)
	}
	for _, inode := range sub.fileSystem.RegularInodeTable {
		if inode.Size > 0 {
			delete(missingObjects, inode.Hash)
		}
	}
	if len(missingObjects) < 1 {
		return true, statusSynced
	}
	// TODO(rgooch): Remove debugging output.
	fmt.Printf("Objects needing to be fetched: %d\n", len(missingObjects))
	var request subproto.FetchRequest
	var reply subproto.FetchResponse
	request.ServerAddress = sub.herd.imageServerAddress
	for hash, _ := range missingObjects {
		request.Hashes = append(request.Hashes, hash)
	}
	err := connection.Call("Subd.Fetch", request, &reply)
	if err != nil {
		fmt.Printf("Error calling\t%s\n", err)
		return false, statusFailedToFetch<commit_msg>Switch to logger in Dominator sub handling code.<commit_after>package herd

import (
	"fmt"
	"github.com/Symantec/Dominator/lib/constants"
	"github.com/Symantec/Dominator/lib/hash"
	subproto "github.com/Symantec/Dominator/proto/sub"
	"net/rpc"
	"strings"
	"time"
)

func (sub *Sub) tryMakeBusy() bool {
	sub.busyMutex.Lock()
	defer sub.busyMutex.Unlock()
	if sub.busy {
		return false
	}
	sub.busy = true
	return true
}

func (sub *Sub) makeUnbusy() {
	sub.busyMutex.Lock()
	defer sub.busyMutex.Unlock()
	sub.busy = false
}

func (sub *Sub) connectAndPoll() {
	sub.status = statusConnecting
	hostname := strings.SplitN(sub.hostname, "*", 2)[0]
	var err error
	connection, err := rpc.DialHTTP("tcp",
		fmt.Sprintf("%s:%d", hostname, constants.SubPortNumber))
	if err != nil {
		sub.status = statusFailedToConnect
		return
	}
	defer connection.Close()
	sub.status = statusWaitingToPoll
	sub.herd.pollSemaphore <- true
	sub.status = statusPolling
	sub.poll(connection)
	<-sub.herd.pollSemaphore
}

func (sub *Sub) poll(connection *rpc.Client) {
	var request subproto.PollRequest
	request.HaveGeneration = sub.generationCount
	var reply subproto.PollResponse
	err := connection.Call("Subd.Poll", request, &reply)
	logger := sub.herd.logger
	if err != nil {
		sub.status = statusFailedToPoll
		logger.Printf("Error calling %s.Poll()\t%s\n", sub.hostname, err)
		return
	}
	sub.lastSuccessfulPoll = time.Now()
	if reply.GenerationCount == 0 {
		sub.fileSystem = nil
		sub.generationCount = 0
	}
	if fs := reply.FileSystem; fs != nil {
		fs.RebuildPointers()
		sub.fileSystem = fs
		sub.generationCount = reply.GenerationCount
		// TODO(rgooch): Remove debugging output.
		fmt.Printf("Polled: %s, GenerationCount=%d\n",
			sub.hostname, reply.GenerationCount)
	}
	if sub.fileSystem == nil {
		sub.status = statusSubNotReady
		return
	}
	if reply.FetchInProgress {
		sub.status = statusFetching
		return
	}
	if reply.UpdateInProgress {
		sub.status = statusUpdating
		return
	}
	if sub.generationCountAtChangeStart == sub.generationCount {
		sub.status = statusWaitingForNextPoll
		return
	}
	if idle, status := sub.fetchMissingObjects(connection,
		sub.requiredImage); !idle {
		sub.status = status
		return
	}
	if idle, status := sub.sendUpdate(connection); !idle {
		sub.status = status
		return
	}
	if idle, status := sub.fetchMissingObjects(connection,
		sub.plannedImage); !idle {
		sub.status = status
		return
	}
	sub.status = statusSynced
}

// Returns true if all required objects are available.
func (sub *Sub) fetchMissingObjects(connection *rpc.Client, imageName string) (
	bool, uint) {
	image := sub.herd.getImage(imageName)
	if image == nil {
		return false, statusImageNotReady
	}
	missingObjects := make(map[hash.Hash]bool)
	for _, inode := range image.FileSystem.RegularInodeTable {
		if inode.Size > 0 {
			missingObjects[inode.Hash] = true
		}
	}
	for _, hash := range sub.fileSystem.ObjectCache {
		delete(missingObjects, hash)
	}
	for _, inode := range sub.fileSystem.RegularInodeTable {
		if inode.Size > 0 {
			delete(missingObjects, inode.Hash)
		}
	}
	if len(missingObjects) < 1 {
		return true, statusSynced
	}
	logger := sub.herd.logger
	logger.Printf("Calling %s.Fetch() for: %d objects\n",
		sub.hostname, len(missingObjects))
	var request subproto.FetchRequest
	var reply subproto.FetchResponse
	request.ServerAddress = sub.herd.imageServerAddress
	for hash, _ := range missingObjects {
		request.Hashes = append(request.Hashes, hash)
	}
	err := connection.Call("Subd.Fetch", request, &reply)
	if err != nil {
		logger.Printf("Error calling %s.Fetch()\t%s\n", sub.hostname, err)
		return false, statusFailedToFetch
{'fraction_non_alphanumeric': 0.12396925227113906, 'fraction_numerical': 0.0019566736547868622, 'mean_word_length': 4.208151382823872, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, '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': '4897977', 'n_tokens_mistral': 2606, 'n_tokens_neox': 2506, 'n_words': 550}