Go语言入门学习之Groupcache源码分析

type Hash func(data []byte) uint32

type Map struct {
	hash     Hash
	replicas int //每个服务器对应的虚拟节点的个数，默认为50个
	keys     []int // Sorted
	hashMap  map[int]string //虚拟节点到实际服务器的映射
}

//创建Map对象
func New(replicas int, fn Hash) *Map {
	m := &Map{
		replicas: replicas,
		hash:     fn,
		hashMap:  make(map[int]string),
	}
	if m.hash == nil {
		m.hash = crc32.ChecksumIEEE
	}
	return m
}

// Returns true if there are no items available.检测当前Map是否为空
func (m *Map) IsEmpty() bool {
	return len(m.keys) == 0
}

// Adds some keys to the hash.添加虚拟节点，
func (m *Map) Add(keys ...string) {
	for _, key := range keys {
		for i := 0; i < m.replicas; i++ {
			hash := int(m.hash([]byte(strconv.Itoa(i) + key)))
			m.keys = append(m.keys, hash)
			m.hashMap[hash] = key
		}
	}
	sort.Ints(m.keys)
}

// Gets the closest item in the hash to the provided key.根据key值获取hash值，获取最靠近hash值的服务器节点，此节点必须大于等于hash
func (m *Map) Get(key string) string {
	if m.IsEmpty() {
		return ""
	}

	hash := int(m.hash([]byte(key)))

	// Binary search for appropriate replica.
	idx := sort.Search(len(m.keys), func(i int) bool { return m.keys[i] >= hash })

	// Means we have cycled back to the first replica.
	if idx == len(m.keys) {
		idx = 0
	}

	return m.hashMap[m.keys[idx]]
}

// Cache is an LRU cache. It is not safe for concurrent access.
type Cache struct {
	// MaxEntries is the maximum number of cache entries before
	// an item is evicted. Zero means no limit.
	MaxEntries int

	// OnEvicted optionally specificies a callback function to be
	// executed when an entry is purged from the cache.
	OnEvicted func(key Key, value interface{})

	ll    *list.List //存储内容的链表
	cache map[interface{}]*list.Element //利用map快速查找内容
}


// RemoveOldest removes the oldest item from the cache.//根据list尾部的数据，去除map和list中的数据
func (c *Cache) RemoveOldest() {
	if c.cache == nil {
		return
	}
	ele := c.ll.Back()
	if ele != nil {
		c.removeElement(ele)
	}
}

//根据list.Element去除map和list中的数据
func (c *Cache) removeElement(e *list.Element) {
	c.ll.Remove(e)
	kv := e.Value.(*entry)
	delete(c.cache, kv.key)
	if c.OnEvicted != nil {
		c.OnEvicted(kv.key, kv.value)
	}
}

// call is an in-flight or completed Do call
type call struct {
	wg  sync.WaitGroup
	val interface{}
	err error
}

// Group represents a class of work and forms a namespace in which
// units of work can be executed with duplicate suppression.
type Group struct {
	mu sync.Mutex       // protects m
	m  map[string]*call // lazily initialized
}

// Do executes and returns the results of the given function, making
// sure that only one execution is in-flight for a given key at a
// time. If a duplicate comes in, the duplicate caller waits for the
// original to complete and receives the same results.
//此函数实现了groupcache.go中的flightGroup接口，功能为保证多个并发查询相同值的时候，只需要其中一个请求访问实际数据，然后其他多个请求一块共享结果
func (g *Group) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[string]*call)
	}
	if c, ok := g.m[key]; ok {
		g.mu.Unlock()
		c.wg.Wait()//如果找到了则等待
		return c.val, c.err
	}
	c := new(call)//如果没找到，则新创建一个call
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()

	c.val, c.err = fn()//获取实际的值
	c.wg.Done()

	g.mu.Lock()
	delete(g.m, key)//当并发请求完数据后，会存储到对应的缓存里，所以此Map里没必要存着此值占用空间
	g.mu.Unlock()

	return c.val, c.err
}

// ProtoGetter is the interface that must be implemented by a peer.//向服务器节点peer查询数据的接口
type ProtoGetter interface {
	Get(context Context, in *pb.GetRequest, out *pb.GetResponse) error
}

// PeerPicker is the interface that must be implemented to locate
// the peer that owns a specific key.
//为通过key值获取节点的接口
type PeerPicker interface {
	// PickPeer returns the peer that owns the specific key
	// and true to indicate that a remote peer was nominated.
	// It returns nil, false if the key owner is the current peer.
	PickPeer(key string) (peer ProtoGetter, ok bool)
}

// HTTPPool implements PeerPicker for a pool of HTTP peers.
type HTTPPool struct {
	// Context optionally specifies a context for the server to use when it
	// receives a request.
	// If nil, the server uses a nil Context.
	Context func(*http.Request) Context

	// Transport optionally specifies an http.RoundTripper for the client
	// to use when it makes a request.
	// If nil, the client uses http.DefaultTransport.
	Transport func(Context) http.RoundTripper

	// this peer's base URL, e.g. "https://example.net:8000"
	self string //当前节点的base url

	// opts specifies the options.
	opts HTTPPoolOptions

	mu          sync.Mutex // guards peers and httpGetters
	peers       *consistenthash.Map //分布式peers，一致性哈希对象，通过此获取key值对应的节点的base url
	httpGetters map[string]*httpGetter // keyed by e.g. "http://10.0.0.2:8008"httpGetter为实现peers.go中ProtoGetter接口的对象，此map的key为节点base url
}

// HTTPPoolOptions are the configurations of a HTTPPool.
type HTTPPoolOptions struct {
	// BasePath specifies the HTTP path that will serve groupcache requests.
	// If blank, it defaults to "/_groupcache/".
	BasePath string //url的基本path

	// Replicas specifies the number of key replicas on the consistent hash.
	// If blank, it defaults to 50.
	Replicas int//一致性哈希的虚拟节点的个数

	// HashFn specifies the hash function of the consistent hash.
	// If blank, it defaults to crc32.ChecksumIEEE.
	HashFn consistenthash.Hash //哈希方法
}

func NewHTTPPool(self string) *HTTPPool {
	p := NewHTTPPoolOpts(self, nil)
	http.Handle(p.opts.BasePath, p)
	return p
}

var httpPoolMade bool

// NewHTTPPoolOpts initializes an HTTP pool of peers with the given options.
// Unlike NewHTTPPool, this function does not register the created pool as an HTTP handler.
// The returned *HTTPPool implements http.Handler and must be registered using http.Handle.
func NewHTTPPoolOpts(self string, o *HTTPPoolOptions) *HTTPPool {
	if httpPoolMade {
		panic("groupcache: NewHTTPPool must be called only once")
	}
	httpPoolMade = true

	p := &HTTPPool{
		self:        self,
		httpGetters: make(map[string]*httpGetter),
	}
	if o != nil {
		p.opts = *o
	}
	if p.opts.BasePath == "" {
		p.opts.BasePath = defaultBasePath//默认url路径
	}
	if p.opts.Replicas == 0 {
		p.opts.Replicas = defaultReplicas//默认虚拟节点个数
	}
	p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn)

	RegisterPeerPicker(func() PeerPicker { return p })
	return p
}
//注册HTTPPool为当前服务器的PeerPicker
func RegisterPeerPicker(fn func() PeerPicker) {
	if portPicker != nil {
		panic("RegisterPeerPicker called more than once")
	}
	portPicker = func(_ string) PeerPicker { return fn() }
}

//httpGetter实现了peers.go的ProtoGetter接口，这个接口实际上是向服务器节点查询数据的接口
type httpGetter struct {
	transport func(Context) http.RoundTripper
	baseURL   string //对应节点的base url
}

var bufferPool = sync.Pool{
	New: func() interface{} { return new(bytes.Buffer) },
}

//Get函数可以通过in里面的key值来将获取的value值存放在Out里面，利用url访问对应的服务器节点获取数据
func (h *httpGetter) Get(context Context, in *pb.GetRequest, out *pb.GetResponse) error{
    u := fmt.Sprintf(
		"%v%v/%v",
		h.baseURL,
		url.QueryEscape(in.GetGroup()),
		url.QueryEscape(in.GetKey()),
	)//获取远程服务器节点的url
	req, err := http.NewRequest("GET", u, nil)//获取对应的数据
	...
}

//设置各个节点到一致性哈希中，并且根据节点设置对应的httpGetter
func (p *HTTPPool) Set(peers ...string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.peers = consistenthash.New(p.opts.Replicas, p.opts.HashFn)
	p.peers.Add(peers...)
	p.httpGetters = make(map[string]*httpGetter, len(peers))
	for _, peer := range peers {
		p.httpGetters[peer] = &httpGetter{transport: p.Transport, baseURL: peer + p.opts.BasePath}
	}
}

//根据key值获取对应的httpGetter
func (p *HTTPPool) PickPeer(key string) (ProtoGetter, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.peers.IsEmpty() {
		return nil, false
	}
	if peer := p.peers.Get(key); peer != p.self {
		return p.httpGetters[peer], true
	}
	return nil, false
}

//为当前服务节点的访问路由函数
func (p *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// Parse request.
	if !strings.HasPrefix(r.URL.Path, p.opts.BasePath) {
		panic("HTTPPool serving unexpected path: " + r.URL.Path)
	}
	parts := strings.SplitN(r.URL.Path[len(p.opts.BasePath):], "/", 2)
	if len(parts) != 2 {
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}
	groupName := parts[0]
	key := parts[1]

	// Fetch the value for this group/key.
	group := GetGroup(groupName)//利用groupname获取对应的Group对象
	if group == nil {
		http.Error(w, "no such group: "+groupName, http.StatusNotFound)
		return
	}
	var ctx Context
	if p.Context != nil {
		ctx = p.Context(r)
	}

	group.Stats.ServerRequests.Add(1)
	var value []byte
	err := group.Get(ctx, key, AllocatingByteSliceSink(&value))//此为函数的重点部分，group对象的Get函数为根据key值获取数据值的总函数，在groupCache.go文档中来讲。
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// Write the value to the response body as a proto message.
	body, err := proto.Marshal(&pb.GetResponse{Value: value})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/x-protobuf")
	w.Write(body)
}

type Group struct {
	name       string //组的名字
	getter     Getter //为使用用户需要实现的部分，此Getter接口主要是用户指定获取原始数据的接口
	peersOnce  sync.Once
	peers      PeerPicker //为之前http.go中NewHTTPPoolOpts注册的HTTPPool对象
	cacheBytes int64 // limit for sum of mainCache and hotCache size
	mainCache cache //主存，为本地服务器节点放置的缓存key-value
	hotCache cache //热存储，设置这个的目的是，因为根据一致性哈希，每个key/value存储在不同的服务器节点的mainCache里，但是如果多次经过此服务器节点二次访问key对应的实际远程服务器节点获取值的时候会造成耗费增大，所以有1/10的概率会将此key-val存储在当前hotCache来加速查询
	loadGroup flightGroup //当多个请求并发访问同一个key对应的val值，并且mainCache和hotCache里都查不到key对应的val时候，LoadGroup实现了将多次并发请求原始数据融合为一次请求，减少原始数据服务器压力
	_ int32 // force Stats to be 8-byte aligned on 32-bit platforms
	// Stats are statistics on the group.
	Stats Stats //服务器节点访问状态
}

//为获取数据值的总流程
func (g *Group) Get(ctx Context, key string, dest Sink) error {
	g.peersOnce.Do(g.initPeers)//初始化peers,为之前注册的HTTPPool对象
	g.Stats.Gets.Add(1)
	if dest == nil {
		return errors.New("groupcache: nil dest Sink")
	}
	value, cacheHit := g.lookupCache(key)//这个是查询本地的mainCache和hotCache,如果有Key对应的val，则返回，没有的话则程序继续向下走

	if cacheHit {
		g.Stats.CacheHits.Add(1)
		return setSinkView(dest, value)
	}
	destPopulated := false
	//load函数主要是分为两个步骤，一是判断当前key是都对应group本身的这个服务器节点，如果是，则直接从用户定义的Getter接口获取数据并存储到当前组mainCache里，如果不是，则从对应的服务器节点获取数据，并有1/10的概率存储在当前组的hotCache里。
	value, destPopulated, err := g.load(ctx, key, dest)
	if err != nil {
		return err
	}
	if destPopulated {
		return nil
	}
	return setSinkView(dest, value)
}

//在缓存查询数据
func (g *Group) lookupCache(key string) (value ByteView, ok bool) {
	if g.cacheBytes <= 0 {
		return
	}
	value, ok = g.mainCache.get(key)
	if ok {
		return
	}
	value, ok = g.hotCache.get(key)
	return
}

func (g *Group) load(ctx Context, key string, dest Sink) (value ByteView, destPopulated bool, err error) {
	g.Stats.Loads.Add(1)
	//Do函数实现了并发合并操作
	viewi, err := g.loadGroup.Do(key, func() (interface{}, error) {
		if value, cacheHit := g.lookupCache(key); cacheHit {
			g.Stats.CacheHits.Add(1)
			return value, nil
		}
		g.Stats.LoadsDeduped.Add(1)
		var value ByteView
		var err error
		//获取对应的服务器节点，如果节点为远程服务器节点，则通过getFromPeer访问远程服务器节点获取数据
		if peer, ok := g.peers.PickPeer(key); ok {
			value, err = g.getFromPeer(ctx, peer, key)
			if err == nil {
				g.Stats.PeerLoads.Add(1)
				return value, nil
			}
			g.Stats.PeerErrors.Add(1)
		}
		//获取的服务器节点问本地服务器节点，所以通过getLocally函数从用户定义的Getter接口获取原始数据
		value, err = g.getLocally(ctx, key, dest)
		if err != nil {
			g.Stats.LocalLoadErrs.Add(1)
			return nil, err
		}
		g.Stats.LocalLoads.Add(1)
		destPopulated = true // only one caller of load gets this return value
		//填充mainCache缓存
		g.populateCache(key, value, &g.mainCache)
		return value, nil
	})
	if err == nil {
		value = viewi.(ByteView)
	}
	return
}