caches.go 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262
  1. // Copyright (C) 2016 The GoHBase Authors. All rights reserved.
  2. // This file is part of GoHBase.
  3. // Use of this source code is governed by the Apache License 2.0
  4. // that can be found in the COPYING file.
  5. package gohbase
  6. import (
  7. "bytes"
  8. "io"
  9. "sync"
  10. log "github.com/sirupsen/logrus"
  11. "github.com/tsuna/gohbase/hrpc"
  12. "modernc.org/b"
  13. )
  14. // clientRegionCache is client -> region cache. Used to quickly
  15. // look up all the regioninfos that map to a specific client
  16. type clientRegionCache struct {
  17. m sync.RWMutex
  18. regions map[hrpc.RegionClient]map[hrpc.RegionInfo]struct{}
  19. }
  20. // put associates a region with client for provided addrss. It returns the client if it's already
  21. // in cache or otherwise instantiates a new one by calling newClient.
  22. // TODO: obvious place for optimization (use map with address as key to lookup exisiting clients)
  23. func (rcc *clientRegionCache) put(addr string, r hrpc.RegionInfo,
  24. newClient func() hrpc.RegionClient) hrpc.RegionClient {
  25. rcc.m.Lock()
  26. for existingClient, regions := range rcc.regions {
  27. // check if client already exists, checking by host and port
  28. // because concurrent callers might try to put the same client
  29. if addr == existingClient.Addr() {
  30. // check client already knows about the region, checking
  31. // by pointer is enough because we make sure that there are
  32. // no regions with the same name around
  33. if _, ok := regions[r]; !ok {
  34. regions[r] = struct{}{}
  35. }
  36. rcc.m.Unlock()
  37. log.WithFields(log.Fields{
  38. "client": existingClient,
  39. }).Debug("region client is already in client's cache")
  40. return existingClient
  41. }
  42. }
  43. // no such client yet
  44. c := newClient()
  45. rcc.regions[c] = map[hrpc.RegionInfo]struct{}{r: struct{}{}}
  46. rcc.m.Unlock()
  47. log.WithField("client", c).Info("added new region client")
  48. return c
  49. }
  50. func (rcc *clientRegionCache) del(r hrpc.RegionInfo) {
  51. rcc.m.Lock()
  52. c := r.Client()
  53. if c != nil {
  54. r.SetClient(nil)
  55. regions := rcc.regions[c]
  56. delete(regions, r)
  57. }
  58. rcc.m.Unlock()
  59. }
  60. func (rcc *clientRegionCache) closeAll() {
  61. rcc.m.Lock()
  62. for client, regions := range rcc.regions {
  63. for region := range regions {
  64. region.MarkUnavailable()
  65. region.SetClient(nil)
  66. }
  67. client.Close()
  68. }
  69. rcc.m.Unlock()
  70. }
  71. func (rcc *clientRegionCache) clientDown(c hrpc.RegionClient) map[hrpc.RegionInfo]struct{} {
  72. rcc.m.Lock()
  73. downregions, ok := rcc.regions[c]
  74. delete(rcc.regions, c)
  75. rcc.m.Unlock()
  76. if ok {
  77. log.WithField("client", c).Info("removed region client")
  78. }
  79. return downregions
  80. }
  81. // key -> region cache.
  82. type keyRegionCache struct {
  83. m sync.RWMutex
  84. // Maps a []byte of a region start key to a hrpc.RegionInfo
  85. regions *b.Tree
  86. }
  87. func (krc *keyRegionCache) get(key []byte) ([]byte, hrpc.RegionInfo) {
  88. krc.m.RLock()
  89. enum, ok := krc.regions.Seek(key)
  90. if ok {
  91. krc.m.RUnlock()
  92. log.Fatalf("WTF: got exact match for region search key %q", key)
  93. return nil, nil
  94. }
  95. k, v, err := enum.Prev()
  96. enum.Close()
  97. krc.m.RUnlock()
  98. if err == io.EOF {
  99. // we are the beginning of the tree
  100. return nil, nil
  101. }
  102. return k.([]byte), v.(hrpc.RegionInfo)
  103. }
  104. func isRegionOverlap(regA, regB hrpc.RegionInfo) bool {
  105. // if region's stop key is empty, it's assumed to be the greatest key
  106. return bytes.Equal(regA.Namespace(), regB.Namespace()) &&
  107. bytes.Equal(regA.Table(), regB.Table()) &&
  108. (len(regB.StopKey()) == 0 || bytes.Compare(regA.StartKey(), regB.StopKey()) < 0) &&
  109. (len(regA.StopKey()) == 0 || bytes.Compare(regA.StopKey(), regB.StartKey()) > 0)
  110. }
  111. func (krc *keyRegionCache) getOverlaps(reg hrpc.RegionInfo) []hrpc.RegionInfo {
  112. var overlaps []hrpc.RegionInfo
  113. var v interface{}
  114. var err error
  115. // deal with empty tree in the beginning so that we don't have to check
  116. // EOF errors for enum later
  117. if krc.regions.Len() == 0 {
  118. return overlaps
  119. }
  120. // check if key created from new region falls into any cached regions
  121. key := createRegionSearchKey(fullyQualifiedTable(reg), reg.StartKey())
  122. enum, ok := krc.regions.Seek(key)
  123. if ok {
  124. log.Fatalf("WTF: found a region with exact name as the search key %q", key)
  125. }
  126. // case 1: landed before the first region in cache
  127. // enum.Prev() returns io.EOF
  128. // enum.Next() returns io.EOF
  129. // SeekFirst() + enum.Next() returns the first region, which has larger start key
  130. // case 2: landed before the second region in cache
  131. // enum.Prev() returns the first region X and moves pointer to -infinity
  132. // enum.Next() returns io.EOF
  133. // SeekFirst() + enum.Next() returns first region X, which has smaller start key
  134. // case 3: landed anywhere after the second region
  135. // enum.Prev() returns the region X before it landed, moves pointer to the region X - 1
  136. // enum.Next() returns X - 1 and move pointer to X, which has smaller start key
  137. enum.Prev()
  138. _, _, err = enum.Next()
  139. if err == io.EOF {
  140. // we are in the beginning of tree, get new enum starting
  141. // from first region
  142. enum.Close()
  143. enum, err = krc.regions.SeekFirst()
  144. if err != nil {
  145. log.Fatalf(
  146. "error seeking first region when getting overlaps for region %v: %v", reg, err)
  147. }
  148. }
  149. _, v, err = enum.Next()
  150. if isRegionOverlap(v.(hrpc.RegionInfo), reg) {
  151. overlaps = append(overlaps, v.(hrpc.RegionInfo))
  152. }
  153. _, v, err = enum.Next()
  154. // now append all regions that overlap until the end of the tree
  155. // or until they don't overlap
  156. for err != io.EOF && isRegionOverlap(v.(hrpc.RegionInfo), reg) {
  157. overlaps = append(overlaps, v.(hrpc.RegionInfo))
  158. _, v, err = enum.Next()
  159. }
  160. enum.Close()
  161. return overlaps
  162. }
  163. // put looks up if there's already region with this name in regions cache
  164. // and if there's, returns it in overlaps and doesn't modify the cache.
  165. // Otherwise, it puts the region and removes all overlaps in case all of
  166. // them are older. Returns a slice of overlapping regions and whether
  167. // passed region was put in the cache.
  168. func (krc *keyRegionCache) put(reg hrpc.RegionInfo) (overlaps []hrpc.RegionInfo, replaced bool) {
  169. krc.m.Lock()
  170. krc.regions.Put(reg.Name(), func(v interface{}, exists bool) (interface{}, bool) {
  171. if exists {
  172. // region is already in cache,
  173. // note: regions with the same name have the same age
  174. overlaps = []hrpc.RegionInfo{v.(hrpc.RegionInfo)}
  175. return nil, false
  176. }
  177. // find all entries that are overlapping with the range of the new region.
  178. overlaps = krc.getOverlaps(reg)
  179. for _, o := range overlaps {
  180. if o.ID() > reg.ID() {
  181. // overlapping region is younger,
  182. // don't replace any regions
  183. // TODO: figure out if there can a case where we might
  184. // have both older and younger overlapping regions, for
  185. // now we only replace if all overlaps are older
  186. return nil, false
  187. }
  188. }
  189. // all overlaps are older, put the new region
  190. replaced = true
  191. return reg, true
  192. })
  193. if !replaced {
  194. krc.m.Unlock()
  195. log.WithFields(log.Fields{
  196. "region": reg,
  197. "overlaps": overlaps,
  198. "replaced": replaced,
  199. }).Debug("region is already in cache")
  200. return
  201. }
  202. // delete overlapping regions
  203. // TODO: in case overlaps are always either younger or older,
  204. // we can just greedily remove them in Put function
  205. for _, o := range overlaps {
  206. krc.regions.Delete(o.Name())
  207. // let region establishers know that they can give up
  208. o.MarkDead()
  209. }
  210. krc.m.Unlock()
  211. log.WithFields(log.Fields{
  212. "region": reg,
  213. "overlaps": overlaps,
  214. "replaced": replaced,
  215. }).Info("added new region")
  216. return
  217. }
  218. func (krc *keyRegionCache) del(reg hrpc.RegionInfo) bool {
  219. krc.m.Lock()
  220. success := krc.regions.Delete(reg.Name())
  221. krc.m.Unlock()
  222. // let region establishers know that they can give up
  223. reg.MarkDead()
  224. log.WithFields(log.Fields{
  225. "region": reg,
  226. }).Debug("removed region")
  227. return success
  228. }