performance_schema/replication_workload/replication_workload_row.go

   1 package replication_workload
   2
   3 import (
   4         "database/sql"
   5         "fmt"
   6         "log"
   7         "sort"
   8         "strings"
   9
  10         "github.com/sjmudd/pstop/lib"
  11 )
  12
  13 type replication_workload_row struct {
  14         NAME string
  15         EVENT_NAME string
  16         OBJECT_NAME string
  17         OPERATION string
  18         SUM_TIMER_WAIT int
  19         SUM_SPINS int
  20         SUM_NUMBER_OF_BYTES int
  21 }
  22
  23 type replication_workload_rows []replication_workload_row
  24
  25 func select_rep_workload_rows(dbh *sql.DB) replication_workload_rows {
  26         var t replication_workload_rows
  27
  28         sql := "SELECT t.NAME, ewc.EVENT_NAME, ewc.OBJECT_NAME, ewc.OPERATION, SUM(ewc.TIMER_WAIT) AS SUM_TIMER_WAIT, SUM(ewc.SPINS) AS SUM_SPINS, SUM(ewc.NUMBER_OF_BYTES) AS SUM_NUMBER_OF_BYTES,  FROM events_waits_history ewc JOIN threads t ON (t.THREAD_ID = ewc.thread_id) WHERE t.NAME LIKE '%slave_sql%' GROUP BY t.NAME, ewc.EVENT_NAME, ewc.OBJECT_NAME, ewc.OPERATION"
  29
  30         rows, err := dbh.Query(sql)
  31         if err != nil {
  32                 log.Fatal(err)
  33         }
  34         defer rows.Close()
  35
  36         for rows.Next() {
  37                 var r replication_workload_row
  38
  39                 if err := rows.Scan(&r.NAME, &r.EVENT_NAME, &r.OBJECT_NAME, &r.OPERATION, &r.SUM_TIMER_WAIT, &r.SUM_SPINS, &r.SUM_NUMBER_OF_BYTES); err != nil {
  40                         log.Fatal(err)
  41                 }
  42                 t = append(t, r)
  43         }
  44         if err := rows.Err(); err != nil {
  45                 log.Fatal(err)
  46         }
  47
  48         return t
  49 }
  50
  51 func (this *replication_workload_row) add(other replication_workload_row) {
  52         this.SUM_TIMER_WAIT += other.SUM_TIMER_WAIT
  53         this.SUM_SPINS += other.SUM_SPINS
  54         this.SUM_NUMBER_OF_BYTES += other.SUM_NUMBER_OF_BYTES
  55 }
  56
  57 func (this *replication_workload_row) subtract(other replication_workload_row) {
  58         this.SUM_TIMER_WAIT -= other.SUM_TIMER_WAIT
  59         this.SUM_SPINS -= other.SUM_SPINS
  60         this.SUM_NUMBER_OF_BYTES -= other.SUM_NUMBER_OF_BYTES
  61 }
  62
  63 func (t replication_workload_rows) Len() int      { return len(t) }
  64 func (t replication_workload_rows) Swap(i, j int) { t[i], t[j] = t[j], t[i] }
  65 // may need to adjust ordering here.!!!
  66 func (t replication_workload_rows) Less(i, j int) bool {
  67         return t[i].SUM_TIMER_WAIT > t[j].SUM_TIMER_WAIT
  68 }
  69
  70 func (t *replication_workload_rows) sort() {
  71         sort.Sort(t)
  72 }
  73
  74 // if the data in t2 is "newer", "has more values" than t then it needs refreshing.
  75 // check this by comparing totals.
  76 func (t replication_workload_rows) needs_refresh(t2 replication_workload_rows) bool {
  77         my_totals := t.totals()
  78         t2_totals := t2.totals()
  79
  80         return my_totals.SUM_TIMER_WAIT > t2_totals.SUM_TIMER_WAIT
  81 }
  82