updating i2c monitoring to use channels

internal/configs/boards.yaml

@@ -0,0 +1,4 @@
+model: raspberrypi
+    bus: 1
+model: beagleboard
+    bus: 2

internal/pkg/I2C/monitoring.go

@@ -5,39 +5,42 @@ import (
     "fmt"
 )
 
-func NewMonitor(b int) *I2CMonitor{
-    m := new(I2CMonitor)
+func NewMonitor(c,dc <-chan int, b int) error {
+    m := &I2CMonitor{Bus:b,Connect:c,Disconnect:dc}
     m.Devices = make(map[int]I2Cdev)
-    m.Bus = b
-    m.Update() //ensure that devices get updated before it gets locked by connected
-    return m
+    s := make(chan struct{})
+    go scan(s)
+    m.Scan = s
+    go m.Monitor()
+    return nil
 }
 
-func (m *I2CMonitor) Update() {
+func (m *I2CMonitor) Update() map[int]bool {
     /*
     gets newly connected sensors and updates any currently dead sensors
     */
     m.mu.Lock()
     defer m.mu.Unlock()
-    // locking right away should be fine
-    fmt.Printf("Getting devices on bus %v\n",m.Bus)
     connected := I2Cconnected(m.Bus)
-    // m.mu.Lock()
-    // defer m.mu.Unlock()
     now := time.Now()
+    newDevs := make(map[int]bool)
     for k,v := range connected {
         if _, ok := m.Devices[k]; ok {   // if address existed
             entry := m.Devices[k]
             entry.Active = v             // update the value
             if v {
                 entry.LastSeen = now     // if its active, update last seen time
+            } else {
+                newDevs[k] = v           // mark the device for purging
             }
             m.Devices[k] = entry
         } else if v {                           // else theres a chance its new
-            fmt.Printf("adding device %v\n",k)
+            fmt.Printf("Device %v found\n",k)
+            newDevs[k] = v              // mark device for adding
             m.Devices[k] = I2Cdev{Active:true, LastSeen:now} // add new sensors
         }
     }
+    return newDevs
 }
 
 func (m *I2CMonitor) Remove(addr int) {
@@ -46,12 +49,26 @@ func (m *I2CMonitor) Remove(addr int) {
     delete(m.Devices, addr)
 }
 
-func (m *I2CMonitor) Connected() []int {
-    // returns list of addresses of connected devices
-    m.mu.Lock()
-    defer m.mu.Unlock()
-    connected := []int{}
-    for a,_ := range m.Devices {
-        connected = append(connected, a)
+func scan(ch <-chan struct{}) {
+    // helper func to time the device updates
+    for true {
+        ch <-struct{}{}     // weird syntax but just empty struct init
+        time.Sleep(5 * time.Second)
     }
-    return connected
+}
+
+func (m *I2CMonitor) Monitor() {
+    // functon that updates the device list and notifies rlc of any changes to sensor composition
+    for {
+        <-m.Scan
+        newdevs := m.Update()
+        for k,v := range newdevs {
+            if v {
+               m.Connect <-k
+           } else {
+               m.Disconnect <-k
+           }
+       }
+   }
+}
+

internal/pkg/I2C/wrappers.go

@@ -16,8 +16,11 @@ type I2Cdev struct {
 }
 
 type I2CMonitor struct {        // used in RLC to synchronize accesses
-    Bus int                 // bus to use {0,1,2}
-    Devices map[int]I2Cdev     // mapping to quickly index addresses to their device structs
+    Bus int                     // bus to use {0,1,2}
+    Devices map[int]I2Cdev      // mapping to quickly index addresses to their device structs
+    Scan <-chan struct{}         // empty struct for efficient sends
+    Connect chan<- int
+    Disconnect chan<- int
     mu sync.Mutex
 }
 

internal/pkg/reactor/rlcoordinator.go

@@ -16,22 +16,23 @@ import (
     pb "FRMS/internal/pkg/grpc"
 )
 
-type Hardware interface {
-    GetIp() string
-    GetId() uint32
-    GetBus() int
-    GetModel() string
-    GetPort() int
+type Hardware struct {
+    Id uint32
+    Ip string
+    Bus int
+    Model string
+    Port int
 }
 
-type I2CMonitor interface {
-    Connected() []int
+type HWinfo interface {
+    Get() (*Hardware, error)
 }
 
 type Coordinator struct {
     Server string
-    HW Hardware         // object to keep track of reactor hw info
-    I2C I2CMonitor         // I2C monitor to keep track of I2C devices
+    HW Hardware
+    Connected <-chan int
+    Disconnected <-chan int
     Sensors map[int]SensorManager
     Err chan error
     mu sync.Mutex
@@ -49,44 +50,76 @@ func NewSensorManager(addr int) SensorManager {
 }
 
 func NewCoordinator(s string,ch chan error) *Coordinator {
-    return &Coordinator{Server:s,Err:ch}
+    c := make(Coordinator{Server:s,Err:ch}
+    c.Connected = make(<-chan int)
+    c.Disconnected = make(<-chan int)
+    return c
 }
 
-func NewI2CMonitor(b int) I2CMonitor {
+func NewI2CMonitor(c, dc <-chan int, b int) error {
     return I2C.NewMonitor(b)
 }
 
-func NewHWMonitor() Hardware {
+func GetHWInfo() Hwinfo {
     return system.NewHWMonitor()
 }
 
 func (c *Coordinator) Start() error {
     // should discover hwinfo and sensors on its own
     // now setting up sensor managers
-    c.HW = NewHWMonitor()
-    fmt.Printf("Reactor at IP addr %v using I2C bus %v\n",c.HW.GetIp(),c.HW.GetBus())
-    response := c.Connect()
+    c.HW, err = GetHWInfo()
+    if err != nil {
+        return err
+    }
+    err = NewI2CMonitor(c.Connect,c.Disconnect,c.Bus)
+    if err != nil {
+        return err
+    }
+    fmt.Printf("Reactor at IP addr %v using I2C bus %v\n",c.Ip,c.Bus)
+    response := c.EstablishConnection()
     for response != nil {
-        fmt.Println("Connection failed!")
-        response = c.Connect()
+        fmt.Println("Connection failed!, Retrying")
+        response = c.EstablishConnection()
     }
     c.Register()
-    c.mu.Lock() // lock to finish setting up devices
+    return nil
+}
+
+func (c *Coordinator) Monitor() {
+    // function to automatically create and destroy sm
+    for {
+        select {
+        case addr := <-c.Connect:
+            //check if sm exists and add sm for addr
+            c.Connected(addr)
+        case addr := <-c.Disconnect:
+            // kill sm for addr
+            c.Disconnected(addr)
+        }
+    }
+}
+
+func (c *Coordinator) Connected(addr int) {
+    c.mu.Lock()
     defer c.mu.Unlock()
-    c.I2C = NewI2CMonitor(c.HW.GetBus())
-    devs := c.I2C.Connected()
-    c.Sensors = make(map[int]SensorManager)
-    for _,d := range devs {
-        // create a goroutine for each active sensor
-        sm := NewSensorManager(d)
-        c.Sensors[d] = sm
+    sm, exists := c.Sensors[addr]
+    if !exists {
+        sm := NewSensorManager(addr)
+        c.Sensors[addr] = sm
         go sm.Start()
-        fmt.Printf("Sensor Manager for addr %v Started!\n",d)
-    }
-    return nil
+    } // ignoring case of existing sm
 }
 
-func (c *Coordinator) Connect() error {
+func (c *Coordinator) Disconnected(addr int) {
+    c.mu.Lock()
+    defer c.mu.Unlock()
+    sm, exists := c.Sensors[addr]
+    if exists {
+        c.Sensors = delete(c.Sensors,addr)
+        sm.Kill()
+    } // not dealing with kill requests for not existing sm
+
+func (c *Coordinator) EstablishConnection() error {
     // function connects to central server and passes hwinfo
     var opts []grpc.DialOption
     opts = append(opts,grpc.WithTransportCredentials(insecure.NewCredentials()))
@@ -96,7 +129,7 @@ func (c *Coordinator) Connect() error {
     }
     defer conn.Close()
     client := pb.NewHandshakeClient(conn)
-    req := &pb.ReactorDiscoveryRequest{Id:c.HW.GetId(),Ip:c.HW.GetIp(),Port:int32(2000),Model:c.HW.GetModel()}
+    req := &pb.ReactorDiscoveryRequest{Id:c.Id,Ip:c.Ip,Port:c.Port,Model:c.Model}
     resp, err := client.ReactorDiscoveryHandler(context.Background(), req)
     if err != nil {
         return err
@@ -110,8 +143,8 @@ func (c *Coordinator) Connect() error {
 }
 
 func (c *Coordinator) Register() error {
-    fmt.Printf("Listing for pings on %v:%v\n",c.HW.GetIp(),c.HW.GetPort())
-    lis, err := net.Listen("tcp", fmt.Sprintf("%v:%v",c.HW.GetIp(),c.HW.GetPort()))
+    fmt.Printf("Listing for pings on %v:%v\n",c.Ip,c.Port)
+    lis, err := net.Listen("tcp", fmt.Sprintf("%v:%v",c.Ip(),c.Port))
     if err != nil {
         return err
     }
@@ -125,7 +158,7 @@ func (c *Coordinator) SensorStatusHandler(ctx context.Context, ping *pb.SensorSt
     c.mu.Lock()
     defer c.mu.Unlock()
     var sensors []*pb.SensorStatus
-    resp := &pb.SensorStatusResponse{Id:c.HW.GetId(),Sensors:sensors}
+    resp := &pb.SensorStatusResponse{Id:c.Id,Sensors:sensors}
     for a, s := range c.Sensors {
         resp.Sensors = append(resp.Sensors,&pb.SensorStatus{Addr:int32(a), Type:s.GetType(), Status:s.GetStatus()})
     }

internal/pkg/sensor/manager.go

@@ -2,19 +2,17 @@ package sensor
 
 import (
     "sync"
+    "FRMS/internal/pkg/I2C"
 )
 
 type Manager struct {
     Addr int
     Type string
     Status bool
-    //I2C I2CMonitor
+    Kill chan<- bool
     mu sync.Mutex
 }
 
-//type I2CMonitor interface {
-
-//}
 
 func (m *Manager) GetStatus() bool{
     m.mu.Lock()

internal/pkg/system/hwinfo.go

@@ -1,8 +1,6 @@
 package system
 
 import (
-    "fmt"
-    "sync"
     "os/exec"
     "bytes"
     "hash/fnv"
@@ -22,58 +20,22 @@ import (
     // *** will just replace info in file everytime
 
 
-type HWMonitor struct {
+type HWinfo struct {
     Ip string
     Id uint32
     Bus int
     Model string
     Port int
-    mu sync.Mutex
 }
 
-func NewHWMonitor() *HWMonitor {
-    h := &HWMonitor{Port:2000}
-    err := h.Start()
-    if err != nil {
-        fmt.Println("hw setup failed " + fmt.Sprint(err))
-    }
-    return h
-}
-
-func (h *HWMonitor) GetPort() int {
-    h.mu.Lock()
-    defer h.mu.Unlock()
-    return h.Port
-}
-
-func (h *HWMonitor) GetIp() string {
-    h.mu.Lock()
-    defer h.mu.Unlock()
-    return h.Ip
-}
-
-func (h *HWMonitor) GetId() uint32 {
-    h.mu.Lock()
-    defer h.mu.Unlock()
-    return h.Id
-}
-
-func (h *HWMonitor) GetBus() int {
-    h.mu.Lock()
-    defer h.mu.Unlock()
-    return h.Bus
-}
-
-func (h *HWMonitor) GetModel() string{
-    h.mu.Lock()
-    defer h.mu.Unlock()
-    return h.Model
+func NewHWinfo() (*HWinfo, error) {
+    h := &HWinfo{Port:2000}
+    err := h.Get()
+    return h, err
 }
 
-func (h *HWMonitor) Start() error {
+func (h *HWinfo) Get() error {
     // responsible for filling out struct
-    h.mu.Lock()
-    defer h.mu.Unlock() // want to ensure we get all values before the RLC reads
     bus := map[string]int{"raspberrypi":1,"beaglebone":2} // eventually will replace this with a config file
 
     ipcmd := "ifconfig eth0 | awk '/inet / {print $2}'"
@@ -89,7 +51,6 @@ func (h *HWMonitor) Start() error {
         cmd.Stderr = &stderr
         err := cmd.Run()
         if err != nil {
-            fmt.Println(fmt.Sprint(err)+": "+stderr.String())
             return err
         }
     }