vendor/google.golang.org/grpc/internal/channelz/channelmap.go - voltha-go - Gitiles

 /*
  *
  * Copyright 2018 gRPC authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  *
  */

 package channelz

 import (
 	"fmt"
 	"sort"
 	"sync"
 	"time"
 )

 // entry represents a node in the channelz database.
 type entry interface {
 	// addChild adds a child e, whose channelz id is id to child list
 	addChild(id int64, e entry)
 	// deleteChild deletes a child with channelz id to be id from child list
 	deleteChild(id int64)
 	// triggerDelete tries to delete self from channelz database. However, if
 	// child list is not empty, then deletion from the database is on hold until
 	// the last child is deleted from database.
 	triggerDelete()
 	// deleteSelfIfReady check whether triggerDelete() has been called before,
 	// and whether child list is now empty. If both conditions are met, then
 	// delete self from database.
 	deleteSelfIfReady()
 	// getParentID returns parent ID of the entry. 0 value parent ID means no parent.
 	getParentID() int64
 	Entity
 }

 // channelMap is the storage data structure for channelz.
 //
 // Methods of channelMap can be divided into two categories with respect to
 // locking.
 //
 // 1. Methods acquire the global lock.
 // 2. Methods that can only be called when global lock is held.
 //
 // A second type of method need always to be called inside a first type of method.
 type channelMap struct {
 	mu               sync.RWMutex
 	topLevelChannels map[int64]struct{}
 	channels         map[int64]*Channel
 	subChannels      map[int64]*SubChannel
 	sockets          map[int64]*Socket
 	servers          map[int64]*Server
 }

 func newChannelMap() *channelMap {
 	return &channelMap{
 		topLevelChannels: make(map[int64]struct{}),
 		channels:         make(map[int64]*Channel),
 		subChannels:      make(map[int64]*SubChannel),
 		sockets:          make(map[int64]*Socket),
 		servers:          make(map[int64]*Server),
 	}
 }

 func (c *channelMap) addServer(id int64, s *Server) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	s.cm = c
 	c.servers[id] = s
 }

 func (c *channelMap) addChannel(id int64, cn *Channel, isTopChannel bool, pid int64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	cn.trace.cm = c
 	c.channels[id] = cn
 	if isTopChannel {
 		c.topLevelChannels[id] = struct{}{}
 	} else if p := c.channels[pid]; p != nil {
 		p.addChild(id, cn)
 	} else {
 		logger.Infof("channel %d references invalid parent ID %d", id, pid)
 	}
 }

 func (c *channelMap) addSubChannel(id int64, sc *SubChannel, pid int64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	sc.trace.cm = c
 	c.subChannels[id] = sc
 	if p := c.channels[pid]; p != nil {
 		p.addChild(id, sc)
 	} else {
 		logger.Infof("subchannel %d references invalid parent ID %d", id, pid)
 	}
 }

 func (c *channelMap) addSocket(s *Socket) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	s.cm = c
 	c.sockets[s.ID] = s
 	if s.Parent == nil {
 		logger.Infof("normal socket %d has no parent", s.ID)
 	}
 	s.Parent.(entry).addChild(s.ID, s)
 }

 // removeEntry triggers the removal of an entry, which may not indeed delete the
 // entry, if it has to wait on the deletion of its children and until no other
 // entity's channel trace references it.  It may lead to a chain of entry
 // deletion. For example, deleting the last socket of a gracefully shutting down
 // server will lead to the server being also deleted.
 func (c *channelMap) removeEntry(id int64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.findEntry(id).triggerDelete()
 }

 // tracedChannel represents tracing operations which are present on both
 // channels and subChannels.
 type tracedChannel interface {
 	getChannelTrace() *ChannelTrace
 	incrTraceRefCount()
 	decrTraceRefCount()
 	getRefName() string
 }

 // c.mu must be held by the caller
 func (c *channelMap) decrTraceRefCount(id int64) {
 	e := c.findEntry(id)
 	if v, ok := e.(tracedChannel); ok {
 		v.decrTraceRefCount()
 		e.deleteSelfIfReady()
 	}
 }

 // c.mu must be held by the caller.
 func (c *channelMap) findEntry(id int64) entry {
 	if v, ok := c.channels[id]; ok {
 		return v
 	}
 	if v, ok := c.subChannels[id]; ok {
 		return v
 	}
 	if v, ok := c.servers[id]; ok {
 		return v
 	}
 	if v, ok := c.sockets[id]; ok {
 		return v
 	}
 	return &dummyEntry{idNotFound: id}
 }

 // c.mu must be held by the caller
 //
 // deleteEntry deletes an entry from the channelMap. Before calling this method,
 // caller must check this entry is ready to be deleted, i.e removeEntry() has
 // been called on it, and no children still exist.
 func (c *channelMap) deleteEntry(id int64) entry {
 	if v, ok := c.sockets[id]; ok {
 		delete(c.sockets, id)
 		return v
 	}
 	if v, ok := c.subChannels[id]; ok {
 		delete(c.subChannels, id)
 		return v
 	}
 	if v, ok := c.channels[id]; ok {
 		delete(c.channels, id)
 		delete(c.topLevelChannels, id)
 		return v
 	}
 	if v, ok := c.servers[id]; ok {
 		delete(c.servers, id)
 		return v
 	}
 	return &dummyEntry{idNotFound: id}
 }

 func (c *channelMap) traceEvent(id int64, desc *TraceEvent) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	child := c.findEntry(id)
 	childTC, ok := child.(tracedChannel)
 	if !ok {
 		return
 	}
 	childTC.getChannelTrace().append(&traceEvent{Desc: desc.Desc, Severity: desc.Severity, Timestamp: time.Now()})
 	if desc.Parent != nil {
 		parent := c.findEntry(child.getParentID())
 		var chanType RefChannelType
 		switch child.(type) {
 		case *Channel:
 			chanType = RefChannel
 		case *SubChannel:
 			chanType = RefSubChannel
 		}
 		if parentTC, ok := parent.(tracedChannel); ok {
 			parentTC.getChannelTrace().append(&traceEvent{
 				Desc:      desc.Parent.Desc,
 				Severity:  desc.Parent.Severity,
 				Timestamp: time.Now(),
 				RefID:     id,
 				RefName:   childTC.getRefName(),
 				RefType:   chanType,
 			})
 			childTC.incrTraceRefCount()
 		}
 	}
 }

 type int64Slice []int64

 func (s int64Slice) Len() int           { return len(s) }
 func (s int64Slice) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
 func (s int64Slice) Less(i, j int) bool { return s[i] < s[j] }

 func copyMap(m map[int64]string) map[int64]string {
 	n := make(map[int64]string)
 	for k, v := range m {
 		n[k] = v
 	}
 	return n
 }

 func (c *channelMap) getTopChannels(id int64, maxResults int) ([]*Channel, bool) {
 	if maxResults <= 0 {
 		maxResults = EntriesPerPage
 	}
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	l := int64(len(c.topLevelChannels))
 	ids := make([]int64, 0, l)

 	for k := range c.topLevelChannels {
 		ids = append(ids, k)
 	}
 	sort.Sort(int64Slice(ids))
 	idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
 	end := true
 	var t []*Channel
 	for _, v := range ids[idx:] {
 		if len(t) == maxResults {
 			end = false
 			break
 		}
 		if cn, ok := c.channels[v]; ok {
 			t = append(t, cn)
 		}
 	}
 	return t, end
 }

 func (c *channelMap) getServers(id int64, maxResults int) ([]*Server, bool) {
 	if maxResults <= 0 {
 		maxResults = EntriesPerPage
 	}
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	ids := make([]int64, 0, len(c.servers))
 	for k := range c.servers {
 		ids = append(ids, k)
 	}
 	sort.Sort(int64Slice(ids))
 	idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
 	end := true
 	var s []*Server
 	for _, v := range ids[idx:] {
 		if len(s) == maxResults {
 			end = false
 			break
 		}
 		if svr, ok := c.servers[v]; ok {
 			s = append(s, svr)
 		}
 	}
 	return s, end
 }

 func (c *channelMap) getServerSockets(id int64, startID int64, maxResults int) ([]*Socket, bool) {
 	if maxResults <= 0 {
 		maxResults = EntriesPerPage
 	}
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	svr, ok := c.servers[id]
 	if !ok {
 		// server with id doesn't exist.
 		return nil, true
 	}
 	svrskts := svr.sockets
 	ids := make([]int64, 0, len(svrskts))
 	sks := make([]*Socket, 0, min(len(svrskts), maxResults))
 	for k := range svrskts {
 		ids = append(ids, k)
 	}
 	sort.Sort(int64Slice(ids))
 	idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= startID })
 	end := true
 	for _, v := range ids[idx:] {
 		if len(sks) == maxResults {
 			end = false
 			break
 		}
 		if ns, ok := c.sockets[v]; ok {
 			sks = append(sks, ns)
 		}
 	}
 	return sks, end
 }

 func (c *channelMap) getChannel(id int64) *Channel {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return c.channels[id]
 }

 func (c *channelMap) getSubChannel(id int64) *SubChannel {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return c.subChannels[id]
 }

 func (c *channelMap) getSocket(id int64) *Socket {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return c.sockets[id]
 }

 func (c *channelMap) getServer(id int64) *Server {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	return c.servers[id]
 }

 type dummyEntry struct {
 	// dummyEntry is a fake entry to handle entry not found case.
 	idNotFound int64
 	Entity
 }

 func (d *dummyEntry) String() string {
 	return fmt.Sprintf("non-existent entity #%d", d.idNotFound)
 }

 func (d *dummyEntry) ID() int64 { return d.idNotFound }

 func (d *dummyEntry) addChild(id int64, e entry) {
 	// Note: It is possible for a normal program to reach here under race
 	// condition.  For example, there could be a race between ClientConn.Close()
 	// info being propagated to addrConn and http2Client. ClientConn.Close()
 	// cancel the context and result in http2Client to error. The error info is
 	// then caught by transport monitor and before addrConn.tearDown() is called
 	// in side ClientConn.Close(). Therefore, the addrConn will create a new
 	// transport. And when registering the new transport in channelz, its parent
 	// addrConn could have already been torn down and deleted from channelz
 	// tracking, and thus reach the code here.
 	logger.Infof("attempt to add child of type %T with id %d to a parent (id=%d) that doesn't currently exist", e, id, d.idNotFound)
 }

 func (d *dummyEntry) deleteChild(id int64) {
 	// It is possible for a normal program to reach here under race condition.
 	// Refer to the example described in addChild().
 	logger.Infof("attempt to delete child with id %d from a parent (id=%d) that doesn't currently exist", id, d.idNotFound)
 }

 func (d *dummyEntry) triggerDelete() {
 	logger.Warningf("attempt to delete an entry (id=%d) that doesn't currently exist", d.idNotFound)
 }

 func (*dummyEntry) deleteSelfIfReady() {
 	// code should not reach here. deleteSelfIfReady is always called on an existing entry.
 }

 func (*dummyEntry) getParentID() int64 {
 	return 0
 }

 // Entity is implemented by all channelz types.
 type Entity interface {
 	isEntity()
 	fmt.Stringer
 	id() int64
 }
	/*
	*
	* Copyright 2018 gRPC authors.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*
	*/

	package channelz

	import (
	"fmt"
	"sort"
	"sync"
	"time"
	)

	// entry represents a node in the channelz database.
	type entry interface {
	// addChild adds a child e, whose channelz id is id to child list
	addChild(id int64, e entry)
	// deleteChild deletes a child with channelz id to be id from child list
	deleteChild(id int64)
	// triggerDelete tries to delete self from channelz database. However, if
	// child list is not empty, then deletion from the database is on hold until
	// the last child is deleted from database.
	triggerDelete()
	// deleteSelfIfReady check whether triggerDelete() has been called before,
	// and whether child list is now empty. If both conditions are met, then
	// delete self from database.
	deleteSelfIfReady()
	// getParentID returns parent ID of the entry. 0 value parent ID means no parent.
	getParentID() int64
	Entity
	}

	// channelMap is the storage data structure for channelz.
	//
	// Methods of channelMap can be divided into two categories with respect to
	// locking.
	//
	// 1. Methods acquire the global lock.
	// 2. Methods that can only be called when global lock is held.
	//
	// A second type of method need always to be called inside a first type of method.
	type channelMap struct {
	mu sync.RWMutex
	topLevelChannels map[int64]struct{}
	channels map[int64]*Channel
	subChannels map[int64]*SubChannel
	sockets map[int64]*Socket
	servers map[int64]*Server
	}

	func newChannelMap() *channelMap {
	return &channelMap{
	topLevelChannels: make(map[int64]struct{}),
	channels: make(map[int64]*Channel),
	subChannels: make(map[int64]*SubChannel),
	sockets: make(map[int64]*Socket),
	servers: make(map[int64]*Server),
	}
	}

	func (c channelMap) addServer(id int64, s Server) {
	c.mu.Lock()
	defer c.mu.Unlock()
	s.cm = c
	c.servers[id] = s
	}

	func (c channelMap) addChannel(id int64, cn Channel, isTopChannel bool, pid int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	cn.trace.cm = c
	c.channels[id] = cn
	if isTopChannel {
	c.topLevelChannels[id] = struct{}{}
	} else if p := c.channels[pid]; p != nil {
	p.addChild(id, cn)
	} else {
	logger.Infof("channel %d references invalid parent ID %d", id, pid)
	}
	}

	func (c channelMap) addSubChannel(id int64, sc SubChannel, pid int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	sc.trace.cm = c
	c.subChannels[id] = sc
	if p := c.channels[pid]; p != nil {
	p.addChild(id, sc)
	} else {
	logger.Infof("subchannel %d references invalid parent ID %d", id, pid)
	}
	}

	func (c channelMap) addSocket(s Socket) {
	c.mu.Lock()
	defer c.mu.Unlock()
	s.cm = c
	c.sockets[s.ID] = s
	if s.Parent == nil {
	logger.Infof("normal socket %d has no parent", s.ID)
	}
	s.Parent.(entry).addChild(s.ID, s)
	}

	// removeEntry triggers the removal of an entry, which may not indeed delete the
	// entry, if it has to wait on the deletion of its children and until no other
	// entity's channel trace references it. It may lead to a chain of entry
	// deletion. For example, deleting the last socket of a gracefully shutting down
	// server will lead to the server being also deleted.
	func (c *channelMap) removeEntry(id int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.findEntry(id).triggerDelete()
	}

	// tracedChannel represents tracing operations which are present on both
	// channels and subChannels.
	type tracedChannel interface {
	getChannelTrace() *ChannelTrace
	incrTraceRefCount()
	decrTraceRefCount()
	getRefName() string
	}

	// c.mu must be held by the caller
	func (c *channelMap) decrTraceRefCount(id int64) {
	e := c.findEntry(id)
	if v, ok := e.(tracedChannel); ok {
	v.decrTraceRefCount()
	e.deleteSelfIfReady()
	}
	}

	// c.mu must be held by the caller.
	func (c *channelMap) findEntry(id int64) entry {
	if v, ok := c.channels[id]; ok {
	return v
	}
	if v, ok := c.subChannels[id]; ok {
	return v
	}
	if v, ok := c.servers[id]; ok {
	return v
	}
	if v, ok := c.sockets[id]; ok {
	return v
	}
	return &dummyEntry{idNotFound: id}
	}

	// c.mu must be held by the caller
	//
	// deleteEntry deletes an entry from the channelMap. Before calling this method,
	// caller must check this entry is ready to be deleted, i.e removeEntry() has
	// been called on it, and no children still exist.
	func (c *channelMap) deleteEntry(id int64) entry {
	if v, ok := c.sockets[id]; ok {
	delete(c.sockets, id)
	return v
	}
	if v, ok := c.subChannels[id]; ok {
	delete(c.subChannels, id)
	return v
	}
	if v, ok := c.channels[id]; ok {
	delete(c.channels, id)
	delete(c.topLevelChannels, id)
	return v
	}
	if v, ok := c.servers[id]; ok {
	delete(c.servers, id)
	return v
	}
	return &dummyEntry{idNotFound: id}
	}

	func (c channelMap) traceEvent(id int64, desc TraceEvent) {
	c.mu.Lock()
	defer c.mu.Unlock()
	child := c.findEntry(id)
	childTC, ok := child.(tracedChannel)
	if !ok {
	return
	}
	childTC.getChannelTrace().append(&traceEvent{Desc: desc.Desc, Severity: desc.Severity, Timestamp: time.Now()})
	if desc.Parent != nil {
	parent := c.findEntry(child.getParentID())
	var chanType RefChannelType
	switch child.(type) {
	case *Channel:
	chanType = RefChannel
	case *SubChannel:
	chanType = RefSubChannel
	}
	if parentTC, ok := parent.(tracedChannel); ok {
	parentTC.getChannelTrace().append(&traceEvent{
	Desc: desc.Parent.Desc,
	Severity: desc.Parent.Severity,
	Timestamp: time.Now(),
	RefID: id,
	RefName: childTC.getRefName(),
	RefType: chanType,
	})
	childTC.incrTraceRefCount()
	}
	}
	}

	type int64Slice []int64

	func (s int64Slice) Len() int { return len(s) }
	func (s int64Slice) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
	func (s int64Slice) Less(i, j int) bool { return s[i] < s[j] }

	func copyMap(m map[int64]string) map[int64]string {
	n := make(map[int64]string)
	for k, v := range m {
	n[k] = v
	}
	return n
	}

	func (c channelMap) getTopChannels(id int64, maxResults int) ([]Channel, bool) {
	if maxResults <= 0 {
	maxResults = EntriesPerPage
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	l := int64(len(c.topLevelChannels))
	ids := make([]int64, 0, l)

	for k := range c.topLevelChannels {
	ids = append(ids, k)
	}
	sort.Sort(int64Slice(ids))
	idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
	end := true
	var t []*Channel
	for _, v := range ids[idx:] {
	if len(t) == maxResults {
	end = false
	break
	}
	if cn, ok := c.channels[v]; ok {
	t = append(t, cn)
	}
	}
	return t, end
	}

	func (c channelMap) getServers(id int64, maxResults int) ([]Server, bool) {
	if maxResults <= 0 {
	maxResults = EntriesPerPage
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	ids := make([]int64, 0, len(c.servers))
	for k := range c.servers {
	ids = append(ids, k)
	}
	sort.Sort(int64Slice(ids))
	idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= id })
	end := true
	var s []*Server
	for _, v := range ids[idx:] {
	if len(s) == maxResults {
	end = false
	break
	}
	if svr, ok := c.servers[v]; ok {
	s = append(s, svr)
	}
	}
	return s, end
	}

	func (c channelMap) getServerSockets(id int64, startID int64, maxResults int) ([]Socket, bool) {
	if maxResults <= 0 {
	maxResults = EntriesPerPage
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	svr, ok := c.servers[id]
	if !ok {
	// server with id doesn't exist.
	return nil, true
	}
	svrskts := svr.sockets
	ids := make([]int64, 0, len(svrskts))
	sks := make([]*Socket, 0, min(len(svrskts), maxResults))
	for k := range svrskts {
	ids = append(ids, k)
	}
	sort.Sort(int64Slice(ids))
	idx := sort.Search(len(ids), func(i int) bool { return ids[i] >= startID })
	end := true
	for _, v := range ids[idx:] {
	if len(sks) == maxResults {
	end = false
	break
	}
	if ns, ok := c.sockets[v]; ok {
	sks = append(sks, ns)
	}
	}
	return sks, end
	}

	func (c channelMap) getChannel(id int64) Channel {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.channels[id]
	}

	func (c channelMap) getSubChannel(id int64) SubChannel {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.subChannels[id]
	}

	func (c channelMap) getSocket(id int64) Socket {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.sockets[id]
	}

	func (c channelMap) getServer(id int64) Server {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.servers[id]
	}

	type dummyEntry struct {
	// dummyEntry is a fake entry to handle entry not found case.
	idNotFound int64
	Entity
	}

	func (d *dummyEntry) String() string {
	return fmt.Sprintf("non-existent entity #%d", d.idNotFound)
	}

	func (d *dummyEntry) ID() int64 { return d.idNotFound }

	func (d *dummyEntry) addChild(id int64, e entry) {
	// Note: It is possible for a normal program to reach here under race
	// condition. For example, there could be a race between ClientConn.Close()
	// info being propagated to addrConn and http2Client. ClientConn.Close()
	// cancel the context and result in http2Client to error. The error info is
	// then caught by transport monitor and before addrConn.tearDown() is called
	// in side ClientConn.Close(). Therefore, the addrConn will create a new
	// transport. And when registering the new transport in channelz, its parent
	// addrConn could have already been torn down and deleted from channelz
	// tracking, and thus reach the code here.
	logger.Infof("attempt to add child of type %T with id %d to a parent (id=%d) that doesn't currently exist", e, id, d.idNotFound)
	}

	func (d *dummyEntry) deleteChild(id int64) {
	// It is possible for a normal program to reach here under race condition.
	// Refer to the example described in addChild().
	logger.Infof("attempt to delete child with id %d from a parent (id=%d) that doesn't currently exist", id, d.idNotFound)
	}

	func (d *dummyEntry) triggerDelete() {
	logger.Warningf("attempt to delete an entry (id=%d) that doesn't currently exist", d.idNotFound)
	}

	func (*dummyEntry) deleteSelfIfReady() {
	// code should not reach here. deleteSelfIfReady is always called on an existing entry.
	}

	func (*dummyEntry) getParentID() int64 {
	return 0
	}

	// Entity is implemented by all channelz types.
	type Entity interface {
	isEntity()
	fmt.Stringer
	id() int64
	}