rtnetmon/internal/monitor/loops.go

//go:build linux
// +build linux

package monitor

import (
	"context"
	"math"
	"math/rand"
	"strings"
	"sync"
	"time"
)

// UIUpdateChan is used to signal UI updates
var UIUpdateChan = make(chan struct{}, 100)

// reachLoop monitors reachability for hosts
func (m *Monitor) reachLoop(ctx context.Context, st *InterfaceStatus, hosts []string) {
	m.logf("Starting reachability monitoring for %s with %d hosts", st.Name, len(hosts))

	// Add random offset to avoid clustering at 1-second intervals
	randomOffset := time.Duration(100+rand.Intn(800)) * time.Millisecond
	m.logf("Reachability monitoring for %s will start after %v offset", st.Name, randomOffset)
	time.Sleep(randomOffset)

	tk := time.NewTicker(time.Second)
	defer tk.Stop()
	for {
		select {
		case <-ctx.Done():
			m.logf("Stopping reachability monitoring for %s", st.Name)
			return
		case <-tk.C:
			var wg sync.WaitGroup
			res := make(map[string]bool, len(hosts))
			mu := sync.Mutex{}
			for _, h := range hosts {
				wg.Add(1)
				go func(host string) {
					defer wg.Done()
					st.mu.Lock()
					st.TotalICMPReq++
					// Increase meter value when a packet is sent
					st.MeterValue++
					if st.MeterValue > m.MaxMeterValue {
						st.MeterValue = m.MaxMeterValue
					}
					st.mu.Unlock()

					ok := m.pingOnce(st.Name, host)
					mu.Lock()
					res[host] = ok
					mu.Unlock()

					st.mu.Lock()
					if ok {
						st.TotalICMPRep++
						// Decrease meter value when a packet is successfully received
						st.MeterValue--
						if st.MeterValue < 0 {
							st.MeterValue = 0
						}
						// Only update spinner when packets are successfully received
						st.Spin()
					} else {
						st.DroppedCount++
						st.LastDrop = time.Now()

						// Track lost packets per host
						st.LostPackets[host]++

						// Trigger UI update on ping failure
						select {
						case UIUpdateChan <- struct{}{}:
						default:
						}
					}
					st.mu.Unlock()
				}(h)
			}
			wg.Wait()

			// Check if reachability status changed
			statusChanged := false
			st.mu.Lock()
			for host, newStatus := range res {
				if oldStatus, ok := st.Reachable[host]; !ok || oldStatus != newStatus {
					statusChanged = true
					break
				}
			}

			st.Reachable = res
			st.LastPing = time.Now()

			// Check if all hosts are reachable
			allReachable := true
			for _, ok := range res {
				if !ok {
					allReachable = false
					break
				}
			}

			// If all hosts are reachable, gradually decay the meter value
			if allReachable && st.MeterValue > 0 {
				st.MeterValue--
			}

			st.mu.Unlock()

			// Always trigger UI update when reachability status changes
			if statusChanged {
				select {
				case UIUpdateChan <- struct{}{}:
				default:
				}
			}
		}
	}
}

// lossLoop monitors packet loss for hosts
func (m *Monitor) lossLoop(ctx context.Context, st *InterfaceStatus, hosts []string) {
	m.logf("Starting packet loss monitoring for %s with %d hosts", st.Name, len(hosts))

	// Add random offset to avoid clustering at periodic intervals
	randomOffset := time.Duration(100+rand.Intn(800)) * time.Millisecond
	m.logf("Packet loss monitoring for %s will start after %v offset", st.Name, randomOffset)
	time.Sleep(randomOffset)

	tk := time.NewTicker(m.PacketLossPeriod)
	defer tk.Stop()
	for {
		select {
		case <-ctx.Done():
			m.logf("Stopping packet loss monitoring for %s", st.Name)
			return
		case <-tk.C:
			var wg sync.WaitGroup
			res := make(map[string]float64, len(hosts))
			mu := sync.Mutex{}
			for _, h := range hosts {
				wg.Add(1)
				go func(host string) {
					defer wg.Done()
					lp := m.lossPercent(st.Name, host)
					mu.Lock()
					res[host] = lp
					mu.Unlock()

					st.mu.Lock()
					if lp == 0 {
						// Only update spinner when there's 0% packet loss
						st.Spin()
					} else {
						// Calculate approximate number of lost packets based on loss percentage
						lostPackets := int(math.Ceil(float64(m.PacketLossPings) * lp))

						// Update dropped count with the number of lost packets
						st.DroppedCount += lostPackets

						// Update last drop time if packets were lost
						if lostPackets > 0 {
							st.LastDrop = time.Now()
						}

						// Track lost packets per host
						st.LostPackets[host] += lostPackets

						// Trigger UI update on packet loss
						select {
						case UIUpdateChan <- struct{}{}:
						default:
						}
					}
					st.mu.Unlock()
				}(h)
			}
			wg.Wait()

			// Check if loss status changed
			statusChanged := false
			st.mu.Lock()
			for host, newLoss := range res {
				if oldLoss, ok := st.Loss[host]; !ok || math.Abs(oldLoss-newLoss) > 0.01 {
					statusChanged = true
					break
				}
			}

			for k, v := range res {
				st.Loss[k] = v
			}
			st.mu.Unlock()

			// Always trigger UI update when loss status changes
			if statusChanged {
				select {
				case UIUpdateChan <- struct{}{}:
				default:
				}
			}
		}
	}
}

// tcpLoop monitors TCP connectivity for hosts
func (m *Monitor) tcpLoop(ctx context.Context, st *InterfaceStatus, hosts []string) {
	m.logf("Starting TCP monitoring for %s with %d hosts", st.Name, len(hosts))

	// Add random offset to avoid clustering at 1-second intervals
	randomOffset := time.Duration(100+rand.Intn(800)) * time.Millisecond
	m.logf("TCP monitoring for %s will start after %v offset", st.Name, randomOffset)
	time.Sleep(randomOffset)

	tk := time.NewTicker(time.Second)
	defer tk.Stop()
	for {
		select {
		case <-ctx.Done():
			m.logf("Stopping TCP monitoring for %s", st.Name)
			return
		case <-tk.C:
			statusChanged := false

			for _, hp := range hosts {
				ms := float64(m.tcpDuration(st.Name, hp).Milliseconds())
				st.mu.Lock()

				// Check if TCP latency significantly changed
				hist := st.TCP[hp]
				if len(hist) > 0 {
					lastMs := hist[len(hist)-1]
					if math.Abs(lastMs-ms) > 20 { // 20ms threshold for significant change
						statusChanged = true
					}
				} else {
					// First measurement
					statusChanged = true
				}

				if ms < float64(m.TCPTimeout.Milliseconds()) {
					// Only update spinner on successful TCP connections
					st.Spin()
				} else {
					// Trigger UI update on TCP timeout
					select {
					case UIUpdateChan <- struct{}{}:
					default:
					}
				}

				if len(hist) >= m.StatsHistory {
					hist = hist[1:]
				}
				st.TCP[hp] = append(hist, ms)

				// Update lost packets for the host (without port)
				hostName := strings.Split(hp, ":")[0]
				if ms >= float64(m.TCPTimeout.Milliseconds()) {
					st.LostPackets[hostName]++
				}

				st.mu.Unlock()
			}

			// Always trigger UI update when TCP status changes significantly
			if statusChanged {
				select {
				case UIUpdateChan <- struct{}{}:
				default:
				}
			}
		}
	}
}