G counter

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
use "collections"

class ref GCounter[A: (Integer[A] val & Unsigned) = U64]
  is (Comparable[GCounter[A]] & Convergent[GCounter[A]] & Replicated)
  """
  A mutable grow-only counter. That is, the value can only be increased.

  It is limited by the maximum value of the used unsigned integer datatype.
  Any operation that would lead to an overflow (if the maximum is the
  maximum value for the used unsigned integer type) will result in the value
  being set to the maximum. So once the maximum is reached, the GCounter will
  never change.

  This data type tracks the state seen from each replica, thus the size of the
  state will grow proportionally with the number of total replicas. New replicas
  may be added as peers at any time, provided that they use unique ids.
  Read-only replicas which never change state and only observe need not use
  unique ids, and should use an id of zero, by convention.

  The state of each replica represents the value incremented so far by that
  particular replica. This local value may only ever increase, never decrease.
  The total value of the counter is the sum of the local value of all replicas.

  When converging state from other replicas, we retain the maximum observed
  value from each replica id. Because a higher value always implies later
  logical time for that replica, and we only keep the highest value seen from
  each replica, we will always retain the latest value seen from each replica.

  Because the view of values from each other replica is eventually consistent,
  the summed value of the overall counter is also eventually consistent.

  All mutator methods accept and return a convergent delta-state.
  """
  var _id: ID
  embed _data: Map[ID, A]
  let _checklist: (DotChecklist | None)

  new ref create(id': ID) =>
    """
    Instantiate the GCounter under the given unique replica id.
    """
    _id        = id'
    _data      = Map[ID, A]
    _checklist = None

  new ref _create_in(ctx: DotContext) =>
    _id        = ctx.id()
    _data      = _data.create()
    _checklist = DotChecklist(ctx)

  fun ref _checklist_write() =>
    match _checklist
    | let c: DotChecklist => c.write()
    end

  fun ref _converge_empty_in(ctx: DotContext box): Bool => // ignore the context
    false

  fun is_empty(): Bool =>
    """
    Return true if the data structure contains no information (bottom state).
    """
    _data.size() == 0

  fun apply(): A =>
    """
    Return the current value of the counter (the sum of all replica values).
    """
    value()

  fun value(): A =>
    """
    Return the current value of the counter (the sum of all replica values).
    """
    var sum = A(0)
    for v in _data.values() do sum = _Math.saturated_sum[A](sum, v) end
    sum

  fun ref _data_update(id': ID, value': A) => _data(id') = value'

  fun ref increment[D: GCounter[A] ref = GCounter[A]](
    value': A = 1,
    delta': D = recover GCounter[A](0) end)
    : D^
  =>
    """
    Increment the counter by the given value.
    Accepts and returns a convergent delta-state.
    """
    let v' =
      _data.upsert(
        _id,
        value',
        {(x: A, y: A): A => _Math.saturated_sum[A](x, y) })
    _checklist_write()
    delta'._data_update(_id, v')
    consume delta'

  fun ref converge(that: GCounter[A] box): Bool =>
    """
    Converge from the given GCounter into this one.
    For each replica state, we select the maximum value seen so far (grow-only).
    Returns true if the convergence added new information to the data structure.
    """
    var changed = false
    for (id, value') in that._data.pairs() do
      // TODO: introduce a stateful upsert in ponyc Map?
      if try value' > _data(id)? else true end then
        _data(id) = value'
        changed = true
      end
    end
    changed

  fun string(): String iso^ =>
    """
    Return a best effort at printing the register. If A is Stringable, use
    the string representation of the value; otherwise print as a question mark.
    """
    iftype A <: Stringable val then
      value().string()
    else
      "?".clone()
    end

  fun eq(that: GCounter[A] box): Bool => value().eq(that.value())
  fun ne(that: GCounter[A] box): Bool => value().ne(that.value())
  fun lt(that: GCounter[A] box): Bool => value().lt(that.value())
  fun le(that: GCounter[A] box): Bool => value().le(that.value())
  fun gt(that: GCounter[A] box): Bool => value().gt(that.value())
  fun ge(that: GCounter[A] box): Bool => value().ge(that.value())

  fun ref from_tokens(that: TokensIterator) ? =>
    """
    Deserialize an instance of this data structure from a stream of tokens.
    """
    var count = that.next[USize]()?

    if count < 1 then error end
    count = count - 1
    _id   = that.next[ID]()?

    if (count % 2) != 0 then error end
    count = count / 2

    // TODO: _data.reserve(count)
    while (count = count - 1) > 0 do
      _data.update(that.next[ID]()?, that.next[A]()?)
    end

  fun ref each_token(tokens: Tokens) =>
    """
    Serialize the data structure, capturing each token into the given Tokens.
    """
    tokens.push(1 + (_data.size() * 2))
    tokens.push(_id)
    for (id, v) in _data.pairs() do
      tokens.push(id)
      tokens.push(v)
    end