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
|
<?php
declare(strict_types=1);
namespace OpenTelemetry\SDK\Metrics\Stream;
use function assert;
use const E_USER_WARNING;
use function extension_loaded;
use GMP;
use function gmp_init;
use function is_int;
use OpenTelemetry\SDK\Metrics\AggregationInterface;
use OpenTelemetry\SDK\Metrics\Data\DataInterface;
use OpenTelemetry\SDK\Metrics\Data\Exemplar;
use OpenTelemetry\SDK\Metrics\Data\Temporality;
use const PHP_INT_SIZE;
use function sprintf;
use function trigger_error;
/**
* @internal
*/
final class SynchronousMetricStream implements MetricStreamInterface
{
private AggregationInterface $aggregation;
private int $timestamp;
private DeltaStorage $delta;
/**
* @psalm-suppress UndefinedDocblockClass
* @phan-suppress PhanUndeclaredTypeProperty
* @var int|GMP
*/
private $readers = 0;
/**
* @psalm-suppress UndefinedDocblockClass
* @phan-suppress PhanUndeclaredTypeProperty
* @var int|GMP
*/
private $cumulative = 0;
public function __construct(AggregationInterface $aggregation, int $startTimestamp)
{
$this->aggregation = $aggregation;
$this->timestamp = $startTimestamp;
$this->delta = new DeltaStorage($aggregation);
}
public function temporality()
{
return Temporality::DELTA;
}
public function timestamp(): int
{
return $this->timestamp;
}
public function push(Metric $metric): void
{
[$this->timestamp, $metric->timestamp] = [$metric->timestamp, $this->timestamp];
$this->delta->add($metric, $this->readers);
}
public function register($temporality): int
{
$reader = 0;
for ($r = $this->readers; ($r & 1) != 0; $r >>= 1, $reader++) {
}
if ($reader === (PHP_INT_SIZE << 3) - 1 && is_int($this->readers)) {
if (!extension_loaded('gmp')) {
trigger_error(sprintf('GMP extension required to register over %d readers', (PHP_INT_SIZE << 3) - 1), E_USER_WARNING);
$reader = PHP_INT_SIZE << 3;
} else {
assert(is_int($this->cumulative));
$this->readers = gmp_init($this->readers);
$this->cumulative = gmp_init($this->cumulative);
}
}
$readerMask = ($this->readers & 1 | 1) << $reader;
$this->readers ^= $readerMask;
if ($temporality === Temporality::CUMULATIVE) {
$this->cumulative ^= $readerMask;
}
return $reader;
}
public function unregister(int $reader): void
{
$readerMask = ($this->readers & 1 | 1) << $reader;
if (($this->readers & $readerMask) == 0) {
return;
}
$this->delta->collect($reader);
$this->readers ^= $readerMask;
if (($this->cumulative & $readerMask) != 0) {
$this->cumulative ^= $readerMask;
}
}
public function collect(int $reader): DataInterface
{
$cumulative = ($this->cumulative >> $reader & 1) != 0;
$metric = $this->delta->collect($reader, $cumulative) ?? new Metric([], [], $this->timestamp);
$temporality = $cumulative
? Temporality::CUMULATIVE
: Temporality::DELTA;
return $this->aggregation->toData(
$metric->attributes,
$metric->summaries,
Exemplar::groupByIndex($metric->exemplars),
$metric->timestamp,
$this->timestamp,
$temporality,
);
}
}
|
