You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@brpc.apache.org by "chenBright (via GitHub)" <gi...@apache.org> on 2026/06/14 03:19:18 UTC
[PR] Add IOBuf<->std::iostream adapters for zero-copy I/O (brpc)
chenBright opened a new pull request, #3341:
URL: https://github.com/apache/brpc/pull/3341
### What problem does this PR solve?
Issue Number: resolve
Problem Summary:
`IOBuf` integrates with protobuf via `IOBufAsZeroCopy{Input,Output}Stream`, but
there is no direct bridge to the `std::iostream` hierarchy. As a result, any
library that consumes/produces `std::istream&` / `std::ostream&` — most notably
`nlohmann::json` — cannot read from or write to an `IOBuf` without an
intermediate `std::string`:
```cpp
auto j = nlohmann::json::parse(req->request_attachment().to_string());
```
This is an obstacle to adopting nlohmann::json in bRPC.
### What is changed and the side effects?
Changed:
Introduce four classes in `src/butil/iobuf.{h,cpp}` that bridge IOBuf and
the standard iostream hierarchy:
- `IOBufAsInputStreamBuf` — read-only `std::streambuf` over an IOBuf,
walking the backing blocks via the public `backing_block()` API. Overrides
`underflow`, `xsgetn`(bulk memcpy across block boundaries to avoid the
default per-byte sbumpc loop), and `showmanyc` (saturating sum of
remaining block sizes).
- `IOBufInputStream` — `std::istream` view that wires the streambuf above.
- `IOBufAsOutputStreamBuf` — append-only `std::streambuf` that delegates to
the existing `IOBufAsZeroCopyOutputStream` for block allocation and the
three-branch `BackUp` (including the shared-block degradation that protects
IOBufs sharing blocks from being corrupted by an in-flight write). Overrides
`overflow`, `xsputn` (block-sized memcpy instead of per-byte sputc),
and `sync` (forwards to shrink()).
- `IOBufOutputStream` — `std::ostream` view that wires the streambuf above;
accepts an optional block_size to allocate dedicated blocks instead of
drawing from the per-thread TLS pool.
Usage with nlohmann::json:
```c++
// Parse: zero string copy
butil::IOBufInputStream in(request_body);
auto j = nlohmann::json::parse(in);
// Serialize: bytes flow straight into IOBuf blocks
butil::IOBufOutputStream os(response_body);
os << j;
```
Side effects:
- Performance effects:
- Breaking backward compatibility:
---
### Check List:
- Please make sure your changes are compilable.
- When providing us with a new feature, it is best to add related tests.
- Please follow [Contributor Covenant Code of Conduct](https://github.com/apache/brpc/blob/master/CODE_OF_CONDUCT.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Add IOBuf<->std::iostream adapters for zero-copy I/O (brpc)
Posted by "chenBright (via GitHub)" <gi...@apache.org>.
chenBright merged PR #3341:
URL: https://github.com/apache/brpc/pull/3341
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Add IOBuf<->std::iostream adapters for zero-copy I/O (brpc)
Posted by "wwbmmm (via GitHub)" <gi...@apache.org>.
wwbmmm commented on PR #3341:
URL: https://github.com/apache/brpc/pull/3341#issuecomment-4700670021
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org
Re: [PR] Add IOBuf<->std::iostream adapters for zero-copy I/O (brpc)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3341:
URL: https://github.com/apache/brpc/pull/3341#discussion_r3409024229
##########
src/butil/iobuf.cpp:
##########
@@ -2025,6 +2027,130 @@ void IOBufAsZeroCopyOutputStream::_release_block() {
_cur_block = NULL;
}
+std::streambuf::int_type IOBufAsInputStreamBuf::underflow() {
+ size_t block_num = _buf.backing_block_num();
+ StringPiece blk;
+ while (_block_index < block_num) {
+ blk = _buf.backing_block(_block_index++);
+ if (!blk.empty()) {
+ break;
+ }
+ }
+ if (blk.empty()) {
+ return traits_type::eof();
+ }
+ // const_cast is safe here: setg() takes char* by API contract, but this
+ // streambuf never writes through it (no overflow/sputc path).
+ char* p = const_cast<char*>(blk.data());
+ setg(p, p, p + blk.size());
+ return traits_type::to_int_type(*gptr());
+}
+
+std::streamsize IOBufAsInputStreamBuf::xsgetn(char* s, std::streamsize n) {
+ auto kIntMax = static_cast<std::streamsize>(std::numeric_limits<int>::max());
+ std::streamsize total = 0;
+ while (total < n) {
+ std::streamsize avail = egptr() - gptr();
+ if (avail == 0) {
+ if (underflow() == traits_type::eof()) {
+ break;
+ }
+ avail = egptr() - gptr();
+ }
Review Comment:
`xsgetn()` computes `egptr() - gptr()` even when the get area has never been initialized (both pointers may be null). Subtracting null/unrelated pointers is undefined behavior and can be hit on the first `read()` call before `underflow()` runs.
##########
src/butil/iobuf.cpp:
##########
@@ -2025,6 +2027,130 @@ void IOBufAsZeroCopyOutputStream::_release_block() {
_cur_block = NULL;
}
+std::streambuf::int_type IOBufAsInputStreamBuf::underflow() {
+ size_t block_num = _buf.backing_block_num();
+ StringPiece blk;
+ while (_block_index < block_num) {
+ blk = _buf.backing_block(_block_index++);
+ if (!blk.empty()) {
+ break;
+ }
+ }
+ if (blk.empty()) {
+ return traits_type::eof();
+ }
+ // const_cast is safe here: setg() takes char* by API contract, but this
+ // streambuf never writes through it (no overflow/sputc path).
+ char* p = const_cast<char*>(blk.data());
+ setg(p, p, p + blk.size());
+ return traits_type::to_int_type(*gptr());
+}
+
+std::streamsize IOBufAsInputStreamBuf::xsgetn(char* s, std::streamsize n) {
+ auto kIntMax = static_cast<std::streamsize>(std::numeric_limits<int>::max());
+ std::streamsize total = 0;
+ while (total < n) {
+ std::streamsize avail = egptr() - gptr();
+ if (avail == 0) {
+ if (underflow() == traits_type::eof()) {
+ break;
+ }
+ avail = egptr() - gptr();
+ }
+ // Cap step at INT_MAX so gbump(int) cannot overflow when a user-data
+ // block exceeds 2GB.
+ std::streamsize step =
+ std::min(std::min(avail, n - total), kIntMax);
+ iobuf::cp(s + total, gptr(), static_cast<size_t>(step));
+ gbump(static_cast<int>(step));
+ total += step;
+ }
+ return total;
+}
+
+std::streamsize IOBufAsInputStreamBuf::showmanyc() {
+ std::streamsize kMax = std::numeric_limits<std::streamsize>::max();
+ std::streamsize n = egptr() - gptr();
+ size_t block_num = _buf.backing_block_num();
+ for (size_t i = _block_index; i < block_num; ++i) {
+ const std::streamsize sz =
+ static_cast<std::streamsize>(_buf.backing_block(i).size());
+ // Saturate instead of overflowing on pathologically large IOBufs.
+ if (n > kMax - sz) {
+ return kMax;
+ }
+ n += sz;
+ }
+ return n;
+}
+
+IOBufAsOutputStreamBuf::~IOBufAsOutputStreamBuf() { shrink(); }
+
+void IOBufAsOutputStreamBuf::shrink() {
+ if (pbase() != NULL) {
+ std::streamsize unused = epptr() - pptr();
+ // _zc.BackUp takes int. A single put area never exceeds one block
+ // (Next() returns int size), so this fits in int by construction;
+ // the cap is purely defensive.
+ int kIntMax = std::numeric_limits<int>::max();
+ _zc.BackUp(unused > kIntMax ? kIntMax : static_cast<int>(unused));
+ setp(NULL, NULL);
+ }
+}
+
+std::streambuf::int_type IOBufAsOutputStreamBuf::overflow(int_type ch) {
+ if (traits_type::eq_int_type(ch, traits_type::eof())) {
+ return traits_type::not_eof(ch);
+ }
+ if (!refresh_put_area()) {
+ return traits_type::eof();
+ }
+ return sputc(traits_type::to_char_type(ch));
+}
+
+std::streamsize IOBufAsOutputStreamBuf::xsputn(
+ const char* s, std::streamsize n) {
+ auto kIntMax = static_cast<std::streamsize>(std::numeric_limits<int>::max());
+ std::streamsize total = 0;
+ while (total < n) {
+ std::streamsize avail = epptr() - pptr();
+ if (avail == 0) {
+ if (!refresh_put_area()) {
+ break;
+ }
+ avail = epptr() - pptr();
+ if (avail == 0) {
+ break;
+ }
+ }
Review Comment:
`xsputn()` computes `epptr() - pptr()` even when the put area has never been initialized (both pointers may be null). This pointer subtraction is undefined behavior and can occur on the first write before `refresh_put_area()` is called.
##########
src/butil/iobuf.h:
##########
@@ -609,6 +611,165 @@ class IOBufAsZeroCopyOutputStream
int64_t _byte_count;
};
+// Wrap IOBuf into a std::streambuf for std::istream-based parsers
+// (e.g. nlohmann::json::parse(std::istream&)).
+//
+// Read-only view: the streambuf never writes to the source IOBuf. Forward-only
+// (seekoff/seekpos are not overridden).
+//
+// NOTE: The source IOBuf MUST NOT be modified during the lifetime of this
+// streambuf, otherwise the StringPieces returned by backing_block() may be
+// invalidated and the stream will read garbage or crash.
+class IOBufAsInputStreamBuf : public std::streambuf {
+public:
+ // `buf' must outlive this streambuf and must not be modified while the
+ // streambuf is in use.
+ explicit IOBufAsInputStreamBuf(const IOBuf& buf) : _buf(buf) {}
+
+protected:
+ int_type underflow() override;
+ std::streamsize xsgetn(char* s, std::streamsize n) override;
+ std::streamsize showmanyc() override;
+
+private:
+ const IOBuf& _buf;
+ size_t _block_index{0};
+};
+
+// std::istream view over an IOBuf. The IOBuf must outlive this stream and
+// must not be modified while the stream is in use. Forward-only — seeking
+// is not supported.
+//
+// Typical use is feeding an IOBuf to a parser that takes std::istream&,
+// e.g. nlohmann::json:
+//
+// butil::IOBufInputStream in(request_body);
+// auto j = nlohmann::json::parse(in);
+//
+// Or formatted extraction:
+//
+// butil::IOBuf buf;
+// buf.append("42 3.14 hello");
+// butil::IOBufInputStream in(buf);
+// int i; double d;
+// std::string s;
+// in >> i >> d >> s;
+//
+// Bulk read into a buffer (goes through xsgetn, copies one block at a time):
+//
+// std::string out(buf.length(), '\0');
+// butil::IOBufInputStream in(buf);
+// in.read(&out[0], out.size());
+class IOBufInputStream : public std::istream {
+public:
+ // `buf' must outlive this stream and must not be modified while the
+ // stream is in use.
+ explicit IOBufInputStream(const IOBuf& buf)
+ : std::istream(NULL), _sb(buf) {
+ rdbuf(&_sb);
+ }
+
+private:
+ IOBufAsInputStreamBuf _sb;
+};
+
+// Wrap IOBuf into a std::streambuf for std::ostream-based serializers
+// (e.g. nlohmann::json's `os << j`). Bytes are appended directly into IOBuf
+// blocks with no intermediate string copy.
+//
+// Internally backed by IOBufAsZeroCopyOutputStream:
+// - by default, blocks are taken from the per-thread TLS pool (8KB);
+// - pass `block_size` to allocate dedicated blocks instead, which avoids
+// fragmenting the TLS pool when many output streams coexist.
+//
+// Append-only — seekoff/seekpos are not overridden.
+//
+// IMPORTANT: The exact length of the source IOBuf only reflects what was
+// written AFTER shrink() / sync() / destruction — `Next()` over-claims the
+// remainder of each block and the unused tail is BackUp'd later. If you need
+// the precise length mid-stream, call sync() (e.g. via `os.flush()` or
+// `_sb.shrink()`).
Review Comment:
The comment suggests calling `_sb.shrink()`, but `_sb` is a private member of `IOBufOutputStream`, so callers can't do that. This makes the usage guidance inaccurate.
##########
test/iobuf_unittest.cpp:
##########
@@ -23,6 +23,9 @@
#include <stdlib.h>
#include <memory>
#include <cstring>
+#if HAS_NLOHMANN_JSON
+#include <nlohmann/json.hpp>
+#endif // HAS_NLOHMANN_JSON
Review Comment:
This file uses `std::min` (and, when JSON tests are enabled, `std::setw`) but doesn't include the corresponding standard headers. Relying on transitive includes can break builds on some toolchains.
##########
src/butil/iobuf.cpp:
##########
@@ -2025,6 +2027,130 @@ void IOBufAsZeroCopyOutputStream::_release_block() {
_cur_block = NULL;
}
+std::streambuf::int_type IOBufAsInputStreamBuf::underflow() {
+ size_t block_num = _buf.backing_block_num();
+ StringPiece blk;
+ while (_block_index < block_num) {
+ blk = _buf.backing_block(_block_index++);
+ if (!blk.empty()) {
+ break;
+ }
+ }
+ if (blk.empty()) {
+ return traits_type::eof();
+ }
+ // const_cast is safe here: setg() takes char* by API contract, but this
+ // streambuf never writes through it (no overflow/sputc path).
+ char* p = const_cast<char*>(blk.data());
+ setg(p, p, p + blk.size());
+ return traits_type::to_int_type(*gptr());
+}
+
+std::streamsize IOBufAsInputStreamBuf::xsgetn(char* s, std::streamsize n) {
+ auto kIntMax = static_cast<std::streamsize>(std::numeric_limits<int>::max());
+ std::streamsize total = 0;
+ while (total < n) {
+ std::streamsize avail = egptr() - gptr();
+ if (avail == 0) {
+ if (underflow() == traits_type::eof()) {
+ break;
+ }
+ avail = egptr() - gptr();
+ }
+ // Cap step at INT_MAX so gbump(int) cannot overflow when a user-data
+ // block exceeds 2GB.
+ std::streamsize step =
+ std::min(std::min(avail, n - total), kIntMax);
+ iobuf::cp(s + total, gptr(), static_cast<size_t>(step));
+ gbump(static_cast<int>(step));
+ total += step;
+ }
+ return total;
+}
+
+std::streamsize IOBufAsInputStreamBuf::showmanyc() {
+ std::streamsize kMax = std::numeric_limits<std::streamsize>::max();
+ std::streamsize n = egptr() - gptr();
Review Comment:
`showmanyc()` computes `egptr() - gptr()` even when no get area has been set yet (both pointers may be null). This is undefined behavior and can be triggered by `in_avail()` before the first extraction/read.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@brpc.apache.org
For additional commands, e-mail: dev-help@brpc.apache.org