本文是关于如何分析共享内存的原理和VCS监控采集的实际情况。边肖觉得挺实用的,所以分享给大家学习。希望你看完这篇文章能有所收获。让我们和边肖一起看看。
00-1010共享内存广泛应用于Redis、Kafka、RabbitMQ等高性能组件。本文主要提供了一个共享内存中广告埋点数据采集的实用场景。
00-1010
一、前言
在Linux中,每个进程都有自己的进程控制块(PCB)和地址空间(Addr Space),每个进程都有对应的页表,页表负责映射进程的虚拟地址和物理地址,并通过内存管理单元(MMU)进行管理。两个不同的虚拟地址通过页表映射到物理空间的相同区域,它们指向的区域是共享内存。
当两个进程通过页表将虚拟地址映射到物理地址时,物理地址中有一个公共的内存区域,即共享内存,两个进程可以同时看到。这样,当一个进程写,另一个进程读时,就可以实现进程间通信。然而,我们想确保一个进程在写入时不能被读取,所以我们使用信号量来实现同步和互斥。
对于共享内存,采用引用计数原则。当进程离开共享内存区时,计数器减1。试管架成功后,计数器增加1。只有计数器变为零,才能删除。当进程终止时,附加到它的共享内存区域将自动分离。
00-1010共享内存可以说是进程间通信最有用的方式,也是IPC最快的形式,因为进程可以直接读写内存,而无需复制任何数据。对于流水线、消息队列等通信方式,需要在内核和用户空间复制数据四次,而共享内存只复制数据:两次,一次从输入文件复制到共享内存区域,一次从共享内存区域复制到输出文件。
事实上,当进程存储在共享内存中时,它们并不总是在读写少量数据后取消映射,然后在有新的通信时重新建立共享内存区域。取而代之的是,共享区一直保留到通信结束,这样数据内容就一直保留在共享内存中,文件不会被写回。未映射时,共享内存的内容通常会写回文件。因此,使用共享内存进行通信的效率非常高。
传统文件
UNIX访问文件的传统方法是用open打开文件。如果多个进程访问同一个文件,每个进程在其自己的地址空间中都包含该文件的副本,这就不必要地浪费了存储空间。
下图说明了两个进程同时读取文件同一页的情况。为了将页面从磁盘读取到缓存,每个进程在内存中执行另一个复制操作,将数据从缓存读取到自己的地址空间。
共享存储映射
现在考虑另一种处理方法:进程A和进程B都将页面映射到自己的地址空间,当进程A第一次访问页面中的数据时,会产生缺页中断。此时,内核将这个页面读入内存,并更新页面表以指向它。之后,当进程B访问同一个页面,出现缺页中断时,该页面已经在内存中,内核只需要将进程B的页表条目指向次页即可。
二、共享内存原理
(1)mmap()系统调用
mmap()系统调用使进程能够通过映射相同的普通文件来共享内存。普通文件映射到进程地址空间后,进程可以像普通内存一样访问文件,无需调用read()、write()等操作。
mmap()系统调用形式如下:
Void * mmap (void * addr,size _ tlen,intprot,intflags,intfd,off _ toffset) mmap用于将文件描述符fd指定的文件的[off,off len]区域映射到调用进程的[addr,addr len]的内存区域:
数字fd即将
映射到进程空间的文件描述字,一般由open()返回,同时,fd可以指定为-1,此时须指定flags参数中的,MAP_ANON,表明进行的是匿名映射(不涉及具体的文件名,避免了文件的创建及打开,很显然只能用于具有亲缘关系的进程间通信)。
len是映射到调用进程地址空间的字节数,它从被映射文件开头offset个字节开始算起。
prot 参数指定共享内存的访问权限。可取如下几个值的或:PROT_READ(可读) , PROT_WRITE (可写), PROT_EXEC (可执行), PROT_NONE(不可访问)。
flags由以下几个常值指定:MAP_SHARED , MAP_PRIVATE , MAP_FIXED,其中,MAP_SHARED , MAP_PRIVATE必选其一,而MAP_FIXED则不推荐使用。
offset参数一般设为0,表示从文件头开始映射。
参数addr指定文件应被映射到进程空间的起始地址,一般被指定一个空指针,此时选择起始地址的任务留给内核来完成。函数的返回值为最后文件映射到进程空间的地址,进程可直接操作起始地址为该值的有效地址。
(2)mmap()返回地址的访问
对mmap()返回地址的访问,linux采用的是页式管理机制。
对于用mmap()映射普通文件来说,进程会在自己的地址空间新增一块空间,空间大小由mmap()的len参数指定,注意,进程并不一定能够对全部新增空间都能进行有效访问。
进程能够访问的有效地址大小取决于文件被映射部分的大小。
简单的说,能够容纳文件被映射部分大小的最少页面个数决定了进程从mmap()返回的地址开始,能够有效访问的地址空间大小。
超过这个空间大小,内核会根据超过的严重程度返回发送不同的信号给进程。可用如下图示说明:
2、分区读写
为了要确保一个进程在写的时候不能被读,我们使用idx来标记可读块。
3、规则,指标和值
下图描述的是从连续内存空间转化成【规则,维度,值】语义的过程:
4、源码分析
5、general.proto
通用监控上报协议:
general.proto syntax = "proto2"; package general; message Data { map<string, string> kv = 1; } message GeneralData { optional string rule_id = 1; repeated Data data = 2; optional int64 count = 3; optional int64 left_size = 4; optional int32 version = 5; }
6、constant.go 配置参数
| 4k protect | magincNum1(4byte) | idx(4byte) | OssMapSz(1024*128byte)*2 | 4*64byte预留长度 | magincNum2(4byte) | 4k protect |
package moni_shm const ( OssShmId uint32 = 0x3eeff00 MagicNum1 uint32 = 0x650a218 MagicNum2 uint32 = 0x138a4f2 CreateShmLock = "/var/run/.oss_shm_lock" OssMapOneAttrCnt = 1024 * 128 //1024 个规则 OssOneAttrEntryCnt = 128 //每个规则有128个指标 EntrySz = 4 OssMapCnt = 2 OneAttrSz = OssOneAttrEntryCnt * EntrySz OssMapSz = OssMapOneAttrCnt * OneAttrSz OssAttrSz = OssMapSz*OssMapCnt + 4 + 4 + 64*4 + 4 defaultIntervalSec = 60 defaultTopic = "moni_general_shared_memory" )
7、util.go 工具类
内存清零工具和"整页"分配:
cd package moni_shm import ( "unsafe" ) //取整分配 func align(actual, to uint64) uint64 { return (actual + to - 1) / to * to } //连续空间清0 func zero(ptr uintptr, bts uint64) { if 0 == bts { return } const sz = 4096 var next uint64 cnt := 0 for ; next+sz <= bts; { //按页清零 arr := (*[sz]byte)(unsafe.Pointer(ptr)) for i := range *arr { (*arr)[i] = 0 } next += sz ptr += uintptr(sz) cnt++ } if next == bts { return } var i uintptr for i = 0; i < uintptr(bts-next); i++ { //剩余空间清零 *(*byte)(unsafe.Pointer(ptr + i)) = 0 } }
8、mgr.go 采集逻辑
共享内存采集逻辑对应 “规则指标和值”:
var ( _basePtr uintptr = 0 _shmUtil = NewShmUtil(OssShmId, OssAttrSz) _intervalSec = defaultIntervalSec _topic = defaultTopic _on bool = false ) func Stat(on bool) { _on = on } func Start() { go collect() //开始采集 } func tryInitBaseptr() error { var err error if _basePtr == 0 { _basePtr, err = _shmUtil.GetData() //获取当前共享内存数据块首地址 if nil != err { logrus.Warnf("init base ptr failed, retrying: %v", err) } } return err } func collect() { var ( cost time.Duration start time.Time first = true ) for { if !first { time.Sleep(time.Second*(time.Duration(_intervalSec)) - cost) //周期对齐 } first = false start = time.Now() if !_on { cost = time.Since(start) continue } if _basePtr == 0 { if err := tryInitBaseptr(); nil != err { cost = time.Since(start) continue } } d := collectOnce() for _, v := range d { moni_report.ProductReportData(*v) } cost = time.Since(start) } } func collectOnce() []*moni_report.ReportData { now := time.Now() var ret []*moni_report.ReportData data := make(map[uint32]*general.GeneralData) d := SwitchAndFetch(_basePtr) logrus.Infof("sending %d data from shm", len(d)) for _, v := range d { ruleId := strconv.FormatUint(uint64(v[0]), 10) dim := strconv.FormatUint(uint64(v[1]), 10) value := strconv.FormatUint(uint64(v[2]), 10) if _, ok := data[v[0]]; !ok { data[v[0]] = &general.GeneralData{ RuleId: proto.String(ruleId), Data: []*general.Data{}, } } data[v[0]].Data = append(data[v[0]].Data, &general.Data{ Kv: map[string]string{ dim: value, "timestamp": strconv.FormatInt(now.Unix()*1000, 10), "ip": viper.GetString("host.inner_ip"), }, }) } logrus.Infof("collect format shm data:%v", data) for _, v := range data { bts, err := proto.Marshal(v) if nil != err { logrus.Errorf("marshal shm data failed: %v", err) continue } ret = append(ret, &moni_report.ReportData{ DataBytes: bts, Topic: _topic, }) } return ret }
9、shmutil.go 共享内存操作
每60秒根据idx值切换可读区,采集后上报后,清零,切换到下一区。
package moni_shm import ( "fmt" "log" "os" "syscall" "unsafe" "github.com/sirupsen/logrus" ) const ( IpcCreate = 00001000 ) var ( ErrNotCreated = fmt.Errorf("shm not created") ErrCreateFailed = fmt.Errorf("shm create failed") ) type shmOpt func(*ShmUtil) func WithCreate(b bool) shmOpt { return func(u *ShmUtil) { u.create = b } } /*共享内存数据结构 |1page mprotect|page align data|1page mprotect| | 4k protect | magincNum1(4byte) | idx(4byte) | OssMapSz(1024*128byte)*2 | 4*64byte预留长度 | magincNum2(4byte) | 4k protect | */ type ShmUtil struct { pageSz int dataSz uint64 total uint64 shmKey uint32 create bool base uintptr data uintptr } func NewShmUtil(key uint32, sz uint64, cfgs ...shmOpt) *ShmUtil { if key == 0 { panic("invalid shm key: 0") } ret := &ShmUtil{ dataSz: sz, shmKey: key, } ret.pageSz = os.Getpagesize() //获取页大小 ret.dataSz = align(ret.dataSz, uint64(ret.pageSz)) //按页分配“包体”大小 ret.total = ret.dataSz + uint64(ret.pageSz)*2 // 总空间大小=包体大小 + 头尾各2页保护地址 for _, c := range cfgs { c(ret) } return ret } func (s *ShmUtil) attachShm(flag int) error { created := false shmid, _, errno := syscall.Syscall(syscall.SYS_SHMGET, uintptr(s.shmKey), uintptr(s.total), uintptr(flag)) //使用已存在的共享内存,返回共享内存标识符 if 0 != errno { return errno } if shmid < 0 { if !s.create { //不允创建,直接返回 return ErrNotCreated } shmid, _, errno = syscall.Syscall(syscall.SYS_SHMGET, uintptr(s.shmKey), uintptr(s.total), uintptr(flag|IpcCreate)) //新创建共享内存 if 0 != errno { return fmt.Errorf("shm create: %v", errno) } if shmid < 0 { return ErrCreateFailed } created = true } addr, _, errno := syscall.Syscall(syscall.SYS_SHMAT, shmid, 0, 0) //挂接共享内存到当前进程 if 0 != errno { return fmt.Errorf("shmat: %v", errno) } if created { zero(addr, s.total)//新创建的共享内存,初始化共享内存数据 } s.base = addr //记录共享内存首地址 用于之后的释放 s.data = s.base + uintptr(s.pageSz) //写数据的起始地址 _, _, errno = syscall.Syscall(syscall.SYS_MPROTECT, s.base, uintptr(s.pageSz), 0) if 0 != errno { //锁定共享内存头,锁指定的内存区间必须包含整个内存页(4K) s.detach() return fmt.Errorf("mprotect head: %v", errno) } _, _, errno = syscall.Syscall(syscall.SYS_MPROTECT, s.data+uintptr(s.dataSz), uintptr(s.pageSz), 0) //锁指定共享内存尾,区间开始的地址start必须是一个内存页的起始地址,并且区间长度len必须是页大小的整数倍。 if 0 != errno { s.detach() return fmt.Errorf("mprotect tail: %v", errno) } return nil } func (s *ShmUtil) detach() { //进程去关联共享内存 if 0 != s.base { syscall.Syscall(syscall.SYS_SHMDT, s.base, 0, 0) s.base = 0 s.data = 0 } } /* 获取内存并且返回数据段起始位置 s.create 决定是否新申请共享内存 */ func (s *ShmUtil) GetData() (uintptr, error) { if s.data != 0 { return s.data, nil } if err := s.attachShm(0666); nil != err { //初始化共享内存,并关联到进程 return 0, err } return s.data, nil } func SwitchAndFetch(ptr uintptr) [][3]uint32 { //从共享内存读取 [][3]uint32{ossid,key,value} if ptr == 0 { return nil } m1 := (*uint32)(unsafe.Pointer(ptr)) m2 := (*uint32)(unsafe.Pointer(ptr + 8 + OssMapSz*2 + 4*64)) if MagicNum1 != *m1 || MagicNum2 != *m2 { logrus.Errorf("magic 1 in header: wrote:%v\tread:%v\n", MagicNum1, *m1) logrus.Errorf("magic 2 in tail: wrote:%v\tread:%v\n", MagicNum2, *m2) return nil } idx := (*uint32)(unsafe.Pointer(ptr + 4)) //切换块标志 old := *idx *idx = 1 - *idx ret := PartialRead(ptr, old) //读取当前idx块数据 zero(ptr+8+uintptr(old)*OssMapSz, OssMapSz) //读完清0 return ret } //根据idx轮流读数据区域 func PartialRead(ptr uintptr, idx uint32) [][3]uint32 { //根据idx获取块起始地址 startPtr := ptr + 8 + uintptr(idx)*OssMapSz ret := ReadOssMap(startPtr) log.Printf("result: %v\n", ret) return ret } func ReadOssMap(ptr uintptr) [][3]uint32 { //1个周期内的指标总容量为 128*1024 = 128k = 13W var ret [][3]uint32 var i uint32 = 0 for i = 0; i < OssMapOneAttrCnt; i++ { //1个周期最多支持1024个业务 for _, v := range ReadOneAttr(ptr) { ret = append(ret, [3]uint32{i, v[0], v[1]}) // [osID,keyID,value] } ptr += OneAttrSz // OneAttrSz = OssOneAttrEntryCnt * EntrySz= 128*4 } return ret } func ReadOneAttr(ptr uintptr) [][2]uint32 { var ret [][2]uint32 var i uint32 = 0 for i = 0; i < OssOneAttrEntryCnt; i++ { //目前默认一个业务下最多有128单维度指标, OssOneAttrEntryCnt = 128 v := *(*uint32)(unsafe.Pointer(ptr)) if v != 0 { ret = append(ret, [2]uint32{i, v}) // [keyID, value] } ptr += EntrySz // 4yte 读取一个指标 } return ret }
以上就是怎么解析共享内存原理与VCS监控采集实战,小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注行业资讯频道。
内容来源网络,如有侵权,联系删除,本文地址:https://www.230890.com/zhan/148967.html