이벤트 모니터링 가능 하도록 코드 수정

consumer/speed_controller.go

@@ -0,0 +1,87 @@
+package consumer
+
+import (
+	"fmt"
+	"os/exec"
+	"amuz.es/src/infra/cpu_ctrl/util"
+	"time"
+	"amuz.es/src/infra/cpu_ctrl/processor"
+)
+
+type fanControl struct {
+	processorCount   int
+	handler          util.Handler
+	fanSpeedConsumer chan processor.FanspeedInfo
+	sampleDuration   time.Duration
+}
+
+type FanControl interface {
+	Consumer() chan<- processor.FanspeedInfo
+	StartControl()
+}
+
+func NewFanControl(processorCount int, sampleDuration time.Duration, handler util.Handler) FanControl {
+	return &fanControl{
+		processorCount:   processorCount,
+		handler:          handler,
+		fanSpeedConsumer: make(chan processor.FanspeedInfo, processorCount),
+		sampleDuration:   sampleDuration,
+	}
+}
+
+func (c *fanControl) Consumer() chan<- processor.FanspeedInfo { return c.fanSpeedConsumer }
+
+func (c *fanControl) StartControl() {
+	defer c.handler.DecreaseWait()
+	defer func() {
+		if err := recover(); err != nil {
+			c.handler.NotifyError(err.(error))
+		}
+	}()
+	defer close(c.fanSpeedConsumer)
+
+	ticker := time.Tick(c.sampleDuration)
+	fanSpeedList := make([]int, c.processorCount)
+
+	for {
+		newFanSpeedList := make([]int, c.processorCount)
+		for {
+			select {
+			case <-ticker:
+				break
+			case changedSpeed := <-c.fanSpeedConsumer:
+				newFanSpeedList[changedSpeed.Id] = changedSpeed.FanSpeed
+			case <-c.handler.Done():
+				return
+			}
+		}
+		if (compareFanSpeed(fanSpeedList, newFanSpeedList)) {
+			fanSpeedList = newFanSpeedList
+			args := make([]string, 0)
+			args = append(args,
+				"raw",
+				"0x3a", "0x01",
+			)
+			for _, item := range fanSpeedList {
+				args = append(args, fmt.Sprintf("0x%x", item))
+			}
+			args = append(args,
+				"0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
+			)
+			cmd := exec.Command("ipmitool", args...)
+			if err := cmd.Run(); err != nil {
+				c.handler.NotifyError(err)
+				return
+			}
+		}
+	}
+}
+
+func compareFanSpeed(old, new []int) bool {
+	for i, v := range old {
+		if v != new[i] { // here is no bounds checking for b[i]
+			return false
+		}
+	}
+	return true
+}

main.go

@@ -12,6 +12,7 @@ import (
 	"amuz.es/src/infra/cpu_ctrl/logger"
 	"amuz.es/src/infra/cpu_ctrl/processor"
 	"amuz.es/src/infra/cpu_ctrl/util"
+	"amuz.es/src/infra/cpu_ctrl/consumer"
 )
 
 var (
@@ -41,9 +42,9 @@ func setMaxProcs() {
 	}
 }
 
-func broadcastTempeture(sender <-chan float64, handler util.Handler, receivers ...<-chan float64) {
+func FanoutTempeture(sender <-chan processor.TempetureInfo, handler util.Handler, receivers ...chan<- processor.TempetureInfo) {
+	handler.IncreaseWait()
 	defer handler.DecreaseWait()
-
 	defer func() {
 		if err := recover(); err != nil {
 			handler.NotifyError(err.(error))
@@ -64,6 +65,7 @@ func broadcastTempeture(sender <-chan float64, handler util.Handler, receivers .
 		}
 	}
 }
+
 func main() {
 
 	var (
@@ -79,21 +81,25 @@ func main() {
 		handler.NotifyError(errors.New("cpu not found!"))
 	}
 
+	fanController := consumer.NewFanControl(processorCount, sampleDuration, handler)
+
 	processors = make([]processor.Processor, 0, processorCount)
 	for i := 0; i < processorCount; i++ {
 		if info, err := processor.NewProcessorInfo(handler, i, sampleDuration,
 			1.5, 0.4, 2.0, 38.0, 0x64, 0x4,
+			nil, fanController.Consumer(),
 		);
 			err != nil {
 			handler.NotifyError(err)
 		} else {
 			processors = append(processors, info)
+			handler.IncreaseWait()
 			go info.StartMonitoring()
 		}
 	}
 
 	handler.IncreaseWait()
-	go TempetureControl(processors, tempetureChange, errorChan, ctx, waiter)
+	go fanController.StartControl()
 
 	signal.Notify(exitSignal, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
 	daemon.NotifyDaemon(daemon.DaemonStarted)

processor/processor.go

@@ -21,13 +21,13 @@ type processor struct {
 	id               int
 	tempeturePath    string
 	tempeture        float64
-	tempetureChanged chan TempetureInfo
+	tempetureChanged chan<- TempetureInfo
 	sampleDuration   time.Duration
 	fanController    pid.Controller
 	fanSpeed         int
 	fanMaxSpeed      int
 	fanMinSpeed      int
-	fanSpeedChanged  chan FanspeedInfo
+	fanSpeedChanged  chan<- FanspeedInfo
 }
 
 type TempetureInfo struct {
@@ -39,8 +39,7 @@ type FanspeedInfo struct {
 	Id       int
 	FanSpeed int
 }
-type TempetureChanged = <-chan TempetureInfo
+
-type FanspeedChanged = <-chan FanspeedInfo
 type Processor interface {
 	Id() int
 	Tempeture() float64
@@ -48,8 +47,6 @@ type Processor interface {
 	FanMaxSpeed() int
 	FanMinSpeed() int
 	StartMonitoring()
-	TempetureNotifier() TempetureChanged
-	FanSpeedNotifier() FanspeedChanged
 	getMaxTempetureOnProcessor(string) (float64)
 	readTempeture(string, chan<- float64, *sync.WaitGroup)
 	normalizeFanspeed(float64) (int)
@@ -83,6 +80,8 @@ func NewProcessorInfo(
 	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 {
@@ -101,25 +100,22 @@ func NewProcessorInfo(
 			handler:          handler,
 			id:               processorId,
 			tempeturePath:    matches[0],
-			tempetureChanged: make(chan TempetureInfo),
+			tempetureChanged: tempetureChanged,
 			sampleDuration:   sampleDuration,
 			fanController:    controller,
-			fanSpeedChanged:  make(chan FanspeedInfo),
+			fanSpeedChanged:  fanSpeedChanged,
 			fanMaxSpeed:      maxNoob,
 			fanMinSpeed:      minNoob,
 		}, nil
 	}
 }
-func (p *processor) Id() int                             { return p.id }
+func (p *processor) Id() int            { return p.id }
-func (p *processor) Tempeture() float64                  { return p.tempeture }
+func (p *processor) Tempeture() float64 { return p.tempeture }
-func (p *processor) FanSpeed() int                       { return p.fanSpeed }
+func (p *processor) FanSpeed() int      { return p.fanSpeed }
-func (p *processor) FanMaxSpeed() int                    { return p.fanMaxSpeed }
+func (p *processor) FanMaxSpeed() int   { return p.fanMaxSpeed }
-func (p *processor) FanMinSpeed() int                    { return p.fanMinSpeed }
+func (p *processor) FanMinSpeed() int   { return p.fanMinSpeed }
-func (p *processor) TempetureNotifier() TempetureChanged { return p.tempetureChanged }
-func (p *processor) FanSpeedNotifier() FanspeedChanged   { return p.fanSpeedChanged }
 
 func (p *processor) StartMonitoring() {
-	p.handler.IncreaseWait()
 	defer p.handler.DecreaseWait()
 	defer func() {
 		if err := recover(); err != nil {
@@ -127,9 +123,6 @@ func (p *processor) StartMonitoring() {
 		}
 	}()
 
-	defer close(p.tempetureChanged)
-	defer close(p.fanSpeedChanged)
-
 	tempeturePathGlob := path.Join(p.tempeturePath, "temp?_input")
 	ticker := time.Tick(p.sampleDuration)
 	log.Infof("Processor %d monitor started with %s", p.id, p.sampleDuration)
@@ -144,18 +137,28 @@ func (p *processor) StartMonitoring() {
 
 			highestTemp = p.getMaxTempetureOnProcessor(tempeturePathGlob)
 			if highestTemp != p.tempeture {
-				p.tempetureChanged <- TempetureInfo{
+				if p.tempetureChanged != nil {
-					Id:        p.id,
+					select {
-					Tempeture: highestTemp,
+					case p.tempetureChanged <- TempetureInfo{
+						Id:        p.id,
+						Tempeture: highestTemp,
+					}:
+					default:
+					}
 				}
 				p.tempeture = highestTemp
 			}
 
 			fanspeed = p.normalizeFanspeed(p.fanController.Update(highestTemp))
 			if fanspeed != p.fanSpeed {
-				p.fanSpeedChanged <- FanspeedInfo{
+				if p.fanSpeedChanged != nil {
-					Id:p.id,
+					select {
-					FanSpeed:fanspeed,
+					case p.fanSpeedChanged <- FanspeedInfo{
+						Id:       p.id,
+						FanSpeed: fanspeed,
+					}:
+					default:
+					}
 				}
 				p.fanSpeed = fanspeed
 			}

processor/speed_controller.go

@@ -1,95 +0,0 @@
-package processor
-
-import (
-	"fmt"
-	"os/exec"
-	"amuz.es/src/infra/cpu_ctrl/pid"
-	"amuz.es/src/infra/cpu_ctrl/util"
-	"time"
-	"golang.org/x/tools/go/gcimporter15/testdata"
-)
-
-type fanControl struct {
-	processorCount   int
-	handler          util.Handler
-	fanSpeedConsumer chan FanspeedInfo
-}
-
-type FanControl interface {
-}
-
-func NewFanControl(processorCount int, handler util.Handler) FanControl {
-	return &fanControl{
-		processorCount:   processorCount,
-		handler:          handler,
-		fanSpeedConsumer: make(chan ),
-	}
-}
-
-func (c *fanControl) StartControl() {
-	c.handler.IncreaseWait()
-	defer c.handler.DecreaseWait()
-	defer func() {
-		if err := recover(); err != nil {
-			c.handler.NotifyError(err.(error))
-		}
-	}()
-
-	defer close(c.fanSpeedConsumer)
-	ticker := time.Tick(p.sampleDuration)
-
-}
-func TempetureControl(
-	handler util.Handler,
-	processors []processor.Processor,
-) {
-	defer handler.DecreaseWait()
-
-	noobs := make([]int, processorCount)
-
-	controllers := make([]pid.Controller, 0, processorCount)
-	for i := 0; i < processorCount; i++ {
-		controller := pid.New(P, I, D)
-		controller.SetSetPoint(SetPoint)
-		controller.SetSampleTime(SampleTime)
-		controller.SetWindupGuard(WindupGuard)
-		controllers = append(controllers, controller)
-	}
-
-	for {
-		select {
-		case change := <-notifier:
-			controller := controllers[change.Id()]
-			adj_noob := int(-controller.Update(change.Tempeture()))
-			if adj_noob < minNoob {
-				adj_noob = minNoob
-			} else if adj_noob > maxNoob {
-				adj_noob = maxNoob
-			}
-			if noobs[change.Id()] == adj_noob {
-				continue
-			}
-			noobs[change.Id()] = adj_noob
-			log.Printf("Processor %d fan 0x%x\n", change.Id, adj_noob)
-
-			args := make([]string, 0)
-			args = append(args,
-				"raw",
-				"0x3a", "0x01",
-			)
-			for _, item := range noobs {
-				args = append(args, fmt.Sprintf("0x%x", item))
-			}
-			args = append(args,
-				"0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
-			)
-			cmd := exec.Command("ipmitool", args...)
-			if err := cmd.Run(); err != nil {
-				errorChan <- err
-				return
-			}
-		case <-ctx.Done():
-			return
-		}
-	}
-}

util/handler.go

@@ -32,7 +32,7 @@ func NewHandler() Handler {
 		ctx:       ctx,
 		canceler:  canceler,
 		waiter:    &sync.WaitGroup{},
-		errorChan: make(chan error),
+		errorChan: make(chan error,1),
 	}
 }
 
@@ -45,7 +45,7 @@ func (h *handler) GracefullWait() {
 	}
 	h.waiter.Wait()
 	close(h.errorChan)
-	
+
 	for remainError := range h.errorChan {
 		log.Errorf("%s", remainError)
 	}