// See the License for the specific language governing permissions and
// limitations under the License.
+// Package statsclient is pure Go implementation of VPP stats API client.
package statsclient
import (
"fmt"
"net"
"os"
+ "path/filepath"
"regexp"
"sync/atomic"
"syscall"
"time"
- "unsafe"
+ "git.fd.io/govpp.git/adapter"
+ "github.com/fsnotify/fsnotify"
"github.com/ftrvxmtrx/fd"
logger "github.com/sirupsen/logrus"
+)
- "git.fd.io/govpp.git/adapter"
+const (
+ // DefaultSocketName is default VPP stats socket file path.
+ DefaultSocketName = adapter.DefaultStatsSocket
+
+ // DefaultSocketRetryPeriod is the time period after the socket availability
+ // will be re-checked
+ DefaultSocketRetryPeriod = 50 * time.Millisecond
+
+ // DefaultSocketRetryTimeout is the maximum time for the stats socket
+ DefaultSocketRetryTimeout = 3 * time.Second
)
var (
Log.Out = os.Stdout
if Debug {
Log.Level = logger.DebugLevel
- Log.Debug("enabled debug mode")
+ Log.Debug("govpp/statsclient: enabled debug mode")
}
}
+func debugf(f string, a ...interface{}) {
+ if Debug {
+ Log.Debugf(f, a...)
+ }
+}
+
+// implements StatsAPI
+var _ adapter.StatsAPI = (*StatsClient)(nil)
+
// StatsClient is the pure Go implementation for VPP stats API.
type StatsClient struct {
- sockAddr string
+ socket string
+ retryPeriod time.Duration
+ retryTimeout time.Duration
+
+ headerData []byte
- currentEpoch int64
- sharedHeader []byte
- directoryVector uintptr
- memorySize int
+ // defines the adapter connection state
+ connected uint32
+
+ // to quit socket monitor
+ done chan struct{}
+
+ statSegment
}
-// NewStatsClient returns new VPP stats API client.
-func NewStatsClient(socketName string) *StatsClient {
- return &StatsClient{
- sockAddr: socketName,
+// Option is a StatsClient option
+type Option func(*StatsClient)
+
+// SetSocketRetryPeriod is and optional parameter to define a custom
+// retry period while waiting for the VPP socket
+func SetSocketRetryPeriod(t time.Duration) Option {
+ return func(c *StatsClient) {
+ c.retryPeriod = t
}
}
-func (c *StatsClient) Connect() error {
- var sockName string
- if c.sockAddr == "" {
- sockName = adapter.DefaultStatsSocket
- } else {
- sockName = c.sockAddr
+// SetSocketRetryTimeout is and optional parameter to define a custom
+// timeout while waiting for the VPP socket
+func SetSocketRetryTimeout(t time.Duration) Option {
+ return func(c *StatsClient) {
+ c.retryTimeout = t
}
+}
- if _, err := os.Stat(sockName); err != nil {
- if os.IsNotExist(err) {
- return fmt.Errorf("stats socket file %q does not exists, ensure that VPP is running with `statseg { ... }` section in config", sockName)
- }
- return fmt.Errorf("stats socket file error: %v", err)
+// NewStatsClient returns a new StatsClient using socket.
+// If socket is empty string DefaultSocketName is used.
+func NewStatsClient(socket string, options ...Option) *StatsClient {
+ if socket == "" {
+ socket = DefaultSocketName
}
+ s := &StatsClient{
+ socket: socket,
+ }
+ for _, option := range options {
+ option(s)
+ }
+ if s.retryPeriod == 0 {
+ s.retryPeriod = DefaultSocketRetryPeriod
+ }
+ if s.retryTimeout == 0 {
+ s.retryTimeout = DefaultSocketRetryTimeout
+ }
+ return s
+}
- if err := c.statSegmentConnect(sockName); err != nil {
+// Connect to validated VPP stats socket and start monitoring
+// socket file changes
+func (sc *StatsClient) Connect() (err error) {
+ if err := sc.waitForSocket(); err != nil {
return err
}
-
+ sc.done = make(chan struct{})
+ if sc.statSegment, err = sc.connect(); err != nil {
+ return err
+ }
+ sc.monitorSocket()
return nil
}
-const statshmFilename = "statshm"
-
-func (c *StatsClient) statSegmentConnect(sockName string) error {
- addr := &net.UnixAddr{
- Net: "unixpacket",
- Name: sockName,
+// Disconnect from the socket, unmap shared memory and terminate
+// socket monitor
+func (sc *StatsClient) Disconnect() error {
+ if sc.headerData == nil {
+ return nil
}
+ if err := syscall.Munmap(sc.headerData); err != nil {
+ Log.Debugf("unmapping shared memory failed: %v", err)
+ return fmt.Errorf("unmapping shared memory failed: %v", err)
+ }
+ sc.headerData = nil
+ sc.done <- struct{}{}
- Log.Debugf("connecting to: %v", addr)
+ Log.Debugf("successfully unmapped shared memory")
+ return nil
+}
- conn, err := net.DialUnix(addr.Net, nil, addr)
- if err != nil {
- Log.Warnf("connecting to socket %s failed: %s", addr, err)
- return err
+func (sc *StatsClient) ListStats(patterns ...string) (entries []adapter.StatIdentifier, err error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
+ accessEpoch := sc.accessStart()
+ if accessEpoch == 0 {
+ return nil, adapter.ErrStatsAccessFailed
}
- defer func() {
- if err := conn.Close(); err != nil {
- Log.Warnf("closing socket failed: %v", err)
- }
- }()
-
- Log.Debugf("connected to socket: %v", addr)
- files, err := fd.Get(conn, 1, []string{statshmFilename})
+ entries, err = sc.getIdentifierEntries(patterns...)
if err != nil {
- return fmt.Errorf("getting file descriptor over socket failed: %v", err)
- } else if len(files) == 0 {
- return fmt.Errorf("no files received over socket")
+ return nil, err
}
- defer func() {
- for _, f := range files {
- if err := f.Close(); err != nil {
- Log.Warnf("closing file %s failed: %v", f.Name(), err)
- }
- }
- }()
- Log.Debugf("received %d files over socket", len(files))
+ if !sc.accessEnd(accessEpoch) {
+ return nil, adapter.ErrStatsDataBusy
+ }
+ return entries, nil
+}
- f := files[0]
+func (sc *StatsClient) DumpStats(patterns ...string) (entries []adapter.StatEntry, err error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
- info, err := f.Stat()
+ accessEpoch := sc.accessStart()
+ if accessEpoch == 0 {
+ return nil, adapter.ErrStatsAccessFailed
+ }
+
+ entries, err = sc.getStatEntries(patterns...)
if err != nil {
- return err
+ return nil, err
+ }
+
+ if !sc.accessEnd(accessEpoch) {
+ return nil, adapter.ErrStatsDataBusy
}
+ return entries, nil
+}
- size := int(info.Size())
+func (sc *StatsClient) PrepareDir(patterns ...string) (*adapter.StatDir, error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
- Log.Debugf("fd: name=%v size=%v", info.Name(), size)
+ accessEpoch := sc.accessStart()
+ if accessEpoch == 0 {
+ return nil, adapter.ErrStatsAccessFailed
+ }
- data, err := syscall.Mmap(int(f.Fd()), 0, size, syscall.PROT_READ, syscall.MAP_SHARED)
+ entries, err := sc.getStatEntries(patterns...)
if err != nil {
- Log.Warnf("mapping shared memory failed: %v", err)
- return fmt.Errorf("mapping shared memory failed: %v", err)
+ return nil, err
}
- Log.Debugf("successfuly mapped shared memory")
+ if !sc.accessEnd(accessEpoch) {
+ return nil, adapter.ErrStatsDataBusy
+ }
- c.sharedHeader = data
- c.memorySize = size
+ dir := &adapter.StatDir{
+ Epoch: accessEpoch,
+ Entries: entries,
+ }
- return nil
+ return dir, nil
}
-func (c *StatsClient) Disconnect() error {
- err := syscall.Munmap(c.sharedHeader)
+func (sc *StatsClient) PrepareDirOnIndex(indexes ...uint32) (*adapter.StatDir, error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
+
+ accessEpoch := sc.accessStart()
+ if accessEpoch == 0 {
+ return nil, adapter.ErrStatsAccessFailed
+ }
+ vector := sc.GetDirectoryVector()
+ if vector == nil {
+ return nil, fmt.Errorf("failed to prepare dir on index: directory vector is nil")
+ }
+ entries, err := sc.getStatEntriesOnIndex(vector, indexes...)
if err != nil {
- Log.Warnf("unmapping shared memory failed: %v", err)
- return fmt.Errorf("unmapping shared memory failed: %v", err)
+ return nil, err
}
- Log.Debugf("successfuly unmapped shared memory")
+ if !sc.accessEnd(accessEpoch) {
+ return nil, adapter.ErrStatsDataBusy
+ }
- return nil
+ dir := &adapter.StatDir{
+ Epoch: accessEpoch,
+ Entries: entries,
+ }
+
+ return dir, nil
}
-func nameMatches(name string, patterns []string) bool {
- if len(patterns) == 0 {
- return true
+// UpdateDir refreshes directory data for all counters
+func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) {
+ if !sc.isConnected() {
+ return adapter.ErrStatsDisconnected
+ }
+ epoch, _ := sc.GetEpoch()
+ if dir.Epoch != epoch {
+ return adapter.ErrStatsDirStale
+ }
+
+ accessEpoch := sc.accessStart()
+ if accessEpoch == 0 {
+ return adapter.ErrStatsAccessFailed
+ }
+ dirVector := sc.GetDirectoryVector()
+ if dirVector == nil {
+ return err
}
- for _, pattern := range patterns {
- matched, err := regexp.MatchString(pattern, name)
- if err == nil && matched {
- return true
+ for i := 0; i < len(dir.Entries); i++ {
+ if err := sc.updateStatOnIndex(&dir.Entries[i], dirVector); err != nil {
+ return err
}
}
- return false
+ if !sc.accessEnd(accessEpoch) {
+ return adapter.ErrStatsDataBusy
+ }
+ return nil
}
-func (c *StatsClient) ListStats(patterns ...string) (statNames []string, err error) {
- sa := c.accessStart()
- if sa == nil {
- return nil, fmt.Errorf("access failed")
+// checks the socket existence and waits for it for the designated
+// time if it is not available immediately
+func (sc *StatsClient) waitForSocket() error {
+ if _, err := os.Stat(sc.socket); err != nil {
+ if os.IsNotExist(err) {
+ n := time.Now()
+ ticker := time.NewTicker(sc.retryPeriod)
+ timeout := time.After(sc.retryTimeout)
+ for {
+ select {
+ case <-ticker.C:
+ if _, err := os.Stat(sc.socket); err == nil {
+ return nil
+ }
+ case <-timeout:
+ return fmt.Errorf("stats socket file %s is not ready within timeout (after %.2f s) ",
+ sc.socket, time.Since(n).Seconds())
+ }
+ }
+ } else {
+ return fmt.Errorf("stats socket error: %v", err)
+ }
}
+ return nil
+}
- dirOffset, _, _ := c.readOffsets()
- Log.Debugf("dirOffset: %v", dirOffset)
-
- vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirOffset]))
- Log.Debugf("vecLen: %v", vecLen)
- Log.Debugf("unsafe.Sizeof(statSegDirectoryEntry{}): %v", unsafe.Sizeof(statSegDirectoryEntry{}))
-
- for i := uint64(0); i < vecLen; i++ {
- offset := uintptr(i) * unsafe.Sizeof(statSegDirectoryEntry{})
- dirEntry := (*statSegDirectoryEntry)(add(unsafe.Pointer(&c.sharedHeader[dirOffset]), offset))
+// connect to the socket and map it into the memory. According to the
+// header version info, an appropriate segment handler is returned
+func (sc *StatsClient) connect() (ss statSegment, err error) {
+ addr := net.UnixAddr{
+ Net: "unixpacket",
+ Name: sc.socket,
+ }
+ Log.Debugf("connecting to: %v", addr)
- nul := bytes.IndexByte(dirEntry.name[:], '\x00')
- if nul < 0 {
- Log.Warnf("no zero byte found for: %q", dirEntry.name[:])
- continue
+ conn, err := net.DialUnix(addr.Net, nil, &addr)
+ if err != nil {
+ Log.Warnf("connecting to socket %s failed: %s", addr, err)
+ return nil, err
+ }
+ defer func() {
+ if err := conn.Close(); err != nil {
+ Log.Warnf("closing socket failed: %v", err)
}
- name := string(dirEntry.name[:nul])
+ }()
+ Log.Debugf("connected to socket")
- Log.Debugf(" %80q (type: %v, data: %d, offset: %d) ", name, dirEntry.directoryType, dirEntry.unionData, dirEntry.offsetVector)
+ files, err := fd.Get(conn, 1, nil)
+ if err != nil {
+ return nil, fmt.Errorf("getting file descriptor over socket failed: %v", err)
+ }
+ if len(files) == 0 {
+ return nil, fmt.Errorf("no files received over socket")
+ }
- if nameMatches(name, patterns) {
- statNames = append(statNames, name)
+ file := files[0]
+ defer func() {
+ if err := file.Close(); err != nil {
+ Log.Warnf("closing file failed: %v", err)
}
+ }()
- // TODO: copy the listed entries elsewhere
+ info, err := file.Stat()
+ if err != nil {
+ return nil, err
}
+ size := info.Size()
- if !c.accessEnd(sa) {
- return nil, adapter.ErrStatDirBusy
+ sc.headerData, err = syscall.Mmap(int(file.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
+ if err != nil {
+ Log.Debugf("mapping shared memory failed: %v", err)
+ return nil, fmt.Errorf("mapping shared memory failed: %v", err)
+ }
+ Log.Debugf("successfully mmapped shared memory segment (size: %v) %v", size, len(sc.headerData))
+
+ version := getVersion(sc.headerData)
+ switch version {
+ case 1:
+ ss = newStatSegmentV1(sc.headerData, size)
+ case 2:
+ ss = newStatSegmentV2(sc.headerData, size)
+ default:
+ return nil, fmt.Errorf("stat segment version is not supported: %v (min: %v, max: %v)",
+ version, minVersion, maxVersion)
}
- c.currentEpoch = sa.epoch
+ // set connected
+ atomic.CompareAndSwapUint32(&sc.connected, 0, 1)
- return statNames, nil
+ return ss, nil
}
-func (c *StatsClient) DumpStats(patterns ...string) (entries []*adapter.StatEntry, err error) {
- epoch, _ := c.readEpoch()
- if c.currentEpoch > 0 && c.currentEpoch != epoch { // TODO: do list stats before dump
- return nil, fmt.Errorf("old data")
+// reconnect disconnects from the socket, re-validates it and
+// connects again
+func (sc *StatsClient) reconnect() (err error) {
+ if err = sc.disconnect(); err != nil {
+ return fmt.Errorf("error disconnecting socket: %v", err)
}
-
- sa := c.accessStart()
- if sa == nil {
- return nil, fmt.Errorf("access failed")
+ if err = sc.waitForSocket(); err != nil {
+ return fmt.Errorf("error while waiting on socket: %v", err)
}
-
- dirOffset, _, _ := c.readOffsets()
- vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirOffset]))
-
- for i := uint64(0); i < vecLen; i++ {
- offset := uintptr(i) * unsafe.Sizeof(statSegDirectoryEntry{})
- dirEntry := (*statSegDirectoryEntry)(add(unsafe.Pointer(&c.sharedHeader[dirOffset]), offset))
-
- entry := c.copyData(dirEntry)
- if nameMatches(entry.Name, patterns) {
- entries = append(entries, &entry)
- }
+ if sc.statSegment, err = sc.connect(); err != nil {
+ return fmt.Errorf("error connecting socket: %v", err)
}
+ return nil
+}
- if !c.accessEnd(sa) {
- return nil, adapter.ErrStatDumpBusy
+// disconnect unmaps socket data from the memory and resets the header
+func (sc *StatsClient) disconnect() error {
+ if !atomic.CompareAndSwapUint32(&sc.connected, 1, 0) {
+ return fmt.Errorf("stats client is already disconnected")
+ }
+ if sc.headerData == nil {
+ return nil
+ }
+ if err := syscall.Munmap(sc.headerData); err != nil {
+ Log.Debugf("unmapping shared memory failed: %v", err)
+ return fmt.Errorf("unmapping shared memory failed: %v", err)
}
+ sc.headerData = nil
- return entries, nil
+ Log.Debugf("successfully unmapped shared memory")
+ return nil
}
-func (c *StatsClient) copyData(dirEntry *statSegDirectoryEntry) (statEntry adapter.StatEntry) {
- name := dirEntry.name[:]
- if nul := bytes.IndexByte(name, '\x00'); nul < 0 {
- Log.Warnf("no zero byte found for: %q", dirEntry.name[:])
- } else {
- name = dirEntry.name[:nul]
+func (sc *StatsClient) monitorSocket() {
+ watcher, err := fsnotify.NewWatcher()
+ if err != nil {
+ Log.Errorf("error starting socket monitor: %v", err)
+ return
}
- statEntry.Name = string(name)
- statEntry.Type = adapter.StatType(dirEntry.directoryType)
-
- Log.Debugf(" - %s (type: %v, data: %v, offset: %v) ", statEntry.Name, statEntry.Type, dirEntry.unionData, dirEntry.offsetVector)
-
- switch statEntry.Type {
- case adapter.ScalarIndex:
- statEntry.Data = adapter.ScalarStat(dirEntry.unionData)
-
- case adapter.ErrorIndex:
- _, errOffset, _ := c.readOffsets()
- offsetVector := unsafe.Pointer(&c.sharedHeader[errOffset])
- vecLen := vectorLen(offsetVector)
-
- var errData adapter.Counter
- for i := uint64(0); i < vecLen; i++ {
- cb := *(*uint64)(add(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))
- errData += val
- }
- statEntry.Data = adapter.ErrorStat(errData)
-
- case adapter.SimpleCounterVector:
- if dirEntry.unionData == 0 {
- Log.Debugf("\toffset is not valid")
- break
- } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) {
- Log.Debugf("\toffset out of range")
- break
- }
-
- simpleCounter := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset
- vecLen := vectorLen(simpleCounter)
- offsetVector := add(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))))
- counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)])
- vecLen2 := vectorLen(counterVec)
- 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)
- }
- }
- statEntry.Data = adapter.SimpleCounterStat(data)
-
- case adapter.CombinedCounterVector:
- if dirEntry.unionData == 0 {
- Log.Debugf("\toffset is not valid")
- break
- } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) {
- Log.Debugf("\toffset out of range")
- break
- }
-
- combinedCounter := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset
- vecLen := vectorLen(combinedCounter)
- offsetVector := add(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))))
- counterVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)])
- vecLen2 := vectorLen(counterVec)
- 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)
- }
- }
- statEntry.Data = adapter.CombinedCounterStat(data)
-
- case adapter.NameVector:
- if dirEntry.unionData == 0 {
- Log.Debugf("\toffset is not valid")
- break
- } else if dirEntry.unionData >= uint64(len(c.sharedHeader)) {
- Log.Debugf("\toffset out of range")
- break
- }
-
- nameVector := unsafe.Pointer(&c.sharedHeader[dirEntry.unionData]) // offset
- vecLen := vectorLen(nameVector)
- offsetVector := add(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))))
- nameVec := unsafe.Pointer(&c.sharedHeader[uintptr(cb)])
- vecLen2 := vectorLen(nameVec)
-
- var nameStr []byte
- for j := uint64(0); j < vecLen2; j++ {
- offset := uintptr(j) * unsafe.Sizeof(byte(0))
- val := *(*byte)(add(nameVec, offset))
- if val > 0 {
- nameStr = append(nameStr, val)
+ go func() {
+ for {
+ select {
+ case event := <-watcher.Events:
+ if event.Op == fsnotify.Remove && event.Name == sc.socket {
+ if err := sc.reconnect(); err != nil {
+ Log.Errorf("error occurred during socket reconnect: %v", err)
+ }
}
+ case err := <-watcher.Errors:
+ Log.Errorf("socket monitor delivered error event: %v", err)
+ case <-sc.done:
+ err := watcher.Close()
+ Log.Debugf("socket monitor closed (error: %v)", err)
+ return
}
- data[i] = adapter.Name(nameStr)
}
- statEntry.Data = adapter.NameStat(data)
+ }()
- default:
- Log.Warnf("Unknown type %d for stat entry: %s", statEntry.Type, statEntry.Name)
+ if err := watcher.Add(filepath.Dir(sc.socket)); err != nil {
+ Log.Errorf("failed to add socket address to the watcher: %v", err)
}
-
- Log.Debugf("\tentry data: %#v", statEntry.Data)
-
- return statEntry
}
-type statDirectoryType int32
-
-func (t statDirectoryType) String() string {
- return adapter.StatType(t).String()
-}
+// Starts monitoring 'inProgress' field. Returns stats segment
+// access epoch when completed, or zero value if not finished
+// within MaxWaitInProgress
+func (sc *StatsClient) accessStart() (epoch int64) {
+ t := time.Now()
-type statSegDirectoryEntry struct {
- directoryType statDirectoryType
- // unionData can represent: offset, index or value
- unionData uint64
- offsetVector uint64
- name [128]byte
+ epoch, inProg := sc.GetEpoch()
+ for inProg {
+ if time.Since(t) > MaxWaitInProgress {
+ return int64(0)
+ }
+ time.Sleep(CheckDelayInProgress)
+ epoch, inProg = sc.GetEpoch()
+ }
+ return epoch
}
-type statSegSharedHeader struct {
- version uint64
- epoch int64
- inProgress int64
- directoryOffset int64
- errorOffset int64
- statsOffset int64
+// AccessEnd returns true if stats data reading was finished, false
+// otherwise
+func (sc *StatsClient) accessEnd(accessEpoch int64) bool {
+ epoch, inProgress := sc.GetEpoch()
+ if accessEpoch != epoch || inProgress {
+ return false
+ }
+ return true
}
-func (c *StatsClient) readVersion() uint64 {
- header := *(*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0]))
- version := atomic.LoadUint64(&header.version)
- return version
+// getStatEntries retrieves all stats matching desired patterns, or all stats if no pattern is provided.
+func (sc *StatsClient) getStatEntries(patterns ...string) (entries []adapter.StatEntry, err error) {
+ vector := sc.GetDirectoryVector()
+ if vector == nil {
+ return nil, fmt.Errorf("failed to get stat entries: directory vector is nil")
+ }
+ indexes, err := sc.listIndexes(vector, patterns...)
+ if err != nil {
+ return nil, err
+ }
+ return sc.getStatEntriesOnIndex(vector, indexes...)
}
-func (c *StatsClient) readEpoch() (int64, bool) {
- header := *(*statSegSharedHeader)(unsafe.Pointer(&c.sharedHeader[0]))
- epoch := atomic.LoadInt64(&header.epoch)
- inprog := atomic.LoadInt64(&header.inProgress)
- return epoch, inprog != 0
+// getIdentifierEntries retrieves all identifiers matching desired patterns, or all identifiers
+// if no pattern is provided.
+func (sc *StatsClient) getIdentifierEntries(patterns ...string) (identifiers []adapter.StatIdentifier, err error) {
+ vector := sc.GetDirectoryVector()
+ if vector == nil {
+ return nil, fmt.Errorf("failed to get identifier entries: directory vector is nil")
+ }
+ indexes, err := sc.listIndexes(vector, patterns...)
+ if err != nil {
+ return nil, err
+ }
+ return sc.getIdentifierEntriesOnIndex(vector, indexes...)
}
-func (c *StatsClient) readOffsets() (dir, err, stat int64) {
- 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
+// getStatEntriesOnIndex retrieves stats on indexes, or all stats if indexes are not defined.
+func (sc *StatsClient) getStatEntriesOnIndex(vector dirVector, indexes ...uint32) (entries []adapter.StatEntry, err error) {
+ dirLen := *(*uint32)(vectorLen(vector))
+ for _, index := range indexes {
+ if index >= dirLen {
+ return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
+ }
+ dirPtr, dirName, dirType := sc.GetStatDirOnIndex(vector, index)
+ if len(dirName) == 0 {
+ return
+ }
+ var t adapter.StatType
+ d := sc.CopyEntryData(dirPtr, ^uint32(0))
+ if d != nil {
+ t = d.Type()
+ }
+ entries = append(entries, adapter.StatEntry{
+ StatIdentifier: adapter.StatIdentifier{
+ Index: index,
+ Name: dirName,
+ },
+ Type: t,
+ Data: d,
+ Symlink: adapter.StatType(dirType) == adapter.Symlink,
+ })
+ }
+ return entries, nil
}
-type statSegAccess struct {
- epoch int64
+// getIdentifierEntriesOnIndex retrieves identifiers on indexes, or all identifiers if indexes are not defined.
+func (sc *StatsClient) getIdentifierEntriesOnIndex(vector dirVector, indexes ...uint32) (identifiers []adapter.StatIdentifier, err error) {
+ dirLen := *(*uint32)(vectorLen(vector))
+ for _, index := range indexes {
+ if index >= dirLen {
+ return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
+ }
+ _, dirName, _ := sc.GetStatDirOnIndex(vector, index)
+ if len(dirName) == 0 {
+ return
+ }
+ identifiers = append(identifiers, adapter.StatIdentifier{
+ Index: index,
+ Name: dirName,
+ })
+ }
+ return identifiers, nil
}
-var maxWaitInProgress = 1 * time.Second
-
-func (c *StatsClient) accessStart() *statSegAccess {
- epoch, inprog := c.readEpoch()
- t := time.Now()
- for inprog {
- if time.Since(t) > maxWaitInProgress {
- return nil
+// listIndexes lists indexes for all stat entries that match any of the regex patterns.
+func (sc *StatsClient) listIndexes(vector dirVector, patterns ...string) (indexes []uint32, err error) {
+ if len(patterns) == 0 {
+ return sc.listIndexesFunc(vector, nil)
+ }
+ var regexes = make([]*regexp.Regexp, len(patterns))
+ for i, pattern := range patterns {
+ r, err := regexp.Compile(pattern)
+ if err != nil {
+ return nil, fmt.Errorf("compiling regexp failed: %v", err)
}
- epoch, inprog = c.readEpoch()
+ regexes[i] = r
}
- return &statSegAccess{
- epoch: epoch,
+ nameMatches := func(name dirName) bool {
+ for _, r := range regexes {
+ if r.Match(name) {
+ return true
+ }
+ }
+ return false
}
+ return sc.listIndexesFunc(vector, nameMatches)
}
-func (c *StatsClient) accessEnd(acc *statSegAccess) bool {
- epoch, inprog := c.readEpoch()
- if acc.epoch != epoch || inprog {
- return false
+// listIndexesFunc lists stats indexes. The optional function
+// argument filters returned values or returns all if empty
+func (sc *StatsClient) listIndexesFunc(vector dirVector, f func(name dirName) bool) (indexes []uint32, err error) {
+ if f == nil {
+ // there is around ~3157 stats, so to avoid too many allocations
+ // we set capacity to 3200 when listing all stats
+ indexes = make([]uint32, 0, 3200)
+ }
+ vecLen := *(*uint32)(vectorLen(vector))
+ for i := uint32(0); i < vecLen; i++ {
+ _, dirName, _ := sc.GetStatDirOnIndex(vector, i)
+ if f != nil {
+ if len(dirName) == 0 || !f(dirName) {
+ continue
+ }
+ }
+ indexes = append(indexes, i)
}
- return true
-}
-type vecHeader struct {
- length uint64
- vectorData [0]uint8
+ return indexes, nil
}
-func vectorLen(v unsafe.Pointer) uint64 {
- vec := *(*vecHeader)(unsafe.Pointer(uintptr(v) - unsafe.Sizeof(uintptr(0))))
- return vec.length
+func (sc *StatsClient) isConnected() bool {
+ return atomic.LoadUint32(&sc.connected) == 1
}
-//go:nosplit
-func add(p unsafe.Pointer, x uintptr) unsafe.Pointer {
- return unsafe.Pointer(uintptr(p) + x)
+// updateStatOnIndex refreshes the entry data.
+func (sc *StatsClient) updateStatOnIndex(entry *adapter.StatEntry, vector dirVector) (err error) {
+ dirLen := *(*uint32)(vectorLen(vector))
+ if entry.Index >= dirLen {
+ return fmt.Errorf("stat entry index %d out of dir vector length (%d)", entry.Index, dirLen)
+ }
+ dirPtr, dirName, dirType := sc.GetStatDirOnIndex(vector, entry.Index)
+ if len(dirName) == 0 ||
+ !bytes.Equal(dirName, entry.Name) ||
+ adapter.StatType(dirType) != entry.Type ||
+ entry.Data == nil {
+ return nil
+ }
+ if err := sc.UpdateEntryData(dirPtr, &entry.Data); err != nil {
+ err = fmt.Errorf("updating stat data for entry %s failed: %v", dirName, err)
+ }
+ return
}
Code Review / govpp.git / blobdiff