Skip to content

Commit 692ec8b

Browse files
johnlannijohnlanni
committed
support wasm redis call
Link: https://code.alibaba-inc.com/Ingress/proxy-wasm-cpp-host/codereview/14429674
1 parent 4d97080 commit 692ec8b

File tree

13 files changed

+153
-18
lines changed

13 files changed

+153
-18
lines changed

‎WORKSPACE

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,14 +8,14 @@ load("@proxy_wasm_cpp_host//bazel:dependencies.bzl", "proxy_wasm_cpp_host_depend
88

99
proxy_wasm_cpp_host_dependencies()
1010

11-
load("@proxy_wasm_cpp_sdk//bazel:repositories.bzl", "proxy_wasm_cpp_sdk_repositories")
11+
load("@proxy_wasm_cpp_sdk//bazel:repositories.bzl", "proxy_wasm_cpp_host_repositories")
1212

13-
proxy_wasm_cpp_sdk_repositories()
13+
proxy_wasm_cpp_host_repositories()
1414

15-
load("@proxy_wasm_cpp_sdk//bazel:dependencies.bzl", "proxy_wasm_cpp_sdk_dependencies")
15+
load("@proxy_wasm_cpp_sdk//bazel:dependencies.bzl", "proxy_wasm_cpp_host_dependencies")
1616

17-
proxy_wasm_cpp_sdk_dependencies()
17+
proxy_wasm_cpp_host_dependencies()
1818

19-
load("@proxy_wasm_cpp_sdk//bazel:dependencies_extra.bzl", "proxy_wasm_cpp_sdk_dependencies_extra")
19+
load("@proxy_wasm_cpp_sdk//bazel:dependencies_extra.bzl", "proxy_wasm_cpp_host_dependencies_extra")
2020

21-
proxy_wasm_cpp_sdk_dependencies_extra()
21+
proxy_wasm_cpp_host_dependencies_extra()

‎bazel/repositories.bzl

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -101,11 +101,10 @@ def proxy_wasm_cpp_host_repositories():
101101
)
102102

103103
maybe(
104-
http_archive,
104+
git_repository,
105105
name = "proxy_wasm_cpp_sdk",
106-
sha256 = "89792fc1abca331f29f99870476a04146de5e82ff903bdffca90e6729c1f2470",
107-
strip_prefix = "proxy-wasm-cpp-sdk-95bb82ce45c41d9100fd1ec15d2ffc67f7f3ceee",
108-
urls = ["https://github.com/proxy-wasm/proxy-wasm-cpp-sdk/archive/95bb82ce45c41d9100fd1ec15d2ffc67f7f3ceee.tar.gz"],
106+
commit = "0a2b9e1928878e1662ad671be4ee9a8d0853243a",
107+
remote = "[email protected]:Ingress/proxy-wasm-cpp-sdk.git",
109108
)
110109

111110
# Test dependencies.

‎include/proxy-wasm/context.h

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,7 @@ class ContextBase : public RootInterface,
136136
public StreamInterface,
137137
public HeaderInterface,
138138
public HttpCallInterface,
139+
public RedisCallInterface,
139140
public GrpcCallInterface,
140141
public GrpcStreamInterface,
141142
public MetricsInterface,
@@ -214,6 +215,10 @@ class ContextBase : public RootInterface,
214215
// Async call response.
215216
void onHttpCallResponse(HttpCallToken token, uint32_t headers, uint32_t body_size,
216217
uint32_t trailers) override;
218+
219+
// Redis
220+
void onRedisCallResponse(RedisCallToken token, uint32_t status, uint32_t response_size) override;
221+
217222
// Grpc
218223
void onGrpcReceiveInitialMetadata(GrpcToken token, uint32_t elements) override;
219224
void onGrpcReceive(GrpcToken token, uint32_t response_size) override;
@@ -276,6 +281,16 @@ class ContextBase : public RootInterface,
276281
return unimplemented();
277282
}
278283

284+
// Redis
285+
WasmResult redisInit(std::string_view /* target */, std::string_view /* username */,
286+
std::string_view /* password */, int /* timeout_millisconds */) override {
287+
return unimplemented();
288+
}
289+
WasmResult redisCall(std::string_view /* target */, std::string_view /* query */,
290+
uint32_t * /* token_ptr */) override {
291+
return unimplemented();
292+
}
293+
279294
// gRPC
280295
WasmResult grpcCall(std::string_view /* grpc_service */, std::string_view /* service_name */,
281296
std::string_view /* method_name */, const Pairs & /* initial_metadata */,

‎include/proxy-wasm/context_interface.h

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ using GrpcToken = uint32_t;
3838
using GrpcStatusCode = uint32_t;
3939
using SharedQueueDequeueToken = uint32_t;
4040
using SharedQueueEnqueueToken = uint32_t;
41+
using RedisCallToken = uint32_t;
4142

4243
// TODO: update SDK and use this.
4344
enum class ProxyAction : uint32_t {
@@ -162,6 +163,12 @@ struct RootInterface : public RootGrpcInterface {
162163
*/
163164
virtual void onHttpCallResponse(HttpCallToken token, uint32_t headers, uint32_t body_size,
164165
uint32_t trailers) = 0;
166+
/**
167+
* Called on a Root Context when a response arrives for an outstanding redisCall().
168+
* @param token is the token returned by the corresponding redisCall().
169+
*/
170+
virtual void onRedisCallResponse(RedisCallToken token, uint32_t status,
171+
uint32_t response_size) = 0;
165172

166173
/**
167174
* Called on a Root Context when an Inter-VM shared queue message has arrived.
@@ -417,6 +424,14 @@ struct HttpCallInterface {
417424
int timeout_milliseconds, HttpCallToken *token_ptr) = 0;
418425
};
419426

427+
struct RedisCallInterface {
428+
virtual ~RedisCallInterface() = default;
429+
virtual WasmResult redisInit(std::string_view target, std::string_view username,
430+
std::string_view password, int timeout_milliseconds) = 0;
431+
virtual WasmResult redisCall(std::string_view target, std::string_view query,
432+
RedisCallToken *token_ptr) = 0;
433+
};
434+
420435
struct GrpcCallInterface {
421436
virtual ~GrpcCallInterface() = default;
422437
/**

‎include/proxy-wasm/exports.h

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,10 @@ Word get_response_body_buffer_bytes(Word start, Word length, Word ptr_ptr, Word
104104
Word http_call(Word uri_ptr, Word uri_size, Word header_pairs_ptr, Word header_pairs_size,
105105
Word body_ptr, Word body_size, Word trailer_pairs_ptr, Word trailer_pairs_size,
106106
Word timeout_milliseconds, Word token_ptr);
107+
Word redis_init(Word service_ptr, Word service_size, Word username_ptr, Word username_size,
108+
Word passowrd_ptr, Word password_size, Word timeout_milliseconds);
109+
Word redis_call(Word service_ptr, Word service_size, Word query_ptr, Word query_size,
110+
Word token_ptr);
107111
Word define_metric(Word metric_type, Word name_ptr, Word name_size, Word metric_id_ptr);
108112
Word increment_metric(Word metric_id, int64_t offset);
109113
Word record_metric(Word metric_id, uint64_t value);
@@ -163,7 +167,8 @@ void emscripten_notify_memory_growth(Word);
163167
_f(get_current_time_nanoseconds) _f(define_metric) \
164168
_f(increment_metric) _f(record_metric) _f(get_metric) \
165169
_f(set_effective_context) _f(done) \
166-
_f(call_foreign_function)
170+
_f(call_foreign_function) _f(redis_init) \
171+
_f(redis_call)
167172

168173
#define FOR_ALL_HOST_FUNCTIONS_ABI_SPECIFIC(_f) \
169174
_f(get_configuration) _f(continue_request) _f(continue_response) _f(clear_route_cache) \

‎include/proxy-wasm/null_plugin.h

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,8 @@ class NullPlugin : public NullVmPlugin {
9292
void onHttpCallResponse(uint64_t context_id, uint64_t token, uint64_t headers, uint64_t body_size,
9393
uint64_t trailers);
9494

95+
void onRedisCallResponse(uint64_t context_id, uint64_t token, uint64_t status, uint64_t response);
96+
9597
void onGrpcReceive(uint64_t context_id, uint64_t token, size_t body_size);
9698
void onGrpcClose(uint64_t context_id, uint64_t token, uint64_t status_code);
9799
void onGrpcReceiveInitialMetadata(uint64_t context_id, uint64_t token, uint64_t headers);

‎include/proxy-wasm/wasm.h

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -176,6 +176,7 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
176176
HttpCall = 0,
177177
GrpcCall = 1,
178178
GrpcStream = 2,
179+
RedisCall = 3,
179180
};
180181
static const uint32_t kCalloutTypeMask = 0x3; // Enough to cover the 3 types.
181182
static const uint32_t kCalloutIncrement = 0x4; // Enough to cover the 3 types.
@@ -188,6 +189,9 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
188189
bool isGrpcStreamId(uint32_t callout_id) {
189190
return (callout_id & kCalloutTypeMask) == static_cast<uint32_t>(CalloutType::GrpcStream);
190191
}
192+
bool isRedisCallId(uint32_t callout_id) {
193+
return (callout_id & kCalloutTypeMask) == static_cast<uint32_t>(CalloutType::RedisCall);
194+
}
191195
uint32_t nextHttpCallId() {
192196
// TODO(PiotrSikora): re-add rollover protection (requires at least 1 billion callouts).
193197
return next_http_call_id_ += kCalloutIncrement;
@@ -200,6 +204,10 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
200204
// TODO(PiotrSikora): re-add rollover protection (requires at least 1 billion callouts).
201205
return next_grpc_stream_id_ += kCalloutIncrement;
202206
}
207+
uint32_t nextRedisCallId() {
208+
// TODO(PiotrSikora): re-add rollover protection (requires at least 1 billion callouts).
209+
return next_redis_call_id_ += kCalloutIncrement;
210+
}
203211

204212
protected:
205213
friend class ContextBase;
@@ -259,6 +267,8 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
259267

260268
WasmCallVoid<5> on_http_call_response_;
261269

270+
WasmCallVoid<4> on_redis_call_response_;
271+
262272
WasmCallVoid<3> on_grpc_receive_;
263273
WasmCallVoid<3> on_grpc_close_;
264274
WasmCallVoid<3> on_grpc_create_initial_metadata_;
@@ -278,9 +288,10 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
278288
_f(on_downstream_connection_close) _f(on_upstream_connection_close) _f(on_request_body) \
279289
_f(on_request_trailers) _f(on_request_metadata) _f(on_response_body) \
280290
_f(on_response_trailers) _f(on_response_metadata) _f(on_http_call_response) \
281-
_f(on_grpc_receive) _f(on_grpc_close) _f(on_grpc_receive_initial_metadata) \
282-
_f(on_grpc_receive_trailing_metadata) _f(on_queue_ready) _f(on_done) \
283-
_f(on_log) _f(on_delete)
291+
_f(on_redis_call_response) _f(on_grpc_receive) _f(on_grpc_close) \
292+
_f(on_grpc_receive_initial_metadata) \
293+
_f(on_grpc_receive_trailing_metadata) _f(on_queue_ready) _f(on_done) \
294+
_f(on_log) _f(on_delete)
284295

285296
// Capabilities which are allowed to be linked to the module. If this is empty, restriction
286297
// is not enforced.
@@ -309,6 +320,7 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
309320
uint32_t next_http_call_id_ = static_cast<uint32_t>(CalloutType::HttpCall);
310321
uint32_t next_grpc_call_id_ = static_cast<uint32_t>(CalloutType::GrpcCall);
311322
uint32_t next_grpc_stream_id_ = static_cast<uint32_t>(CalloutType::GrpcStream);
323+
uint32_t next_redis_call_id_ = static_cast<uint32_t>(CalloutType::RedisCall);
312324

313325
// Actions to be done after the call into the VM returns.
314326
std::deque<std::function<void()>> after_vm_call_actions_;

‎include/proxy-wasm/wasm_api_impl.h

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,23 @@ inline WasmResult proxy_http_call(const char *uri_ptr, size_t uri_size, void *he
213213
WR(trailer_pairs_ptr), WS(trailer_pairs_size),
214214
WS(timeout_milliseconds), WR(token_ptr)));
215215
}
216+
217+
// Redis
218+
// Returns token, used in callback onRedisCallResponse
219+
inline WasmResult proxy_redis_call(const char *service_ptr, size_t service_size,
220+
const char *query_ptr, size_t query_size, uint32_t *token_ptr) {
221+
return wordToWasmResult(exports::redis_call(WR(service_ptr), WS(service_size), WR(query_ptr),
222+
WS(query_size), WR(token_ptr)));
223+
}
224+
inline WasmResult proxy_redis_init(const char *service_ptr, size_t service_size,
225+
const char *username_ptr, size_t username_size,
226+
const char *password_ptr, size_t password_size,
227+
uint32_t timeout_milliseconds) {
228+
return wordToWasmResult(exports::redis_init(WR(service_ptr), WS(service_size), WR(username_ptr),
229+
WS(username_size), WR(password_ptr),
230+
WS(password_size), WS(timeout_milliseconds)));
231+
}
232+
216233
// gRPC
217234
// Returns token, used in gRPC callbacks (onGrpc...)
218235
inline WasmResult proxy_grpc_call(const char *service_ptr, size_t service_size,

‎include/proxy-wasm/wasm_vm.h

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -70,9 +70,9 @@ using WasmCallWord = std::function<WasmCallInFuncType<N, Word, ContextBase *, Wo
7070

7171
#define FOR_ALL_WASM_VM_EXPORTS(_f) \
7272
_f(proxy_wasm::WasmCallVoid<0>) _f(proxy_wasm::WasmCallVoid<1>) _f(proxy_wasm::WasmCallVoid<2>) \
73-
_f(proxy_wasm::WasmCallVoid<3>) _f(proxy_wasm::WasmCallVoid<5>) \
74-
_f(proxy_wasm::WasmCallWord<1>) _f(proxy_wasm::WasmCallWord<2>) \
75-
_f(proxy_wasm::WasmCallWord<3>)
73+
_f(proxy_wasm::WasmCallVoid<3>) _f(proxy_wasm::WasmCallVoid<4>) \
74+
_f(proxy_wasm::WasmCallVoid<5>) _f(proxy_wasm::WasmCallWord<1>) \
75+
_f(proxy_wasm::WasmCallWord<2>) _f(proxy_wasm::WasmCallWord<3>)
7676

7777
// These are templates and its helper for constructing signatures of functions callbacks from Wasm
7878
// VMs.
@@ -139,7 +139,13 @@ enum class Cloneable {
139139
InstantiatedModule // VMs can be cloned from an instantiated module.
140140
};
141141

142-
enum class AbiVersion { ProxyWasm_0_1_0, ProxyWasm_0_2_0, ProxyWasm_0_2_1, ProxyWasm_0_2_100, Unknown };
142+
enum class AbiVersion {
143+
ProxyWasm_0_1_0,
144+
ProxyWasm_0_2_0,
145+
ProxyWasm_0_2_1,
146+
ProxyWasm_0_2_100,
147+
Unknown
148+
};
143149

144150
class NullPlugin;
145151

‎src/context.cc

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
// limitations under the License.
1515

1616
#include <cassert>
17+
#include <cstdint>
1718
#include <deque>
1819
#include <map>
1920
#include <memory>
@@ -422,6 +423,14 @@ void ContextBase::onHttpCallResponse(uint32_t token, uint32_t headers, uint32_t
422423
wasm_->on_http_call_response_(this, id_, token, headers, body_size, trailers);
423424
}
424425

426+
void ContextBase::onRedisCallResponse(uint32_t token, uint32_t status, uint32_t response_size) {
427+
if (isFailed() || !wasm_->on_redis_call_response_) {
428+
return;
429+
}
430+
DeferAfterCallActions actions(this);
431+
wasm_->on_redis_call_response_(this, id_, token, status, response_size);
432+
}
433+
425434
void ContextBase::onQueueReady(uint32_t token) {
426435
if (!isFailed() && wasm_->on_queue_ready_) {
427436
DeferAfterCallActions actions(this);

‎src/exports.cc

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -533,6 +533,35 @@ Word http_call(Word uri_ptr, Word uri_size, Word header_pairs_ptr, Word header_p
533533
return result;
534534
}
535535

536+
Word redis_init(Word service_ptr, Word service_size, Word username_ptr, Word username_size,
537+
Word passowrd_ptr, Word password_size, Word timeout_milliseconds) {
538+
auto *context = contextOrEffectiveContext()->root_context();
539+
auto service = context->wasmVm()->getMemory(service_ptr, service_size);
540+
auto username = context->wasmVm()->getMemory(username_ptr, username_size);
541+
auto password = context->wasmVm()->getMemory(passowrd_ptr, password_size);
542+
if (!service || !username || !password) {
543+
return WasmResult::InvalidMemoryAccess;
544+
}
545+
return context->redisInit(service.value(), username.value(), password.value(),
546+
timeout_milliseconds);
547+
}
548+
549+
Word redis_call(Word service_ptr, Word service_size, Word query_ptr, Word query_size,
550+
Word token_ptr) {
551+
auto *context = contextOrEffectiveContext()->root_context();
552+
auto service = context->wasmVm()->getMemory(service_ptr, service_size);
553+
auto query = context->wasmVm()->getMemory(query_ptr, query_size);
554+
uint32_t token = 0;
555+
// NB: try to write the token to verify the memory before starting the async
556+
// operation.
557+
if (!context->wasm()->setDatatype(token_ptr, token)) {
558+
return WasmResult::InvalidMemoryAccess;
559+
}
560+
auto result = context->redisCall(service.value(), query.value(), &token);
561+
context->wasm()->setDatatype(token_ptr, token);
562+
return result;
563+
}
564+
536565
Word define_metric(Word metric_type, Word name_ptr, Word name_size, Word metric_id_ptr) {
537566
auto *context = contextOrEffectiveContext();
538567
auto name = context->wasmVm()->getMemory(name_ptr, name_size);

‎src/null/null_plugin.cc

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,20 @@ void NullPlugin::getFunction(std::string_view function_name, WasmCallVoid<3> *f)
127127
}
128128
}
129129

130+
void NullPlugin::getFunction(std::string_view function_name, WasmCallVoid<4> *f) {
131+
auto *plugin = this;
132+
if (function_name == "proxy_on_redis_call_response") {
133+
*f = [plugin](ContextBase *context, Word context_id, Word token, Word status,
134+
Word response_size) {
135+
SaveRestoreContext saved_context(context);
136+
plugin->onRedisCallResponse(context_id, token, status, response_size);
137+
};
138+
} else if (!wasm_vm_->integration()->getNullVmFunction(function_name, false, 4, this, f)) {
139+
error("Missing getFunction for: " + std::string(function_name));
140+
*f = nullptr;
141+
}
142+
}
143+
130144
void NullPlugin::getFunction(std::string_view function_name, WasmCallVoid<5> *f) {
131145
auto *plugin = this;
132146
if (function_name == "proxy_on_http_call_response") {
@@ -441,6 +455,12 @@ void NullPlugin::onHttpCallResponse(uint64_t context_id, uint64_t token, uint64_
441455
getRootContext(context_id)->onHttpCallResponse(token, headers, body_size, trailers);
442456
}
443457

458+
void NullPlugin::onRedisCallResponse(uint64_t context_id, uint64_t token, uint64_t status,
459+
uint64_t response_size) {
460+
getRootContext(context_id)
461+
->onRedisCallResponse(token, static_cast<RedisStatus>(status), response_size);
462+
}
463+
444464
void NullPlugin::onGrpcReceive(uint64_t context_id, uint64_t token, size_t body_size) {
445465
getRootContext(context_id)->onGrpcReceive(token, body_size);
446466
}

‎src/wasm.cc

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,8 @@ void WasmBase::registerCallbacks() {
129129
_REGISTER_PROXY(continue_stream);
130130
_REGISTER_PROXY(close_stream);
131131
_REGISTER_PROXY(get_log_level);
132+
_REGISTER_PROXY(redis_init);
133+
_REGISTER_PROXY(redis_call);
132134
}
133135
#undef _REGISTER_PROXY
134136

@@ -179,8 +181,12 @@ void WasmBase::getFunctions() {
179181
_GET_PROXY_ABI(on_request_headers, _abi_02);
180182
_GET_PROXY_ABI(on_response_headers, _abi_02);
181183
_GET_PROXY(on_foreign_function);
184+
_GET_PROXY(on_redis_call_response);
182185
} else if (abiVersion() == AbiVersion::ProxyWasm_0_2_100) {
183186
_GET_PROXY_ABI(on_request_headers, _abi_03);
187+
_GET_PROXY_ABI(on_response_headers, _abi_02);
188+
_GET_PROXY(on_foreign_function);
189+
_GET_PROXY(on_redis_call_response);
184190
}
185191
#undef _GET_PROXY_ABI
186192
#undef _GET_PROXY

0 commit comments

Comments
 (0)