ItemList.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516
  1. package redis3
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "github.com/redis/go-redis/v9"
  7. "github.com/seaweedfs/seaweedfs/weed/util/skiplist"
  8. )
  9. type ItemList struct {
  10. skipList *skiplist.SkipList
  11. batchSize int
  12. client redis.UniversalClient
  13. prefix string
  14. }
  15. func newItemList(client redis.UniversalClient, prefix string, store skiplist.ListStore, batchSize int) *ItemList {
  16. return &ItemList{
  17. skipList: skiplist.New(store),
  18. batchSize: batchSize,
  19. client: client,
  20. prefix: prefix,
  21. }
  22. }
  23. /*
  24. Be reluctant to create new nodes. Try to fit into either previous node or next node.
  25. Prefer to add to previous node.
  26. There are multiple cases after finding the name for greater or equal node
  27. 1. found and node.Key == name
  28. The node contains a batch with leading key the same as the name
  29. nothing to do
  30. 2. no such node found or node.Key > name
  31. if no such node found
  32. prevNode = list.LargestNode
  33. // case 2.1
  34. if previousNode contains name
  35. nothing to do
  36. // prefer to add to previous node
  37. if prevNode != nil {
  38. // case 2.2
  39. if prevNode has capacity
  40. prevNode.add name, and save
  41. return
  42. // case 2.3
  43. split prevNode by name
  44. }
  45. // case 2.4
  46. // merge into next node. Avoid too many nodes if adding data in reverse order.
  47. if nextNode is not nil and nextNode has capacity
  48. delete nextNode.Key
  49. nextNode.Key = name
  50. nextNode.batch.add name
  51. insert nodeNode.Key
  52. return
  53. // case 2.5
  54. if prevNode is nil
  55. insert new node with key = name, value = batch{name}
  56. return
  57. */
  58. func (nl *ItemList) canAddMember(node *skiplist.SkipListElementReference, name string) (alreadyContains bool, nodeSize int, err error) {
  59. ctx := context.Background()
  60. pipe := nl.client.TxPipeline()
  61. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  62. countOperation := pipe.ZLexCount(ctx, key, "-", "+")
  63. scoreOperationt := pipe.ZScore(ctx, key, name)
  64. if _, err = pipe.Exec(ctx); err != nil && err != redis.Nil {
  65. return false, 0, err
  66. }
  67. if err == redis.Nil {
  68. err = nil
  69. }
  70. alreadyContains = scoreOperationt.Err() == nil
  71. nodeSize = int(countOperation.Val())
  72. return
  73. }
  74. func (nl *ItemList) WriteName(name string) error {
  75. lookupKey := []byte(name)
  76. prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
  77. if err != nil {
  78. return err
  79. }
  80. // case 1: the name already exists as one leading key in the batch
  81. if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
  82. return nil
  83. }
  84. var prevNodeReference *skiplist.SkipListElementReference
  85. if !found {
  86. prevNodeReference = nl.skipList.GetLargestNodeReference()
  87. }
  88. if nextNode != nil && prevNode == nil {
  89. prevNodeReference = nextNode.Prev
  90. }
  91. if prevNodeReference != nil {
  92. alreadyContains, nodeSize, err := nl.canAddMember(prevNodeReference, name)
  93. if err != nil {
  94. return err
  95. }
  96. if alreadyContains {
  97. // case 2.1
  98. return nil
  99. }
  100. // case 2.2
  101. if nodeSize < nl.batchSize {
  102. return nl.NodeAddMember(prevNodeReference, name)
  103. }
  104. // case 2.3
  105. x := nl.NodeInnerPosition(prevNodeReference, name)
  106. y := nodeSize - x
  107. addToX := x <= y
  108. // add to a new node
  109. if x == 0 || y == 0 {
  110. if err := nl.ItemAdd(lookupKey, 0, name); err != nil {
  111. return err
  112. }
  113. return nil
  114. }
  115. if addToX {
  116. // collect names before name, add them to X
  117. namesToX, err := nl.NodeRangeBeforeExclusive(prevNodeReference, name)
  118. if err != nil {
  119. return nil
  120. }
  121. // delete skiplist reference to old node
  122. if _, err := nl.skipList.DeleteByKey(prevNodeReference.Key); err != nil {
  123. return err
  124. }
  125. // add namesToY and name to a new X
  126. namesToX = append(namesToX, name)
  127. if err := nl.ItemAdd([]byte(namesToX[0]), 0, namesToX...); err != nil {
  128. return nil
  129. }
  130. // remove names less than name from current Y
  131. if err := nl.NodeDeleteBeforeExclusive(prevNodeReference, name); err != nil {
  132. return nil
  133. }
  134. // point skip list to current Y
  135. if err := nl.ItemAdd(lookupKey, prevNodeReference.ElementPointer); err != nil {
  136. return nil
  137. }
  138. return nil
  139. } else {
  140. // collect names after name, add them to Y
  141. namesToY, err := nl.NodeRangeAfterExclusive(prevNodeReference, name)
  142. if err != nil {
  143. return nil
  144. }
  145. // add namesToY and name to a new Y
  146. namesToY = append(namesToY, name)
  147. if err := nl.ItemAdd(lookupKey, 0, namesToY...); err != nil {
  148. return nil
  149. }
  150. // remove names after name from current X
  151. if err := nl.NodeDeleteAfterExclusive(prevNodeReference, name); err != nil {
  152. return nil
  153. }
  154. return nil
  155. }
  156. }
  157. // case 2.4
  158. if nextNode != nil {
  159. nodeSize := nl.NodeSize(nextNode.Reference())
  160. if nodeSize < nl.batchSize {
  161. if id, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
  162. return err
  163. } else {
  164. if err := nl.ItemAdd(lookupKey, id, name); err != nil {
  165. return err
  166. }
  167. }
  168. return nil
  169. }
  170. }
  171. // case 2.5
  172. // now prevNode is nil
  173. return nl.ItemAdd(lookupKey, 0, name)
  174. }
  175. /*
  176. // case 1: exists in nextNode
  177. if nextNode != nil && nextNode.Key == name {
  178. remove from nextNode, update nextNode
  179. // TODO: merge with prevNode if possible?
  180. return
  181. }
  182. if nextNode is nil
  183. prevNode = list.Largestnode
  184. if prevNode == nil and nextNode.Prev != nil
  185. prevNode = load(nextNode.Prev)
  186. // case 2: does not exist
  187. // case 2.1
  188. if prevNode == nil {
  189. return
  190. }
  191. // case 2.2
  192. if prevNameBatch does not contain name {
  193. return
  194. }
  195. // case 3
  196. delete from prevNameBatch
  197. if prevNameBatch + nextNode < capacityList
  198. // case 3.1
  199. merge
  200. else
  201. // case 3.2
  202. update prevNode
  203. */
  204. func (nl *ItemList) DeleteName(name string) error {
  205. lookupKey := []byte(name)
  206. prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
  207. if err != nil {
  208. return err
  209. }
  210. // case 1
  211. if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
  212. if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
  213. return err
  214. }
  215. if err := nl.NodeDeleteMember(nextNode.Reference(), name); err != nil {
  216. return err
  217. }
  218. minName := nl.NodeMin(nextNode.Reference())
  219. if minName == "" {
  220. return nl.NodeDelete(nextNode.Reference())
  221. }
  222. return nl.ItemAdd([]byte(minName), nextNode.Id)
  223. }
  224. if !found {
  225. prevNode, err = nl.skipList.GetLargestNode()
  226. if err != nil {
  227. return err
  228. }
  229. }
  230. if nextNode != nil && prevNode == nil {
  231. prevNode, err = nl.skipList.LoadElement(nextNode.Prev)
  232. if err != nil {
  233. return err
  234. }
  235. }
  236. // case 2
  237. if prevNode == nil {
  238. // case 2.1
  239. return nil
  240. }
  241. if !nl.NodeContainsItem(prevNode.Reference(), name) {
  242. return nil
  243. }
  244. // case 3
  245. if err := nl.NodeDeleteMember(prevNode.Reference(), name); err != nil {
  246. return err
  247. }
  248. prevSize := nl.NodeSize(prevNode.Reference())
  249. if prevSize == 0 {
  250. if _, err := nl.skipList.DeleteByKey(prevNode.Key); err != nil {
  251. return err
  252. }
  253. return nil
  254. }
  255. nextSize := nl.NodeSize(nextNode.Reference())
  256. if nextSize > 0 && prevSize+nextSize < nl.batchSize {
  257. // case 3.1 merge nextNode and prevNode
  258. if _, err := nl.skipList.DeleteByKey(nextNode.Key); err != nil {
  259. return err
  260. }
  261. nextNames, err := nl.NodeRangeBeforeExclusive(nextNode.Reference(), "")
  262. if err != nil {
  263. return err
  264. }
  265. if err := nl.NodeAddMember(prevNode.Reference(), nextNames...); err != nil {
  266. return err
  267. }
  268. return nl.NodeDelete(nextNode.Reference())
  269. } else {
  270. // case 3.2 update prevNode
  271. // no action to take
  272. return nil
  273. }
  274. return nil
  275. }
  276. func (nl *ItemList) ListNames(startFrom string, visitNamesFn func(name string) bool) error {
  277. lookupKey := []byte(startFrom)
  278. prevNode, nextNode, found, err := nl.skipList.FindGreaterOrEqual(lookupKey)
  279. if err != nil {
  280. return err
  281. }
  282. if found && bytes.Compare(nextNode.Key, lookupKey) == 0 {
  283. prevNode = nil
  284. }
  285. if !found {
  286. prevNode, err = nl.skipList.GetLargestNode()
  287. if err != nil {
  288. return err
  289. }
  290. }
  291. if prevNode != nil {
  292. if !nl.NodeScanInclusiveAfter(prevNode.Reference(), startFrom, visitNamesFn) {
  293. return nil
  294. }
  295. }
  296. for nextNode != nil {
  297. if !nl.NodeScanInclusiveAfter(nextNode.Reference(), startFrom, visitNamesFn) {
  298. return nil
  299. }
  300. nextNode, err = nl.skipList.LoadElement(nextNode.Next[0])
  301. if err != nil {
  302. return err
  303. }
  304. }
  305. return nil
  306. }
  307. func (nl *ItemList) RemoteAllListElement() error {
  308. t := nl.skipList
  309. nodeRef := t.StartLevels[0]
  310. for nodeRef != nil {
  311. node, err := t.LoadElement(nodeRef)
  312. if err != nil {
  313. return err
  314. }
  315. if node == nil {
  316. return nil
  317. }
  318. if err := t.DeleteElement(node); err != nil {
  319. return err
  320. }
  321. if err := nl.NodeDelete(node.Reference()); err != nil {
  322. return err
  323. }
  324. nodeRef = node.Next[0]
  325. }
  326. return nil
  327. }
  328. func (nl *ItemList) NodeContainsItem(node *skiplist.SkipListElementReference, item string) bool {
  329. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  330. _, err := nl.client.ZScore(context.Background(), key, item).Result()
  331. if err == redis.Nil {
  332. return false
  333. }
  334. if err == nil {
  335. return true
  336. }
  337. return false
  338. }
  339. func (nl *ItemList) NodeSize(node *skiplist.SkipListElementReference) int {
  340. if node == nil {
  341. return 0
  342. }
  343. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  344. return int(nl.client.ZLexCount(context.Background(), key, "-", "+").Val())
  345. }
  346. func (nl *ItemList) NodeAddMember(node *skiplist.SkipListElementReference, names ...string) error {
  347. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  348. var members []redis.Z
  349. for _, name := range names {
  350. members = append(members, redis.Z{
  351. Score: 0,
  352. Member: name,
  353. })
  354. }
  355. return nl.client.ZAddNX(context.Background(), key, members...).Err()
  356. }
  357. func (nl *ItemList) NodeDeleteMember(node *skiplist.SkipListElementReference, name string) error {
  358. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  359. return nl.client.ZRem(context.Background(), key, name).Err()
  360. }
  361. func (nl *ItemList) NodeDelete(node *skiplist.SkipListElementReference) error {
  362. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  363. return nl.client.Del(context.Background(), key).Err()
  364. }
  365. func (nl *ItemList) NodeInnerPosition(node *skiplist.SkipListElementReference, name string) int {
  366. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  367. return int(nl.client.ZLexCount(context.Background(), key, "-", "("+name).Val())
  368. }
  369. func (nl *ItemList) NodeMin(node *skiplist.SkipListElementReference) string {
  370. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  371. slice := nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
  372. Min: "-",
  373. Max: "+",
  374. Offset: 0,
  375. Count: 1,
  376. }).Val()
  377. if len(slice) > 0 {
  378. s := slice[0]
  379. return s
  380. }
  381. return ""
  382. }
  383. func (nl *ItemList) NodeScanInclusiveAfter(node *skiplist.SkipListElementReference, startFrom string, visitNamesFn func(name string) bool) bool {
  384. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  385. if startFrom == "" {
  386. startFrom = "-"
  387. } else {
  388. startFrom = "[" + startFrom
  389. }
  390. names := nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
  391. Min: startFrom,
  392. Max: "+",
  393. }).Val()
  394. for _, n := range names {
  395. if !visitNamesFn(n) {
  396. return false
  397. }
  398. }
  399. return true
  400. }
  401. func (nl *ItemList) NodeRangeBeforeExclusive(node *skiplist.SkipListElementReference, stopAt string) ([]string, error) {
  402. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  403. if stopAt == "" {
  404. stopAt = "+"
  405. } else {
  406. stopAt = "(" + stopAt
  407. }
  408. return nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
  409. Min: "-",
  410. Max: stopAt,
  411. }).Result()
  412. }
  413. func (nl *ItemList) NodeRangeAfterExclusive(node *skiplist.SkipListElementReference, startFrom string) ([]string, error) {
  414. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  415. if startFrom == "" {
  416. startFrom = "-"
  417. } else {
  418. startFrom = "(" + startFrom
  419. }
  420. return nl.client.ZRangeByLex(context.Background(), key, &redis.ZRangeBy{
  421. Min: startFrom,
  422. Max: "+",
  423. }).Result()
  424. }
  425. func (nl *ItemList) NodeDeleteBeforeExclusive(node *skiplist.SkipListElementReference, stopAt string) error {
  426. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  427. if stopAt == "" {
  428. stopAt = "+"
  429. } else {
  430. stopAt = "(" + stopAt
  431. }
  432. return nl.client.ZRemRangeByLex(context.Background(), key, "-", stopAt).Err()
  433. }
  434. func (nl *ItemList) NodeDeleteAfterExclusive(node *skiplist.SkipListElementReference, startFrom string) error {
  435. key := fmt.Sprintf("%s%dm", nl.prefix, node.ElementPointer)
  436. if startFrom == "" {
  437. startFrom = "-"
  438. } else {
  439. startFrom = "(" + startFrom
  440. }
  441. return nl.client.ZRemRangeByLex(context.Background(), key, startFrom, "+").Err()
  442. }
  443. func (nl *ItemList) ItemAdd(lookupKey []byte, idIfKnown int64, names ...string) error {
  444. if id, err := nl.skipList.InsertByKey(lookupKey, idIfKnown, nil); err != nil {
  445. return err
  446. } else {
  447. if len(names) > 0 {
  448. return nl.NodeAddMember(&skiplist.SkipListElementReference{
  449. ElementPointer: id,
  450. Key: lookupKey,
  451. }, names...)
  452. }
  453. }
  454. return nil
  455. }