diff options
Diffstat (limited to 'adapter/statsclient/stat_segment.go')
-rw-r--r-- | adapter/statsclient/stat_segment.go | 415 |
1 files changed, 243 insertions, 172 deletions
diff --git a/adapter/statsclient/stat_segment.go b/adapter/statsclient/stat_segment.go index e8d20b0..9f028eb 100644 --- a/adapter/statsclient/stat_segment.go +++ b/adapter/statsclient/stat_segment.go @@ -17,9 +17,7 @@ package statsclient import ( "fmt" "net" - "sync/atomic" "syscall" - "time" "unsafe" "github.com/ftrvxmtrx/fd" @@ -28,41 +26,70 @@ import ( ) var ( - maxWaitInProgress = time.Second * 1 + ErrStatDataLenIncorrect = fmt.Errorf("stat data length incorrect") ) -type statDirectoryType int32 - -func (t statDirectoryType) String() string { - return adapter.StatType(t).String() -} +const ( + minVersion = 0 + maxVersion = 1 +) -type statSegDirectoryEntry struct { - directoryType statDirectoryType - // unionData can represent: offset, index or value - unionData uint64 - offsetVector uint64 - name [128]byte +func checkVersion(ver uint64) error { + if ver < minVersion { + return fmt.Errorf("stat segment version is too old: %v (minimal version: %v)", ver, minVersion) + } else if ver > maxVersion { + return fmt.Errorf("stat segment version is not supported: %v (minimal version: %v)", ver, maxVersion) + } + return nil } type statSegment struct { sharedHeader []byte memorySize int64 - // oldHeader defines version 0 for stat segment - // and is used for VPP 19.04 - oldHeader bool + // legacyVersion represents stat segment version 0 + // and is used as fallback for VPP 19.04 + legacyVersion bool +} + +func (c *statSegment) getStatDirVector() unsafe.Pointer { + dirOffset, _, _ := c.getOffsets() + return unsafe.Pointer(&c.sharedHeader[dirOffset]) +} + +func (c *statSegment) getStatDirIndex(p unsafe.Pointer, index uint32) *statSegDirectoryEntry { + return (*statSegDirectoryEntry)(unsafe.Pointer(uintptr(p) + uintptr(index)*unsafe.Sizeof(statSegDirectoryEntry{}))) +} + +func (c *statSegment) getHeader() (header statSegSharedHeader) { + if c.legacyVersion { + return statSegHeaderLegacy(c.sharedHeader) + } + return statSegHeader(c.sharedHeader) +} + +func (c *statSegment) getEpoch() (int64, bool) { + h := c.getHeader() + return h.epoch, h.inProgress != 0 +} + +func (c *statSegment) getOffsets() (dir, err, stat int64) { + h := c.getHeader() + return h.directoryOffset, h.errorOffset, h.statsOffset } func (c *statSegment) connect(sockName string) error { - addr := &net.UnixAddr{ + if c.sharedHeader != nil { + return fmt.Errorf("already connected") + } + + addr := net.UnixAddr{ Net: "unixpacket", Name: sockName, } - Log.Debugf("connecting to: %v", addr) - conn, err := net.DialUnix(addr.Net, nil, addr) + conn, err := net.DialUnix(addr.Net, nil, &addr) if err != nil { Log.Warnf("connecting to socket %s failed: %s", addr, err) return err @@ -82,84 +109,102 @@ func (c *statSegment) connect(sockName string) error { if len(files) == 0 { return fmt.Errorf("no files received over socket") } + + file := files[0] defer func() { - for _, f := range files { - if err := f.Close(); err != nil { - Log.Warnf("closing file %s failed: %v", f.Name(), err) - } + if err := file.Close(); err != nil { + Log.Warnf("closing file failed: %v", err) } }() - Log.Debugf("received %d files over socket", len(files)) - - f := files[0] - - info, err := f.Stat() + info, err := file.Stat() if err != nil { return err } - size := info.Size() - Log.Debugf("fd: name=%v size=%v", info.Name(), size) - - data, err := syscall.Mmap(int(f.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED) + data, err := syscall.Mmap(int(file.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED) if err != nil { - Log.Warnf("mapping shared memory failed: %v", err) + Log.Debugf("mapping shared memory failed: %v", err) return fmt.Errorf("mapping shared memory failed: %v", err) } - Log.Debugf("successfuly mapped shared memory") - c.sharedHeader = data c.memorySize = size - header := c.readHeader() - Log.Debugf("stat segment header: %+v", header) + Log.Debugf("successfuly mmapped shared memory segment (size: %v)", size) - // older VPP (<=19.04) did not have version in stat segment header - // we try to provide fallback support by skipping it in header - if header.version > MaxVersion && header.inProgress > 1 && header.epoch == 0 { - h := c.readHeaderOld() - Log.Debugf("statsclient: falling back to old stat segment version (VPP <=19.04): %+v", h) - c.oldHeader = true + hdr := statSegHeader(c.sharedHeader) + Log.Debugf("stat segment header: %+v", hdr) + + if hdr.legacyVersion() { + c.legacyVersion = true + hdr = statSegHeaderLegacy(c.sharedHeader) + Log.Debugf("falling back to legacy version (VPP <=19.04) of stat segment (header: %+v)", hdr) + } + + if err := checkVersion(hdr.version); err != nil { + return err } return nil } func (c *statSegment) disconnect() error { + if c.sharedHeader == nil { + return nil + } + if err := syscall.Munmap(c.sharedHeader); err != nil { - Log.Warnf("unmapping shared memory failed: %v", err) + Log.Debugf("unmapping shared memory failed: %v", err) return fmt.Errorf("unmapping shared memory failed: %v", err) } + c.sharedHeader = nil Log.Debugf("successfuly unmapped shared memory") - return nil } -func (c *statSegment) copyData(dirEntry *statSegDirectoryEntry) adapter.Stat { - switch typ := adapter.StatType(dirEntry.directoryType); typ { +type statDirectoryType int32 + +func (t statDirectoryType) String() string { + return adapter.StatType(t).String() +} + +type statSegDirectoryEntry struct { + directoryType statDirectoryType + // unionData can represent: + // - offset + // - index + // - value + unionData uint64 + offsetVector uint64 + name [128]byte +} + +func (c *statSegment) copyEntryData(dirEntry *statSegDirectoryEntry) adapter.Stat { + dirType := adapter.StatType(dirEntry.directoryType) + + switch dirType { case adapter.ScalarIndex: return adapter.ScalarStat(dirEntry.unionData) case adapter.ErrorIndex: - _, errOffset, _ := c.readOffsets() + _, errOffset, _ := c.getOffsets() offsetVector := unsafe.Pointer(&c.sharedHeader[errOffset]) var errData adapter.Counter - if c.oldHeader { + if c.legacyVersion { // error were not vector (per-worker) in VPP 19.04 offset := uintptr(dirEntry.unionData) * unsafe.Sizeof(uint64(0)) - val := *(*adapter.Counter)(add(offsetVector, offset)) + val := *(*adapter.Counter)(statSegPointer(offsetVector, offset)) errData = val } else { vecLen := vectorLen(offsetVector) for i := uint64(0); i < vecLen; i++ { - cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) offset := uintptr(cb) + uintptr(dirEntry.unionData)*unsafe.Sizeof(adapter.Counter(0)) - val := *(*adapter.Counter)(add(unsafe.Pointer(&c.sharedHeader[0]), offset)) + val := *(*adapter.Counter)(statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), offset)) errData += val } } @@ -167,83 +212,82 @@ func (c *statSegment) copyData(dirEntry *statSegDirectoryEntry) adapter.Stat { case adapter.SimpleCounterVector: if dirEntry.unionData == 0 { - Log.Debugf("\toffset is not valid") + debugf("offset invalid for %s", dirEntry.name) break } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) { - Log.Debugf("\toffset out of range") + debugf("offset out of range for %s", dirEntry.name) break } - simpleCounter := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset - vecLen := vectorLen(simpleCounter) - offsetVector := add(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) + vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirEntry.unionData])) + offsetVector := statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) data := make([][]adapter.Counter, vecLen) for i := uint64(0); i < vecLen; i++ { - cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)]) vecLen2 := vectorLen(counterVec) + data[i] = make([]adapter.Counter, vecLen2) for j := uint64(0); j < vecLen2; j++ { offset := uintptr(j) * unsafe.Sizeof(adapter.Counter(0)) - val := *(*adapter.Counter)(add(counterVec, offset)) - data[i] = append(data[i], val) + val := *(*adapter.Counter)(statSegPointer(counterVec, offset)) + data[i][j] = val } } return adapter.SimpleCounterStat(data) case adapter.CombinedCounterVector: if dirEntry.unionData == 0 { - Log.Debugf("\toffset is not valid") + debugf("offset invalid for %s", dirEntry.name) break } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) { - Log.Debugf("\toffset out of range") + debugf("offset out of range for %s", dirEntry.name) break } - combinedCounter := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset - vecLen := vectorLen(combinedCounter) - offsetVector := add(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) + vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirEntry.unionData])) + offsetVector := statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) data := make([][]adapter.CombinedCounter, vecLen) for i := uint64(0); i < vecLen; i++ { - cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)]) vecLen2 := vectorLen(counterVec) + data[i] = make([]adapter.CombinedCounter, vecLen2) for j := uint64(0); j < vecLen2; j++ { offset := uintptr(j) * unsafe.Sizeof(adapter.CombinedCounter{}) - val := *(*adapter.CombinedCounter)(add(counterVec, offset)) - data[i] = append(data[i], val) + val := *(*adapter.CombinedCounter)(statSegPointer(counterVec, offset)) + data[i][j] = val } } return adapter.CombinedCounterStat(data) case adapter.NameVector: if dirEntry.unionData == 0 { - Log.Debugf("\toffset is not valid") + debugf("offset invalid for %s", dirEntry.name) break } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) { - Log.Debugf("\toffset out of range") + debugf("offset out of range for %s", dirEntry.name) break } - nameVector := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset - vecLen := vectorLen(nameVector) - offsetVector := add(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) + vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirEntry.unionData])) + offsetVector := statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) data := make([]adapter.Name, vecLen) for i := uint64(0); i < vecLen; i++ { - cb := *(*uint64)(add(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) if cb == 0 { - Log.Debugf("\tname vector cb out of range") + debugf("name vector out of range for %s (%v)", dirEntry.name, i) continue } nameVec := unsafe.Pointer(&c.sharedHeader[cb]) vecLen2 := vectorLen(nameVec) - var nameStr []byte + nameStr := make([]byte, 0, vecLen2) for j := uint64(0); j < vecLen2; j++ { offset := uintptr(j) * unsafe.Sizeof(byte(0)) - val := *(*byte)(add(nameVec, offset)) + val := *(*byte)(statSegPointer(nameVec, offset)) if val > 0 { nameStr = append(nameStr, val) } @@ -253,116 +297,143 @@ func (c *statSegment) copyData(dirEntry *statSegDirectoryEntry) adapter.Stat { return adapter.NameStat(data) default: - Log.Warnf("Unknown type %d for stat entry: %q", dirEntry.directoryType, dirEntry.name) + // TODO: monitor occurrences with metrics + debugf("Unknown type %d for stat entry: %q", dirEntry.directoryType, dirEntry.name) } return nil } -type sharedHeaderBase struct { - epoch int64 - inProgress int64 - directoryOffset int64 - errorOffset int64 - statsOffset int64 -} +func (c *statSegment) updateEntryData(dirEntry *statSegDirectoryEntry, stat *adapter.Stat) error { + switch (*stat).(type) { + case adapter.ScalarStat: + *stat = adapter.ScalarStat(dirEntry.unionData) -type statSegSharedHeader struct { - version uint64 - sharedHeaderBase -} + case adapter.ErrorStat: + _, errOffset, _ := c.getOffsets() + offsetVector := unsafe.Pointer(&c.sharedHeader[errOffset]) -func (c *statSegment) readHeaderOld() (header statSegSharedHeader) { - h := (*sharedHeaderBase)(unsafe.Pointer(&c.sharedHeader[0])) - header.version = 0 - header.epoch = atomic.LoadInt64(&h.epoch) - header.inProgress = atomic.LoadInt64(&h.inProgress) - header.directoryOffset = atomic.LoadInt64(&h.directoryOffset) - header.errorOffset = atomic.LoadInt64(&h.errorOffset) - header.statsOffset = atomic.LoadInt64(&h.statsOffset) - return -} + var errData adapter.Counter + if c.legacyVersion { + // error were not vector (per-worker) in VPP 19.04 + offset := uintptr(dirEntry.unionData) * unsafe.Sizeof(uint64(0)) + val := *(*adapter.Counter)(statSegPointer(offsetVector, offset)) + errData = val + } else { + vecLen := vectorLen(offsetVector) + for i := uint64(0); i < vecLen; i++ { + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + offset := uintptr(cb) + uintptr(dirEntry.unionData)*unsafe.Sizeof(adapter.Counter(0)) + val := *(*adapter.Counter)(statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), offset)) + errData += val + } + } + *stat = adapter.ErrorStat(errData) -func (c *statSegment) readHeader() (header statSegSharedHeader) { - h := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0])) - header.version = atomic.LoadUint64(&h.version) - header.epoch = atomic.LoadInt64(&h.epoch) - header.inProgress = atomic.LoadInt64(&h.inProgress) - header.directoryOffset = atomic.LoadInt64(&h.directoryOffset) - header.errorOffset = atomic.LoadInt64(&h.errorOffset) - header.statsOffset = atomic.LoadInt64(&h.statsOffset) - return -} + case adapter.SimpleCounterStat: + if dirEntry.unionData == 0 { + debugf("offset invalid for %s", dirEntry.name) + break + } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) { + debugf("offset out of range for %s", dirEntry.name) + break + } -func (c *statSegment) readVersion() uint64 { - if c.oldHeader { - return 0 - } - header := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0])) - version := atomic.LoadUint64(&header.version) - return version -} + vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirEntry.unionData])) + offsetVector := statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) -func (c *statSegment) readEpoch() (int64, bool) { - if c.oldHeader { - h := c.readHeaderOld() - return h.epoch, h.inProgress != 0 - } - header := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0])) - epoch := atomic.LoadInt64(&header.epoch) - inprog := atomic.LoadInt64(&header.inProgress) - return epoch, inprog != 0 -} + data := (*stat).(adapter.SimpleCounterStat) + if uint64(len(data)) != vecLen { + return ErrStatDataLenIncorrect + } + for i := uint64(0); i < vecLen; i++ { + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)]) + vecLen2 := vectorLen(counterVec) + simpData := data[i] + if uint64(len(simpData)) != vecLen2 { + return ErrStatDataLenIncorrect + } + for j := uint64(0); j < vecLen2; j++ { + offset := uintptr(j) * unsafe.Sizeof(adapter.Counter(0)) + val := *(*adapter.Counter)(statSegPointer(counterVec, offset)) + simpData[j] = val + } + } -func (c *statSegment) readOffsets() (dir, err, stat int64) { - if c.oldHeader { - h := c.readHeaderOld() - return h.directoryOffset, h.errorOffset, h.statsOffset - } - header := (*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0])) - dirOffset := atomic.LoadInt64(&header.directoryOffset) - errOffset := atomic.LoadInt64(&header.errorOffset) - statOffset := atomic.LoadInt64(&header.statsOffset) - return dirOffset, errOffset, statOffset -} + case adapter.CombinedCounterStat: + if dirEntry.unionData == 0 { + debugf("offset invalid for %s", dirEntry.name) + break + } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) { + debugf("offset out of range for %s", dirEntry.name) + break + } -type statSegAccess struct { - epoch int64 -} + vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirEntry.unionData])) + offsetVector := statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) -func (c *statSegment) accessStart() *statSegAccess { - epoch, inprog := c.readEpoch() - t := time.Now() - for inprog { - if time.Since(t) > maxWaitInProgress { - return nil + data := (*stat).(adapter.CombinedCounterStat) + if uint64(len(data)) != vecLen { + return ErrStatDataLenIncorrect + } + for i := uint64(0); i < vecLen; i++ { + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)]) + vecLen2 := vectorLen(counterVec) + combData := data[i] + if uint64(len(combData)) != vecLen2 { + return ErrStatDataLenIncorrect + } + for j := uint64(0); j < vecLen2; j++ { + offset := uintptr(j) * unsafe.Sizeof(adapter.CombinedCounter{}) + val := *(*adapter.CombinedCounter)(statSegPointer(counterVec, offset)) + combData[j] = val + } } - epoch, inprog = c.readEpoch() - } - return &statSegAccess{ - epoch: epoch, - } -} -func (c *statSegment) accessEnd(acc *statSegAccess) bool { - epoch, inprog := c.readEpoch() - if acc.epoch != epoch || inprog { - return false - } - return true -} + case adapter.NameStat: + if dirEntry.unionData == 0 { + debugf("offset invalid for %s", dirEntry.name) + break + } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) { + debugf("offset out of range for %s", dirEntry.name) + break + } -type vecHeader struct { - length uint64 - vectorData [0]uint8 -} + vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirEntry.unionData])) + offsetVector := statSegPointer(unsafe.Pointer(&c.sharedHeader[0]), uintptr(dirEntry.offsetVector)) -func vectorLen(v unsafe.Pointer) uint64 { - vec := *(*vecHeader)(unsafe.Pointer(uintptr(v) - unsafe.Sizeof(uintptr(0)))) - return vec.length -} + data := (*stat).(adapter.NameStat) + if uint64(len(data)) != vecLen { + return ErrStatDataLenIncorrect + } + for i := uint64(0); i < vecLen; i++ { + cb := *(*uint64)(statSegPointer(offsetVector, uintptr(i)*unsafe.Sizeof(uint64(0)))) + if cb == 0 { + continue + } + nameVec := unsafe.Pointer(&c.sharedHeader[cb]) + vecLen2 := vectorLen(nameVec) + + nameData := data[i] + if uint64(len(nameData))+1 != vecLen2 { + return ErrStatDataLenIncorrect + } + for j := uint64(0); j < vecLen2; j++ { + offset := uintptr(j) * unsafe.Sizeof(byte(0)) + val := *(*byte)(statSegPointer(nameVec, offset)) + if val == 0 { + break + } + nameData[j] = val + } + } -//go:nosplit -func add(p unsafe.Pointer, x uintptr) unsafe.Pointer { - return unsafe.Pointer(uintptr(p) + x) + default: + if Debug { + Log.Debugf("Unrecognized stat type %T for stat entry: %v", stat, dirEntry.name) + } + } + return nil } |