Skip to content

Commit 778ad85

Browse files
duncanpoduncanpo
committed
Add a timeout to asynchronous callbacks to prevent deadlocks
1 parent 9f752e5 commit 778ad85

File tree

10 files changed

+132
-41
lines changed

10 files changed

+132
-41
lines changed

‎api/metrics/+opentelemetry/+metrics/AsynchronousInstrument.m‎

Lines changed: 42 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,10 @@
1313
Callbacks % Callback function, called at each data export
1414
end
1515

16+
properties (Constant, Hidden)
17+
DefaultTimeout = seconds(30)
18+
end
19+
1620
properties (Access=private)
1721
Proxy % Proxy object to interface C++ code
1822
end
@@ -29,16 +33,42 @@
2933
end
3034

3135
methods
32-
function addCallback(obj, callback)
36+
function addCallback(obj, callback, optionnames, optionvalues)
3337
% ADDCALLBACK Add a callback function
3438
% ADDCALLBACK(INST, CALLBACK) adds a callback function to
35-
% collect metrics at every export. CALLBACK is specified as a
39+
% collect metrics at every export. CALLBACK is specified as a
3640
% function handle, and must accept no input and return one
3741
% output of type opentelemetry.metrics.ObservableResult.
3842
%
43+
% ADDCALLBACK(INST, CALLBACK, "Timeout", TIMEOUT)
44+
% also specifies the maximum time before callback is timed
45+
% out and its results not get recorded. TIMEOUT must be a
46+
% positive duration scalar.
47+
%
3948
% See also REMOVECALLBACK, OPENTELEMETRY.METRICS.OBSERVABLERESULT
49+
arguments
50+
obj
51+
callback
52+
end
53+
arguments (Repeating)
54+
optionnames
55+
optionvalues
56+
end
57+
4058
if isa(callback, "function_handle")
41-
obj.Proxy.addCallback(callback);
59+
% parse name-value pairs
60+
validnames = "Timeout";
61+
timeout = obj.DefaultTimeout;
62+
for i = 1:length(optionnames)
63+
try
64+
validatestring(optionnames{i}, validnames);
65+
catch
66+
continue
67+
end
68+
timeout = optionvalues{i};
69+
end
70+
timeout = obj.mustBeScalarPositiveDurationTimeout(timeout);
71+
obj.Proxy.addCallback(callback, milliseconds(timeout));
4272
% append to Callbacks property
4373
if isempty(obj.Callbacks)
4474
obj.Callbacks = callback;
@@ -47,7 +77,7 @@ function addCallback(obj, callback)
4777
else
4878
obj.Callbacks = [obj.Callbacks, {callback}];
4979
end
50-
end
80+
end
5181
end
5282

5383
function removeCallback(obj, callback)
@@ -78,4 +108,12 @@ function removeCallback(obj, callback)
78108
end
79109
end
80110
end
111+
112+
methods (Static)
113+
function timeout = mustBeScalarPositiveDurationTimeout(timeout)
114+
if ~(isscalar(timeout) && isa(timeout, "duration") && timeout > 0)
115+
timeout = opentelemetry.metrics.AsynchronousInstrument.DefaultTimeout;
116+
end
117+
end
118+
end
81119
end

‎api/metrics/+opentelemetry/+metrics/Meter.m‎

Lines changed: 43 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,8 @@
108108
histogram = opentelemetry.metrics.Histogram(HistogramProxy, name, description, unit);
109109
end
110110

111-
function obscounter = createObservableCounter(obj, callback, name, description, unit)
111+
function obscounter = createObservableCounter(obj, callback, name, ...
112+
description, unit, timeout)
112113
% CREATEOBSERVABLECOUNTER Create an observable counter
113114
% C = CREATEOBSERVABLECOUNTER(M, CALLBACK, NAME) creates an
114115
% observable counter with the specified callback function
@@ -117,9 +118,14 @@
117118
% output of type opentelemetry.metrics.ObservableResult.
118119
% The counter's value can only increase but not decrease.
119120
%
120-
% C = CREATEOBSERVABLECOUNTER(M, CALLBACK NAME, DESCRIPTION, UNIT)
121+
% C = CREATEOBSERVABLECOUNTER(M, CALLBACK, NAME, DESCRIPTION, UNIT)
121122
% also specifies a description and a unit.
122123
%
124+
% C = CREATEOBSERVABLECOUNTER(M, CALLBACK, NAME, DESCRIPTION, UNIT, TIMEOUT)
125+
% also specifies the maximum time before callback is timed
126+
% out and its results not get recorded. TIMEOUT must be a
127+
% duration.
128+
%
123129
% See also OPENTELEMETRY.METRICS.OBSERVABLERESULT,
124130
% CREATEOBSERVABLEUPDOWNCOUNTER, CREATEOBSERVABLEGAUGE, CREATECOUNTER
125131
arguments
@@ -128,17 +134,20 @@
128134
name
129135
description = ""
130136
unit = ""
137+
timeout = opentelemetry.metrics.ObservableCounter.DefaultTimeout
131138
end
132139

133-
[callback, name, description, unit] = processAsynchronousInputs(...
134-
callback, name, description, unit);
135-
id = obj.Proxy.createObservableCounter(name, description, unit, callback);
140+
[callback, name, description, unit, timeout] = processAsynchronousInputs(...
141+
callback, name, description, unit, timeout);
142+
id = obj.Proxy.createObservableCounter(name, description, unit, ...
143+
callback, milliseconds(timeout));
136144
ObservableCounterproxy = libmexclass.proxy.Proxy("Name", ...
137145
"libmexclass.opentelemetry.ObservableCounterProxy", "ID", id);
138146
obscounter = opentelemetry.metrics.ObservableCounter(ObservableCounterproxy, name, description, unit, callback);
139147
end
140148

141-
function obsudcounter = createObservableUpDownCounter(obj, callback, name, description, unit)
149+
function obsudcounter = createObservableUpDownCounter(obj, callback, ...
150+
name, description, unit, timeout)
142151
% CREATEOBSERVABLEUPDOWNCOUNTER Create an observable UpDownCounter
143152
% C = CREATEOBSERVABLEUPDOWNCOUNTER(M, CALLBACK, NAME)
144153
% creates an observable UpDownCounter with the specified
@@ -149,7 +158,12 @@
149158
%
150159
% C = CREATEOBSERVABLEUPDOWNCOUNTER(M, CALLBACK, NAME, DESCRIPTION, UNIT)
151160
% also specifies a description and a unit.
152-
%
161+
%
162+
% C = CREATEOBSERVABLEUPDOWNCOUNTER(M, CALLBACK, NAME, DESCRIPTION, UNIT, TIMEOUT)
163+
% also specifies the maximum time before callback is timed
164+
% out and its results not get recorded. TIMEOUT must be a
165+
% duration.
166+
%
153167
% See also OPENTELEMETRY.METRICS.OBSERVABLERESULT,
154168
% CREATEOBSERVABLECOUNTER, CREATEOBSERVABLEGAUGE, CREATEUPDOWNCOUNTER
155169
arguments
@@ -158,18 +172,21 @@
158172
name
159173
description = ""
160174
unit = ""
175+
timeout = opentelemetry.metrics.ObservableUpDownCounter.DefaultTimeout
161176
end
162177

163-
[callback, name, description, unit] = processAsynchronousInputs(...
164-
callback, name, description, unit);
165-
id = obj.Proxy.createObservableUpDownCounter(name, description, unit, callback);
178+
[callback, name, description, unit, timeout] = processAsynchronousInputs(...
179+
callback, name, description, unit, timeout);
180+
id = obj.Proxy.createObservableUpDownCounter(name, description, ...
181+
unit, callback, milliseconds(timeout));
166182
ObservableUpDownCounterproxy = libmexclass.proxy.Proxy("Name", ...
167183
"libmexclass.opentelemetry.ObservableUpDownCounterProxy", "ID", id);
168184
obsudcounter = opentelemetry.metrics.ObservableUpDownCounter(...
169185
ObservableUpDownCounterproxy, name, description, unit, callback);
170186
end
171187

172-
function obsgauge = createObservableGauge(obj, callback, name, description, unit)
188+
function obsgauge = createObservableGauge(obj, callback, name, ...
189+
description, unit, timeout)
173190
% CREATEOBSERVABLEGAUGE Create an observable gauge
174191
% C = CREATEOBSERVABLEGAUGE(M, CALLBACK, NAME) creates an
175192
% observable gauge with the specified callback function
@@ -179,9 +196,14 @@
179196
% A gauge's value can increase or decrease but it should
180197
% never be summed in aggregation.
181198
%
182-
% C = CREATEOBSERVABLEGAUGE(M, CALLBACK NAME, DESCRIPTION, UNIT)
199+
% C = CREATEOBSERVABLEGAUGE(M, CALLBACK, NAME, DESCRIPTION, UNIT)
183200
% also specifies a description and a unit.
184-
%
201+
%
202+
% C = CREATEOBSERVABLEGAUGE(M, CALLBACK, NAME, DESCRIPTION, UNIT, TIMEOUT)
203+
% also specifies the maximum time before callback is timed
204+
% out and its results not get recorded. TIMEOUT must be a
205+
% positive duration scalar.
206+
%
185207
% See also OPENTELEMETRY.METRICS.OBSERVABLERESULT,
186208
% CREATEOBSERVABLECOUNTER, CREATEOBSERVABLEUPDOWNCOUNTER
187209
arguments
@@ -190,11 +212,13 @@
190212
name
191213
description = ""
192214
unit = ""
215+
timeout = opentelemetry.metrics.ObservableGauge.DefaultTimeout
193216
end
194217

195-
[callback, name, description, unit] = processAsynchronousInputs(...
196-
callback, name, description, unit);
197-
id = obj.Proxy.createObservableGauge(name, description, unit, callback);
218+
[callback, name, description, unit, timeout] = processAsynchronousInputs(...
219+
callback, name, description, unit, timeout);
220+
id = obj.Proxy.createObservableGauge(name, description, unit, ...
221+
callback, milliseconds(timeout));
198222
ObservableGaugeproxy = libmexclass.proxy.Proxy("Name", ...
199223
"libmexclass.opentelemetry.ObservableGaugeProxy", "ID", id);
200224
obsgauge = opentelemetry.metrics.ObservableGauge(...
@@ -211,10 +235,11 @@
211235
unit = mustBeScalarString(unit);
212236
end
213237

214-
function [callback, name, description, unit] = processAsynchronousInputs(...
215-
callback, name, description, unit)
238+
function [callback, name, description, unit, timeout] = processAsynchronousInputs(...
239+
callback, name, description, unit, timeout)
216240
[name, description, unit] = processSynchronousInputs(name, description, unit);
217241
if ~isa(callback, "function_handle")
218242
callback = []; % callback is invalid, set to empty double
219243
end
244+
timeout = opentelemetry.metrics.AsynchronousInstrument.mustBeScalarPositiveDurationTimeout(timeout);
220245
end

‎api/metrics/include/opentelemetry-matlab/metrics/AsynchronousCallbackInput.h‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,21 @@
22

33
#pragma once
44

5+
#include <chrono>
6+
57
#include "MatlabDataArray.hpp"
68
#include "mex.hpp"
79

810
namespace libmexclass::opentelemetry {
911
struct AsynchronousCallbackInput
1012
{
1113
AsynchronousCallbackInput(const matlab::data::Array& fh,
14+
const std::chrono::milliseconds& timeout,
1215
const std::shared_ptr<matlab::engine::MATLABEngine> eng)
13-
: FunctionHandle(fh), MexEngine(eng) {}
16+
: FunctionHandle(fh), Timeout(timeout), MexEngine(eng) {}
1417

1518
matlab::data::Array FunctionHandle;
19+
std::chrono::milliseconds Timeout;
1620
const std::shared_ptr<matlab::engine::MATLABEngine> MexEngine;
1721
};
1822
} // namespace libmexclass::opentelemetry

‎api/metrics/include/opentelemetry-matlab/metrics/AsynchronousInstrumentProxy.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#pragma once
44

55
#include <list>
6+
#include <chrono>
67

78
#include "opentelemetry-matlab/metrics/AsynchronousCallbackInput.h"
89

@@ -25,7 +26,7 @@ class AsynchronousInstrumentProxy : public libmexclass::proxy::Proxy {
2526

2627
// This method should ideally be an overloaded version of addCallback. However, addCallback is a registered
2728
// method and REGISTER_METHOD macro doesn't like overloaded methods. Rename to avoid overloading.
28-
void addCallback_helper(const matlab::data::Array& callback);
29+
void addCallback_helper(const matlab::data::Array& callback, const std::chrono::milliseconds& timeout);
2930

3031
void removeCallback(libmexclass::proxy::method::Context& context);
3132

‎api/metrics/include/opentelemetry-matlab/metrics/AsynchronousInstrumentProxyFactory.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
// Copyright 2023-2024 The MathWorks, Inc.
22

33
#pragma once
4+
#include <chrono>
45

56
#include "libmexclass/proxy/Proxy.h"
67

@@ -21,7 +22,7 @@ class AsynchronousInstrumentProxyFactory {
2122

2223
std::shared_ptr<libmexclass::proxy::Proxy> create(AsynchronousInstrumentType type,
2324
const matlab::data::Array& callback, const std::string& name, const std::string& description,
24-
const std::string& unit);
25+
const std::string& unit, const std::chrono::milliseconds& timeout);
2526

2627
private:
2728

‎api/metrics/src/AsynchronousInstrumentProxy.cpp‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,13 @@ namespace libmexclass::opentelemetry {
1010

1111

1212
void AsynchronousInstrumentProxy::addCallback(libmexclass::proxy::method::Context& context){
13-
addCallback_helper(context.inputs[0]);
13+
matlab::data::TypedArray<double> timeout_mda = context.inputs[1];
14+
addCallback_helper(context.inputs[0], std::chrono::milliseconds(static_cast<int64_t>(timeout_mda[0])));
1415
}
1516

16-
void AsynchronousInstrumentProxy::addCallback_helper(const matlab::data::Array& callback){
17-
AsynchronousCallbackInput arg(callback, MexEngine);
17+
void AsynchronousInstrumentProxy::addCallback_helper(const matlab::data::Array& callback,
18+
const std::chrono::milliseconds& timeout){
19+
AsynchronousCallbackInput arg(callback, timeout, MexEngine);
1820
CallbackInputs.push_back(arg);
1921
CppInstrument->AddCallback(MeasurementFetcher::Fetcher, static_cast<void*>(&CallbackInputs.back()));
2022
}

‎api/metrics/src/AsynchronousInstrumentProxyFactory.cpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,8 @@
77

88
namespace libmexclass::opentelemetry {
99
std::shared_ptr<libmexclass::proxy::Proxy> AsynchronousInstrumentProxyFactory::create(AsynchronousInstrumentType type,
10-
const matlab::data::Array& callback, const std::string& name, const std::string& description, const std::string& unit) {
10+
const matlab::data::Array& callback, const std::string& name, const std::string& description, const std::string& unit,
11+
const std::chrono::milliseconds& timeout) {
1112
std::shared_ptr<libmexclass::proxy::Proxy> proxy;
1213
switch(type) {
1314
case AsynchronousInstrumentType::ObservableCounter:
@@ -31,7 +32,7 @@ std::shared_ptr<libmexclass::proxy::Proxy> AsynchronousInstrumentProxyFactory::c
3132
}
3233
// add callback
3334
if (!callback.isEmpty()) {
34-
std::static_pointer_cast<AsynchronousInstrumentProxy>(proxy)->addCallback_helper(callback);
35+
std::static_pointer_cast<AsynchronousInstrumentProxy>(proxy)->addCallback_helper(callback, timeout);
3536
}
3637
return proxy;
3738
}

‎api/metrics/src/MeasurementFetcher.cpp‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
// Copyright 2023-2024 The MathWorks, Inc.
22

3+
#include <chrono>
4+
35
#include "MatlabDataArray.hpp"
46
#include "mex.hpp"
57
#include "cppmex/detail/mexErrorDispatch.hpp"
@@ -30,11 +32,21 @@ void MeasurementFetcher::Fetcher(metrics_api::ObserverResult observer_result, vo
3032
nostd::shared_ptr<metrics_api::ObserverResultT<double>>>(observer_result))
3133
{
3234
auto arg = static_cast<AsynchronousCallbackInput*>(in);
35+
auto callback_timeout = arg->Timeout;
36+
const std::chrono::seconds property_timeout(1); // for getProperty, use a fixed timeout of 1 second, should be sufficient
3337
auto future = arg->MexEngine->fevalAsync(u"opentelemetry.metrics.collectObservableMetrics",
3438
arg->FunctionHandle);
3539
try {
40+
auto status = future.wait_for(callback_timeout);
41+
if (status != std::future_status::ready) {
42+
return;
43+
}
3644
matlab::data::ObjectArray resultobj = future.get();
3745
auto futureresult = arg->MexEngine->getPropertyAsync(resultobj, 0, u"Results");
46+
status = futureresult.wait_for(property_timeout);
47+
if (status != std::future_status::ready) {
48+
return;
49+
}
3850
matlab::data::CellArray resultdata = futureresult.get();
3951
size_t n = resultdata.getNumberOfElements();
4052
size_t i = 0;

‎api/metrics/src/MeterProxy.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,11 @@ void MeterProxy::createAsynchronous(libmexclass::proxy::method::Context& context
5555
matlab::data::StringArray unit_mda = context.inputs[2];
5656
std::string unit = static_cast<std::string>(unit_mda[0]);
5757
matlab::data::Array callback_mda = context.inputs[3];
58+
matlab::data::TypedArray<double> timeout_mda = context.inputs[4];
59+
auto timeout = std::chrono::milliseconds(static_cast<int64_t>(timeout_mda[0])); // milliseconds
5860

5961
AsynchronousInstrumentProxyFactory proxyfactory(CppMeter, MexEngine);
60-
auto proxy = proxyfactory.create(type, callback_mda, name, description, unit);
62+
auto proxy = proxyfactory.create(type, callback_mda, name, description, unit, timeout);
6163

6264
// obtain a proxy ID
6365
libmexclass::proxy::ID proxyid = libmexclass::proxy::ProxyManager::manageProxy(proxy);

0 commit comments

Comments
 (0)