services/keep-balance/block_state.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "sync"
   9
  10         "git.curoverse.com/arvados.git/sdk/go/arvados"
  11 )
  12
  13 // Replica is a file on disk (or object in an S3 bucket, or blob in an
  14 // Azure storage container, etc.) as reported in a keepstore index
  15 // response.
  16 type Replica struct {
  17         *KeepMount
  18         Mtime int64
  19 }
  20
  21 // BlockState indicates the desired storage class and number of
  22 // replicas (according to the collections we know about) and the
  23 // replicas actually stored (according to the keepstore indexes we
  24 // know about).
  25 type BlockState struct {
  26         RefCount int
  27         Replicas []Replica
  28         Desired  map[string]int
  29         // TODO: Support combinations of classes ("private + durable")
  30         // by replacing the map[string]int with a map[*[]string]int
  31         // here, where the map keys come from a pool of semantically
  32         // distinct class combinations.
  33         //
  34         // TODO: Use a pool of semantically distinct Desired maps to
  35         // conserve memory (typically there are far more BlockState
  36         // objects in memory than distinct Desired profiles).
  37 }
  38
  39 var defaultClasses = []string{"default"}
  40
  41 func (bs *BlockState) addReplica(r Replica) {
  42         bs.Replicas = append(bs.Replicas, r)
  43 }
  44
  45 func (bs *BlockState) increaseDesired(classes []string, n int) {
  46         bs.RefCount++
  47         if len(classes) == 0 {
  48                 classes = defaultClasses
  49         }
  50         for _, class := range classes {
  51                 if bs.Desired == nil {
  52                         bs.Desired = map[string]int{class: n}
  53                 } else if d, ok := bs.Desired[class]; !ok || d < n {
  54                         bs.Desired[class] = n
  55                 }
  56         }
  57 }
  58
  59 // BlockStateMap is a goroutine-safe wrapper around a
  60 // map[arvados.SizedDigest]*BlockState.
  61 type BlockStateMap struct {
  62         entries map[arvados.SizedDigest]*BlockState
  63         mutex   sync.Mutex
  64 }
  65
  66 // NewBlockStateMap returns a newly allocated BlockStateMap.
  67 func NewBlockStateMap() *BlockStateMap {
  68         return &BlockStateMap{
  69                 entries: make(map[arvados.SizedDigest]*BlockState),
  70         }
  71 }
  72
  73 // return a BlockState entry, allocating a new one if needed. (Private
  74 // method: not goroutine-safe.)
  75 func (bsm *BlockStateMap) get(blkid arvados.SizedDigest) *BlockState {
  76         // TODO? Allocate BlockState structs a slice at a time,
  77         // instead of one at a time.
  78         blk := bsm.entries[blkid]
  79         if blk == nil {
  80                 blk = &BlockState{}
  81                 bsm.entries[blkid] = blk
  82         }
  83         return blk
  84 }
  85
  86 // Apply runs f on each entry in the map.
  87 func (bsm *BlockStateMap) Apply(f func(arvados.SizedDigest, *BlockState)) {
  88         bsm.mutex.Lock()
  89         defer bsm.mutex.Unlock()
  90
  91         for blkid, blk := range bsm.entries {
  92                 f(blkid, blk)
  93         }
  94 }
  95
  96 // AddReplicas updates the map to indicate that mnt has a replica of
  97 // each block in idx.
  98 func (bsm *BlockStateMap) AddReplicas(mnt *KeepMount, idx []arvados.KeepServiceIndexEntry) {
  99         bsm.mutex.Lock()
 100         defer bsm.mutex.Unlock()
 101
 102         for _, ent := range idx {
 103                 bsm.get(ent.SizedDigest).addReplica(Replica{
 104                         KeepMount: mnt,
 105                         Mtime:     ent.Mtime,
 106                 })
 107         }
 108 }
 109
 110 // IncreaseDesired updates the map to indicate the desired replication
 111 // for the given blocks in the given storage class is at least n.
 112 func (bsm *BlockStateMap) IncreaseDesired(classes []string, n int, blocks []arvados.SizedDigest) {
 113         bsm.mutex.Lock()
 114         defer bsm.mutex.Unlock()
 115
 116         for _, blkid := range blocks {
 117                 bsm.get(blkid).increaseDesired(classes, n)
 118         }
 119 }