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
|
<?php
declare(strict_types=1);
namespace OpenTelemetry\SDK\Metrics\Stream;
use function assert;
use GMP;
use OpenTelemetry\SDK\Metrics\AggregationInterface;
/**
* @internal
*/
final class DeltaStorage
{
private AggregationInterface $aggregation;
private Delta $head;
public function __construct(AggregationInterface $aggregation)
{
$this->aggregation = $aggregation;
$this->head = new Delta(new Metric([], [], 0), 0);
/** @phan-suppress-next-line PhanTypeObjectUnsetDeclaredProperty */
unset($this->head->metric);
}
/**
* @psalm-suppress UndefinedDocblockClass
* @phan-suppress PhanUndeclaredTypeParameter
* @param int|GMP $readers
*/
public function add(Metric $metric, $readers): void
{
/** @phpstan-ignore-next-line */
if ($readers == 0) {
return;
}
if (($this->head->prev->readers ?? null) != $readers) {
$this->head->prev = new Delta($metric, $readers, $this->head->prev);
} else {
assert($this->head->prev !== null);
$this->mergeInto($this->head->prev->metric, $metric);
}
}
public function collect(int $reader, bool $retain = false): ?Metric
{
$n = null;
for ($d = $this->head; $d->prev; $d = $d->prev) {
if (($d->prev->readers >> $reader & 1) != 0) {
if ($n !== null) {
assert($n->prev !== null);
$n->prev->readers ^= $d->prev->readers;
$this->mergeInto($d->prev->metric, $n->prev->metric);
$this->tryUnlink($n);
if ($n->prev === $d->prev) {
continue;
}
}
$n = $d;
}
}
$metric = $n->prev->metric ?? null;
if (!$retain && $n) {
assert($n->prev !== null);
$n->prev->readers ^= ($n->prev->readers & 1 | 1) << $reader;
$this->tryUnlink($n);
}
return $metric;
}
private function tryUnlink(Delta $n): void
{
assert($n->prev !== null);
/** @phpstan-ignore-next-line */
if ($n->prev->readers == 0) {
$n->prev = $n->prev->prev;
return;
}
for ($c = $n->prev->prev;
$c && ($n->prev->readers & $c->readers) == 0;
$c = $c->prev) {
}
if ($c && $n->prev->readers === $c->readers) {
$this->mergeInto($c->metric, $n->prev->metric);
$n->prev = $n->prev->prev;
}
}
private function mergeInto(Metric $into, Metric $metric): void
{
foreach ($metric->summaries as $k => $summary) {
$into->attributes[$k] ??= $metric->attributes[$k];
$into->summaries[$k] = isset($into->summaries[$k])
? $this->aggregation->merge($into->summaries[$k], $summary)
: $summary;
}
$into->exemplars += $metric->exemplars;
}
}
|
