Skip to content

Commit 53cdd22

Browse files
committed
up
1 parent 4ecd0e7 commit 53cdd22

3 files changed

Lines changed: 482 additions & 22 deletions

File tree

‎extension/llm/serving/serving_runtime.cpp‎

Lines changed: 65 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
#include <utility>
1717

1818
#include <executorch/extension/llm/batching/runner.h>
19+
#include <executorch/runtime/platform/log.h>
1920

2021
namespace executorch {
2122
namespace extension {
@@ -35,7 +36,8 @@ struct ServingRuntime::Impl {
3536
struct Command {
3637
Operation operation;
3738
std::string key;
38-
std::promise<LifecycleResult> completion;
39+
// Queue transfers must not destroy user captures under mutex_.
40+
std::unique_ptr<LifecycleCallback> completion;
3941
};
4042

4143
Impl(
@@ -53,12 +55,44 @@ struct ServingRuntime::Impl {
5355
}
5456
}
5557

58+
static void complete(
59+
std::unique_ptr<LifecycleCallback> completion,
60+
LifecycleResult result) {
61+
if (!completion || !*completion) {
62+
return;
63+
}
64+
#if ET_HAS_EXCEPTIONS
65+
try {
66+
#endif
67+
(*completion)(std::move(result));
68+
#if ET_HAS_EXCEPTIONS
69+
} catch (...) {
70+
ET_LOG(Error, "Lifecycle completion callback threw");
71+
}
72+
#endif
73+
}
74+
5675
std::future<LifecycleResult> submit(Operation operation, std::string key) {
57-
Command command{operation, std::move(key), {}};
58-
auto future = command.completion.get_future();
76+
auto completion = std::make_shared<std::promise<LifecycleResult>>();
77+
auto future = completion->get_future();
78+
submit(
79+
operation,
80+
std::move(key),
81+
[completion = std::move(completion)](LifecycleResult result) {
82+
completion->set_value(std::move(result));
83+
});
84+
return future;
85+
}
86+
87+
void
88+
submit(Operation operation, std::string key, LifecycleCallback on_complete) {
89+
Command command{
90+
operation,
91+
std::move(key),
92+
std::make_unique<LifecycleCallback>(std::move(on_complete))};
93+
LifecycleResult error;
5994
{
6095
std::lock_guard<std::mutex> lock(mutex_);
61-
LifecycleResult error;
6296
if (!valid_config_) {
6397
error =
6498
ServingError{ErrorCode::InvalidArgument, "invalid runtime config"};
@@ -74,15 +108,16 @@ struct ServingRuntime::Impl {
74108
error = ServingError{
75109
ErrorCode::CapacityExceeded, "lifecycle operation limit reached"};
76110
}
77-
if (error) {
78-
command.completion.set_value(std::move(error));
79-
return future;
111+
if (!error) {
112+
inbox_.push_back(std::move(command));
113+
++outstanding_;
80114
}
81-
inbox_.push_back(std::move(command));
82-
++outstanding_;
83115
}
84-
cv_.notify_one();
85-
return future;
116+
if (error) {
117+
complete(std::move(command.completion), std::move(error));
118+
} else {
119+
cv_.notify_one();
120+
}
86121
}
87122

88123
LifecycleResult process(Command& command) {
@@ -190,8 +225,8 @@ struct ServingRuntime::Impl {
190225
result = ServingError{ErrorCode::NotReady, "runtime is stopping"};
191226
}
192227
--outstanding_;
193-
command.completion.set_value(std::move(result));
194228
}
229+
complete(std::move(command.completion), std::move(result));
195230
}
196231
decltype(sessions_) retired;
197232
{
@@ -260,16 +295,34 @@ ServingRuntime::~ServingRuntime() {
260295
shutdown();
261296
}
262297

298+
void ServingRuntime::open_session_async(
299+
std::string key,
300+
LifecycleCallback on_complete) {
301+
impl_->submit(Impl::Operation::Open, std::move(key), std::move(on_complete));
302+
}
303+
263304
std::future<LifecycleResult> ServingRuntime::open_session_async(
264305
std::string key) {
265306
return impl_->submit(Impl::Operation::Open, std::move(key));
266307
}
267308

309+
void ServingRuntime::close_session_async(
310+
std::string key,
311+
LifecycleCallback on_complete) {
312+
impl_->submit(Impl::Operation::Close, std::move(key), std::move(on_complete));
313+
}
314+
268315
std::future<LifecycleResult> ServingRuntime::close_session_async(
269316
std::string key) {
270317
return impl_->submit(Impl::Operation::Close, std::move(key));
271318
}
272319

320+
void ServingRuntime::reset_session_async(
321+
std::string key,
322+
LifecycleCallback on_complete) {
323+
impl_->submit(Impl::Operation::Reset, std::move(key), std::move(on_complete));
324+
}
325+
273326
std::future<LifecycleResult> ServingRuntime::reset_session_async(
274327
std::string key) {
275328
return impl_->submit(Impl::Operation::Reset, std::move(key));

‎extension/llm/serving/serving_runtime.h‎

Lines changed: 22 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
#pragma once
1010

1111
#include <cstddef>
12+
#include <functional>
1213
#include <future>
1314
#include <memory>
1415
#include <optional>
@@ -28,13 +29,15 @@ struct ET_EXPERIMENTAL ServingRuntimeConfig {
2829
std::size_t max_sessions = 1;
2930
// Reported metadata; 0 means unknown.
3031
std::size_t max_context_length = 0;
31-
// Queued plus executing lifecycle operations. Must be non-zero.
32+
// Queued plus executing lifecycle operations, excluding completion
33+
// callbacks. Must be non-zero.
3234
std::size_t max_pending_operations = 64;
3335
};
3436

3537
// nullopt acknowledges success. Dropping a future does not cancel its
3638
// operation.
3739
using LifecycleResult ET_EXPERIMENTAL = std::optional<ServingError>;
40+
using LifecycleCallback ET_EXPERIMENTAL = std::function<void(LifecycleResult)>;
3841

3942
class ET_EXPERIMENTAL ServingRuntime {
4043
public:
@@ -55,34 +58,45 @@ class ET_EXPERIMENTAL ServingRuntime {
5558
ServingRuntime& operator=(ServingRuntime&&) = delete;
5659

5760
// Any thread. Keys must be non-empty. Accepted operations run in admission
58-
// order, including operations submitted before a previous future completes.
59-
// Queue saturation returns CapacityExceeded without changing session state.
61+
// order. Queue saturation reports CapacityExceeded without changing state.
62+
// For all lifecycle operations, callbacks run once on control after
63+
// processing and admission release, or inline on the caller for immediate
64+
// rejection. They run without runtime locks and may race with submission's
65+
// return. Callbacks must do short, nonblocking work: no waits for runtime
66+
// work or runtime shutdown/destruction. Exceptions are logged, not retried.
67+
// An empty callback discards the result. Futures adapt this same completion
68+
// path.
69+
//
6070
// Opens may queue during initialization; success means an empty session is
6171
// owned, or the key was already open. An initial open failure releases its
6272
// slot. Opening an unavailable key fails with NotReady; use reset to retry.
73+
void open_session_async(std::string key, LifecycleCallback on_complete);
6374
std::future<LifecycleResult> open_session_async(std::string key);
6475

6576
// Idempotent, including absent keys. Success releases the logical slot and
6677
// requests Session closure through RAII, not synchronous physical cleanup.
78+
void close_session_async(std::string key, LifecycleCallback on_complete);
6779
std::future<LifecycleResult> close_session_async(std::string key);
6880

6981
// Cold replacement, retaining the key and logical slot throughout. Success
7082
// means reopening completed. Failure destroys the old state and leaves the
7183
// key unavailable, still reserving its slot until reset succeeds or close
7284
// releases it. A missing key returns SessionNotFound. Never uses prefix
7385
// reuse.
86+
void reset_session_async(std::string key, LifecycleCallback on_complete);
7487
std::future<LifecycleResult> reset_session_async(std::string key);
7588

7689
// A synchronized snapshot. Ready means initialized and accepting operations,
7790
// not that capacity is available. Active slots include opening, reopening,
7891
// and unavailable keys, but not commands still waiting to reserve a slot.
7992
ServingInfo info() const;
8093

81-
// Idempotent, including concurrent callers. Stops admission, settles all
82-
// accepted futures, and joins the control and Runner threads. No new work is
83-
// accepted afterward. Call only from external threads, never engine
84-
// callbacks. Destruction calls shutdown; as usual, object lifetime must be
85-
// externally synchronized against callers still accessing the runtime.
94+
// Idempotent, including concurrent callers. Stops admission, completes all
95+
// accepted operations, and joins the control and Runner threads. Accepted
96+
// callbacks have returned and their runtime-owned captures are released.
97+
// No new work is accepted afterward. Call only from external threads, never
98+
// engine or lifecycle callbacks. Destruction calls shutdown; as usual, object
99+
// lifetime must be synchronized against callers still accessing the runtime.
86100
void shutdown();
87101

88102
private:

0 commit comments

Comments
 (0)