MOSN 源码解析 - 共享内存模型

func Alloc(name string, size int) (*ShmSpan, error) {
	...
	return NewShmSpan(name, data), nil
}

func Free(span *ShmSpan) error {
	Clear(span.name)
	return syscall.Munmap(span.origin)
}

func Clear(name string) error {
	return os.Remove(path(name))
}

type ShmSpan struct {
	origin []byte // mmap 返回的数组
	name   string // span 名, 创建时指定

	data   uintptr // 保存 mmap 内存段的首指针
	offset int // span 已经使用的字节长度
	size   int // span 大小
}

func initializeMetrics(config v2.MetricsConfig) {
	// init shm zone
	if config.ShmZone != "" && config.ShmSize > 0 {
		shm.InitDefaultMetricsZone(config.ShmZone, int(config.ShmSize), store.GetMosnState() != store.Active_Reconfiguring)
        ...

{
  ...
  "metrics": {
    ...
    "shm_zone": "文件名",
    "shm_size": "共享内存文件大小"
  },
  ...
}

type zone struct {
	span *shm.ShmSpan

	
	ref   *uint32

	set *hashSet // mutex + ref = 64bit, so atomic ops has no problem
}
...
type hashSet struct {
	entry []hashEntry
	meta  *meta
	slots []uint32
}
...
type hashEntry struct {
	metricsEntry
	next uint32

	// Prevents false sharing on widespread platforms with
	// 128 mod (cache line size) = 0 .
	pad [128 - unsafe.Sizeof(metricsEntry{})%128 - 4]byte
}

func newSharedMetrics(name string, size int) (*zone, error) {
	alignedSize := align(size, pageSize)

    // 申请 ShmSpan
	span, err := shm.Alloc(name, alignedSize)
	if err != nil {
		return nil, err
	}
	// 1. mutex and ref
	// 从 span 里取 4 个字节做互斥锁
	mutex, err := span.Alloc(4)
	if err != nil {
		return nil, err
	}

    // 从 span 里取 4 个字节做引用计数
	ref, err := span.Alloc(4)
	if err != nil {
		return nil, err
	}

	zone := &zone{
		span:  span,
		mutex: (*uint32)(unsafe.Pointer(mutex)),
		ref:   (*uint32)(unsafe.Pointer(ref)),
	}

	// 2. hashSet
	// 划分哈希表过程

	// assuming that 100 entries with 50 slots, so the ratio of occupied memory is
	// entries:slots  = 100 x 128 : 50 x 4 = 64 : 1
	// so assuming slots memory size is N, total allocated memory size is M, then we have:
	// M - 1024 < 65N + 28 <= M

    // 计算 slot 的数量和内存占用大小
	slotsNum := (alignedSize - 28) / (65 * 4)
	slotsSize := slotsNum * 4
	// 计算 entry 数量和内存占用大小
	entryNum := slotsNum * 2
	entrySize := slotsSize * 64
	
	// 哈希表内存大小 = entry 内存占用 + 20 字节 + slot 内存占用大小
	hashSegSize := entrySize + 20 + slotsSize
	hashSegment, err := span.Alloc(hashSegSize)
	if err != nil {
		return nil, err
	}

	// if zones's ref > 0, no need to initialize hashset's value
	// 初始化哈希表结构
	set, err := newHashSet(hashSegment, hashSegSize, entryNum, slotsNum, atomic.LoadUint32(zone.ref) == 0)
	if err != nil {
		return nil, err
	}
	zone.set = set

	// add ref
	atomic.AddUint32(zone.ref, 1)

	return zone, nil
}

type hashSet struct {
	entry []hashEntry
	meta  *meta
	slots []uint32
}

...


type meta struct {
	cap       uint32
	size      uint32
	freeIndex uint32

	slotsNum uint32
	bytesNum uint32
}

func NewShmCounterFunc(name string) func() gometrics.Counter {
	return func() gometrics.Counter {
		if defaultZone != nil {
			if entry, err := defaultZone.alloc(name); err == nil {
			...

func (z *zone) alloc(name string) (*hashEntry, error) {
	z.lock()
	defer z.unlock()

	entry, create := z.set.Alloc(name)
	...

func (s *hashSet) Alloc(name string) (*hashEntry, bool) {
	// 1. search existed slots and entries
	// 计算 hash 值作为 slot index
	h := hash(name)
	slot := h % s.meta.slotsNum

	// name convert if length exceeded
	if len(name) > maxNameLength {
		// if name is longer than max length, use hash_string as leading character
		// and the remaining maxNameLength - len(hash_string) bytes follows
		hStr := strconv.Itoa(int(h))
		name = hStr + name[len(hStr)+len(name)-maxNameLength:]
	}

	nameBytes := []byte(name)

    // 查找链表找到对应的 entry
	var entry *hashEntry
	for index := s.slots[slot]; index != sentinel; {
		entry = &s.entry[index]

		if entry.equalName(nameBytes) {
			return entry, false
		}

		index = entry.next
	}

	// 2. create new entry
	// 如果找不到, 创建新的 entry
	if s.meta.size >= s.meta.cap {
		return nil, false
	}

    // 创建新的 entry 从 hashset 的 meta 信息里拿 next free index
	newIndex := s.meta.freeIndex
	newEntry := &s.entry[newIndex]
	newEntry.assignName(nameBytes)
	newEntry.ref = 1

	if entry == nil {
		// 所以是链表头,保存 index 到 slot
		s.slots[slot] = newIndex
	} else {
		// 否则保存在上一个 entry 的 next 字段内
		entry.next = newIndex
	}

	s.meta.size++
	// 设置 next free index
	s.meta.freeIndex = newEntry.next
	// 设置队尾
	newEntry.next = sentinel

	return newEntry, true
}

func NewShmCounterFunc(name string) func() gometrics.Counter {
	return func() gometrics.Counter {
		if defaultZone != nil {
			if entry, err := defaultZone.alloc(name); err == nil {
			    return ShmCounter(unsafe.Pointer(&entry.value))
			}
			...

func initializeMetrics(config v2.MetricsConfig) {
	// init shm zone
	if config.ShmZone != "" && config.ShmSize > 0 {
		shm.InitDefaultMetricsZone(config.ShmZone, int(config.ShmSize), store.GetMosnState() != store.Active_Reconfiguring)
        ...

func createMetricsZone(name string, size int, clear bool) *zone {
	if clear {
		shm.Clear(name)
	}
    ...

func newSharedMetrics(name string, size int) (*zone, error) {
    ...
	set, err := newHashSet(hashSegment, hashSegSize, entryNum, slotsNum, atomic.LoadUint32(zone.ref) == 0)
	...

func newHashSet(segment uintptr, bytesNum, cap, slotsNum int, init bool) (*hashSet, error) {
	set := &hashSet{}
	...

func (z *zone) lock() {
	times := 0

	// 5ms spin interval, 5 times burst
	for {
		if atomic.CompareAndSwapUint32(z.mutex, 0, pid) {
			return
		}
	...