cpu_ctrl/processor/processor.go

package processor

import (
	"github.com/shirou/gopsutil/cpu"
	"strconv"
	"path/filepath"
	"fmt"
	"errors"
	"time"
	"sync"
	"path"
	"amuz.es/src/infra/cpu_ctrl/logger"
	"io/ioutil"
	"strings"
	"amuz.es/src/infra/cpu_ctrl/pid"
	"amuz.es/src/infra/cpu_ctrl/util"
)

type processor struct {
	handler          util.Handler
	id               int
	tempeturePath    string
	tempeture        float64
	tempetureChanged chan<- TempetureInfo
	sampleDuration   time.Duration
	fanController    pid.Controller
	fanSpeed         int
	fanMaxSpeed      int
	fanMinSpeed      int
	fanSpeedChanged  chan<- FanspeedInfo
}

type TempetureInfo struct {
	Id        int
	Tempeture float64
	At        time.Time
}

type FanspeedInfo struct {
	Id       int
	FanSpeed int
	At       time.Time
}

type Processor interface {
	Id() int
	Tempeture() float64
	FanSpeed() int
	FanMaxSpeed() int
	FanMinSpeed() int
	StartMonitoring()
	getMaxTempetureOnProcessor(string) (float64)
	readTempeture(string, chan<- float64, *sync.WaitGroup)
	normalizeFanspeed(float64) (int)
}

var (
	log = logger.NewLogger("processor")
)

func GetProcessorCount() (maxcpu int) {
	stat, err := cpu.Info()
	if err != nil {
		panic(err)
	}
	for _, info := range stat {
		physicalId, err := strconv.Atoi(info.PhysicalID)
		if err != nil {
			return
		} else if maxcpu < physicalId {
			maxcpu = physicalId
		}
	}
	maxcpu += 1
	return
}

func NewProcessorInfo(
	handler util.Handler,
	processorId int,
	sampleDuration time.Duration,
	P, I, D,
	setPoint float64,
	maxNoob, minNoob int,
	tempetureChanged chan<- TempetureInfo,
	fanSpeedChanged chan<- FanspeedInfo,
) (Processor, error) {

	if matches, err := filepath.Glob(fmt.Sprintf("/sys/devices/platform/coretemp.%d/hwmon/hwmon?", processorId)); err != nil {
		return nil, err
	} else if matches == nil {
		return nil, errors.New("hwmon not found!")
	} else {
		windupGuard := float64(maxNoob - minNoob)

		controller := pid.New(P, I, D)
		controller.SetSetPoint(setPoint)
		controller.SetSampleTime(sampleDuration)
		controller.SetWindupGuard(windupGuard)

		return &processor{
			handler:          handler,
			id:               processorId,
			tempeturePath:    matches[0],
			tempetureChanged: tempetureChanged,
			sampleDuration:   sampleDuration,
			fanController:    controller,
			fanSpeedChanged:  fanSpeedChanged,
			fanMaxSpeed:      maxNoob,
			fanMinSpeed:      minNoob,
		}, nil
	}
}
func (p *processor) Id() int            { return p.id }
func (p *processor) Tempeture() float64 { return p.tempeture }
func (p *processor) FanSpeed() int      { return p.fanSpeed }
func (p *processor) FanMaxSpeed() int   { return p.fanMaxSpeed }
func (p *processor) FanMinSpeed() int   { return p.fanMinSpeed }

func (p *processor) StartMonitoring() {
	defer p.handler.DecreaseWait()
	defer func() {
		if err := recover(); err != nil {
			p.handler.NotifyError(err.(error))
		}
	}()

	tempeturePathGlob := path.Join(p.tempeturePath, "temp?_input")
	ticker := time.Tick(p.sampleDuration)
	defer log.Infof("Processor %d monitor stopped", p.id)
	log.Infof("Processor %d monitor started with %s", p.id, p.sampleDuration)

	for {
		select {
		case now := <-ticker:
			var (
				highestTemp float64
				fanspeed    int
			)

			/*
			SELECT mean("tempeture") as tempeture FROM "processor_tempeture" WHERE $timeFilter GROUP BY "processor", time(5s) fill(previous)
			*/
			highestTemp = p.getMaxTempetureOnProcessor(tempeturePathGlob)
			if p.tempetureChanged != nil {
				select {
				case p.tempetureChanged <- TempetureInfo{
					Id:        p.id,
					Tempeture: highestTemp,
					At:        now,
				}:
				default:
				}
			}
			p.tempeture = highestTemp
			log.Debugf("processor %d : tempeture changed %f", p.id, highestTemp)

			/*
			SELECT mean("noob") as noob FROM "processor_cooling_fanspeed" WHERE $timeFilter GROUP BY "processor", time(5s) fill(previous)			CREATE CONTINUOUS QUERY "processor_cooling_fanspeed_5s" ON "core" BEGIN SELECT max("noob") AS "mean_noob" INTO  "processor_cooling_fanspeed_5s" FROM "processor_cooling_fanspeed" GROUP BY "processor", time(5s) fill(previous) END
			 */
			fanspeed = p.normalizeFanspeed(p.fanController.Update(highestTemp))
			if p.fanSpeedChanged != nil {
				select {
				case p.fanSpeedChanged <- FanspeedInfo{
					Id:       p.id,
					FanSpeed: fanspeed,
					At:       now,
				}:
				default:
				}
			}
			p.fanSpeed = fanspeed
			log.Debugf("processor %d : fan changed 0x%x", p.id, fanspeed)
		case <-p.handler.Done():
			return
		}
	}
}

func (p *processor) getMaxTempetureOnProcessor(tempeturePathGlob string) (maxTempeture float64) {
	matches, err := filepath.Glob(tempeturePathGlob)
	if err != nil {
		panic(err)
	}

	tempetureReadWaiter := &sync.WaitGroup{}
	queue := make(chan float64, len(matches)-1)

	// exclude package temp
	for _, tempetureFilePath := range matches[1:] {
		tempetureReadWaiter.Add(1)
		go p.readTempeture(tempetureFilePath, queue, tempetureReadWaiter)
	}
	tempetureReadWaiter.Wait()
	close(queue)
	for sense := range queue {
		if maxTempeture < sense {
			maxTempeture = sense
		}
	}
	return
}

func (p *processor) readTempeture(path string, senseChan chan<- float64, waiter *sync.WaitGroup) {
	defer waiter.Done()
	if dat, err := ioutil.ReadFile(path); err != nil {
		p.handler.NotifyError(err)
	} else if tempetureSense, err := strconv.Atoi(strings.TrimSpace(string(dat))); err != nil {
		p.handler.NotifyError(err)
	} else {
		senseChan <- float64(tempetureSense) / 1000.0
	}
}

func (p *processor) normalizeFanspeed(response float64) (adjusted int) {
	adjusted = int(-response)
	if adjusted < p.fanMinSpeed {
		adjusted = p.fanMinSpeed
	} else if adjusted > p.fanMaxSpeed {
		adjusted = p.fanMaxSpeed
	}
	return
}