diff --git a/Readme.md b/Readme.md index 7413e42..4e62c4f 100644 --- a/Readme.md +++ b/Readme.md @@ -96,19 +96,19 @@ message-specific payload. ### Message Types -| Value | Name | arg1 | arg2 | Payload | -| ------ | --------------------- | --------------------------- | ----------- | --------------------------------------- | -| `0x00` | UNKNOWN | — | — | — | -| `0x01` | DATA | — | `target_id` | Raw value bytes | -| `0x02` | CONFIGURATION_REQUEST | — | — | Empty | -| `0x03` | CLAIM | `0`=request, `1`=ack | — | `ClaimPayload` (request) or empty (ack) | -| `0x04` | HEARTBEAT | — | — | Empty | -| `0x05` | TRANSACTION | `0`=data, `1`=configuration | — | Sequence of `DataDescriptor`+payload | -| `0x7F` | LOG | log level (1–7) | — | UTF-8 string (max 255 bytes) | -| `0x80` | SERVICE_ADVERTISEMENT | — | — | CBOR-encoded service JSON | -| `0x81` | SERVICE_QUERY | — | — | — | -| `0x82` | DESCRIPTION_REQUEST | offset into the omitted desc | `max_size` | `DescriptionRequestPayload` (ip:port) | -| `0x83` | DESCRIPTION_PART | offset into the omitted desc | `total_size`| Window of the omitted description | +| Value | Name | arg1 | arg2 | Payload | +| ------ | --------------------- | ---------------------------- | ------------ | --------------------------------------- | +| `0x00` | UNKNOWN | — | — | — | +| `0x01` | DATA | — | `target_id` | Raw value bytes | +| `0x02` | CONFIGURATION_REQUEST | — | — | Empty | +| `0x03` | CLAIM | `0`=request, `1`=ack | — | `ClaimPayload` (request) or empty (ack) | +| `0x04` | HEARTBEAT | — | — | Empty | +| `0x05` | TRANSACTION | `0`=data, `1`=configuration | — | Sequence of `DataDescriptor`+payload | +| `0x7F` | LOG | log level (1–7) | — | UTF-8 string (max 255 bytes) | +| `0x80` | SERVICE_ADVERTISEMENT | — | — | CBOR-encoded service JSON | +| `0x81` | SERVICE_QUERY | — | — | — | +| `0x82` | DESCRIPTION_REQUEST | offset into the omitted desc | `max_size` | `DescriptionRequestPayload` (ip:port) | +| `0x83` | DESCRIPTION_PART | offset into the omitted desc | `total_size` | Window of the omitted description | `DESCRIPTION_REQUEST` is sent unicast by a host that needs the part of the schema the advertisement could not carry (see *Service description* below); the service answers with one @@ -245,20 +245,46 @@ Services are defined in a JSON file. The code generator (`codegen/`) reads this } ``` -**Valid types:** `char`, `uint8_t`, `uint16_t`, `uint32_t`, `uint64_t`, `int8_t`, `int16_t`, `int32_t`, `int64_t`, `float`, `double`, `blob` (registers only), or fixed-length arrays of any scalar type as `type[N]`. +**Valid types:** `char`, `uint8_t`, `uint16_t`, `uint32_t`, `uint64_t`, `int8_t`, `int16_t`, `int32_t`, `int64_t`, `float`, `double`, `blob` (registers, inputs, outputs, RPC parameters and RPC return values), or fixed-length arrays of any scalar type as `type[N]`. + +`blob` is variable length: as a parameter it is passed as `(pointer, length)` into the received packet and is +not copied, as an input or output it is passed/sent as `(pointer, length)` with the length in bytes, as a +register it is the configuration blob, and as a return value it uses the shared buffer pool +(see [Shared Buffer Pool](#shared-buffer-pool)). A `type[N]` return above 128 bytes is pooled the same way, so a +service never reserves stack for a payload it rarely sends. + +Switching a parameter or return from a fixed `type[N]` to `blob` changes its wire layout, so the service needs a +version bump (`"version": 2`). Generated clients refuse a mismatch instead of sending a call with the wrong +layout: the C++ and Python clients always compare against the schema they were given, the Go client compares +only when `ExpectVersion` is set (generated clients set it, a hand-written one has to do the same). Service and +clients have to be updated together. **IDs** must be unique within each section (`inputs`, `outputs`, `registers`). **Generated API (service side):** - `void On{Name}Changed(const T* value, uint32_t length)` — called when an input arrives (array types) - `void On{Name}Changed(const T& value)` — called when an input arrives (scalar types) +- `void On{Name}Changed(const uint8_t* value, uint32_t length)` — called when a blob input arrives, `length` in bytes - `bool Send{Name}(const T* data, uint32_t length)` — send an output (array types) - `bool Send{Name}(const T& data)` — send an output (scalar types) +- `bool Send{Name}(const uint8_t* data, uint32_t length)` — send a blob output, `length` in bytes - `bool OnRegister{Name}Changed(const void* data, size_t length)` — called when a register is set (blob registers) - Struct `{Name}` with `value`, `length` (arrays), and `valid` fields — for non-blob registers +- `void RPC{Name}(uint16_t call_id, ...)` — one pure virtual per function; a `blob` parameter arrives as + `const uint8_t* , uint32_t Len` and is only valid until the handler returns + (copy it if you need it longer) +- A `type[N]` return above the stack threshold and a `blob` return append `uint8_t* data, + uint16_t* response_length`. A `blob` return is always `uint8_t`, a `type[N]` return uses its element + type; `*response_length` is a byte count for `blob` and an element count for `type[N]` **Generated API (interface side — same JSON, different template):** - Inputs and outputs swap roles: outputs become callbacks, inputs become send methods. +- Every `blob` parameter gets a frame budget helper, e.g. `MaxFileWriteData(PathLen)` in C++ and + `MaxFileWriteData(Path []byte)` in Go: how many bytes the parameter can carry in one call, with the + other blob parameters as arguments. Chunking a value needs exactly that number, and the framing that + limits it belongs to the generator, not to every client. The helpers saturate: a call whose other + parameters already use up the packet reports 0, so a chunker has to stop instead of sending. Every + chunk is `min(bytes left, budget)` long - the budget is an upper bound, only the last chunk is shorter. **CMake integration:** ```cmake @@ -275,6 +301,61 @@ Then inherit from the generated `MyServiceNameBase` or `MyInterfaceNameBase` cla --- +## Shared Buffer Pool + +An RPC return value is limited by `max_packet_size - sizeof(XbotHeader)` = 1448 bytes, but a service +thread stack is much smaller. Array return values above a small threshold therefore do not live on +the stack (that is what used to force the fixed size of a transfer type); the generated dispatcher +borrows a block from a pool that all services share: + +```cpp +#include + +auto lease = xbot::service::buffer::acquire(required_bytes); +if (!lease) { + SendRpcResponse(call_id, xbot::datatypes::RpcStatus::BUSY, nullptr, 0); + return; +} +fill(lease.data(), lease.size()); // lease.size() is the capacity, not the request +SendRpcResponse(call_id, xbot::datatypes::RpcStatus::SUCCESS, lease.data(), used_bytes); +``` + +Rules that keep the pool usable: + +- A lease is **move-only** and returns its block in the destructor. Do not keep it beyond the call + that borrowed it: the pool is shared by all services, and a lease held across a flash operation or + across RPCs starves the next caller. +- `acquire()` is **non-blocking** on purpose. It returns an empty lease when every block is out, when + no storage has been configured yet, and when the request is larger than a block; a handler that + follows the pattern above answers BUSY in all three cases - an RPC handler must never wait for + another service's transfer. With the default of two blocks, one outstanding return does not exhaust + the pool. A block is only held while a return value is filled and copied into a packet. +- Release happens after the handler returned; `SendRpcResponse()` copies the payload into a packet. +- No heap is involved: the storage is static. + +Sizing is a platform decision. `xbot::config::buffer_block_size` (1536 by default, must hold one +packet payload - asserted at build time) and `buffer_block_count` (2) only size the framework +internal default storage. A platform that brings its own storage - e.g. a section in a RAM region +that is otherwise unused - calls + +```cpp +alignas(xbot::service::buffer::block_alignment) uint8_t storage[2 * 1536] + __attribute__((section(".sram4"))); +xbot::service::buffer::initialize(storage, 1536, 2); +``` + +once, before the services start, and sets `XBOT_BUFFER_DEFAULT_POOL=0` so the default array is not +linked in at all. `initialize()` is not thread safe and refuses to run while a lease is out. + +Parameters hit the same stack limit from the other direction: a fixed `type[N]` parameter is copied into a +stack array by the generated dispatcher, which is why a variable length payload had to be declared as +e.g. `uint8_t[256]`. A `blob` parameter is passed as `(pointer, length)` into the received packet instead - +the contract `handleData()` and `setRegister()` already use - so it costs no stack and no copy, and its +length is only bounded by one packet. The generated dispatcher takes returns above 128 bytes and blob +returns from the pool; small returns stay on the stack, so the pool is not touched in the hot path. + +--- + ## Python Interface `xbot_service_interface_py` is a pure-Python client for xBot services — no codegen step required. It talks directly diff --git a/codegen/templates/ServiceInterfaceTemplate.cpp b/codegen/templates/ServiceInterfaceTemplate.cpp index 44add45..13ca8b6 100644 --- a/codegen/templates/ServiceInterfaceTemplate.cpp +++ b/codegen/templates/ServiceInterfaceTemplate.cpp @@ -28,7 +28,9 @@ void ServiceTemplateInterfaceBase::OnData(uint16_t service_id, uint64_t timestam /*[[[cog for o in service['outputs']: cog.outl(f"case {o['id']}:"); - if o['is_array']: + if o['is_blob']: + cog.outl(f"{o['callback_name']}(static_cast(payload), length);") + elif o['is_array']: cog.outl(f"if(length % sizeof({o['type']}) != 0) {{"); cog.outl(" spdlog::error(\"Invalid data size\");"); cog.outl(" return;"); @@ -56,6 +58,9 @@ void ServiceTemplateInterfaceBase::OnData(uint16_t service_id, uint64_t timestam } OnExampleOutput2Changed(*static_cast(payload)); break; + case 2: + OnExampleOutputBlobChanged(static_cast(payload), length); + break; //[[[end]]] default: return; @@ -96,7 +101,11 @@ bool ServiceTemplateInterfaceBase::SetRegisterRegister4Optional(const uint32_t & /*[[[cog # Generate send function implementations. for input in service["inputs"]: - if input['is_array']: + if input['is_blob']: + cog.outl(f"bool {service['interface_class_name']}::{input['send_method_name']}(const uint8_t* data, uint32_t length) {{") + cog.outl(f" return SendData({input['id']}, data, length, false);") + cog.outl("}") + elif input['is_array']: cog.outl(f"bool {service['interface_class_name']}::{input['send_method_name']}(const {input['type']}* data, uint32_t length) {{") cog.outl(f" return SendData({input['id']}, data, length*sizeof({input['type']}), false);") cog.outl("}") @@ -111,6 +120,9 @@ bool ServiceTemplateInterfaceBase::SendExampleInput1(const char* data, uint32_t bool ServiceTemplateInterfaceBase::SendExampleInput2(const uint32_t &data) { return SendData(1, &data, sizeof(uint32_t), false); } +bool ServiceTemplateInterfaceBase::SendExampleInputBlob(const uint8_t* data, uint32_t length) { + return SendData(2, data, length, false); +} //[[[end]]] /*[[[cog @@ -119,11 +131,15 @@ for func in service["functions"]: # Build method signature sig_params = [] for p in func["parameters"]: - if p['is_array']: + if p['is_blob']: + sig_params.append(f"const uint8_t* {p['name']}, uint32_t {p['name']}Len") + elif p['is_array']: sig_params.append(f"const {p['type']}* {p['name']}, uint32_t {p['name']}Len") else: sig_params.append(f"const {p['type']}& {p['name']}") - if func['return_is_array']: + if func['return_is_blob']: + sig_params.append("uint8_t* data, uint16_t& result_length") + elif func['return_is_array']: sig_params.append(f"{func['return_base_type']}* data, uint16_t& result_length") elif func["return_type"] != "void": sig_params.append(f"{func['return_type']}& result") @@ -135,9 +151,13 @@ for func in service["functions"]: if func["parameters"]: cog.outl(" std::vector params_buf;") for p in func["parameters"]: - if p['is_array']: + if p['is_blob'] or p['is_array']: + if p['is_blob']: + byte_len_expr = f"{p['name']}Len" + else: + byte_len_expr = f"{p['name']}Len * sizeof({p['type']})" cog.outl(f" {{") - cog.outl(f" const size_t byte_len = {p['name']}Len * sizeof({p['type']});") + cog.outl(f" const size_t byte_len = {byte_len_expr};") cog.outl(f" const size_t off = params_buf.size();") cog.outl(f" params_buf.resize(off + sizeof(xbot::datatypes::DataDescriptor) + byte_len);") cog.outl(f" auto* desc = reinterpret_cast(params_buf.data() + off);") @@ -158,7 +178,12 @@ for func in service["functions"]: params_call = "nullptr, 0" # Call SendRpc with appropriate response buffer - if func['return_is_array']: + if func['return_is_blob']: + cog.outl(f" size_t resp_size = result_length;") + cog.outl(f" const auto r = SendRpc({func['id']}, {params_call}, data, &resp_size, timeout_ms);") + cog.outl(f" if (r != RPC_OK) return false;") + cog.outl(f" result_length = static_cast(resp_size);") + elif func['return_is_array']: cog.outl(f" size_t resp_size = result_length * sizeof({func['return_base_type']});") cog.outl(f" const auto r = SendRpc({func['id']}, {params_call}, reinterpret_cast(data), &resp_size, timeout_ms);") cog.outl(f" if (r != RPC_OK) return false;") @@ -257,4 +282,28 @@ bool ServiceTemplateInterfaceBase::CallMixedParamsWithReturn(const char* Name, u result = resp_buf; return true; } +bool ServiceTemplateInterfaceBase::CallBlobParamsWithBlobReturn(const uint8_t* Data, uint32_t DataLen, uint8_t* data, uint16_t& result_length, uint32_t timeout_ms) { + std::vector params_buf; + { + const size_t byte_len = DataLen; + const size_t off = params_buf.size(); + params_buf.resize(off + sizeof(xbot::datatypes::DataDescriptor) + byte_len); + auto* desc = reinterpret_cast(params_buf.data() + off); + desc->target_id = 0; desc->reserved = 0; desc->payload_size = static_cast(byte_len); + if (byte_len > 0 && Data == nullptr) return false; + memcpy(params_buf.data() + off + sizeof(xbot::datatypes::DataDescriptor), Data, byte_len); + } + size_t resp_size = result_length; + const auto r = SendRpc(4, params_buf.data(), params_buf.size(), data, &resp_size, timeout_ms); + if (r != RPC_OK) return false; + result_length = static_cast(resp_size); + return true; +} +bool ServiceTemplateInterfaceBase::CallLargeArrayReturn(uint8_t* data, uint16_t& result_length, uint32_t timeout_ms) { + size_t resp_size = result_length * sizeof(uint8_t); + const auto r = SendRpc(5, nullptr, 0, reinterpret_cast(data), &resp_size, timeout_ms); + if (r != RPC_OK) return false; + result_length = static_cast(resp_size / sizeof(uint8_t)); + return true; +} //[[[end]]] diff --git a/codegen/templates/ServiceInterfaceTemplate.hpp b/codegen/templates/ServiceInterfaceTemplate.hpp index eff31bf..ba48ae9 100644 --- a/codegen/templates/ServiceInterfaceTemplate.hpp +++ b/codegen/templates/ServiceInterfaceTemplate.hpp @@ -18,6 +18,7 @@ cog.outl(f"#define {service['interface_class_name'].upper()}_HPP") #include #include +#include /*[[[cog xbot_codegen.generateEnums(service) @@ -52,13 +53,16 @@ class ServiceTemplateInterfaceBase : public xbot::serviceif::ServiceInterfaceBas /*[[[cog # Generate send functions for each input. for input in service["inputs"]: - if input['is_array']: + if input['is_blob']: + cog.outl(f"bool {input['send_method_name']}(const uint8_t* data, uint32_t length);") + elif input['is_array']: cog.outl(f"bool {input['send_method_name']}(const {input['type']}* data, uint32_t length);") else: cog.outl(f"bool {input['send_method_name']}(const {input['type']} &data);") ]]]*/ bool SendExampleInput1(const char* data, uint32_t length); bool SendExampleInput2(const uint32_t &data); + bool SendExampleInputBlob(const uint8_t* data, uint32_t length); //[[[end]]] /*[[[cog @@ -66,11 +70,15 @@ class ServiceTemplateInterfaceBase : public xbot::serviceif::ServiceInterfaceBas for func in service["functions"]: params = [] for p in func["parameters"]: - if p['is_array']: + if p['is_blob']: + params.append(f"const uint8_t* {p['name']}, uint32_t {p['name']}Len") + elif p['is_array']: params.append(f"const {p['type']}* {p['name']}, uint32_t {p['name']}Len") else: params.append(f"const {p['type']}& {p['name']}") - if func['return_is_array']: + if func['return_is_blob']: + params.append("uint8_t* data, uint16_t& result_length") + elif func['return_is_array']: params.append(f"{func['return_base_type']}* data, uint16_t& result_length") elif func["return_type"] != "void": params.append(f"{func['return_type']}& result") @@ -82,6 +90,64 @@ class ServiceTemplateInterfaceBase : public xbot::serviceif::ServiceInterfaceBas bool CallScalarParamsWithReturn(const float& Speed, const uint32_t& Count, const uint8_t& Enable, int32_t& result, uint32_t timeout_ms = 1000); bool CallArrayParamNoReturn(const char* Label, uint32_t LabelLen, uint32_t timeout_ms = 1000); bool CallMixedParamsWithReturn(const char* Name, uint32_t NameLen, const float& Value, uint8_t& result, uint32_t timeout_ms = 1000); + bool CallBlobParamsWithBlobReturn(const uint8_t* Data, uint32_t DataLen, uint8_t* data, uint16_t& result_length, uint32_t timeout_ms = 1000); + bool CallLargeArrayReturn(uint8_t* data, uint16_t& result_length, uint32_t timeout_ms = 1000); + //[[[end]]] + + /*[[[cog + # A blob parameter is chunked by the caller, so the frame budget belongs next to the + # call that has the limit: one packet of body minus the framing of the call. Every + # subtraction saturates, so a call whose parameters use up the whole packet reports 0 + # instead of a wrapped (huge) budget. + blob_funcs = [f for f in service["functions"] if any(p['is_blob'] for p in f["parameters"])] + if blob_funcs: + cog.outl(" // Saturating subtraction for the frame budgets below: a call whose parameters") + cog.outl(" // use up the whole packet reports 0 instead of a wrapped (huge) budget.") + cog.outl(" static constexpr size_t RemainingBudget(size_t budget) {") + cog.outl(" return budget;") + cog.outl(" }") + cog.outl(" static constexpr size_t RemainingBudget(size_t budget, size_t used) {") + cog.outl(" return (used < budget) ? (budget - used) : 0;") + cog.outl(" }") + cog.outl(" template ") + cog.outl(" static constexpr size_t RemainingBudget(size_t budget, size_t used, Rest... rest) {") + cog.outl(" return RemainingBudget((used < budget) ? (budget - used) : 0, rest...);") + cog.outl(" }") + for func in blob_funcs: + blobs = [p for p in func["parameters"] if p['is_blob']] + fixed = sum( + (p['max_length'] * xbot_codegen.typeSize(service, p['type'])) if p['is_array'] + else xbot_codegen.typeSize(service, p['type']) + for p in func["parameters"] if not p['is_blob']) + for target in blobs: + others = [p for p in blobs if p['name'] != target['name']] + args = ", ".join(f"size_t {p['name']}Len" for p in others) + frame = (f"xbot::config::max_packet_size - sizeof(xbot::datatypes::XbotHeader) - " + f"{len(func['parameters'])} * sizeof(xbot::datatypes::DataDescriptor)") + used = ([f"static_cast({fixed})"] if fixed else []) + [f"{p['name']}Len" for p in others] + budget = f"RemainingBudget({frame}, " + ", ".join(used) + ")" if used else frame + name = f"Max{func['name']}{target['name']}" + cog.outl(f" // {name} is the largest {target['name']} payload one {func['name']} call can carry.") + cog.outl(f" static constexpr size_t {name}({args}) {{") + cog.outl(f" return {budget};") + cog.outl(" }") + ]]]*/ + // Saturating subtraction for the frame budgets below: a call whose parameters + // use up the whole packet reports 0 instead of a wrapped (huge) budget. + static constexpr size_t RemainingBudget(size_t budget) { + return budget; + } + static constexpr size_t RemainingBudget(size_t budget, size_t used) { + return (used < budget) ? (budget - used) : 0; + } + template + static constexpr size_t RemainingBudget(size_t budget, size_t used, Rest... rest) { + return RemainingBudget((used < budget) ? (budget - used) : 0, rest...); + } + // MaxBlobParamsWithBlobReturnData is the largest Data payload one BlobParamsWithBlobReturn call can carry. + static constexpr size_t MaxBlobParamsWithBlobReturnData() { + return xbot::config::max_packet_size - sizeof(xbot::datatypes::XbotHeader) - 1 * sizeof(xbot::datatypes::DataDescriptor); + } //[[[end]]] /*[[[cog @@ -104,7 +170,12 @@ class ServiceTemplateInterfaceBase : public xbot::serviceif::ServiceInterfaceBas /*[[[cog # Generate callback functions for each service output. for output in service["outputs"]: - if output['is_array']: + if output['is_blob']: + cog.outl(f"virtual void {output['callback_name']}(const uint8_t* new_value, uint32_t length) {{") + cog.outl(f"(void)new_value;") + cog.outl(f"(void)length;") + cog.outl(f"}};") + elif output['is_array']: cog.outl(f"virtual void {output['callback_name']}(const {output['type']}* new_value, uint32_t length) {{") cog.outl(f"(void)new_value;") cog.outl(f"(void)length;") @@ -117,6 +188,10 @@ class ServiceTemplateInterfaceBase : public xbot::serviceif::ServiceInterfaceBas (void)length; }; virtual void OnExampleOutput2Changed(const uint32_t &new_value) {(void) new_value;}; + virtual void OnExampleOutputBlobChanged(const uint8_t* new_value, uint32_t length) { + (void)new_value; + (void)length; + }; //[[[end]]] private: @@ -140,6 +215,11 @@ cog.outl(f"\u002f*\n{service['service_json']}\n*\u002f") "id": 1, "name": "ExampleInput2", "type": "uint32_t" + }, + { + "id": 2, + "name": "ExampleInputBlob", + "type": "blob" } ], "outputs": [ @@ -152,6 +232,11 @@ cog.outl(f"\u002f*\n{service['service_json']}\n*\u002f") "id": 1, "name": "ExampleOutput2", "type": "uint32_t" + }, + { + "id": 2, + "name": "ExampleOutputBlob", + "type": "blob" } ], "registers": [ @@ -234,6 +319,24 @@ cog.outl(f"\u002f*\n{service['service_json']}\n*\u002f") } ], "return_type": "uint8_t" + }, + { + "id": 4, + "name": "BlobParamsWithBlobReturn", + "parameters": [ + { + "id": 0, + "name": "Data", + "type": "blob" + } + ], + "return_type": "blob" + }, + { + "id": 5, + "name": "LargeArrayReturn", + "parameters": [], + "return_type": "uint8_t[256]" } ], "enums": [ diff --git a/codegen/templates/ServiceTemplate.cpp b/codegen/templates/ServiceTemplate.cpp index e17fec4..bd865da 100644 --- a/codegen/templates/ServiceTemplate.cpp +++ b/codegen/templates/ServiceTemplate.cpp @@ -3,14 +3,17 @@ /*[[[cog import cog -from xbot_codegen import toCamelCase, loadService +from xbot_codegen import MAX_RPC_RETURN_BYTES, loadService, needsBufferPool, pooledReturnBytes, toCamelCase service = loadService(service_file) cog.outl(f'#include "{service["class_name"]}.hpp"') +if needsBufferPool(service): + cog.outl('#include ') ]]]*/ #include "ServiceTemplateBase.hpp" +#include //[[[end]]] #include #include @@ -47,7 +50,10 @@ void ServiceTemplateBase::handleData(uint16_t target_id, const void *payload, si /*[[[cog for i in service['inputs']: cog.outl(f"case {i['id']}:"); - if i['is_array']: + if i['is_blob']: + cog.outl(f"{i['callback_name']}(static_cast(payload), length);") + cog.outl("return;") + elif i['is_array']: cog.outl(f"if (length % sizeof({i['type']}) != 0) {{"); cog.outl(" ULOG_ARG_ERROR(&service_id_, \"Invalid data size\");"); cog.outl(" return;"); @@ -77,6 +83,9 @@ void ServiceTemplateBase::handleData(uint16_t target_id, const void *payload, si } OnExampleInput2Changed(*static_cast(payload)); return; + case 2: + OnExampleInputBlobChanged(static_cast(payload), length); + return; //[[[end]]] } } @@ -189,7 +198,11 @@ bool ServiceTemplateBase::AdvertiseServiceImpl() { /*[[[cog # Generate send function implementations. for output in service["outputs"]: - if output['is_array']: + if output['is_blob']: + cog.outl(f"bool {service['class_name']}::{output['send_method_name']}(const uint8_t* data, uint32_t length) {{") + cog.outl(f" return SendData({output['id']}, data, length);") + cog.outl("}") + elif output['is_array']: cog.outl(f"bool {service['class_name']}::{output['send_method_name']}(const {output['type']}* data, uint32_t length) {{") cog.outl(f" return SendData({output['id']}, data, length*sizeof({output['type']}));") cog.outl("}") @@ -204,6 +217,9 @@ bool ServiceTemplateBase::SendExampleOutput1(const char* data, uint32_t length) bool ServiceTemplateBase::SendExampleOutput2(const uint32_t &data) { return SendData(1, &data, sizeof(uint32_t)); } +bool ServiceTemplateBase::SendExampleOutputBlob(const uint8_t* data, uint32_t length) { + return SendData(2, data, length); +} //[[[end]]] /*[[[cog @@ -350,9 +366,14 @@ if service["functions"]: cog.outl(" switch (function_id) {") for func in service["functions"]: cog.outl(f" case {func['id']}: {{") - # Declare parameter variables + # Declare parameter variables. A blob parameter is not copied: the handler gets a + # pointer into the received packet (the contract handleData() and setRegister() + # already use), which stays valid until the handler returns. for p in func["parameters"]: - if p['is_array']: + if p['is_blob']: + cog.outl(f" const uint8_t* param_{p['name']} = nullptr;") + cog.outl(f" uint32_t param_{p['name']}Len = 0;") + elif p['is_array']: cog.outl(f" {p['type']} param_{p['name']}[{p['max_length']}] = {{}};") cog.outl(f" uint32_t param_{p['name']}Len = 0;") else: @@ -366,11 +387,16 @@ if service["functions"]: cog.outl(" while (offset + sizeof(xbot::datatypes::DataDescriptor) <= len) {") cog.outl(" const auto* desc = reinterpret_cast(buf + offset);") cog.outl(" offset += sizeof(xbot::datatypes::DataDescriptor);") - cog.outl(" if (offset + desc->payload_size > len) { parse_ok = false; break; }") + cog.outl(" // Bounds check against the remaining bytes: offset + payload_size can wrap on a 32 bit target") + cog.outl(" if (desc->payload_size > len - offset) { parse_ok = false; break; }") cog.outl(" switch (desc->target_id) {") for p in func["parameters"]: cog.outl(f" case {p['id']}:") - if p['is_array']: + if p['is_blob']: + cog.outl(f" param_{p['name']} = buf + offset;") + cog.outl(f" param_{p['name']}Len = desc->payload_size;") + cog.outl(f" params_received++;") + elif p['is_array']: cog.outl(f" if (desc->payload_size % sizeof({p['type']}) == 0 &&") cog.outl(f" desc->payload_size <= sizeof(param_{p['name']})) {{") cog.outl(f" memcpy(param_{p['name']}, buf + offset, desc->payload_size);") @@ -394,11 +420,37 @@ if service["functions"]: # Build call argument list — call_id first, then params, then return buffer for arrays call_args = ["call_id"] for p in func["parameters"]: - if p['is_array']: + if p['is_blob'] or p['is_array']: call_args.append(f"param_{p['name']}, param_{p['name']}Len") else: call_args.append(f"param_{p['name']}") - if func['return_is_array']: + # Return storage: small arrays stay on the stack, everything larger and blob + # returns come from the shared buffer pool. The lease is released when this case + # ends, so the handler has to call SendRpcResponse() before it returns. + pooled_bytes = pooledReturnBytes(service, func) + if func['return_is_blob']: + cog.outl(f" constexpr uint16_t kMaxRpcReturn = {MAX_RPC_RETURN_BYTES};") + cog.outl(" static_assert(kMaxRpcReturn <= xbot::config::buffer_block_size,") + cog.outl(' "blob return does not fit into a buffer pool block");') + cog.outl(" auto rpc_ret_lease = xbot::service::buffer::acquire(kMaxRpcReturn);") + cog.outl(" if (!rpc_ret_lease) {") + cog.outl(" SendRpcResponse(call_id, xbot::datatypes::RpcStatus::BUSY, nullptr, 0);") + cog.outl(" return;") + cog.outl(" }") + cog.outl(" uint16_t rpc_ret_len = kMaxRpcReturn;") + call_args.append("rpc_ret_lease.data(), &rpc_ret_len") + elif pooled_bytes is not None: + cog.outl(f" static_assert({pooled_bytes} <= xbot::config::buffer_block_size,") + cog.outl(' "return value does not fit into a buffer pool block");') + cog.outl(f" auto rpc_ret_lease = xbot::service::buffer::acquire({pooled_bytes});") + cog.outl(" if (!rpc_ret_lease) {") + cog.outl(" SendRpcResponse(call_id, xbot::datatypes::RpcStatus::BUSY, nullptr, 0);") + cog.outl(" return;") + cog.outl(" }") + cog.outl(f" auto* rpc_ret_buf = reinterpret_cast<{func['return_base_type']}*>(rpc_ret_lease.data());") + cog.outl(f" uint16_t rpc_ret_len = {func['return_max_length']};") + call_args.append("rpc_ret_buf, &rpc_ret_len") + elif func['return_is_array']: cog.outl(f" {func['return_base_type']} rpc_ret_buf[{func['return_max_length']}] = {{}};") cog.outl(f" uint16_t rpc_ret_len = {func['return_max_length']};") call_args.append("rpc_ret_buf, &rpc_ret_len") @@ -406,6 +458,8 @@ if service["functions"]: # Set max response size before dispatching to user code if func['return_type'] == 'void': cog.outl(f" rpc_max_response_size_ = 0;") + elif func['return_is_blob']: + cog.outl(f" rpc_max_response_size_ = kMaxRpcReturn;") elif func['return_is_array']: cog.outl(f" rpc_max_response_size_ = {func['return_max_length']} * sizeof({func['return_base_type']});") else: @@ -437,7 +491,8 @@ void ServiceTemplateBase::dispatchRpcCall(uint8_t function_id, uint16_t call_id, while (offset + sizeof(xbot::datatypes::DataDescriptor) <= len) { const auto* desc = reinterpret_cast(buf + offset); offset += sizeof(xbot::datatypes::DataDescriptor); - if (offset + desc->payload_size > len) { parse_ok = false; break; } + // Bounds check against the remaining bytes: offset + payload_size can wrap on a 32 bit target + if (desc->payload_size > len - offset) { parse_ok = false; break; } switch (desc->target_id) { case 0: if (desc->payload_size == sizeof(float)) { @@ -478,7 +533,8 @@ void ServiceTemplateBase::dispatchRpcCall(uint8_t function_id, uint16_t call_id, while (offset + sizeof(xbot::datatypes::DataDescriptor) <= len) { const auto* desc = reinterpret_cast(buf + offset); offset += sizeof(xbot::datatypes::DataDescriptor); - if (offset + desc->payload_size > len) { parse_ok = false; break; } + // Bounds check against the remaining bytes: offset + payload_size can wrap on a 32 bit target + if (desc->payload_size > len - offset) { parse_ok = false; break; } switch (desc->target_id) { case 0: if (desc->payload_size % sizeof(char) == 0 && @@ -510,7 +566,8 @@ void ServiceTemplateBase::dispatchRpcCall(uint8_t function_id, uint16_t call_id, while (offset + sizeof(xbot::datatypes::DataDescriptor) <= len) { const auto* desc = reinterpret_cast(buf + offset); offset += sizeof(xbot::datatypes::DataDescriptor); - if (offset + desc->payload_size > len) { parse_ok = false; break; } + // Bounds check against the remaining bytes: offset + payload_size can wrap on a 32 bit target + if (desc->payload_size > len - offset) { parse_ok = false; break; } switch (desc->target_id) { case 0: if (desc->payload_size % sizeof(char) == 0 && @@ -538,6 +595,58 @@ void ServiceTemplateBase::dispatchRpcCall(uint8_t function_id, uint16_t call_id, RPCMixedParamsWithReturn(call_id, param_Name, param_NameLen, param_Value); return; } + case 4: { + const uint8_t* param_Data = nullptr; + uint32_t param_DataLen = 0; + size_t offset = 0; + bool parse_ok = true; + uint8_t params_received = 0; + while (offset + sizeof(xbot::datatypes::DataDescriptor) <= len) { + const auto* desc = reinterpret_cast(buf + offset); + offset += sizeof(xbot::datatypes::DataDescriptor); + // Bounds check against the remaining bytes: offset + payload_size can wrap on a 32 bit target + if (desc->payload_size > len - offset) { parse_ok = false; break; } + switch (desc->target_id) { + case 0: + param_Data = buf + offset; + param_DataLen = desc->payload_size; + params_received++; + break; + default: break; + } + offset += desc->payload_size; + } + if (!parse_ok || params_received != 1) { + SendRpcResponse(call_id, xbot::datatypes::RpcStatus::ERROR, nullptr, 0); + return; + } + constexpr uint16_t kMaxRpcReturn = 1448; + static_assert(kMaxRpcReturn <= xbot::config::buffer_block_size, + "blob return does not fit into a buffer pool block"); + auto rpc_ret_lease = xbot::service::buffer::acquire(kMaxRpcReturn); + if (!rpc_ret_lease) { + SendRpcResponse(call_id, xbot::datatypes::RpcStatus::BUSY, nullptr, 0); + return; + } + uint16_t rpc_ret_len = kMaxRpcReturn; + rpc_max_response_size_ = kMaxRpcReturn; + RPCBlobParamsWithBlobReturn(call_id, param_Data, param_DataLen, rpc_ret_lease.data(), &rpc_ret_len); + return; + } + case 5: { + static_assert(256 <= xbot::config::buffer_block_size, + "return value does not fit into a buffer pool block"); + auto rpc_ret_lease = xbot::service::buffer::acquire(256); + if (!rpc_ret_lease) { + SendRpcResponse(call_id, xbot::datatypes::RpcStatus::BUSY, nullptr, 0); + return; + } + auto* rpc_ret_buf = reinterpret_cast(rpc_ret_lease.data()); + uint16_t rpc_ret_len = 256; + rpc_max_response_size_ = 256 * sizeof(uint8_t); + RPCLargeArrayReturn(call_id, rpc_ret_buf, &rpc_ret_len); + return; + } default: SendRpcResponse(call_id, xbot::datatypes::RpcStatus::ERROR, nullptr, 0); } diff --git a/codegen/templates/ServiceTemplate.hpp b/codegen/templates/ServiceTemplate.hpp index bf910ba..41fe5ad 100644 --- a/codegen/templates/ServiceTemplate.hpp +++ b/codegen/templates/ServiceTemplate.hpp @@ -95,7 +95,7 @@ class ServiceTemplateBase : public xbot::service::Service { 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x54, 0x65, 0x6D, 0x70, 0x6C, 0x61, 0x74, 0x65, 0x67, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6F, 0x6E, 0x01, 0x66, - 0x69, 0x6E, 0x70, 0x75, 0x74, 0x73, 0x82, 0xA3, + 0x69, 0x6E, 0x70, 0x75, 0x74, 0x73, 0x83, 0xA3, 0x62, 0x69, 0x64, 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x6D, 0x45, 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65, 0x49, 0x6E, 0x70, 0x75, 0x74, 0x31, 0x64, @@ -105,113 +105,154 @@ class ServiceTemplateBase : public xbot::service::Service { 0x6D, 0x45, 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65, 0x49, 0x6E, 0x70, 0x75, 0x74, 0x32, 0x64, 0x74, 0x79, 0x70, 0x65, 0x68, 0x75, 0x69, 0x6E, 0x74, - 0x33, 0x32, 0x5F, 0x74, 0x67, 0x6F, 0x75, 0x74, - 0x70, 0x75, 0x74, 0x73, 0x82, 0xA3, 0x62, 0x69, - 0x64, 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x6E, - 0x45, 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65, 0x4F, - 0x75, 0x74, 0x70, 0x75, 0x74, 0x31, 0x64, 0x74, - 0x79, 0x70, 0x65, 0x69, 0x63, 0x68, 0x61, 0x72, - 0x5B, 0x31, 0x30, 0x30, 0x5D, 0xA3, 0x62, 0x69, - 0x64, 0x01, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x6E, - 0x45, 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65, 0x4F, - 0x75, 0x74, 0x70, 0x75, 0x74, 0x32, 0x64, 0x74, - 0x79, 0x70, 0x65, 0x68, 0x75, 0x69, 0x6E, 0x74, - 0x33, 0x32, 0x5F, 0x74, 0x69, 0x72, 0x65, 0x67, - 0x69, 0x73, 0x74, 0x65, 0x72, 0x73, 0x84, 0xA3, - 0x62, 0x69, 0x64, 0x00, 0x64, 0x6E, 0x61, 0x6D, - 0x65, 0x69, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, - 0x65, 0x72, 0x31, 0x64, 0x74, 0x79, 0x70, 0x65, - 0x68, 0x63, 0x68, 0x61, 0x72, 0x5B, 0x34, 0x32, - 0x5D, 0xA3, 0x62, 0x69, 0x64, 0x01, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x69, 0x52, 0x65, 0x67, 0x69, - 0x73, 0x74, 0x65, 0x72, 0x32, 0x64, 0x74, 0x79, - 0x70, 0x65, 0x68, 0x75, 0x69, 0x6E, 0x74, 0x33, - 0x32, 0x5F, 0x74, 0xA3, 0x62, 0x69, 0x64, 0x02, - 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x69, 0x52, 0x65, - 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x33, 0x64, + 0x33, 0x32, 0x5F, 0x74, 0xA3, 0x62, 0x69, 0x64, + 0x02, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x70, 0x45, + 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65, 0x49, 0x6E, + 0x70, 0x75, 0x74, 0x42, 0x6C, 0x6F, 0x62, 0x64, 0x74, 0x79, 0x70, 0x65, 0x64, 0x62, 0x6C, 0x6F, - 0x62, 0xA4, 0x62, 0x69, 0x64, 0x03, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x72, 0x52, 0x65, 0x67, 0x69, - 0x73, 0x74, 0x65, 0x72, 0x34, 0x20, 0x4F, 0x70, - 0x74, 0x69, 0x6F, 0x6E, 0x61, 0x6C, 0x64, 0x74, - 0x79, 0x70, 0x65, 0x68, 0x75, 0x69, 0x6E, 0x74, - 0x33, 0x32, 0x5F, 0x74, 0x68, 0x6F, 0x70, 0x74, - 0x69, 0x6F, 0x6E, 0x61, 0x6C, 0xF5, 0x69, 0x66, - 0x75, 0x6E, 0x63, 0x74, 0x69, 0x6F, 0x6E, 0x73, - 0x84, 0xA4, 0x62, 0x69, 0x64, 0x00, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x70, 0x4E, 0x6F, 0x50, 0x61, - 0x72, 0x61, 0x6D, 0x73, 0x4E, 0x6F, 0x52, 0x65, + 0x62, 0x67, 0x6F, 0x75, 0x74, 0x70, 0x75, 0x74, + 0x73, 0x83, 0xA3, 0x62, 0x69, 0x64, 0x00, 0x64, + 0x6E, 0x61, 0x6D, 0x65, 0x6E, 0x45, 0x78, 0x61, + 0x6D, 0x70, 0x6C, 0x65, 0x4F, 0x75, 0x74, 0x70, + 0x75, 0x74, 0x31, 0x64, 0x74, 0x79, 0x70, 0x65, + 0x69, 0x63, 0x68, 0x61, 0x72, 0x5B, 0x31, 0x30, + 0x30, 0x5D, 0xA3, 0x62, 0x69, 0x64, 0x01, 0x64, + 0x6E, 0x61, 0x6D, 0x65, 0x6E, 0x45, 0x78, 0x61, + 0x6D, 0x70, 0x6C, 0x65, 0x4F, 0x75, 0x74, 0x70, + 0x75, 0x74, 0x32, 0x64, 0x74, 0x79, 0x70, 0x65, + 0x68, 0x75, 0x69, 0x6E, 0x74, 0x33, 0x32, 0x5F, + 0x74, 0xA3, 0x62, 0x69, 0x64, 0x02, 0x64, 0x6E, + 0x61, 0x6D, 0x65, 0x71, 0x45, 0x78, 0x61, 0x6D, + 0x70, 0x6C, 0x65, 0x4F, 0x75, 0x74, 0x70, 0x75, + 0x74, 0x42, 0x6C, 0x6F, 0x62, 0x64, 0x74, 0x79, + 0x70, 0x65, 0x64, 0x62, 0x6C, 0x6F, 0x62, 0x69, + 0x72, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, + 0x73, 0x84, 0xA3, 0x62, 0x69, 0x64, 0x00, 0x64, + 0x6E, 0x61, 0x6D, 0x65, 0x69, 0x52, 0x65, 0x67, + 0x69, 0x73, 0x74, 0x65, 0x72, 0x31, 0x64, 0x74, + 0x79, 0x70, 0x65, 0x68, 0x63, 0x68, 0x61, 0x72, + 0x5B, 0x34, 0x32, 0x5D, 0xA3, 0x62, 0x69, 0x64, + 0x01, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x69, 0x52, + 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x32, + 0x64, 0x74, 0x79, 0x70, 0x65, 0x68, 0x75, 0x69, + 0x6E, 0x74, 0x33, 0x32, 0x5F, 0x74, 0xA3, 0x62, + 0x69, 0x64, 0x02, 0x64, 0x6E, 0x61, 0x6D, 0x65, + 0x69, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, + 0x72, 0x33, 0x64, 0x74, 0x79, 0x70, 0x65, 0x64, + 0x62, 0x6C, 0x6F, 0x62, 0xA4, 0x62, 0x69, 0x64, + 0x03, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x72, 0x52, + 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x34, + 0x20, 0x4F, 0x70, 0x74, 0x69, 0x6F, 0x6E, 0x61, + 0x6C, 0x64, 0x74, 0x79, 0x70, 0x65, 0x68, 0x75, + 0x69, 0x6E, 0x74, 0x33, 0x32, 0x5F, 0x74, 0x68, + 0x6F, 0x70, 0x74, 0x69, 0x6F, 0x6E, 0x61, 0x6C, + 0xF5, 0x69, 0x66, 0x75, 0x6E, 0x63, 0x74, 0x69, + 0x6F, 0x6E, 0x73, 0x86, 0xA4, 0x62, 0x69, 0x64, + 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x70, 0x4E, + 0x6F, 0x50, 0x61, 0x72, 0x61, 0x6D, 0x73, 0x4E, + 0x6F, 0x52, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x6A, + 0x70, 0x61, 0x72, 0x61, 0x6D, 0x65, 0x74, 0x65, + 0x72, 0x73, 0x80, 0x6B, 0x72, 0x65, 0x74, 0x75, + 0x72, 0x6E, 0x5F, 0x74, 0x79, 0x70, 0x65, 0x64, + 0x76, 0x6F, 0x69, 0x64, 0xA4, 0x62, 0x69, 0x64, + 0x01, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x76, 0x53, + 0x63, 0x61, 0x6C, 0x61, 0x72, 0x50, 0x61, 0x72, + 0x61, 0x6D, 0x73, 0x57, 0x69, 0x74, 0x68, 0x52, + 0x65, 0x74, 0x75, 0x72, 0x6E, 0x6A, 0x70, 0x61, + 0x72, 0x61, 0x6D, 0x65, 0x74, 0x65, 0x72, 0x73, + 0x83, 0xA3, 0x62, 0x69, 0x64, 0x00, 0x64, 0x6E, + 0x61, 0x6D, 0x65, 0x65, 0x53, 0x70, 0x65, 0x65, + 0x64, 0x64, 0x74, 0x79, 0x70, 0x65, 0x65, 0x66, + 0x6C, 0x6F, 0x61, 0x74, 0xA3, 0x62, 0x69, 0x64, + 0x01, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x65, 0x43, + 0x6F, 0x75, 0x6E, 0x74, 0x64, 0x74, 0x79, 0x70, + 0x65, 0x68, 0x75, 0x69, 0x6E, 0x74, 0x33, 0x32, + 0x5F, 0x74, 0xA3, 0x62, 0x69, 0x64, 0x02, 0x64, + 0x6E, 0x61, 0x6D, 0x65, 0x66, 0x45, 0x6E, 0x61, + 0x62, 0x6C, 0x65, 0x64, 0x74, 0x79, 0x70, 0x65, + 0x67, 0x75, 0x69, 0x6E, 0x74, 0x38, 0x5F, 0x74, + 0x6B, 0x72, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x5F, + 0x74, 0x79, 0x70, 0x65, 0x67, 0x69, 0x6E, 0x74, + 0x33, 0x32, 0x5F, 0x74, 0xA4, 0x62, 0x69, 0x64, + 0x02, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x72, 0x41, + 0x72, 0x72, 0x61, 0x79, 0x50, 0x61, 0x72, 0x61, + 0x6D, 0x4E, 0x6F, 0x52, 0x65, 0x74, 0x75, 0x72, + 0x6E, 0x6A, 0x70, 0x61, 0x72, 0x61, 0x6D, 0x65, + 0x74, 0x65, 0x72, 0x73, 0x81, 0xA3, 0x62, 0x69, + 0x64, 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x65, + 0x4C, 0x61, 0x62, 0x65, 0x6C, 0x64, 0x74, 0x79, + 0x70, 0x65, 0x68, 0x63, 0x68, 0x61, 0x72, 0x5B, + 0x36, 0x34, 0x5D, 0x6B, 0x72, 0x65, 0x74, 0x75, + 0x72, 0x6E, 0x5F, 0x74, 0x79, 0x70, 0x65, 0x64, + 0x76, 0x6F, 0x69, 0x64, 0xA4, 0x62, 0x69, 0x64, + 0x03, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x75, 0x4D, + 0x69, 0x78, 0x65, 0x64, 0x50, 0x61, 0x72, 0x61, + 0x6D, 0x73, 0x57, 0x69, 0x74, 0x68, 0x52, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x6A, 0x70, 0x61, 0x72, - 0x61, 0x6D, 0x65, 0x74, 0x65, 0x72, 0x73, 0x80, + 0x61, 0x6D, 0x65, 0x74, 0x65, 0x72, 0x73, 0x82, + 0xA3, 0x62, 0x69, 0x64, 0x00, 0x64, 0x6E, 0x61, + 0x6D, 0x65, 0x64, 0x4E, 0x61, 0x6D, 0x65, 0x64, + 0x74, 0x79, 0x70, 0x65, 0x68, 0x63, 0x68, 0x61, + 0x72, 0x5B, 0x33, 0x32, 0x5D, 0xA3, 0x62, 0x69, + 0x64, 0x01, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x65, + 0x56, 0x61, 0x6C, 0x75, 0x65, 0x64, 0x74, 0x79, + 0x70, 0x65, 0x65, 0x66, 0x6C, 0x6F, 0x61, 0x74, 0x6B, 0x72, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x5F, - 0x74, 0x79, 0x70, 0x65, 0x64, 0x76, 0x6F, 0x69, - 0x64, 0xA4, 0x62, 0x69, 0x64, 0x01, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x76, 0x53, 0x63, 0x61, 0x6C, - 0x61, 0x72, 0x50, 0x61, 0x72, 0x61, 0x6D, 0x73, - 0x57, 0x69, 0x74, 0x68, 0x52, 0x65, 0x74, 0x75, - 0x72, 0x6E, 0x6A, 0x70, 0x61, 0x72, 0x61, 0x6D, - 0x65, 0x74, 0x65, 0x72, 0x73, 0x83, 0xA3, 0x62, - 0x69, 0x64, 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, - 0x65, 0x53, 0x70, 0x65, 0x65, 0x64, 0x64, 0x74, - 0x79, 0x70, 0x65, 0x65, 0x66, 0x6C, 0x6F, 0x61, - 0x74, 0xA3, 0x62, 0x69, 0x64, 0x01, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x65, 0x43, 0x6F, 0x75, 0x6E, - 0x74, 0x64, 0x74, 0x79, 0x70, 0x65, 0x68, 0x75, - 0x69, 0x6E, 0x74, 0x33, 0x32, 0x5F, 0x74, 0xA3, - 0x62, 0x69, 0x64, 0x02, 0x64, 0x6E, 0x61, 0x6D, - 0x65, 0x66, 0x45, 0x6E, 0x61, 0x62, 0x6C, 0x65, - 0x64, 0x74, 0x79, 0x70, 0x65, 0x67, 0x75, 0x69, - 0x6E, 0x74, 0x38, 0x5F, 0x74, 0x6B, 0x72, 0x65, + 0x74, 0x79, 0x70, 0x65, 0x67, 0x75, 0x69, 0x6E, + 0x74, 0x38, 0x5F, 0x74, 0xA4, 0x62, 0x69, 0x64, + 0x04, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x78, 0x18, + 0x42, 0x6C, 0x6F, 0x62, 0x50, 0x61, 0x72, 0x61, + 0x6D, 0x73, 0x57, 0x69, 0x74, 0x68, 0x42, 0x6C, + 0x6F, 0x62, 0x52, 0x65, 0x74, 0x75, 0x72, 0x6E, + 0x6A, 0x70, 0x61, 0x72, 0x61, 0x6D, 0x65, 0x74, + 0x65, 0x72, 0x73, 0x81, 0xA3, 0x62, 0x69, 0x64, + 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x64, 0x44, + 0x61, 0x74, 0x61, 0x64, 0x74, 0x79, 0x70, 0x65, + 0x64, 0x62, 0x6C, 0x6F, 0x62, 0x6B, 0x72, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x5F, 0x74, 0x79, 0x70, - 0x65, 0x67, 0x69, 0x6E, 0x74, 0x33, 0x32, 0x5F, - 0x74, 0xA4, 0x62, 0x69, 0x64, 0x02, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x72, 0x41, 0x72, 0x72, 0x61, - 0x79, 0x50, 0x61, 0x72, 0x61, 0x6D, 0x4E, 0x6F, - 0x52, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x6A, 0x70, - 0x61, 0x72, 0x61, 0x6D, 0x65, 0x74, 0x65, 0x72, - 0x73, 0x81, 0xA3, 0x62, 0x69, 0x64, 0x00, 0x64, - 0x6E, 0x61, 0x6D, 0x65, 0x65, 0x4C, 0x61, 0x62, - 0x65, 0x6C, 0x64, 0x74, 0x79, 0x70, 0x65, 0x68, - 0x63, 0x68, 0x61, 0x72, 0x5B, 0x36, 0x34, 0x5D, - 0x6B, 0x72, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x5F, - 0x74, 0x79, 0x70, 0x65, 0x64, 0x76, 0x6F, 0x69, - 0x64, 0xA4, 0x62, 0x69, 0x64, 0x03, 0x64, 0x6E, - 0x61, 0x6D, 0x65, 0x75, 0x4D, 0x69, 0x78, 0x65, - 0x64, 0x50, 0x61, 0x72, 0x61, 0x6D, 0x73, 0x57, - 0x69, 0x74, 0x68, 0x52, 0x65, 0x74, 0x75, 0x72, + 0x65, 0x64, 0x62, 0x6C, 0x6F, 0x62, 0xA4, 0x62, + 0x69, 0x64, 0x05, 0x64, 0x6E, 0x61, 0x6D, 0x65, + 0x70, 0x4C, 0x61, 0x72, 0x67, 0x65, 0x41, 0x72, + 0x72, 0x61, 0x79, 0x52, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x6A, 0x70, 0x61, 0x72, 0x61, 0x6D, 0x65, - 0x74, 0x65, 0x72, 0x73, 0x82, 0xA3, 0x62, 0x69, - 0x64, 0x00, 0x64, 0x6E, 0x61, 0x6D, 0x65, 0x64, - 0x4E, 0x61, 0x6D, 0x65, 0x64, 0x74, 0x79, 0x70, - 0x65, 0x68, 0x63, 0x68, 0x61, 0x72, 0x5B, 0x33, - 0x32, 0x5D, 0xA3, 0x62, 0x69, 0x64, 0x01, 0x64, - 0x6E, 0x61, 0x6D, 0x65, 0x65, 0x56, 0x61, 0x6C, - 0x75, 0x65, 0x64, 0x74, 0x79, 0x70, 0x65, 0x65, - 0x66, 0x6C, 0x6F, 0x61, 0x74, 0x6B, 0x72, 0x65, + 0x74, 0x65, 0x72, 0x73, 0x80, 0x6B, 0x72, 0x65, 0x74, 0x75, 0x72, 0x6E, 0x5F, 0x74, 0x79, 0x70, - 0x65, 0x67, 0x75, 0x69, 0x6E, 0x74, 0x38, 0x5F, - 0x74, 0x65, 0x65, 0x6E, 0x75, 0x6D, 0x73, 0x82, - 0xA3, 0x62, 0x69, 0x64, 0x70, 0x45, 0x78, 0x61, - 0x6D, 0x70, 0x6C, 0x65, 0x45, 0x6E, 0x75, 0x6D, - 0x43, 0x6C, 0x61, 0x73, 0x73, 0x69, 0x62, 0x61, - 0x73, 0x65, 0x5F, 0x74, 0x79, 0x70, 0x65, 0x67, - 0x75, 0x69, 0x6E, 0x74, 0x38, 0x5F, 0x74, 0x66, - 0x76, 0x61, 0x6C, 0x75, 0x65, 0x73, 0xA3, 0x66, - 0x56, 0x41, 0x4C, 0x55, 0x45, 0x31, 0x00, 0x66, - 0x56, 0x41, 0x4C, 0x55, 0x45, 0x32, 0x01, 0x66, - 0x56, 0x41, 0x4C, 0x55, 0x45, 0x33, 0x02, 0xA4, - 0x62, 0x69, 0x64, 0x72, 0x45, 0x78, 0x61, 0x6D, - 0x70, 0x6C, 0x65, 0x42, 0x69, 0x74, 0x6D, 0x61, - 0x73, 0x6B, 0x45, 0x6E, 0x75, 0x6D, 0x69, 0x62, - 0x61, 0x73, 0x65, 0x5F, 0x74, 0x79, 0x70, 0x65, - 0x67, 0x75, 0x69, 0x6E, 0x74, 0x38, 0x5F, 0x74, - 0x67, 0x62, 0x69, 0x74, 0x6D, 0x61, 0x73, 0x6B, - 0xF5, 0x66, 0x76, 0x61, 0x6C, 0x75, 0x65, 0x73, - 0xA3, 0x66, 0x56, 0x41, 0x4C, 0x55, 0x45, 0x31, - 0x00, 0x66, 0x56, 0x41, 0x4C, 0x55, 0x45, 0x32, - 0x01, 0x66, 0x56, 0x41, 0x4C, 0x55, 0x45, 0x33, - 0x02 + 0x65, 0x6C, 0x75, 0x69, 0x6E, 0x74, 0x38, 0x5F, + 0x74, 0x5B, 0x32, 0x35, 0x36, 0x5D, 0x65, 0x65, + 0x6E, 0x75, 0x6D, 0x73, 0x82, 0xA3, 0x62, 0x69, + 0x64, 0x70, 0x45, 0x78, 0x61, 0x6D, 0x70, 0x6C, + 0x65, 0x45, 0x6E, 0x75, 0x6D, 0x43, 0x6C, 0x61, + 0x73, 0x73, 0x69, 0x62, 0x61, 0x73, 0x65, 0x5F, + 0x74, 0x79, 0x70, 0x65, 0x67, 0x75, 0x69, 0x6E, + 0x74, 0x38, 0x5F, 0x74, 0x66, 0x76, 0x61, 0x6C, + 0x75, 0x65, 0x73, 0xA3, 0x66, 0x56, 0x41, 0x4C, + 0x55, 0x45, 0x31, 0x00, 0x66, 0x56, 0x41, 0x4C, + 0x55, 0x45, 0x32, 0x01, 0x66, 0x56, 0x41, 0x4C, + 0x55, 0x45, 0x33, 0x02, 0xA4, 0x62, 0x69, 0x64, + 0x72, 0x45, 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65, + 0x42, 0x69, 0x74, 0x6D, 0x61, 0x73, 0x6B, 0x45, + 0x6E, 0x75, 0x6D, 0x69, 0x62, 0x61, 0x73, 0x65, + 0x5F, 0x74, 0x79, 0x70, 0x65, 0x67, 0x75, 0x69, + 0x6E, 0x74, 0x38, 0x5F, 0x74, 0x67, 0x62, 0x69, + 0x74, 0x6D, 0x61, 0x73, 0x6B, 0xF5, 0x66, 0x76, + 0x61, 0x6C, 0x75, 0x65, 0x73, 0xA3, 0x66, 0x56, + 0x41, 0x4C, 0x55, 0x45, 0x31, 0x00, 0x66, 0x56, + 0x41, 0x4C, 0x55, 0x45, 0x32, 0x01, 0x66, 0x56, + 0x41, 0x4C, 0x55, 0x45, 0x33, 0x02 + }; + + // Blocks that did not fit into the advertisement; served on request + // (DESCRIPTION_REQUEST -> DESCRIPTION_PART), so a host without generated + // classes can still fetch the complete schema. + static constexpr unsigned char SERVICE_OMITTED_CBOR[] = { + 0xA0 }; + static constexpr size_t SERVICE_OMITTED_CBOR_SIZE = 0; + + const unsigned char* GetOmittedDescription(size_t& size) override { + size = SERVICE_OMITTED_CBOR_SIZE; + return SERVICE_OMITTED_CBOR; + } //[[[end]]] private: @@ -239,7 +280,11 @@ class ServiceTemplateBase : public xbot::service::Service { /*[[[cog # Generate callback functions for each input. for input in service["inputs"]: - if input['is_array']: + if input['is_blob']: + cog.outl(f"virtual void {input['callback_name']}(const uint8_t* new_value, uint32_t length) {{") + cog.outl(" (void)new_value;") + cog.outl(" (void)length;") + elif input['is_array']: cog.outl(f"virtual void {input['callback_name']}(const {input['type']}* new_value, uint32_t length) {{") cog.outl(" (void)new_value;") cog.outl(" (void)length;") @@ -257,17 +302,25 @@ class ServiceTemplateBase : public xbot::service::Service { (void)new_value; } + virtual void OnExampleInputBlobChanged(const uint8_t* new_value, uint32_t length) { + (void)new_value; + (void)length; + } + //[[[end]]] /*[[[cog # Generate send functions for each output. for output in service["outputs"]: - if output['is_array']: + if output['is_blob']: + cog.outl(f"bool {output['send_method_name']}(const uint8_t* data, uint32_t length);") + elif output['is_array']: cog.outl(f"bool {output['send_method_name']}(const {output['type']}* data, uint32_t length);") else: cog.outl(f"bool {output['send_method_name']}(const {output['type']} &data);") ]]]*/ bool SendExampleOutput1(const char* data, uint32_t length); bool SendExampleOutput2(const uint32_t &data); + bool SendExampleOutputBlob(const uint8_t* data, uint32_t length); //[[[end]]] /*[[[cog # Generate register struct @@ -315,17 +368,24 @@ class ServiceTemplateBase : public xbot::service::Service { /*[[[cog # Generate RPC virtual for each function. # call_id is passed so the implementation can respond asynchronously via SendRpcResponse(). + # Blob parameters are passed as (pointer, length) into the received packet: no copy and + # no size limit, valid until the handler returns. # For array return types: data buffer (max capacity) and response_length (init to max) are # appended. The implementation fills data, sets *response_length to actual element count, # then calls SendRpcResponse(call_id, SUCCESS, data, *response_length * sizeof(T)). + # A blob return works the same way, but *response_length is a byte count. for func in service["functions"]: params = ["uint16_t call_id"] for p in func["parameters"]: - if p['is_array']: + if p['is_blob']: + params.append(f"const uint8_t* {p['name']}, uint32_t {p['name']}Len") + elif p['is_array']: params.append(f"const {p['type']}* {p['name']}, uint32_t {p['name']}Len") else: params.append(f"{p['type']} {p['name']}") - if func['return_is_array']: + if func['return_is_blob']: + params.append("uint8_t* data, uint16_t* response_length") + elif func['return_is_array']: params.append(f"{func['return_base_type']}* data, uint16_t* response_length") params_str = ", ".join(params) cog.outl(f"virtual void RPC{func['name']}({params_str}) = 0;") @@ -334,6 +394,8 @@ class ServiceTemplateBase : public xbot::service::Service { virtual void RPCScalarParamsWithReturn(uint16_t call_id, float Speed, uint32_t Count, uint8_t Enable) = 0; virtual void RPCArrayParamNoReturn(uint16_t call_id, const char* Label, uint32_t LabelLen) = 0; virtual void RPCMixedParamsWithReturn(uint16_t call_id, const char* Name, uint32_t NameLen, float Value) = 0; + virtual void RPCBlobParamsWithBlobReturn(uint16_t call_id, const uint8_t* Data, uint32_t DataLen, uint8_t* data, uint16_t* response_length) = 0; + virtual void RPCLargeArrayReturn(uint16_t call_id, uint8_t* data, uint16_t* response_length) = 0; //[[[end]]] }; @@ -354,6 +416,11 @@ cog.outl(f"\u002f*\n{service['service_json']}\n*\u002f") "id": 1, "name": "ExampleInput2", "type": "uint32_t" + }, + { + "id": 2, + "name": "ExampleInputBlob", + "type": "blob" } ], "outputs": [ @@ -366,6 +433,11 @@ cog.outl(f"\u002f*\n{service['service_json']}\n*\u002f") "id": 1, "name": "ExampleOutput2", "type": "uint32_t" + }, + { + "id": 2, + "name": "ExampleOutputBlob", + "type": "blob" } ], "registers": [ @@ -448,6 +520,24 @@ cog.outl(f"\u002f*\n{service['service_json']}\n*\u002f") } ], "return_type": "uint8_t" + }, + { + "id": 4, + "name": "BlobParamsWithBlobReturn", + "parameters": [ + { + "id": 0, + "name": "Data", + "type": "blob" + } + ], + "return_type": "blob" + }, + { + "id": 5, + "name": "LargeArrayReturn", + "parameters": [], + "return_type": "uint8_t[256]" } ], "enums": [ diff --git a/codegen/templates/service.json b/codegen/templates/service.json index 8014f8b..4b426bb 100644 --- a/codegen/templates/service.json +++ b/codegen/templates/service.json @@ -11,6 +11,11 @@ "id": 1, "name": "ExampleInput2", "type": "uint32_t" + }, + { + "id": 2, + "name": "ExampleInputBlob", + "type": "blob" } ], "outputs": [ @@ -23,6 +28,11 @@ "id": 1, "name": "ExampleOutput2", "type": "uint32_t" + }, + { + "id": 2, + "name": "ExampleOutputBlob", + "type": "blob" } ], "registers": [ @@ -81,6 +91,20 @@ { "id": 1, "name": "Value", "type": "float" } ], "return_type": "uint8_t" + }, + { + "id": 4, + "name": "BlobParamsWithBlobReturn", + "parameters": [ + { "id": 0, "name": "Data", "type": "blob" } + ], + "return_type": "blob" + }, + { + "id": 5, + "name": "LargeArrayReturn", + "parameters": [], + "return_type": "uint8_t[256]" } ], "enums": [ diff --git a/codegen/xbot_codegen/xbot_codegen.py b/codegen/xbot_codegen/xbot_codegen.py index 179e143..60cc261 100644 --- a/codegen/xbot_codegen/xbot_codegen.py +++ b/codegen/xbot_codegen/xbot_codegen.py @@ -71,6 +71,58 @@ def array_type_attrs(max_length): return {"is_array": False} +# sizeof of the scalar types in the generated C++ code (and on the wire). +TYPE_SIZES = { + "char": 1, + "uint8_t": 1, + "int8_t": 1, + "uint16_t": 2, + "int16_t": 2, + "uint32_t": 4, + "int32_t": 4, + "float": 4, + "uint64_t": 8, + "int64_t": 8, + "double": 8, +} + +# Largest RPC body that fits into one packet: max_packet_size(1472) - sizeof(XbotHeader)(24). +# A return value can never be larger, which is what the shared buffer pool is sized for. +MAX_RPC_RETURN_BYTES = 1472 - 24 + +# Array returns up to this size stay on the service thread stack. Larger ones and blob +# returns are taken from the shared buffer pool (see xbot-service/Buffer.hpp), so a +# service never has to reserve that stack space permanently. +RPC_STACK_RETURN_MAX = 128 + + +def typeSize(service, type_name): + """sizeof of a scalar or enum type in the generated C++ code.""" + for enum in service["enums"]: + if enum["id"] == type_name: + return TYPE_SIZES[enum["base_type"]] + if type_name in TYPE_SIZES: + return TYPE_SIZES[type_name] + raise Exception(f"Cannot determine the size of type '{type_name}'") + + +def pooledReturnBytes(service, func): + """Bytes a return value needs from the buffer pool, or None if it fits on the stack.""" + if func["return_type"] == "void": + return None + if func["return_is_blob"]: + return MAX_RPC_RETURN_BYTES + if func["return_is_array"]: + bytes_needed = func["return_max_length"] * typeSize(service, func["return_base_type"]) + return bytes_needed if bytes_needed > RPC_STACK_RETURN_MAX else None + return None + + +def needsBufferPool(service): + """True if any generated return value comes from the buffer pool.""" + return any(pooledReturnBytes(service, func) is not None for func in service["functions"]) + + def common_attrs(json, valid_types, callback_name, send_method_name): id = int(json['id']) name = toCamelCase(json['name']) @@ -81,6 +133,7 @@ def common_attrs(json, valid_types, callback_name, send_method_name): "id": id, "name": name, "type": type, + "is_blob": type == "blob", "callback_name": callback_name.format(name), "send_method_name": send_method_name.format(name), **array_type_attrs(max_length) @@ -183,7 +236,7 @@ def loadService(path: str) -> dict: # Transform the input definitions seen_input_names = set() for json_input in json_service["inputs"]: - input = common_attrs(json_input, valid_types, "On{}Changed", "Send{}") + input = common_attrs(json_input, valid_types + ["blob"], "On{}Changed", "Send{}") if input["name"] in seen_input_names: raise Exception( f"Duplicate normalized input name '{input['name']}' " @@ -194,7 +247,7 @@ def loadService(path: str) -> dict: # Transform the output definitions seen_output_names = set() for json_output in json_service["outputs"]: - output = common_attrs(json_output, valid_types, "On{}Changed", "Send{}") + output = common_attrs(json_output, valid_types + ["blob"], "On{}Changed", "Send{}") if output["name"] in seen_output_names: raise Exception( f"Duplicate normalized output name '{output['name']}' " @@ -218,17 +271,30 @@ def loadService(path: str) -> dict: raise Exception(f"Duplicate normalized function name '{func_name}' after name normalization") seen_func_names.add(func_name) return_type_str = json_func.get("return_type", "void") + return_is_blob = False if return_type_str == "void": return_base_type, return_max_length = None, None else: return_base_type, return_max_length = parse_type(return_type_str) - if return_base_type not in valid_types: + if return_base_type not in valid_types + ["blob"]: raise Exception(f"Illegal return type for function '{func_name}': {return_type_str}") + if return_base_type == "blob": + # Variable length and therefore bounded by one packet only + # (MAX_RPC_RETURN_BYTES); the generated dispatcher takes the buffer + # from the shared pool. + return_is_blob = True if return_max_length is not None: if return_max_length <= 0 or return_max_length > MAX_ARRAY_LENGTH: raise Exception( f"Illegal array return length {return_max_length} for function '{func_name}' " f"return type '{return_type_str}': must be 1..{MAX_ARRAY_LENGTH}") + # A return value travels in one packet, so a schema that declares more can + # never be answered. Fail here instead of at runtime. + return_bytes = return_max_length * typeSize(service, return_base_type) + if return_bytes > MAX_RPC_RETURN_BYTES: + raise Exception( + f"Function '{func_name}' return type '{return_type_str}' is {return_bytes} bytes, " + f"more than one packet can carry ({MAX_RPC_RETURN_BYTES})") return_is_array = return_max_length is not None check_unique_ids(json_func.get("parameters", [])) @@ -243,12 +309,13 @@ def loadService(path: str) -> dict: f"in function '{func_name}'") seen_param_names.add(param_name) param_type, param_max_length = parse_type(json_param["type"]) - if param_type not in valid_types: + if param_type not in valid_types + ["blob"]: raise Exception(f"Illegal parameter type in function '{func_name}': {param_type}") params.append({ "id": int(json_param["id"]), "name": param_name, "type": param_type, + "is_blob": param_type == "blob", **array_type_attrs(param_max_length), }) @@ -258,6 +325,7 @@ def loadService(path: str) -> dict: "return_type": return_type_str, "return_base_type": return_base_type, "return_is_array": return_is_array, + "return_is_blob": return_is_blob, "return_max_length": return_max_length, "parameters": params, }) diff --git a/include/xbot/config.hpp b/include/xbot/config.hpp index ef8d5ac..b987ce6 100644 --- a/include/xbot/config.hpp +++ b/include/xbot/config.hpp @@ -69,6 +69,38 @@ static constexpr uint32_t heartbeat_jitter = 100'000; static_assert(max_log_length > 100); +/** + * Settings for the shared RPC buffer pool (see xbot-service/Buffer.hpp). + * + * XBOT_BUFFER_DEFAULT_POOL: 1 lets the framework allocate the pool storage + * itself (hosts, tests, examples). A platform that brings its own storage - + * e.g. a section in a RAM region that is otherwise unused - sets this to 0 and + * calls buffer::initialize() instead, so the static array is not linked in. + */ +#ifndef XBOT_BUFFER_BLOCK_SIZE +#define XBOT_BUFFER_BLOCK_SIZE 1536 +#endif +#ifndef XBOT_BUFFER_BLOCK_COUNT +#define XBOT_BUFFER_BLOCK_COUNT 2 +#endif +#ifndef XBOT_BUFFER_DEFAULT_POOL +#define XBOT_BUFFER_DEFAULT_POOL 1 +#endif + +/** + * Size of one block in the shared buffer pool. A return value travels in one + * packet, so a block has to hold max_packet_size - sizeof(XbotHeader) = 1448 + * bytes; 1536 leaves headroom (asserted in Buffer.cpp). + */ +static constexpr uint16_t buffer_block_size = XBOT_BUFFER_BLOCK_SIZE; +/** + * Number of blocks that can be out at the same time. A block is only held while + * an RPC fills and sends its return value, so this is about concurrency, not + * about the number of services: two blocks leave room for a slow transfer next + * to a quick one. + */ +static constexpr uint8_t buffer_block_count = XBOT_BUFFER_BLOCK_COUNT; + namespace service { static constexpr uint32_t io_thread_stack_size = 5000; } diff --git a/libxbot-service/CMakeLists.txt b/libxbot-service/CMakeLists.txt index ed204c4..f3d3db3 100644 --- a/libxbot-service/CMakeLists.txt +++ b/libxbot-service/CMakeLists.txt @@ -16,6 +16,7 @@ add_library(xbot-service STATIC src/Scheduler.cpp src/ServiceIo.cpp src/DataSource.cpp + src/Buffer.cpp ) target_compile_options(xbot-service PRIVATE -Wall -Wextra -Werror -Wno-volatile) diff --git a/libxbot-service/include/xbot-service/Buffer.hpp b/libxbot-service/include/xbot-service/Buffer.hpp new file mode 100644 index 0000000..c7f4dce --- /dev/null +++ b/libxbot-service/include/xbot-service/Buffer.hpp @@ -0,0 +1,108 @@ +// +// Shared buffer pool for RPC payloads. +// + +#ifndef XBOT_BUFFER_HPP +#define XBOT_BUFFER_HPP + +#include +#include + +namespace xbot::service::buffer { + +/** @brief Highest number of blocks the pool bitmap can hold. */ +static constexpr uint8_t max_block_count = 32; + +/** @brief Alignment of every block and of the storage passed to initialize(). */ +static constexpr size_t block_alignment = 8; + +/** @brief Index that marks "no block". */ +static constexpr uint8_t invalid_index = 0xFF; + +/** + * @brief A block borrowed from the shared buffer pool. + * + * A lease owns one block until it is released or destroyed. It is move-only, so + * a block can never be released twice and the destructor returns it to the pool. + * + * Do not keep a lease beyond the call that borrowed it. The pool is shared by + * all services: a lease that is held across a flash erase or across RPCs starves + * the next caller, which then has to answer BUSY. + */ +class Lease { + public: + Lease() = default; + Lease(const Lease &) = delete; + Lease &operator=(const Lease &) = delete; + Lease(Lease &&other) noexcept; + Lease &operator=(Lease &&other) noexcept; + ~Lease(); + + /** @brief Start of the block, nullptr for an empty lease. */ + uint8_t *data() const { + return data_; + } + + /** @brief Capacity of the block in bytes (block size, not the requested size). */ + size_t size() const { + return size_; + } + + /** @brief True if this lease owns a block. */ + explicit operator bool() const { + return data_ != nullptr; + } + + /** @brief Return the block to the pool. Idempotent. */ + void release(); + + private: + friend Lease acquire(size_t bytes); + Lease(uint8_t *data, size_t size, uint8_t index) : data_(data), size_(size), index_(index) { + } + + uint8_t *data_ = nullptr; + size_t size_ = 0; + uint8_t index_ = invalid_index; +}; + +/** + * @brief Register the storage of the pool. + * + * Call once, before the first acquire() and before the services start; it is not + * thread safe. A platform that brings its own storage (e.g. a section in RAM that + * is otherwise unused) calls this with that storage and sets + * XBOT_BUFFER_DEFAULT_POOL to 0, which removes the framework default array. + * + * @param storage start of the storage, block_alignment aligned + * @param block_size bytes per block, must hold the largest RPC return value + * @param block_count number of blocks, 1..max_block_count + * @return false if an argument is unusable or a lease is still out + */ +bool initialize(void *storage, size_t block_size, size_t block_count); + +/** + * @brief Borrow a block that holds at least @p bytes. + * + * Deliberately non-blocking: an RPC handler must never wait for a slow transfer + * on another service, it answers BUSY instead. A timeout would not help either, + * because a block is only held while a return value is filled and copied into a + * packet, i.e. for microseconds, not for the duration of an upload. + * + * @return An empty lease if the request is larger than a block or every block is + * out. + */ +Lease acquire(size_t bytes); + +/** @brief Block size in bytes. */ +size_t blockSize(); + +/** @brief Number of blocks in the pool. */ +size_t blockCount(); + +/** @brief Number of blocks that are currently free. */ +size_t available(); + +} // namespace xbot::service::buffer + +#endif // XBOT_BUFFER_HPP diff --git a/libxbot-service/src/Buffer.cpp b/libxbot-service/src/Buffer.cpp new file mode 100644 index 0000000..2f0eb19 --- /dev/null +++ b/libxbot-service/src/Buffer.cpp @@ -0,0 +1,173 @@ +// +// Shared buffer pool for RPC payloads. +// + +#include + +#include +#include +#include +#include +#include + +namespace { + +using xbot::service::buffer::invalid_index; +using xbot::service::buffer::Lease; + +// A return value has to fit into one packet, so a block never has to be larger +// than that. Keeping the invariant here catches a pool that was sized for +// something smaller. +static_assert(xbot::config::buffer_block_size >= xbot::config::max_packet_size - sizeof(xbot::datatypes::XbotHeader), + "buffer_block_size must hold the largest possible RPC return value"); +static_assert(xbot::config::buffer_block_count >= 1U, "the buffer pool needs at least one block"); +static_assert(xbot::config::buffer_block_count <= xbot::service::buffer::max_block_count, + "buffer_block_count exceeds the pool bitmap"); +static_assert((xbot::config::buffer_block_size % xbot::service::buffer::block_alignment) == 0, + "buffer_block_size must be a multiple of block_alignment"); + +#if XBOT_BUFFER_DEFAULT_POOL +// Storage for platforms that do not bring their own (hosts, tests, examples). A +// firmware that places the pool in its own RAM section sets +// XBOT_BUFFER_DEFAULT_POOL to 0 and calls buffer::initialize() instead. +alignas(xbot::service::buffer::block_alignment) uint8_t + default_storage[xbot::config::buffer_block_count * xbot::config::buffer_block_size]{}; +#endif + +// Constant initialized on purpose: acquire() may run before a dynamic +// initializer had a chance to run (services start right after the IO thread). +#if XBOT_BUFFER_DEFAULT_POOL +uint8_t *storage_ = default_storage; +size_t block_size_ = xbot::config::buffer_block_size; +size_t block_count_ = xbot::config::buffer_block_count; +#else +uint8_t *storage_ = nullptr; +size_t block_size_ = 0; +size_t block_count_ = 0; +#endif + +// One bit per block, set while the block is out. The compare-and-swap loop keeps +// the pool lock free: no mutex means no priority inversion, and - more +// importantly - no port has to provide a try-lock, so every existing port keeps +// working unchanged. +std::atomic used_mask_{0}; + +// Guards the "not initialized" log line only, so a plain bool is enough: the +// worst case is a duplicated line. +bool warned_uninitialized_ = false; + +uint32_t blockBit(uint8_t index) { + return uint32_t{1} << index; +} + +} // namespace + +Lease::Lease(Lease &&other) noexcept : data_(other.data_), size_(other.size_), index_(other.index_) { + other.data_ = nullptr; + other.size_ = 0; + other.index_ = invalid_index; +} + +Lease &Lease::operator=(Lease &&other) noexcept { + if (this != &other) { + release(); + data_ = other.data_; + size_ = other.size_; + index_ = other.index_; + other.data_ = nullptr; + other.size_ = 0; + other.index_ = invalid_index; + } + return *this; +} + +Lease::~Lease() { + release(); +} + +void Lease::release() { + if (data_ == nullptr) { + return; + } + // Release: everything the handler wrote into the block is visible before the + // next owner reads it. + used_mask_.fetch_and(~blockBit(index_), std::memory_order_release); + data_ = nullptr; + size_ = 0; + index_ = invalid_index; +} + +bool xbot::service::buffer::initialize(void *storage, size_t block_size, size_t block_count) { + if (storage == nullptr || block_size == 0U || block_count == 0U) { + return false; + } + if (block_count > max_block_count) { + return false; + } + if ((reinterpret_cast(storage) % block_alignment) != 0U) { + return false; + } + if ((block_size % block_alignment) != 0U) { + return false; // a block stride that is not aligned would misalign every later lease + } + if (used_mask_.load(std::memory_order_relaxed) != 0U) { + return false; // a lease is still out, keep the current storage alive + } + storage_ = static_cast(storage); + block_size_ = block_size; + block_count_ = block_count; + ULOG_INFO("Buffer pool: %u blocks of %u bytes", static_cast(block_count), + static_cast(block_size)); + return true; +} + +xbot::service::buffer::Lease xbot::service::buffer::acquire(size_t bytes) { + if (storage_ == nullptr) { + if (!warned_uninitialized_) { + warned_uninitialized_ = true; + ULOG_WARNING("Buffer pool is not initialized, call xbot::service::buffer::initialize()"); + } + return {}; + } + if (bytes > block_size_) { + return {}; + } + + uint32_t mask = used_mask_.load(std::memory_order_relaxed); + for (;;) { + uint8_t index = invalid_index; + for (uint8_t i = 0; i < block_count_; ++i) { + if ((mask & blockBit(i)) == 0U) { + index = i; + break; + } + } + if (index == invalid_index) { + return {}; // every block is out, the caller answers BUSY + } + if (used_mask_.compare_exchange_weak(mask, mask | blockBit(index), std::memory_order_acquire, + std::memory_order_relaxed)) { + return {storage_ + static_cast(index) * block_size_, block_size_, index}; + } + // compare_exchange_weak refreshed mask with the current value, try again + } +} + +size_t xbot::service::buffer::blockSize() { + return block_size_; +} + +size_t xbot::service::buffer::blockCount() { + return block_count_; +} + +size_t xbot::service::buffer::available() { + size_t used = 0; + const uint32_t mask = used_mask_.load(std::memory_order_relaxed); + for (uint8_t i = 0; i < block_count_; ++i) { + if ((mask & blockBit(i)) != 0U) { + ++used; + } + } + return block_count_ - used; +} diff --git a/libxbot-service/test/BufferTests/BufferTests.cpp b/libxbot-service/test/BufferTests/BufferTests.cpp new file mode 100644 index 0000000..2ac19d9 --- /dev/null +++ b/libxbot-service/test/BufferTests/BufferTests.cpp @@ -0,0 +1,118 @@ +// +// Tests for the shared RPC buffer pool. +// + +#include +#include + +#include "CppUTest/TestHarness.h" + +using namespace xbot::service; + +TEST_GROUP(BufferTests) { + // The tests bring their own storage, so they do not depend on (or disturb) + // the framework default pool. + alignas(buffer::block_alignment) uint8_t storage_[2 * 128]{}; + + static constexpr size_t kBlocks = 2; + static constexpr size_t kBlockSize = 128; + + void setup() override { + CHECK_TRUE(buffer::initialize(storage_, kBlockSize, kBlocks)); + CHECK_EQUAL(kBlocks, buffer::available()); + } +}; + +TEST(BufferTests, AcquireReturnsBlockCapacity) { + buffer::Lease lease = buffer::acquire(100); + CHECK_TRUE(lease); + CHECK_EQUAL(kBlockSize, lease.size()); + CHECK_TRUE(lease.data() == storage_); + CHECK_EQUAL(kBlocks - 1, buffer::available()); +} + +TEST(BufferTests, AcquireUsesDistinctBlocks) { + buffer::Lease first = buffer::acquire(10); + buffer::Lease second = buffer::acquire(10); + CHECK_TRUE(first); + CHECK_TRUE(second); + CHECK_TRUE(first.data() != second.data()); + CHECK_EQUAL(static_cast(0), buffer::available()); +} + +TEST(BufferTests, ExhaustedPoolReturnsEmptyLease) { + buffer::Lease first = buffer::acquire(10); + buffer::Lease second = buffer::acquire(10); + buffer::Lease third = buffer::acquire(10); + CHECK_TRUE(first); + CHECK_TRUE(second); + CHECK_FALSE(third); // the caller answers BUSY +} + +TEST(BufferTests, ReleaseReturnsBlock) { + { + buffer::Lease lease = buffer::acquire(10); + CHECK_TRUE(lease); + lease.release(); + CHECK_FALSE(lease); + CHECK_EQUAL(kBlocks, buffer::available()); + } + buffer::Lease again = buffer::acquire(10); + CHECK_TRUE(again); +} + +TEST(BufferTests, DestructorReleasesBlock) { + { + buffer::Lease lease = buffer::acquire(10); + CHECK_TRUE(lease); + } + CHECK_EQUAL(kBlocks, buffer::available()); +} + +TEST(BufferTests, OversizedRequestReturnsEmptyLease) { + CHECK_FALSE(buffer::acquire(kBlockSize + 1)); + CHECK_EQUAL(kBlocks, buffer::available()); +} + +TEST(BufferTests, MoveTransfersOwnership) { + buffer::Lease source = buffer::acquire(10); + CHECK_TRUE(source); + + buffer::Lease target = std::move(source); + CHECK_FALSE(source); + CHECK_TRUE(target); + CHECK_EQUAL(kBlocks - 1, buffer::available()); + + target = std::move(source); // move assign from an empty lease releases target + CHECK_FALSE(target); + CHECK_EQUAL(kBlocks, buffer::available()); +} + +TEST(BufferTests, ReleaseIsIdempotent) { + buffer::Lease lease = buffer::acquire(10); + lease.release(); + lease.release(); + CHECK_EQUAL(kBlocks, buffer::available()); +} + +TEST(BufferTests, InitializeRejectsBadArguments) { + CHECK_FALSE(buffer::initialize(nullptr, kBlockSize, kBlocks)); + CHECK_FALSE(buffer::initialize(storage_, 0, kBlocks)); + CHECK_FALSE(buffer::initialize(storage_, kBlockSize, 0)); + CHECK_FALSE(buffer::initialize(storage_, kBlockSize, buffer::max_block_count + 1)); + CHECK_FALSE(buffer::initialize(storage_ + 1, kBlockSize, kBlocks)); // unaligned storage + CHECK_FALSE(buffer::initialize(storage_, kBlockSize + 1, kBlocks)); // unaligned block stride +} + +TEST(BufferTests, InitializeRefusesWhileALeaseIsOut) { + buffer::Lease lease = buffer::acquire(10); + CHECK_TRUE(lease); + CHECK_FALSE(buffer::initialize(storage_, kBlockSize, kBlocks)); + lease.release(); + CHECK_TRUE(buffer::initialize(storage_, kBlockSize, kBlocks)); +} + +TEST(BufferTests, BlockSizeAndCountReportTheInitializedPool) { + CHECK_EQUAL(kBlockSize, buffer::blockSize()); + CHECK_EQUAL(kBlocks, buffer::blockCount()); +} diff --git a/libxbot-service/test/CMakeLists.txt b/libxbot-service/test/CMakeLists.txt index 3b45982..cec883f 100644 --- a/libxbot-service/test/CMakeLists.txt +++ b/libxbot-service/test/CMakeLists.txt @@ -2,6 +2,8 @@ add_executable(AllTests all_tests.cpp QueueTests/QueueTests.cpp + BufferTests/BufferTests.cpp + ${PROJECT_SOURCE_DIR}/src/Buffer.cpp ${PROJECT_SOURCE_DIR}/src/portable/linux/queue.cpp ) @@ -9,6 +11,7 @@ target_include_directories(AllTests PRIVATE . ${PROJECT_SOURCE_DIR}/include + ${PROJECT_SOURCE_DIR}/../include ) target_compile_options(AllTests @@ -18,6 +21,7 @@ target_compile_options(AllTests target_link_libraries(AllTests PRIVATE CppUTest::CppUTestExt + ulog ) if(CPPUTEST_TEST_DISCOVERY OR NOT DEFINED CPPUTEST_TEST_DISCOVERY) diff --git a/libxbot-service/test/all_tests.cpp b/libxbot-service/test/all_tests.cpp index 225b300..64ceec4 100644 --- a/libxbot-service/test/all_tests.cpp +++ b/libxbot-service/test/all_tests.cpp @@ -4,6 +4,7 @@ #include "CppUTest/CommandLineTestRunner.h" IMPORT_TEST_GROUP(QueueTests); +IMPORT_TEST_GROUP(BufferTests); int main(int argc, char** argv) { return RUN_ALL_TESTS(argc, argv); diff --git a/xbot_service_interface_go/codegen/emit.go b/xbot_service_interface_go/codegen/emit.go index 319fe7a..6b78138 100644 --- a/xbot_service_interface_go/codegen/emit.go +++ b/xbot_service_interface_go/codegen/emit.go @@ -148,6 +148,9 @@ func (g *generator) decodeReturn(t, fn string) (string, []string, error) { switch { case t == "" || t == "void": return "", nil, nil + case base == "blob": + // Variable length return: hand the payload back as is. + return "[]byte", []string{"\treturn append([]byte(nil), payload...), nil"}, nil case base == "char" && isArray: return "string", []string{"\treturn charValue(payload), nil"}, nil case isArray && base == "uint8_t": diff --git a/xbot_service_interface_go/codegen/generate_test.go b/xbot_service_interface_go/codegen/generate_test.go index c9d98e3..9cea5e3 100644 --- a/xbot_service_interface_go/codegen/generate_test.go +++ b/xbot_service_interface_go/codegen/generate_test.go @@ -57,7 +57,7 @@ func TestLoadRejects(t *testing.T) { cases := map[string]string{ "missing type": `{"version": 1}`, "unknown type": `{"type": "X", "version": 1, "inputs": [{"id": 0, "name": "A", "type": "uint7_t"}]}`, - "blob as input": `{"type": "X", "version": 1, "inputs": [{"id": 0, "name": "A", "type": "blob"}]}`, + "blob array": `{"type": "X", "version": 1, "inputs": [{"id": 0, "name": "A", "type": "blob[4]"}]}`, "duplicate ids": `{"type": "X", "version": 1, "inputs": [{"id": 1, "name": "A", "type": "uint8_t"}, {"id": 1, "name": "B", "type": "uint8_t"}]}`, "name collision": `{"type": "X", "version": 1, "inputs": [{"id": 0, "name": "My Input", "type": "uint8_t"}, {"id": 1, "name": "MyInput", "type": "uint8_t"}]}`, "unknown enum base": `{"type": "X", "version": 1, "enums": [{"id": "E", "base_type": "uint7_t", "values": {"A": 0}}]}`, @@ -69,6 +69,11 @@ func TestLoadRejects(t *testing.T) { "functions normalize equal": `{"type": "X", "version": 1, "functions": [ {"id": 0, "name": "SoundId", "return_type": "void"}, {"id": 1, "name": "SoundID", "return_type": "void"}]}`, + "blob budget helpers collide": `{"type": "X", "version": 1, "functions": [ + {"id": 0, "name": "Write", "return_type": "void", + "parameters": [{"id": 0, "name": "PathData", "type": "blob"}]}, + {"id": 1, "name": "WritePath", "return_type": "void", + "parameters": [{"id": 0, "name": "Data", "type": "blob"}]}]}`, "enum values normalize equal": `{"type": "X", "version": 1, "enums": [{"id": "E", "base_type": "uint8_t", "values": {"IDLE": 0, "Idle": 1}}]}`, "enum types normalize equal": `{"type": "X", "version": 1, "enums": [ @@ -88,6 +93,98 @@ func TestLoadRejects(t *testing.T) { } } +// TestBlobParameters pins that blob parameters and returns work: a blob parameter is +// sent as is (no padding to a fixed size) and a blob return is handed back as the raw +// payload. +func TestBlobParameters(t *testing.T) { + path := t.TempDir() + "/service.json" + schema := `{"type": "X", "version": 1, "functions": [ + {"id": 0, "name": "Echo", "return_type": "blob", + "parameters": [{"id": 0, "name": "Data", "type": "blob"}]}, + {"id": 1, "name": "Write", "return_type": "int32_t", + "parameters": [ + {"id": 0, "name": "Path", "type": "blob"}, + {"id": 1, "name": "Offset", "type": "uint32_t"}, + {"id": 2, "name": "Data", "type": "blob"}, + {"id": 3, "name": "Hash", "type": "uint32_t"}]}]}` + if err := os.WriteFile(path, []byte(schema), 0o644); err != nil { + t.Fatal(err) + } + s, err := Load(path) + if err != nil { + t.Fatalf("Load: %v", err) + } + src, err := Generate(s, Options{}) + if err != nil { + t.Fatalf("Generate: %v", err) + } + for _, want := range []string{ + "func (c *Client) Echo(ctx context.Context, Data []byte) ([]byte, error)", + "func (c *Client) Write(ctx context.Context, Path []byte, Offset uint32, Data []byte, Hash uint32) (int32, error)", + "xbot.Bytes(0, Data)", + "xbot.Bytes(0, Path)", + "return append([]byte(nil), payload...)", + // One helper per blob parameter, with the framing of this call subtracted. + // The budget saturates: a call whose other parameters use up the packet has + // no room left for this one, and a negative int is not a budget. + "func MaxEchoData() int {", + "func MaxWriteData(Path []byte) int {", + "return max(0, xbot.MaxPacketSize-xbot.HeaderSize-4*xbot.DescriptorSize-8-len(Path))", + } { + if !bytes.Contains(src, []byte(want)) { + t.Errorf("generated code does not contain %q", want) + } + } + // A blob must not be padded to a declared size - that was the whole point. + for _, notWant := range []string{"byteParam(Data", "charParam(Path"} { + if bytes.Contains(src, []byte(notWant)) { + t.Errorf("generated code still pads a blob parameter: found %q", notWant) + } + } +} + +// TestBlobHelperNames pins the identifier of the frame budget helper: it is built +// from the normalized RPC name, not from the raw one, or an RPC called "Write Data" +// would emit `func MaxWrite DataData()` and Generate would fail while formatting. +func TestBlobHelperNames(t *testing.T) { + path := t.TempDir() + "/service.json" + schema := `{"type": "X", "version": 1, "functions": [ + {"id": 0, "name": "Write Data", "return_type": "int32_t", + "parameters": [{"id": 0, "name": "Data", "type": "blob"}]}]}` + if err := os.WriteFile(path, []byte(schema), 0o644); err != nil { + t.Fatal(err) + } + s, err := Load(path) + if err != nil { + t.Fatalf("Load: %v", err) + } + src, err := Generate(s, Options{}) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if !bytes.Contains(src, []byte("func MaxWriteDataData() int {")) { + t.Errorf("generated code does not contain the normalized helper name: %s", src) + } +} + +// path already carries a pointer and a length, only the schema rejected it. +func TestBlobInputOutput(t *testing.T) { + path := t.TempDir() + "/service.json" + schema := `{"type": "X", "version": 1, + "inputs": [{"id": 0, "name": "Data", "type": "blob"}], + "outputs": [{"id": 0, "name": "Extra", "type": "blob"}]}` + if err := os.WriteFile(path, []byte(schema), 0o644); err != nil { + t.Fatal(err) + } + s, err := Load(path) + if err != nil { + t.Fatalf("blob input/output rejected: %v", err) + } + if _, err := Generate(s, Options{}); err != nil { + t.Fatalf("Generate: %v", err) + } +} + // TestNumericArrays pins that numeric arrays (T[n]) are supported: they are // padded to n elements on send and decoded from n elements on receive. func TestNumericArrays(t *testing.T) { diff --git a/xbot_service_interface_go/codegen/render.go b/xbot_service_interface_go/codegen/render.go index fa6217c..8f94e64 100644 --- a/xbot_service_interface_go/codegen/render.go +++ b/xbot_service_interface_go/codegen/render.go @@ -97,6 +97,7 @@ func (g *generator) run() error { return err } } + g.emitFrameBudgets() for _, r := range s.Registers { if err := g.emitRegister(r); err != nil { return err @@ -177,6 +178,82 @@ type charCheck struct { length int } +// paramBytes is how many bytes a fixed size parameter occupies on the wire, i.e. what it +// takes away from the budget of a blob parameter of the same call. +func (g *generator) paramBytes(p Parameter) (int, error) { + base, length, isArray := splitType(p.Type) + wire := base + if eb := g.enumBase(base); eb != "" { + wire = eb + } + size, ok := typeSizes[wire] + if base == "char" { + size, ok = 1, true // not in typeSizes: char is the natural character type + } + if !ok { + return 0, fmt.Errorf("codegen: unsupported type %q in parameter %s", p.Type, p.Name) + } + if isArray { + return size * length, nil + } + return size, nil +} + +// emitFrameBudgets writes one helper per blob parameter: the largest payload it can +// carry, given the lengths of the other variable parameters. A caller that chunks a value +// needs exactly this number, and the framing that limits it (one descriptor per parameter +// plus the fixed payloads) is knowledge that belongs here, not in every client. +func (g *generator) emitFrameBudgets() { + for _, fn := range g.schema.Functions { + var blobs []Parameter + fixed := 0 + for _, p := range fn.Parameters { + if base, _, _ := splitType(p.Type); base == "blob" { + blobs = append(blobs, p) + continue + } + if size, err := g.paramBytes(p); err == nil { + fixed += size + } // an unsupported type is reported by emitFunction already + } + if len(blobs) == 0 { + continue + } + + // The helper name is built from the normalized RPC name: an RPC may be called + // "Write Data" with a parameter "Data", and MaxWrite DataData is not an identifier. + fnName := goName(fn.Name) + for _, target := range blobs { + targetName := goName(target.Name) + var args []string + var variable strings.Builder + for _, other := range blobs { + if other.Name == target.Name { + continue + } + otherName := goName(other.Name) + args = append(args, otherName+" []byte") + variable.WriteString(" - len(" + otherName + ")") + } + + budget := fmt.Sprintf("%d*xbot.DescriptorSize", len(fn.Parameters)) + if fixed > 0 { + budget += fmt.Sprintf(" - %d", fixed) + } + + g.f("// Max%s%s is the largest %s payload one %s call can carry.", fnName, targetName, targetName, fnName) + g.line("//") + g.line("// One packet of RPC body minus the framing of the call: a descriptor per") + g.line("// parameter and the payloads of the fixed parameters. A call whose parameters") + g.line("// already use up the packet reports 0, never a negative budget.") + g.f("func Max%s%s(%s) int {", fnName, targetName, strings.Join(args, ", ")) + g.f("\treturn max(0, xbot.MaxPacketSize - xbot.HeaderSize - %s%s)", budget, variable.String()) + g.line("}") + g.line("") + } + } +} + // emitFunction writes one RPC method. func (g *generator) emitFunction(fn Function) error { name := goName(fn.Name) diff --git a/xbot_service_interface_go/codegen/schema.go b/xbot_service_interface_go/codegen/schema.go index c70f2ad..20e1e36 100644 --- a/xbot_service_interface_go/codegen/schema.go +++ b/xbot_service_interface_go/codegen/schema.go @@ -132,16 +132,14 @@ func (s *Schema) validate() error { for _, set := range []struct { what string fields []Field - blob bool }{ - {"input", s.Inputs, false}, {"output", s.Outputs, false}, {"register", s.Registers, true}, + {"input", s.Inputs}, {"output", s.Outputs}, {"register", s.Registers}, } { for _, f := range set.fields { base, _, isArray := splitType(f.Type) if base == "blob" { - if !set.blob { - return fmt.Errorf("%s %s: \"blob\" is only allowed for registers", set.what, f.Name) - } + // Variable length, same rule as the C++ generator: allowed for + // inputs, outputs and registers, but never as an array. if isArray { return fmt.Errorf("%s %s: \"blob\" cannot be an array", set.what, f.Name) } diff --git a/xbot_service_interface_go/codegen/validate_names.go b/xbot_service_interface_go/codegen/validate_names.go index c0545fe..d212b35 100644 --- a/xbot_service_interface_go/codegen/validate_names.go +++ b/xbot_service_interface_go/codegen/validate_names.go @@ -52,6 +52,18 @@ func (s *Schema) validateGeneratedNames() error { if err := reserve(method, name, "function "+fn.Name); err != nil { return err } + // A blob parameter's frame budget helper is a package level function too, and + // distinct RPC names can produce the same one: Write(Path blob) and + // WritePath(Data blob) both give MaxWritePathData. + for _, p := range fn.Parameters { + if base, _, _ := splitType(p.Type); base != "blob" { + continue + } + what := "the frame budget helper of parameter " + p.Name + " of function " + fn.Name + if err := reserve(pkg, "Max"+name+goName(p.Name), what); err != nil { + return err + } + } } for _, r := range s.Registers { if err := reserve(method, "Set"+goName(r.Name), "register "+r.Name); err != nil { diff --git a/xbot_service_interface_go/xbot/conn.go b/xbot_service_interface_go/xbot/conn.go index c172035..1f67304 100644 --- a/xbot_service_interface_go/xbot/conn.go +++ b/xbot_service_interface_go/xbot/conn.go @@ -324,6 +324,13 @@ func (c *Conn) startCall(functionID uint8, params []Param) (uint16, error) { // send writes one header+payload packet to the service. func (c *Conn) send(msgType MessageType, arg1 uint8, arg2 uint16, payload []byte) error { + // One packet has to fit into a single, unfragmented datagram (the low level + // runs with IP reassembly disabled, an oversized packet is dropped silently). + // Splitting a large value is the caller's job, e.g. into offset addressed chunks. + if HeaderSize+len(payload) > MaxPacketSize { + return fmt.Errorf("xbot: payload of %d bytes exceeds the packet limit of %d bytes", + len(payload), MaxPacketSize-HeaderSize) + } hdr := Header{ ProtocolVersion: ProtocolVersion, MessageType: msgType, diff --git a/xbot_service_interface_go/xbot/conn_test.go b/xbot_service_interface_go/xbot/conn_test.go index 9ff2264..f35e044 100644 --- a/xbot_service_interface_go/xbot/conn_test.go +++ b/xbot_service_interface_go/xbot/conn_test.go @@ -305,3 +305,22 @@ func TestCheckAdvertisement(t *testing.T) { } } } + +// TestSendRejectsOversizedPayload makes sure a payload that does not fit into one +// unfragmented datagram is rejected locally instead of being dropped silently by the +// low level (which runs with IP reassembly disabled). +func TestSendRejectsOversizedPayload(t *testing.T) { + _, addr := startFakeService(t, respondWith(0, nil)) + c, err := Dial(context.Background(), Options{ServiceID: 1, Endpoint: addr}) + if err != nil { + t.Fatalf("Dial: %v", err) + } + t.Cleanup(func() { _ = c.Close() }) + + if err := c.send(MsgData, 0, 0, make([]byte, MaxPacketSize-HeaderSize+1)); err == nil { + t.Fatal("expected an error for a payload above the packet limit") + } + if err := c.send(MsgData, 0, 0, make([]byte, MaxPacketSize-HeaderSize)); err != nil { + t.Fatalf("payload at the packet limit rejected: %v", err) + } +} diff --git a/xbot_service_interface_py/tests/test_rpc.py b/xbot_service_interface_py/tests/test_rpc.py index 97f96bd..ebe1fe5 100644 --- a/xbot_service_interface_py/tests/test_rpc.py +++ b/xbot_service_interface_py/tests/test_rpc.py @@ -61,6 +61,14 @@ ], 'return_type': 'uint8_t', }, + { + 'id': 4, + 'name': 'BlobReturn', + 'parameters': [ + {'id': 0, 'name': 'Data', 'type': 'blob'}, + ], + 'return_type': 'blob', + }, ], } @@ -78,7 +86,7 @@ class TestSchemaFunctions: def test_functions_parsed(self): s = ServiceSchema.from_dict(RPC_DESC) - assert len(s.functions) == 4 + assert len(s.functions) == 5 def test_no_functions_key(self): s = ServiceSchema.from_dict(NO_FUNC_DESC) @@ -267,6 +275,11 @@ def _respond_async(si, call_id, status, payload=b''): ready = threading.Event() def _do(): ready.wait() + # A response that arrives before the call is registered is dropped by + # _on_rpc_response, so wait for it instead of racing the caller. + deadline = time.time() + 1.0 + while time.time() < deadline and not si._rpc_call_active: + time.sleep(0.001) si._on_rpc_response(call_id, status, payload) t = threading.Thread(target=_do, daemon=False) t.start() @@ -308,6 +321,27 @@ def test_call_uint8_return(self): result = si.call_uint8_return(1, timeout_ms=500) assert result == 1 + def test_call_blob_return(self): + si = make_rpc_si() + payload = b'\x00\x01\x02binary' + _respond_async(si, call_id=1, status=0, payload=payload) + result = si.call_blob_return(b'ping', timeout_ms=500) + assert result == payload + + def test_call_blob_param_is_not_padded(self): + si = make_rpc_si() + _respond_async(si, call_id=1, status=0, payload=b'') + si.call_blob_return(b'ping', timeout_ms=500) + + params_bytes = si._io.send_rpc_call.call_args[0][3] + tid, _, size = struct.unpack_from(' dict: diff --git a/xbot_service_interface_py/xbot_service_interface/interface.py b/xbot_service_interface_py/xbot_service_interface/interface.py index f89f10b..925dd15 100644 --- a/xbot_service_interface_py/xbot_service_interface/interface.py +++ b/xbot_service_interface_py/xbot_service_interface/interface.py @@ -7,7 +7,7 @@ from .exceptions import UnknownChannelError, IncompatibleServiceError, RpcError, RpcBusyError, RpcTimeoutError from .schema import ServiceSchema from .serialization import pack_value, unpack_value, sizeof_type, parse_type_string -from .datatypes import pack_descriptor +from .datatypes import pack_descriptor, HEADER_SIZE, MAX_UDP_PAYLOAD log = logging.getLogger(__name__) @@ -374,6 +374,14 @@ def _call_rpc(self, fn: dict, args: tuple, timeout_ms: int): raw = pack_value(param['type_str'], arg, enums) params_bytes += pack_descriptor(param['id'], len(raw)) + raw + # The request has to fit into one datagram. The low level drops a larger one without + # reporting anything, which would surface as a timeout - so fail here instead. + max_params = MAX_UDP_PAYLOAD - HEADER_SIZE + if len(params_bytes) > max_params: + raise ValueError( + f"RPC {fn['name']!r} parameters are {len(params_bytes)} bytes, " + f"more than one packet can carry ({max_params})") + with self._rpc_lock: with self._rpc_condition: counter = (self._rpc_call_counter + 1) & 0xFFFF @@ -410,12 +418,16 @@ def _call_rpc(self, fn: dict, args: tuple, timeout_ms: int): if fn['return_type'] == 'void': return None - _, is_array, _ = parse_type_string(fn['return_type']) - max_bytes = sizeof_type(fn['return_type']) + base, is_array, _ = parse_type_string(fn['return_type']) + if base == 'blob': + # Variable length return, bounded by one packet only. + max_bytes = MAX_UDP_PAYLOAD - HEADER_SIZE + else: + max_bytes = sizeof_type(fn['return_type']) if len(payload) > max_bytes: raise RpcError( 2, f"RPC {fn['name']!r} response too large: got {len(payload)} bytes, max {max_bytes}") - if not is_array and len(payload) < max_bytes: + if base != 'blob' and not is_array and len(payload) < max_bytes: raise RpcError( 2, f"RPC {fn['name']!r} response too small: got {len(payload)} bytes, need {max_bytes}") return unpack_value(fn['return_type'], payload, enums)