services/keep-balance/change_set.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package keepbalance
   6
   7 import (
   8         "encoding/json"
   9         "fmt"
  10         "sync"
  11
  12         "git.arvados.org/arvados.git/sdk/go/arvados"
  13         "git.arvados.org/arvados.git/services/keepstore"
  14 )
  15
  16 // Pull is a request to retrieve a block from a remote server, and
  17 // store it locally.
  18 type Pull struct {
  19         arvados.SizedDigest
  20         From *KeepService
  21         To   *KeepMount
  22 }
  23
  24 // MarshalJSON formats a pull request the way keepstore wants to see
  25 // it.
  26 func (p Pull) MarshalJSON() ([]byte, error) {
  27         return json.Marshal(keepstore.PullListItem{
  28                 Locator:   string(p.SizedDigest),
  29                 Servers:   []string{p.From.URLBase()},
  30                 MountUUID: p.To.KeepMount.UUID,
  31         })
  32 }
  33
  34 // Trash is a request to delete a block.
  35 type Trash struct {
  36         arvados.SizedDigest
  37         Mtime int64
  38         From  *KeepMount
  39 }
  40
  41 // MarshalJSON formats a trash request the way keepstore wants to see
  42 // it, i.e., as a bare locator with no +size hint.
  43 func (t Trash) MarshalJSON() ([]byte, error) {
  44         return json.Marshal(keepstore.TrashListItem{
  45                 Locator:    string(t.SizedDigest),
  46                 BlockMtime: t.Mtime,
  47                 MountUUID:  t.From.KeepMount.UUID,
  48         })
  49 }
  50
  51 // ChangeSet is a set of change requests that will be sent to a
  52 // keepstore server.
  53 type ChangeSet struct {
  54         PullLimit  int
  55         TrashLimit int
  56
  57         Pulls           []Pull
  58         PullsDeferred   int // number that weren't added because of PullLimit
  59         Trashes         []Trash
  60         TrashesDeferred int // number that weren't added because of TrashLimit
  61         mutex           sync.Mutex
  62 }
  63
  64 // AddPull adds a Pull operation.
  65 func (cs *ChangeSet) AddPull(p Pull) {
  66         cs.mutex.Lock()
  67         if len(cs.Pulls) < cs.PullLimit {
  68                 cs.Pulls = append(cs.Pulls, p)
  69         } else {
  70                 cs.PullsDeferred++
  71         }
  72         cs.mutex.Unlock()
  73 }
  74
  75 // AddTrash adds a Trash operation
  76 func (cs *ChangeSet) AddTrash(t Trash) {
  77         cs.mutex.Lock()
  78         if len(cs.Trashes) < cs.TrashLimit {
  79                 cs.Trashes = append(cs.Trashes, t)
  80         } else {
  81                 cs.TrashesDeferred++
  82         }
  83         cs.mutex.Unlock()
  84 }
  85
  86 // String implements fmt.Stringer.
  87 func (cs *ChangeSet) String() string {
  88         cs.mutex.Lock()
  89         defer cs.mutex.Unlock()
  90         return fmt.Sprintf("ChangeSet{Pulls:%d, Trashes:%d} Deferred{Pulls:%d Trashes:%d}", len(cs.Pulls), len(cs.Trashes), cs.PullsDeferred, cs.TrashesDeferred)
  91 }