adapter/statsclient/stat_segment.go

   1 //  Copyright (c) 2019 Cisco and/or its affiliates.
   2 //
   3 //  Licensed under the Apache License, Version 2.0 (the "License");
   4 //  you may not use this file except in compliance with the License.
   5 //  You may obtain a copy of the License at:
   6 //
   7 //      http://www.apache.org/licenses/LICENSE-2.0
   8 //
   9 //  Unless required by applicable law or agreed to in writing, software
  10 //  distributed under the License is distributed on an "AS IS" BASIS,
  11 //  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 //  See the License for the specific language governing permissions and
  13 //  limitations under the License.
  14
  15 package statsclient
  16
  17 import (
  18         "fmt"
  19         "net"
  20         "sync/atomic"
  21         "syscall"
  22         "time"
  23         "unsafe"
  24
  25         "github.com/ftrvxmtrx/fd"
  26
  27         "git.fd.io/govpp.git/adapter"
  28 )
  29
  30 var (
  31         maxWaitInProgress = time.Second * 1
  32 )
  33
  34 type statDirectoryType int32
  35
  36 func (t statDirectoryType) String() string {
  37         return adapter.StatType(t).String()
  38 }
  39
  40 type statSegDirectoryEntry struct {
  41         directoryType statDirectoryType
  42         // unionData can represent: offset, index or value
  43         unionData    uint64
  44         offsetVector uint64
  45         name         [128]byte
  46 }
  47
  48 type statSegment struct {
  49         sharedHeader []byte
  50         memorySize   int64
  51
  52         // oldHeader defines version 0 for stat segment
  53         // and is used for VPP 19.04
  54         oldHeader bool
  55 }
  56
  57 func (c *statSegment) connect(sockName string) error {
  58         addr := &net.UnixAddr{
  59                 Net:  "unixpacket",
  60                 Name: sockName,
  61         }
  62
  63         Log.Debugf("connecting to: %v", addr)
  64
  65         conn, err := net.DialUnix(addr.Net, nil, addr)
  66         if err != nil {
  67                 Log.Warnf("connecting to socket %s failed: %s", addr, err)
  68                 return err
  69         }
  70         defer func() {
  71                 if err := conn.Close(); err != nil {
  72                         Log.Warnf("closing socket failed: %v", err)
  73                 }
  74         }()
  75
  76         Log.Debugf("connected to socket")
  77
  78         files, err := fd.Get(conn, 1, nil)
  79         if err != nil {
  80                 return fmt.Errorf("getting file descriptor over socket failed: %v", err)
  81         }
  82         if len(files) == 0 {
  83                 return fmt.Errorf("no files received over socket")
  84         }
  85         defer func() {
  86                 for _, f := range files {
  87                         if err := f.Close(); err != nil {
  88                                 Log.Warnf("closing file %s failed: %v", f.Name(), err)
  89                         }
  90                 }
  91         }()
  92
  93         Log.Debugf("received %d files over socket", len(files))
  94
  95         f := files[0]
  96
  97         info, err := f.Stat()
  98         if err != nil {
  99                 return err
 100         }
 101
 102         size := info.Size()
 103
 104         Log.Debugf("fd: name=%v size=%v", info.Name(), size)
 105
 106         data, err := syscall.Mmap(int(f.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
 107         if err != nil {
 108                 Log.Warnf("mapping shared memory failed: %v", err)
 109                 return fmt.Errorf("mapping shared memory failed: %v", err)
 110         }
 111
 112         Log.Debugf("successfuly mapped shared memory")
 113
 114         c.sharedHeader = data
 115         c.memorySize = size
 116
 117         header := c.readHeader()
 118         Log.Debugf("stat segment header: %+v", header)
 119
 120         // older VPP (<=19.04) did not have version in stat segment header
 121         // we try to provide fallback support by skipping it in header
 122         if header.version > MaxVersion && header.inProgress > 1 && header.epoch == 0 {
 123                 h := c.readHeaderOld()
 124                 Log.Debugf("statsclient: falling back to old stat segment version (VPP <=19.04): %+v", h)
 125                 c.oldHeader = true
 126         }
 127
 128         return nil
 129 }
 130
 131 func (c *statSegment) disconnect() error {
 132         if err := syscall.Munmap(c.sharedHeader); err != nil {
 133                 Log.Warnf("unmapping shared memory failed: %v", err)
 134                 return fmt.Errorf("unmapping shared memory failed: %v", err)
 135         }
 136
 137         Log.Debugf("successfuly unmapped shared memory")
 138
 139         return nil
 140 }
 141
 142 func (c *statSegment) copyData(dirEntry *statSegDirectoryEntry) adapter.Stat {
 143         switch typ := adapter.StatType(dirEntry.directoryType); typ {
 144         case adapter.ScalarIndex:
 145                 return adapter.ScalarStat(dirEntry.unionData)
 146
 147         case adapter.ErrorIndex:
 148                 _, errOffset, _ := c.readOffsets()
 149                 offsetVector := unsafe.Pointer(&c.sharedHeader[errOffset])
 150
 151                 var errData adapter.Counter
 152                 if c.oldHeader {
 153                         // error were not vector (per-worker) in VPP 19.04
 154                         offset := uintptr(dirEntry.unionData) * unsafe.Sizeof(uint64(0))
 155                         val := *(*adapter.Counter)(add(offsetVector, offset))
 156                         errData = val
 157                 } else {
 158                         vecLen := vectorLen(offsetVector)
 159                         for i := uint64(0); i < vecLen; i++ {
 160                                 cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0))))
 161                                 offset := uintptr(cb) + uintptr(dirEntry.unionData)*unsafe.Sizeof(adapter.Counter(0))
 162                                 val := *(*adapter.Counter)(add(unsafe.Pointer(&c.sharedHeader[0]), offset))
 163                                 errData += val
 164                         }
 165                 }
 166                 return adapter.ErrorStat(errData)
 167
 168         case adapter.SimpleCounterVector:
 169                 if dirEntry.unionData == 0 {
 170                         Log.Debugf("\toffset is not valid")
 171                         break
 172                 } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) {
 173                         Log.Debugf("\toffset out of range")
 174                         break
 175                 }
 176
 177                 simpleCounter := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset
 178                 vecLen := vectorLen(simpleCounter)
 179                 offsetVector := add(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector))
 180
 181                 data := make([][]adapter.Counter, vecLen)
 182                 for i := uint64(0); i < vecLen; i++ {
 183                         cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0))))
 184                         counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)])
 185                         vecLen2 := vectorLen(counterVec)
 186                         for j := uint64(0); j < vecLen2; j++ {
 187                                 offset := uintptr(j) * unsafe.Sizeof(adapter.Counter(0))
 188                                 val := *(*adapter.Counter)(add(counterVec, offset))
 189                                 data[i] = append(data[i], val)
 190                         }
 191                 }
 192                 return adapter.SimpleCounterStat(data)
 193
 194         case adapter.CombinedCounterVector:
 195                 if dirEntry.unionData == 0 {
 196                         Log.Debugf("\toffset is not valid")
 197                         break
 198                 } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) {
 199                         Log.Debugf("\toffset out of range")
 200                         break
 201                 }
 202
 203                 combinedCounter := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset
 204                 vecLen := vectorLen(combinedCounter)
 205                 offsetVector := add(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector))
 206
 207                 data := make([][]adapter.CombinedCounter, vecLen)
 208                 for i := uint64(0); i < vecLen; i++ {
 209                         cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0))))
 210                         counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)])
 211                         vecLen2 := vectorLen(counterVec)
 212                         for j := uint64(0); j < vecLen2; j++ {
 213                                 offset := uintptr(j) * unsafe.Sizeof(adapter.CombinedCounter{})
 214                                 val := *(*adapter.CombinedCounter)(add(counterVec, offset))
 215                                 data[i] = append(data[i], val)
 216                         }
 217                 }
 218                 return adapter.CombinedCounterStat(data)
 219
 220         case adapter.NameVector:
 221                 if dirEntry.unionData == 0 {
 222                         Log.Debugf("\toffset is not valid")
 223                         break
 224                 } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) {
 225                         Log.Debugf("\toffset out of range")
 226                         break
 227                 }
 228
 229                 nameVector := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset
 230                 vecLen := vectorLen(nameVector)
 231                 offsetVector := add(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector))
 232
 233                 data := make([]adapter.Name, vecLen)
 234                 for i := uint64(0); i < vecLen; i++ {
 235                         cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0))))
 236                         if cb == 0 {
 237                                 Log.Debugf("\tname vector cb out of range")
 238                                 continue
 239                         }
 240                         nameVec := unsafe.Pointer(&c.sharedHeader[cb])
 241                         vecLen2 := vectorLen(nameVec)
 242
 243                         var nameStr []byte
 244                         for j := uint64(0); j < vecLen2; j++ {
 245                                 offset := uintptr(j) * unsafe.Sizeof(byte(0))
 246                                 val := *(*byte)(add(nameVec, offset))
 247                                 if val > 0 {
 248                                         nameStr = append(nameStr, val)
 249                                 }
 250                         }
 251                         data[i] = adapter.Name(nameStr)
 252                 }
 253                 return adapter.NameStat(data)
 254
 255         default:
 256                 Log.Warnf("Unknown type %d for stat entry: %q", dirEntry.directoryType, dirEntry.name)
 257         }
 258
 259         return nil
 260 }
 261
 262 type sharedHeaderBase struct {
 263         epoch           int64
 264         inProgress      int64
 265         directoryOffset int64
 266         errorOffset     int64
 267         statsOffset     int64
 268 }
 269
 270 type statSegSharedHeader struct {
 271         version uint64
 272         sharedHeaderBase
 273 }
 274
 275 func (c *statSegment) readHeaderOld() (header statSegSharedHeader) {
 276         h := (*sharedHeaderBase)(unsafe.Pointer(&c.sharedHeader[0]))
 277         header.version = 0
 278         header.epoch = atomic.LoadInt64(&h.epoch)
 279         header.inProgress = atomic.LoadInt64(&h.inProgress)
 280         header.directoryOffset = atomic.LoadInt64(&h.directoryOffset)
 281         header.errorOffset = atomic.LoadInt64(&h.errorOffset)
 282         header.statsOffset = atomic.LoadInt64(&h.statsOffset)
 283         return
 284 }
 285
 286 func (c *statSegment) readHeader() (header statSegSharedHeader) {
 287         h := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0]))
 288         header.version = atomic.LoadUint64(&h.version)
 289         header.epoch = atomic.LoadInt64(&h.epoch)
 290         header.inProgress = atomic.LoadInt64(&h.inProgress)
 291         header.directoryOffset = atomic.LoadInt64(&h.directoryOffset)
 292         header.errorOffset = atomic.LoadInt64(&h.errorOffset)
 293         header.statsOffset = atomic.LoadInt64(&h.statsOffset)
 294         return
 295 }
 296
 297 func (c *statSegment) readVersion() uint64 {
 298         if c.oldHeader {
 299                 return 0
 300         }
 301         header := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0]))
 302         version := atomic.LoadUint64(&header.version)
 303         return version
 304 }
 305
 306 func (c *statSegment) readEpoch() (int64, bool) {
 307         if c.oldHeader {
 308                 h := c.readHeaderOld()
 309                 return h.epoch, h.inProgress != 0
 310         }
 311         header := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0]))
 312         epoch := atomic.LoadInt64(&header.epoch)
 313         inprog := atomic.LoadInt64(&header.inProgress)
 314         return epoch, inprog != 0
 315 }
 316
 317 func (c *statSegment) readOffsets() (dir, err, stat int64) {
 318         if c.oldHeader {
 319                 h := c.readHeaderOld()
 320                 return h.directoryOffset, h.errorOffset, h.statsOffset
 321         }
 322         header := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0]))
 323         dirOffset := atomic.LoadInt64(&header.directoryOffset)
 324         errOffset := atomic.LoadInt64(&header.errorOffset)
 325         statOffset := atomic.LoadInt64(&header.statsOffset)
 326         return dirOffset, errOffset, statOffset
 327 }
 328
 329 type statSegAccess struct {
 330         epoch int64
 331 }
 332
 333 func (c *statSegment) accessStart() *statSegAccess {
 334         epoch, inprog := c.readEpoch()
 335         t := time.Now()
 336         for inprog {
 337                 if time.Since(t) > maxWaitInProgress {
 338                         return nil
 339                 }
 340                 epoch, inprog = c.readEpoch()
 341         }
 342         return &statSegAccess{
 343                 epoch: epoch,
 344         }
 345 }
 346
 347 func (c *statSegment) accessEnd(acc *statSegAccess) bool {
 348         epoch, inprog := c.readEpoch()
 349         if acc.epoch != epoch || inprog {
 350                 return false
 351         }
 352         return true
 353 }
 354
 355 type vecHeader struct {
 356         length     uint64
 357         vectorData [0]uint8
 358 }
 359
 360 func vectorLen(v unsafe.Pointer) uint64 {
 361         vec := *(*vecHeader)(unsafe.Pointer(uintptr(v) - unsafe.Sizeof(uintptr(0))))
 362         return vec.length
 363 }
 364
 365 //go:nosplit
 366 func add(p unsafe.Pointer, x uintptr) unsafe.Pointer {
 367         return unsafe.Pointer(uintptr(p) + x)
 368 }