<commit_before>package master

import (
	"encoding/json"
	"log"
	"math"
	"math/rand"
	"net"
	"net/http"
	"os"
	"os/signal"
	"time"

	"github.com/aaronang/cong-the-ripper/lib"
	"github.com/aws/aws-sdk-go/service/ec2"
)

type Master struct {
	port             string
	svc              *ec2.EC2 // safe to be used concurrently
	instances        map[string]slave
	jobs             map[int]*job
	jobsChan         chan lib.Job
	heartbeatChan    chan lib.Heartbeat
	statusChan       chan chan statusSummary
	newTasks         []*lib.Task
	scheduledTasks   []*lib.Task
	scheduleTicker   *time.Ticker // channel to instruct the main loop to schedule tasks
	controllerTicker *time.Ticker
	controller       controller
	taskSize         int
	quit             chan bool
}

type slave struct {
	tasks    []*lib.Task
	maxSlots int
	instance *ec2.Instance // can't populate this easily
}

type controller struct {
	dt       time.Duration
	kp       float64
	kd       float64
	ki       float64
	prevErr  float64
	integral float64
}

type statusSummary struct {
	instances map[string]slave // TODO *ec2.Instance probably won't serialise
	jobs      map[int64]*job
}

// Init creates the master object
func Init(port string) Master {
	// set some defaults
	return Master{
		port:             port,
		svc:              nil, // initialised in Run
		instances:        make(map[string]slave),
		jobs:             make(map[int]*job),
		jobsChan:         make(chan lib.Job),
		heartbeatChan:    make(chan lib.Heartbeat),
		statusChan:       make(chan chan statusSummary),
		newTasks:         make([]*lib.Task, 0),
		scheduledTasks:   make([]*lib.Task, 0),
		scheduleTicker:   nil, // initialised in Run
		controllerTicker: nil, //initialised in Run
		controller: controller{
			dt:       time.Minute * 2,
			kp:       1,
			kd:       0,
			ki:       0,
			prevErr:  0,
			integral: 0,
		},
		taskSize: 6400 * 1000 * 1000,
		quit:     make(chan bool),
	}
}

// Run starts the master
func (m *Master) Run() {
	// initialise the nils
	m.initAWS()
	m.controllerTicker = time.NewTicker(m.controller.dt)
	// TODO test how this performs when a lot of tasks get submitted.
	m.scheduleTicker = time.NewTicker(time.Duration(100/(len(m.newTasks)+1)) * time.Millisecond)

	// setup and run http
	go func() {
		http.HandleFunc(lib.JobsCreatePath, makeJobsHandler(m.jobsChan))
		http.HandleFunc(lib.HeartbeatPath, makeHeartbeatHandler(m.heartbeatChan))
		http.HandleFunc(lib.StatusPath, makeStatusHandler(m.statusChan))
		log.Println("Running master on port", m.port)
		e := http.ListenAndServe(m.port, nil)
		if e != nil {
			log.Fatalln(e)
		}
	}()

	// send message to m.quit on interrupt
	go func() {
		sigchan := make(chan os.Signal, 10)
		signal.Notify(sigchan, os.Interrupt)
		<-sigchan
		m.quit <- true
	}()

	// main loop
	for {
		select {
		case <-m.controllerTicker.C:
			// run one iteration of the controller
			m.runController()
		case <-m.scheduleTicker.C:
			// we shedule the tasks when something is in this channel<commit_msg>Fix port parameter of ListenAndServe<commit_after>package master

import (
	"encoding/json"
	"log"
	"math"
	"math/rand"
	"net"
	"net/http"
	"os"
	"os/signal"
	"time"

	"github.com/aaronang/cong-the-ripper/lib"
	"github.com/aws/aws-sdk-go/service/ec2"
)

type Master struct {
	port             string
	svc              *ec2.EC2 // safe to be used concurrently
	instances        map[string]slave
	jobs             map[int]*job
	jobsChan         chan lib.Job
	heartbeatChan    chan lib.Heartbeat
	statusChan       chan chan statusSummary
	newTasks         []*lib.Task
	scheduledTasks   []*lib.Task
	scheduleTicker   *time.Ticker // channel to instruct the main loop to schedule tasks
	controllerTicker *time.Ticker
	controller       controller
	taskSize         int
	quit             chan bool
}

type slave struct {
	tasks    []*lib.Task
	maxSlots int
	instance *ec2.Instance // can't populate this easily
}

type controller struct {
	dt       time.Duration
	kp       float64
	kd       float64
	ki       float64
	prevErr  float64
	integral float64
}

type statusSummary struct {
	instances map[string]slave // TODO *ec2.Instance probably won't serialise
	jobs      map[int64]*job
}

// Init creates the master object
func Init(port string) Master {
	// set some defaults
	return Master{
		port:             port,
		svc:              nil, // initialised in Run
		instances:        make(map[string]slave),
		jobs:             make(map[int]*job),
		jobsChan:         make(chan lib.Job),
		heartbeatChan:    make(chan lib.Heartbeat),
		statusChan:       make(chan chan statusSummary),
		newTasks:         make([]*lib.Task, 0),
		scheduledTasks:   make([]*lib.Task, 0),
		scheduleTicker:   nil, // initialised in Run
		controllerTicker: nil, //initialised in Run
		controller: controller{
			dt:       time.Minute * 2,
			kp:       1,
			kd:       0,
			ki:       0,
			prevErr:  0,
			integral: 0,
		},
		taskSize: 6400 * 1000 * 1000,
		quit:     make(chan bool),
	}
}

// Run starts the master
func (m *Master) Run() {
	// initialise the nils
	m.initAWS()
	m.controllerTicker = time.NewTicker(m.controller.dt)
	// TODO test how this performs when a lot of tasks get submitted.
	m.scheduleTicker = time.NewTicker(time.Duration(100/(len(m.newTasks)+1)) * time.Millisecond)

	// setup and run http
	go func() {
		http.HandleFunc(lib.JobsCreatePath, makeJobsHandler(m.jobsChan))
		http.HandleFunc(lib.HeartbeatPath, makeHeartbeatHandler(m.heartbeatChan))
		http.HandleFunc(lib.StatusPath, makeStatusHandler(m.statusChan))
		log.Println("Running master on port", m.port)
		e := http.ListenAndServe(":"+m.port, nil)
		if e != nil {
			log.Fatalln(e)
		}
	}()

	// send message to m.quit on interrupt
	go func() {
		sigchan := make(chan os.Signal, 10)
		signal.Notify(sigchan, os.Interrupt)
		<-sigchan
		m.quit <- true
	}()

	// main loop
	for {
		select {
		case <-m.controllerTicker.C:
			// run one iteration of the controller
			m.runController()
		case <-m.scheduleTicker.C:
			// we shedule the tasks when something is in this channel
{'fraction_non_alphanumeric': 0.11042944785276074, 'fraction_numerical': 0.014259658431437573, 'mean_word_length': 2.7028852056476365, 'pattern_counts': {'":': 1, '<': 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': '1161923', 'n_tokens_mistral': 2266, 'n_tokens_neox': 2025, 'n_words': 547}