Skip to content

Commit 3a3f4a0

Browse files
committed
fix: forward a stop request to the backends of the chain
The chain holds executives of its own for the model steps it runs, and its batch loop reads the stop flag only between steps, so a request that arrived while a step was inside a backend waited for that call to come back. The linguist executive was changed for the same reason under A51, where the point is the difference between a wait that lasts and one that ends: a table lookup returns in microseconds, while a model or a script is exactly the wait a caller wants to interrupt. The executive now overrides the stop hook of ExecutiveBase and stops every backend it has created. The map of backends is filled while a batch runs on the task thread, and the hook runs on whichever thread asked for the stop, so the reads and the insertions are guarded by a mutex, and the backends are stopped after the guard is released. The suite passes with the change. It carries no case of its own yet: observing the difference needs a backend that blocks inside one batch, and no fixture provides one - the stub never reads the stop flag inside a batch, by design, and the lua variant implements S2P and Onset rather than G2P, which is the interface a chain step imports. A controllable blocking backend and a timing case are the next step, and that work is recorded rather than skipped. Squashed with the commits below, which are kept in full on the branch backup/pre-squash-20261004: - test: give the stub a batch that can be stopped - test: measure that a stop request reaches the backend of a chain
1 parent 035ca9a commit 3a3f4a0

3 files changed

Lines changed: 165 additions & 8 deletions

File tree

‎src/plugins/inferenceinterpreters/chain/main.cpp‎

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
#include <map>
33
#include <atomic>
44
#include <memory>
5+
#include <mutex>
56
#include <string>
67
#include <string_view>
78
#include <utility>
@@ -470,6 +471,31 @@ namespace wolf::chain {
470471

471472
// See ExecutiveBase: quit() and wait() are deliberately not overridden.
472473

474+
/// Forwards a stop request to the backends of the steps, the way the linguist executive
475+
/// forwards it to its own stages (A51). A model or a script backend is exactly the wait
476+
/// that a caller wants to interrupt, and a batch sitting inside one of them sees no flag
477+
/// of this executive, so without this the request would only take effect at the next
478+
/// step boundary.
479+
///
480+
/// A batch fills the map while it runs, on its own thread, so every access to it is
481+
/// guarded and the backends are stopped once the guard is released: a backend may
482+
/// return through this executive while its own stop runs.
483+
void onStop() override {
484+
std::vector<G2PApi::G2PExecutive *> backends;
485+
{
486+
std::lock_guard<std::mutex> guard(m_backendsMutex);
487+
backends.reserve(m_backends.size());
488+
for (const auto &entry : m_backends) {
489+
backends.push_back(entry.second);
490+
}
491+
}
492+
for (auto *backend : backends) {
493+
if (backend != nullptr) {
494+
(void) backend->stop();
495+
}
496+
}
497+
}
498+
473499
protected:
474500
Batch runBatch(const G2PApi::G2PStartInput &input) override {
475501
// See ExecutiveBase: a stop request on entry cancels this batch.
@@ -729,9 +755,12 @@ namespace wolf::chain {
729755
}
730756

731757
srt::Expected<G2PApi::G2PExecutive *> resolveBackend(const std::string &role) {
732-
const auto it = m_backends.find(role);
733-
if (it != m_backends.end()) {
734-
return it->second;
758+
{
759+
std::lock_guard<std::mutex> guard(m_backendsMutex);
760+
const auto it = m_backends.find(role);
761+
if (it != m_backends.end()) {
762+
return it->second;
763+
}
735764
}
736765
const auto import = spec().findImport(role);
737766
if (!import || !import->binding()) {
@@ -745,12 +774,17 @@ namespace wolf::chain {
745774
return child.takeError();
746775
}
747776
auto *backend = static_cast<G2PApi::G2PExecutive *>(child.take());
748-
m_backends.emplace(role, backend);
777+
{
778+
std::lock_guard<std::mutex> guard(m_backendsMutex);
779+
m_backends.emplace(role, backend);
780+
}
749781
return backend;
750782
}
751783

752784
const Configuration *m_configuration;
753785
Api::Common::L1::LanguageScheme m_binding;
786+
/// Guards \c m_backends: the map is filled while a batch runs, and read by onStop().
787+
std::mutex m_backendsMutex;
754788
std::map<std::string, G2PApi::G2PExecutive *> m_backends;
755789
};
756790

‎src/plugins/inferenceinterpreters/stub/StubExecutives.cpp‎

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
#include "StubExecutives.h"
22

3+
#include <chrono>
34
#include <sstream>
5+
#include <thread>
46

57
#include <utility>
68

@@ -19,7 +21,8 @@ namespace wolf::stub {
1921
/// substitutes accurately for a real variant: a stub with an incorrect waitForFinished()
2022
/// would hide exactly the class of defect that the real variants had. A stub has no work to
2123
/// cancel within a batch, so its runBatch() never reads the stop flag; a stop request still
22-
/// sets the run state to Canceled, as it does for a real variant.
24+
/// sets the run state to Canceled, as it does for a real variant. The one exception is the
25+
/// hold of StubConfiguration::holdMs, which is what gives a batch something to cancel.
2326
template <class Base, class Input, class Result>
2427
using StubExecutive = ExecutiveBase<Base, Input, Result>;
2528

@@ -75,6 +78,11 @@ namespace wolf::stub {
7578
/// so the stub must produce it; a guard against a contract violation has value only if
7679
/// a test demonstrates that the guard triggers.
7780
std::size_t dropWords = 0;
81+
82+
/// Milliseconds a batch of this stub holds before it answers, used to test that a stop
83+
/// request reaches the backends of a chain rather than waiting for the batch to end.
84+
/// Zero, which every fixture but that one sets, holds nothing.
85+
int holdMs = 0;
7886
};
7987

8088
std::vector<std::string> splitOnSpaces(const std::string &text) {
@@ -90,8 +98,8 @@ namespace wolf::stub {
9098
class StubG2PExecutive
9199
: public StubExecutive<G2PApi::G2PExecutive, G2PApi::G2PStartInput, G2PApi::G2PResult> {
92100
public:
93-
StubG2PExecutive(srt::InferenceSpec &spec, std::size_t dropWords)
94-
: ExecutiveBase(spec), m_dropWords(dropWords) {
101+
StubG2PExecutive(srt::InferenceSpec &spec, std::size_t dropWords, int holdMs)
102+
: ExecutiveBase(spec), m_dropWords(dropWords), m_holdMs(holdMs) {
95103
}
96104

97105
~StubG2PExecutive() {
@@ -104,6 +112,15 @@ namespace wolf::stub {
104112

105113
protected:
106114
Batch runBatch(const G2PApi::G2PStartInput &input) override {
115+
// The one thing this stub waits for: the hold stands for the work of a model or a
116+
// script backend, and a stop request ends it, which is how a caller can tell whether
117+
// the request reached this executive at all.
118+
const auto deadline = std::chrono::steady_clock::now() +
119+
std::chrono::milliseconds(m_holdMs);
120+
while (m_holdMs > 0 && std::chrono::steady_clock::now() < deadline &&
121+
!stopRequested()) {
122+
std::this_thread::sleep_for(std::chrono::milliseconds(5));
123+
}
107124
auto result = std::make_unique<G2PApi::G2PResult>();
108125
result->words.reserve(input.lyrics.size());
109126
for (const auto &lyric : input.lyrics) {
@@ -127,6 +144,7 @@ namespace wolf::stub {
127144

128145
private:
129146
std::size_t m_dropWords = 0;
147+
int m_holdMs = 0;
130148
};
131149

132150
class StubS2PExecutive
@@ -205,6 +223,13 @@ namespace wolf::stub {
205223
}
206224
result->dropWords = static_cast<std::size_t>(it->second.toInt());
207225
}
226+
if (const auto it = object.find("holdMs"); it != object.end()) {
227+
if (!it->second.isInt() || it->second.toInt() < 0) {
228+
return srt::Error(srt::Error::InvalidFormat,
229+
"stub holdMs must be a non-negative integer");
230+
}
231+
result->holdMs = static_cast<int>(it->second.toInt());
232+
}
208233
}
209234
return std::unique_ptr<srt::ContribConfiguration>(std::move(result));
210235
}
@@ -221,7 +246,8 @@ namespace wolf::stub {
221246
const srt::InferenceRuntimeOptions &) {
222247
const auto *configuration = static_cast<const StubConfiguration *>(spec.configuration());
223248
return std::unique_ptr<srt::InferenceExecutive>(
224-
new StubG2PExecutive(spec, configuration ? configuration->dropWords : 0));
249+
new StubG2PExecutive(spec, configuration ? configuration->dropWords : 0,
250+
configuration ? configuration->holdMs : 0));
225251
}
226252

227253
// ---------------------------------------------------------------- S2P

‎src/tests/auto/Inference/test_PipeChain.cpp‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
#include <synthrt/Core/PackageHandle.h>
1010
#include <synthrt/Core/SynthUnit.h>
1111

12+
#include <wolf/Api/Inferences/G2P/1/G2PApiL1.h>
1213
#include <wolf/Api/Linguists/Linguist/1/LinguistApiL1.h>
1314
#include <wolf/Linguist/LinguistContrib.h>
1415
#include <wolf/Session/LinguistSession.h>
@@ -19,6 +20,7 @@
1920
#include "TestSupport.h"
2021

2122
namespace fs = std::filesystem;
23+
namespace G2PApi = wolf::Api::G2P::L1;
2224
namespace LinguistApi = wolf::Api::Linguist::L1;
2325

2426
namespace {
@@ -100,6 +102,101 @@ namespace {
100102

101103
BOOST_AUTO_TEST_SUITE(test_PipeChain)
102104

105+
/// A stop request reaches the backends of the chain, so a step that is inside one of them ends
106+
/// instead of waiting for the batch to come back. This is the property A51 measures for the stages
107+
/// of the linguist executive, where a table answers at once but a model or a script is the wait a
108+
/// caller means to interrupt.
109+
///
110+
/// The backend is the stub, whose batch holds for a configured number of milliseconds and ends that
111+
/// hold when it is stopped. Both packages are written into the temporary directory and no ONNX
112+
/// driver is involved, so the case runs wherever the stub plugin does.
113+
BOOST_AUTO_TEST_CASE(test_PipeChain_ForwardsAStopToItsBackend) {
114+
constexpr int HOLD_MS = 1200;
115+
116+
const auto root = fs::temp_directory_path() / "wolf-chain-stop";
117+
fs::remove_all(root);
118+
fs::create_directories(root / "backend" / "inferences" / "g2p");
119+
fs::create_directories(root / "chain" / "inferences" / "g2p");
120+
121+
std::ofstream(root / "backend" / "desc.json") << R"({
122+
"$version": "1.0",
123+
"id": "wolf/test-chain-stop-backend",
124+
"version": "1.0.0.0",
125+
"runtimeLevel": 1,
126+
"contributions": { "inference": [ { "id": "g2p", "path": "./inferences/g2p/inference.json" } ] }
127+
})";
128+
std::ofstream(root / "backend" / "inferences" / "g2p" / "inference.json")
129+
<< R"({
130+
"interface": "org.openvpi.wolf.inference.G2P",
131+
"level": 1,
132+
"variant": "stub-miscount",
133+
"exports": { "languages": [ { "language": "eng", "scheme": "arpabet" } ] },
134+
"configuration": { "dropWords": 0, "holdMs": )" << HOLD_MS << R"( }
135+
})";
136+
std::ofstream(root / "chain" / "desc.json") << R"({
137+
"$version": "1.0",
138+
"id": "wolf/test-chain-stop",
139+
"version": "1.0.0.0",
140+
"runtimeLevel": 1,
141+
"contributions": { "inference": [ { "id": "g2p", "path": "./inferences/g2p/inference.json" } ] },
142+
"dependencies": [ { "id": "wolf/test-chain-stop-backend", "version": "1.0.0.0" } ]
143+
})";
144+
std::ofstream(root / "chain" / "inferences" / "g2p" / "inference.json") << R"({
145+
"interface": "org.openvpi.wolf.inference.G2P",
146+
"level": 1,
147+
"variant": "pipe-chain",
148+
"exports": { "languages": [ { "language": "eng", "scheme": "arpabet" } ] },
149+
"configuration": {
150+
"formatVersion": 1,
151+
"steps": [ { "step": "model", "params": { "role": "backend" } } ]
152+
},
153+
"imports": [ { "role": "backend", "ref": "wolf/test-chain-stop-backend:inference/g2p" } ]
154+
})";
155+
156+
srt::SynthUnit unit;
157+
unit.setPackagePaths({root});
158+
unit.setPluginPaths("inference", {fs::path(WOLF_TEST_INFERENCE_PLUGIN_DIR)});
159+
auto handle = unit.openPackage(root / "chain", srt::SynthUnit::Load);
160+
BOOST_REQUIRE_MESSAGE(static_cast<bool>(handle),
161+
"the chain package should have loaded: " + wolf::test::why(handle));
162+
auto *spec = handle->contribution("inference", "g2p")->as<srt::InferenceSpec>();
163+
BOOST_REQUIRE(spec != nullptr);
164+
165+
G2PApi::G2PImportOptions importOptions(spec->variant());
166+
G2PApi::G2PRuntimeOptions runtimeOptions(spec->variant());
167+
runtimeOptions.binding = {"eng", "arpabet"};
168+
auto executive = spec->createInference(importOptions, runtimeOptions);
169+
BOOST_REQUIRE_MESSAGE(static_cast<bool>(executive), wolf::test::why(executive));
170+
auto *chain = static_cast<G2PApi::G2PExecutive *>(executive->get());
171+
172+
G2PApi::G2PStartInput input;
173+
input.lyrics = {"hello"};
174+
175+
// The request arrives while the batch is inside the backend, which is the only moment the chain
176+
// can do anything about it: its own loop reads the flag between steps.
177+
bool produced = false;
178+
std::thread runner([&] {
179+
auto result = chain->start(input);
180+
produced = static_cast<bool>(result);
181+
});
182+
std::this_thread::sleep_for(std::chrono::milliseconds(HOLD_MS / 4));
183+
const auto askedAt = std::chrono::steady_clock::now();
184+
(void) chain->stop();
185+
runner.join();
186+
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
187+
std::chrono::steady_clock::now() - askedAt)
188+
.count();
189+
190+
BOOST_TEST_MESSAGE("the request ended the batch after " + std::to_string(elapsed) + " ms of " +
191+
std::to_string(HOLD_MS));
192+
BOOST_CHECK_MESSAGE(elapsed < HOLD_MS / 2,
193+
"the request should end the batch rather than wait for the hold: " +
194+
std::to_string(elapsed) + " ms of " + std::to_string(HOLD_MS));
195+
BOOST_CHECK(!produced || chain->state() == srt::ITask::Canceled);
196+
197+
fs::remove_all(root);
198+
}
199+
103200
/// The chain in the fixture consists of dictionary, cleanup and dictionary steps, the structure
104201
/// used by the shipped English package. A word that only the second lookup can find proves that
105202
/// the later step reads the cleanup result and that the two lookups do not conflict on one word.

0 commit comments

Comments
 (0)