This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 9aa41f79 Support flow-controlled gRPC client requests (#3430)
9aa41f79 is described below
commit 9aa41f79f2bc39ad081e8e2ab80fe3b4dead328b
Author: Xiaofeng Wang <[email protected]>
AuthorDate: Sun Aug 16 15:22:43 2026 +0800
Support flow-controlled gRPC client requests (#3430)
* Support flow-controlled gRPC client requests
- Split client request DATA frames according to the peer's connection and
stream flow-control windows.
- Buffer unsent DATA and resume transmission when WINDOW_UPDATE restores
capacity.
- Track pending request bytes per H2 connection and apply
socket_max_unwritten_bytes as an upper bound.
- Reject or reroute new requests once the pending DATA limit is reached.
- Release buffered DATA when an RPC fails, times out, or its stream is
removed.
- Add tests for fragmented transmission, deferred DATA flushing,
pending-byte accounting, and buffer cleanup.
* Fix pending HTTP/2 data limit race
- Check pending DATA capacity atomically with client stream insertion.
- Leave stream and window state unchanged when the limit is exceeded.
- Use an ephemeral port in the gRPC flow-control test.
- Avoid accessing an empty payload buffer in H2 frame tests.
---
src/brpc/policy/http2_rpc_protocol.cpp | 395 +++++++++++++++++++++----------
src/brpc/policy/http2_rpc_protocol.h | 26 +-
test/brpc_grpc_protocol_unittest.cpp | 50 ++++
test/brpc_h2_unsent_message_unittest.cpp | 212 +++++++++++++++++
test/brpc_http_rpc_protocol_unittest.cpp | 22 +-
5 files changed, 566 insertions(+), 139 deletions(-)
diff --git a/src/brpc/policy/http2_rpc_protocol.cpp
b/src/brpc/policy/http2_rpc_protocol.cpp
index d527a055..eb0e3531 100644
--- a/src/brpc/policy/http2_rpc_protocol.cpp
+++ b/src/brpc/policy/http2_rpc_protocol.cpp
@@ -28,6 +28,7 @@ DECLARE_bool(http_verbose);
DECLARE_int32(http_verbose_max_body_length);
DECLARE_int32(health_check_interval);
DECLARE_bool(usercode_in_pthread);
+DECLARE_int64(socket_max_unwritten_bytes);
namespace policy {
@@ -147,6 +148,12 @@ static int WriteAck(Socket* s, const void* data, size_t n)
{
return s->Write(&sendbuf, &wopt);
}
+static int WriteAck(Socket* s, butil::IOBuf* data) {
+ Socket::WriteOptions wopt;
+ wopt.ignore_eovercrowded = true;
+ return s->Write(data, &wopt);
+}
+
// [ https://tools.ietf.org/html/rfc7540#section-6.5.1 ]
enum H2SettingsIdentifier {
@@ -269,18 +276,13 @@ inline bool AddWindowSize(butil::atomic<int64_t>*
window_size, int64_t diff) {
// If a sender receives a WINDOW_UPDATE that causes a flow-control window
// to exceed this maximum, it MUST terminate either the stream or the
connection,
// as appropriate.
- int64_t before_add = window_size->fetch_add(diff,
butil::memory_order_relaxed);
- if ((((before_add | diff) >> 31) & 1) == 0) {
- // two positive int64_t, check positive overflow
- if ((before_add + diff) & (1 << 31)) {
- return false;
- }
- }
- if ((((before_add & diff) >> 31) & 1) == 1) {
- // two negative int64_t, check negaitive overflow
- if (((before_add + diff) & (1 << 31)) == 0) {
- return false;
- }
+ const int64_t before_add =
+ window_size->fetch_add(diff, butil::memory_order_relaxed);
+ const int64_t after_add = before_add + diff;
+ if (after_add > std::numeric_limits<int32_t>::max() ||
+ after_add < std::numeric_limits<int32_t>::min()) {
+ window_size->fetch_sub(diff, butil::memory_order_relaxed);
+ return false;
}
// window_size being negative is OK
return true;
@@ -323,20 +325,19 @@ inline H2Context::FrameHandler
FindFrameHandler(H2FrameType type) {
H2Context::H2Context(Socket* socket, const Server* server)
: _socket(socket)
- // Maximize the window size to make sending big request possible before
- // receving the remote settings.
- , _remote_window_left(H2Settings::MAX_WINDOW_SIZE)
+ , _remote_window_left(H2Settings::DEFAULT_INITIAL_WINDOW_SIZE)
, _conn_state(H2_CONNECTION_UNINITIALIZED)
, _last_received_stream_id(-1)
, _last_sent_stream_id(1)
, _goaway_stream_id(-1)
, _remote_settings_received(false)
+ , _pending_data_size(0)
, _deferred_window_update(0) {
// Stop printing the field which is useless for remote settings.
_remote_settings.connection_window_size = 0;
- // Maximize the window size to make sending big request possible before
- // receving the remote settings.
- _remote_settings.stream_window_size = H2Settings::MAX_WINDOW_SIZE;
+ // SETTINGS_INITIAL_WINDOW_SIZE defaults to 65535 until the peer sends a
+ // different value. Larger requests are resumed by WINDOW_UPDATE.
+ _remote_settings.stream_window_size =
H2Settings::DEFAULT_INITIAL_WINDOW_SIZE;
if (server) {
_unack_local_settings = server->options().h2_settings;
} else {
@@ -370,6 +371,16 @@ int H2Context::Init() {
return 0;
}
+H2Settings H2Context::remote_settings() const {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ return _remote_settings;
+}
+
+size_t H2Context::VolatilePendingStreamSize() const {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ return _pending_streams.size();
+}
+
H2StreamContext* H2Context::RemoveStreamAndDeferWU(int stream_id) {
H2StreamContext* sctx = NULL;
{
@@ -377,6 +388,8 @@ H2StreamContext* H2Context::RemoveStreamAndDeferWU(int
stream_id) {
if (!_pending_streams.erase(stream_id, &sctx)) {
return NULL;
}
+ CHECK_GE(_pending_data_size, sctx->_pending_data.size());
+ _pending_data_size -= sctx->_pending_data.size();
}
// The remote stream will not send any more data, sending back the
// stream-level WINDOW_UPDATE is pointless, just move the value into
@@ -394,6 +407,7 @@ void H2Context::RemoveGoAwayStreams(
std::unique_lock<butil::Mutex> mu(_stream_mutex);
_goaway_stream_id = goaway_stream_id;
_pending_streams.swap(tmp);
+ _pending_data_size = 0;
}
for (StreamMap::const_iterator it = tmp.begin(); it != tmp.end();
++it) {
out_streams->push_back(it->second);
@@ -408,6 +422,9 @@ void H2Context::RemoveGoAwayStreams(
}
}
for (size_t i = 0; i < out_streams->size(); ++i) {
+ CHECK_GE(_pending_data_size,
+ (*out_streams)[i]->_pending_data.size());
+ _pending_data_size -= (*out_streams)[i]->_pending_data.size();
_pending_streams.erase((*out_streams)[i]->stream_id());
}
}
@@ -429,6 +446,9 @@ int H2Context::TryToInsertStream(int stream_id,
H2StreamContext* ctx) {
}
H2StreamContext*& sctx = _pending_streams[stream_id];
if (sctx == NULL) {
+ // Synchronize creation with SETTINGS_INITIAL_WINDOW_SIZE updates.
+ ctx->_remote_window_left.store(_remote_settings.stream_window_size,
+ butil::memory_order_relaxed);
sctx = ctx;
return 0;
}
@@ -872,43 +892,27 @@ H2ParseResult H2Context::OnSettings(
_local_settings = _unack_local_settings;
return MakeH2Message(NULL);
}
- const int64_t old_stream_window_size = _remote_settings.stream_window_size;
- if (!_remote_settings_received) {
- // To solve the problem that sender can't send large request before
receving
- // remote setting, the initial window size of stream/connection is set
to
- // MAX_WINDOW_SIZE(see constructor of H2Context).
- // As a result, in the view of remote side, window size is 65535 by
default so
- // it may not send its stream size to sender, making stream size still
be
- // MAX_WINDOW_SIZE. In this case we need to revert this value to
default.
- H2Settings tmp_settings;
- if (!ParseH2Settings(&tmp_settings, it, frame_head.payload_size)) {
- LOG(ERROR) << "Fail to parse from SETTINGS";
- return MakeH2Error(H2_PROTOCOL_ERROR);
- }
- _remote_settings = tmp_settings;
- _remote_window_left.fetch_sub(
- H2Settings::MAX_WINDOW_SIZE -
H2Settings::DEFAULT_INITIAL_WINDOW_SIZE,
- butil::memory_order_relaxed);
- _remote_settings_received = true;
- } else {
+ int64_t window_diff = 0;
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ const int64_t old_stream_window_size =
+ _remote_settings.stream_window_size;
if (!ParseH2Settings(&_remote_settings, it, frame_head.payload_size)) {
LOG(ERROR) << "Fail to parse from SETTINGS";
return MakeH2Error(H2_PROTOCOL_ERROR);
}
- }
- const int64_t window_diff =
- static_cast<int64_t>(_remote_settings.stream_window_size)
- - old_stream_window_size;
- if (window_diff) {
- // Do not update the connection flow-control window here, which can
only
- // be changed using WINDOW_UPDATE frames.
- // https://tools.ietf.org/html/rfc7540#section-6.9.2
- // TODO(gejun): Has race conditions with AppendAndDestroySelf
- std::unique_lock<butil::Mutex> mu(_stream_mutex);
- for (StreamMap::const_iterator it = _pending_streams.begin();
- it != _pending_streams.end(); ++it) {
- if (!AddWindowSize(&it->second->_remote_window_left, window_diff))
{
- return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ _remote_settings_received = true;
+ window_diff = static_cast<int64_t>(_remote_settings.stream_window_size)
+ - old_stream_window_size;
+ if (window_diff) {
+ // SETTINGS_INITIAL_WINDOW_SIZE changes all existing stream
windows,
+ // but never the connection-level flow-control window.
+ for (StreamMap::const_iterator it = _pending_streams.begin();
+ it != _pending_streams.end(); ++it) {
+ if (!AddWindowSize(&it->second->_remote_window_left,
+ window_diff)) {
+ return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ }
}
}
}
@@ -919,6 +923,9 @@ H2ParseResult H2Context::OnSettings(
LOG(WARNING) << "Fail to respond settings with ack to " << *_socket;
return MakeH2Error(H2_PROTOCOL_ERROR);
}
+ if (window_diff > 0 && !FlushPendingData(0)) {
+ return MakeH2Error(H2_PROTOCOL_ERROR);
+ }
return MakeH2Message(NULL);
}
@@ -1026,27 +1033,51 @@ H2ParseResult H2Context::OnWindowUpdate(
return MakeH2Error(H2_PROTOCOL_ERROR);
}
if (frame_head.stream_id == 0) {
- if (!AddWindowSize(&_remote_window_left, inc)) {
- LOG(ERROR) << "Invalid connection-level window_size_increment=" <<
inc;
- return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ if (!AddWindowSize(&_remote_window_left, inc)) {
+ LOG(ERROR) << "Invalid connection-level
window_size_increment=" << inc;
+ return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ }
+ }
+ if (!FlushPendingData(0)) {
+ return MakeH2Error(H2_PROTOCOL_ERROR);
}
return MakeH2Message(NULL);
} else {
- H2StreamContext* sctx = FindStream(frame_head.stream_id);
- if (sctx == NULL) {
- RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
- return MakeH2Message(NULL);
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ H2StreamContext** psctx =
_pending_streams.seek(frame_head.stream_id);
+ if (psctx == nullptr) {
+ RPC_VLOG << "Fail to find stream_id=" << frame_head.stream_id;
+ return MakeH2Message(nullptr);
+ }
+ if (!AddWindowSize(&(*psctx)->_remote_window_left, inc)) {
+ LOG(ERROR) << "Invalid stream-level window_size_increment=" <<
inc
+ << " to remote_window_left="
+ <<
(*psctx)->_remote_window_left.load(butil::memory_order_relaxed);
+ return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ }
}
- if (!AddWindowSize(&sctx->_remote_window_left, inc)) {
- LOG(ERROR) << "Invalid stream-level window_size_increment=" << inc
- << " to remote_window_left=" <<
sctx->_remote_window_left.load(butil::memory_order_relaxed);
- return MakeH2Error(H2_FLOW_CONTROL_ERROR);
+ if (!FlushPendingData(frame_head.stream_id)) {
+ return MakeH2Error(H2_PROTOCOL_ERROR);
}
return MakeH2Message(NULL);
}
}
void H2Context::Describe(std::ostream& os, const DescribeOptions& opt) const {
+ H2Settings remote_settings;
+ bool remote_settings_received = false;
+ size_t pending_stream_size = 0;
+ size_t pending_data_size = 0;
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ remote_settings = _remote_settings;
+ remote_settings_received = _remote_settings_received;
+ pending_stream_size = _pending_streams.size();
+ pending_data_size = _pending_data_size;
+ }
if (opt.verbose) {
os << '\n';
}
@@ -1058,8 +1089,8 @@ void H2Context::Describe(std::ostream& os, const
DescribeOptions& opt) const {
<< _deferred_window_update.load(butil::memory_order_relaxed)
<< sep << "remote_conn_window_left="
<< _remote_window_left.load(butil::memory_order_relaxed)
- << sep << "remote_settings=" << _remote_settings
- << sep << "remote_settings_received=" << _remote_settings_received
+ << sep << "remote_settings=" << remote_settings
+ << sep << "remote_settings_received=" << remote_settings_received
<< sep << "local_settings=" << _local_settings
<< sep << "hpacker={";
IndentingOStream os2(os, 2);
@@ -1071,7 +1102,8 @@ void H2Context::Describe(std::ostream& os, const
DescribeOptions& opt) const {
abandoned_size = _abandoned_streams.size();
}
os << sep << "abandoned_streams=" << abandoned_size
- << sep << "pending_streams=" << VolatilePendingStreamSize();
+ << sep << "pending_streams=" << pending_stream_size
+ << sep << "pending_data_size=" << pending_data_size;
if (opt.verbose) {
os << '\n';
}
@@ -1205,28 +1237,6 @@ void H2StreamContext::SetState(H2StreamState state) {
}
#endif
-bool H2StreamContext::ConsumeWindowSize(int64_t size) {
- // This method is guaranteed to be called in AppendAndDestroySelf() which
- // is run sequentially. As a result, _remote_window_left of this stream
- // context will not be decremented (may be incremented) because following
- // AppendAndDestroySelf() are not run yet.
- // This fact is important to make window_size changes to stream and
- // connection contexts transactionally.
- if (_remote_window_left.load(butil::memory_order_relaxed) < size) {
- return false;
- }
- if (!MinusWindowSize(&_conn_ctx->_remote_window_left, size)) {
- return false;
- }
- int64_t after_sub = _remote_window_left.fetch_sub(size,
butil::memory_order_relaxed) - size;
- if (after_sub < 0) {
- LOG(FATAL) << "Impossible, the http2 impl is buggy";
- _remote_window_left.fetch_add(size, butil::memory_order_relaxed);
- return false;
- }
- return true;
-}
-
int H2StreamContext::ConsumeHeaders(butil::IOBufBytesIterator& it) {
HPacker& hpacker = _conn_ctx->hpacker();
HttpHeader& h = header();
@@ -1316,43 +1326,53 @@ int
H2StreamContext::ConsumeHeaders(butil::IOBufBytesIterator& it) {
const CommonStrings* get_common_strings();
-static void PackH2Message(butil::IOBuf* out,
+static void PackH2Headers(butil::IOBuf* out,
butil::IOBuf& headers,
- butil::IOBuf& trailer_headers,
- const butil::IOBuf& data,
int stream_id,
- H2Context* conn_ctx) {
- const H2Settings& remote_settings = conn_ctx->remote_settings();
+ uint32_t max_frame_size,
+ bool end_stream) {
char headbuf[FRAME_HEAD_SIZE];
H2FrameHead headers_head = {
(uint32_t)headers.size(), H2_FRAME_HEADERS, 0, stream_id};
- if (data.empty() && trailer_headers.empty()) {
+ if (end_stream) {
headers_head.flags |= H2_FLAGS_END_STREAM;
}
- if (headers_head.payload_size <= remote_settings.max_frame_size) {
+ if (headers_head.payload_size <= max_frame_size) {
headers_head.flags |= H2_FLAGS_END_HEADERS;
SerializeFrameHead(headbuf, headers_head);
out->append(headbuf, sizeof(headbuf));
out->append(butil::IOBuf::Movable(headers));
} else {
- headers_head.payload_size = remote_settings.max_frame_size;
+ headers_head.payload_size = max_frame_size;
SerializeFrameHead(headbuf, headers_head);
out->append(headbuf, sizeof(headbuf));
headers.cutn(out, headers_head.payload_size);
H2FrameHead cont_head = {0, H2_FRAME_CONTINUATION, 0, stream_id};
while (!headers.empty()) {
- if (headers.size() <= remote_settings.max_frame_size) {
+ if (headers.size() <= max_frame_size) {
cont_head.flags |= H2_FLAGS_END_HEADERS;
cont_head.payload_size = headers.size();
} else {
- cont_head.payload_size = remote_settings.max_frame_size;
+ cont_head.payload_size = max_frame_size;
}
SerializeFrameHead(headbuf, cont_head);
out->append(headbuf, FRAME_HEAD_SIZE);
headers.cutn(out, cont_head.payload_size);
}
}
+}
+
+static void PackH2Message(butil::IOBuf* out,
+ butil::IOBuf& headers,
+ butil::IOBuf& trailer_headers,
+ const butil::IOBuf& data,
+ int stream_id,
+ H2Context* conn_ctx) {
+ const H2Settings& remote_settings = conn_ctx->remote_settings();
+ char headbuf[FRAME_HEAD_SIZE];
+ PackH2Headers(out, headers, stream_id, remote_settings.max_frame_size,
+ data.empty() && trailer_headers.empty());
if (!data.empty()) {
H2FrameHead data_head = {0, H2_FRAME_DATA, 0, stream_id};
butil::IOBufBytesIterator it(data);
@@ -1388,6 +1408,135 @@ static void PackH2Message(butil::IOBuf* out,
}
}
+void H2Context::AppendPendingDataLocked(H2StreamContext* sctx,
+ butil::IOBuf* out) {
+ CHECK(sctx != nullptr);
+ const uint32_t max_frame_size = _remote_settings.max_frame_size;
+ char headbuf[FRAME_HEAD_SIZE];
+ while (!sctx->_pending_data.empty()) {
+ const int64_t conn_window =
+ _remote_window_left.load(butil::memory_order_relaxed);
+ const int64_t stream_window =
+ sctx->_remote_window_left.load(butil::memory_order_relaxed);
+ if (conn_window <= 0 || stream_window <= 0) {
+ break;
+ }
+ const size_t payload_size = std::min(
+ sctx->_pending_data.size(),
+ std::min(static_cast<size_t>(max_frame_size),
+ static_cast<size_t>(std::min(conn_window,
stream_window))));
+ CHECK_GT(payload_size, 0u);
+ _remote_window_left.fetch_sub(payload_size,
butil::memory_order_relaxed);
+ sctx->_remote_window_left.fetch_sub(payload_size,
+ butil::memory_order_relaxed);
+
+ H2FrameHead data_head = {
+ static_cast<uint32_t>(payload_size), H2_FRAME_DATA, 0,
+ sctx->stream_id()};
+ if (payload_size == sctx->_pending_data.size()) {
+ data_head.flags |= H2_FLAGS_END_STREAM;
+ }
+ SerializeFrameHead(headbuf, data_head);
+ out->append(headbuf, sizeof(headbuf));
+ sctx->_pending_data.cutn(out, payload_size);
+ CHECK_GE(_pending_data_size, payload_size);
+ _pending_data_size -= payload_size;
+ }
+}
+
+butil::Status H2Context::TryToInsertClientStream(
+ int stream_id, H2StreamContext* sctx, const butil::IOBuf& data,
+ butil::IOBuf* out) {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ if (_goaway_stream_id >= 0 && stream_id > _goaway_stream_id) {
+ return butil::Status(ELOGOFF, "the connection just issued GOAWAY");
+ }
+ if (_pending_streams.seek(stream_id) != nullptr) {
+ return butil::Status(EINTERNAL,
+ "Fail to insert existing stream_id");
+ }
+ if (_pending_streams.size() >= _remote_settings.max_concurrent_streams) {
+ return butil::Status(
+ ELIMIT, "Pending Stream count exceeds max concurrent stream");
+ }
+
+ sctx->_remote_window_left.store(_remote_settings.stream_window_size,
+ butil::memory_order_relaxed);
+ const int64_t conn_window =
+ _remote_window_left.load(butil::memory_order_relaxed);
+ const int64_t stream_window =
+ sctx->_remote_window_left.load(butil::memory_order_relaxed);
+ size_t sendable_size = 0;
+ if (conn_window > 0 && stream_window > 0) {
+ sendable_size = std::min(
+ data.size(),
+ static_cast<size_t>(std::min(conn_window, stream_window)));
+ }
+ const size_t pending_size = data.size() - sendable_size;
+ if (FLAGS_socket_max_unwritten_bytes > 0) {
+ const auto limit =
+ static_cast<size_t>(FLAGS_socket_max_unwritten_bytes);
+ // Check and reserve pending bytes under the same lock. Otherwise,
+ // concurrent requests may all observe available capacity before any
+ // of them adds its unsent DATA.
+ if (_pending_data_size > limit ||
+ pending_size > limit - _pending_data_size) {
+ return butil::Status(EOVERCROWDED,
+ "Too much pending HTTP/2 request data");
+ }
+ }
+
+ // Mutate stream and window state only after all failure checks above.
+ _pending_streams[stream_id] = sctx;
+ if (!data.empty()) {
+ CHECK(sctx->_pending_data.empty());
+ sctx->_pending_data = data;
+ _pending_data_size += data.size();
+ AppendPendingDataLocked(sctx, out);
+ }
+ return butil::Status::OK();
+}
+
+void H2Context::ClearPendingData(int stream_id) {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ H2StreamContext** psctx = _pending_streams.seek(stream_id);
+ if (psctx == nullptr) {
+ return;
+ }
+ CHECK_GE(_pending_data_size, (*psctx)->_pending_data.size());
+ _pending_data_size -= (*psctx)->_pending_data.size();
+ (*psctx)->_pending_data.clear();
+}
+
+bool H2Context::PendingDataOvercrowded() const {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ return FLAGS_socket_max_unwritten_bytes > 0 &&
+ _pending_data_size >=
+ static_cast<size_t>(FLAGS_socket_max_unwritten_bytes);
+}
+
+bool H2Context::FlushPendingData(int stream_id) {
+ butil::IOBuf out;
+ {
+ std::unique_lock<butil::Mutex> mu(_stream_mutex);
+ if (stream_id != 0) {
+ H2StreamContext** psctx = _pending_streams.seek(stream_id);
+ if (psctx != nullptr) {
+ AppendPendingDataLocked(*psctx, &out);
+ }
+ } else {
+ for (StreamMap::const_iterator it = _pending_streams.begin();
+ it != _pending_streams.end(); ++it) {
+ if (_remote_window_left.load(butil::memory_order_relaxed) <=
0) {
+ break;
+ }
+ AppendPendingDataLocked(it->second, &out);
+ }
+ }
+ }
+ return out.empty() || WriteAck(_socket, &out) == 0;
+}
+
H2UnsentRequest* H2UnsentRequest::New(Controller* c) {
const HttpHeader& h = c->http_request();
const CommonStrings* const common = get_common_strings();
@@ -1485,12 +1634,13 @@ void
H2UnsentRequest::DestroyStreamUserData(SocketUniquePtr& sending_sock,
int error_code,
bool /*end_of_rpc*/) {
RemoveRefOnQuit deref_self(this);
- if (sending_sock != NULL && error_code != 0) {
+ if (sending_sock != nullptr && error_code != 0) {
CHECK_EQ(cntl, _cntl);
std::unique_lock<butil::Mutex> mu(_mutex);
_cntl = NULL;
if (_stream_id != 0) {
H2Context* ctx =
static_cast<H2Context*>(sending_sock->parsing_context());
+ ctx->ClearPendingData(_stream_id);
ctx->AddAbandonedStream(_stream_id);
}
}
@@ -1534,11 +1684,6 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out,
Socket* socket) {
out->append(settingsbuf, nb);
}
- // TODO(zhujiashun): also check this in server push
- if (ctx->VolatilePendingStreamSize() >
ctx->remote_settings().max_concurrent_streams) {
- return butil::Status(ELIMIT, "Pending Stream count exceeds max
concurrent stream");
- }
-
// Although the critical section looks huge, it should rarely be contended
// since timeout of RPC is much larger than the delay of sending.
std::unique_lock<butil::Mutex> mu(_mutex);
@@ -1557,30 +1702,15 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf*
out, Socket* socket) {
}
_sctx->Init(ctx, id);
- // check flow control restriction
- if (!_cntl->request_attachment().empty()) {
- const int64_t data_size = _cntl->request_attachment().size();
- if (!_sctx->ConsumeWindowSize(data_size)) {
- return butil::Status(ELIMIT, "remote_window_left is not enough,
data_size=%" PRId64, data_size);
- }
- }
-
- const int rc = ctx->TryToInsertStream(id, _sctx.get());
- if (rc < 0) {
- return butil::Status(EINTERNAL, "Fail to insert existing stream_id");
- } else if (rc > 0) {
- return butil::Status(ELOGOFF, "the connection just issued GOAWAY");
- }
- _stream_id = _sctx->stream_id();
- // After calling TryToInsertStream, the ownership of _sctx is transferred
to ctx
- _sctx.release();
+ H2StreamContext* const sctx = _sctx.get();
HPacker& hpacker = ctx->hpacker();
butil::IOBufAppender appender;
HPackOptions options;
options.encode_name = FLAGS_h2_hpack_encode_name;
options.encode_value = FLAGS_h2_hpack_encode_value;
- if (ctx->remote_settings().header_table_size == 0) {
+ const H2Settings remote_settings = ctx->remote_settings();
+ if (remote_settings.header_table_size == 0) {
options.index_policy = HPACK_NEVER_INDEX_HEADER;
}
@@ -1597,8 +1727,24 @@ H2UnsentRequest::AppendAndDestroySelf(butil::IOBuf* out,
Socket* socket) {
}
butil::IOBuf frag;
appender.move_to(frag);
- butil::IOBuf dummy_buf;
- PackH2Message(out, frag, dummy_buf, _cntl->request_attachment(),
_stream_id, ctx);
+ const butil::IOBuf& request_data = _cntl->request_attachment();
+ PackH2Headers(out, frag, id, remote_settings.max_frame_size,
+ request_data.empty());
+ const butil::Status insert_status =
+ ctx->TryToInsertClientStream(id, sctx, request_data, out);
+ if (!insert_status.ok()) {
+ return insert_status;
+ }
+ _stream_id = id;
+ // TryToInsertClientStream transfers ownership of _sctx to ctx on success.
+ _sctx.release();
+ const int64_t conn_wu = ctx->ReleaseDeferredWindowUpdate();
+ if (conn_wu > 0) {
+ char winbuf[FRAME_HEAD_SIZE + 4];
+ SerializeFrameHead(winbuf, 4, H2_FRAME_WINDOW_UPDATE, 0, 0);
+ SaveUint32(winbuf + FRAME_HEAD_SIZE, conn_wu);
+ out->append(winbuf, sizeof(winbuf));
+ }
return butil::Status::OK();
}
@@ -1821,7 +1967,8 @@ void PackH2Request(butil::IOBuf*,
static bool IsH2SocketValid(Socket* s) {
H2Context* c = static_cast<H2Context*>(s->parsing_context());
- return (c == NULL || !c->RunOutStreams());
+ return c == nullptr ||
+ (!c->RunOutStreams() && !c->PendingDataOvercrowded());
}
StreamUserData* H2GlobalStreamCreator::OnCreatingStream(
diff --git a/src/brpc/policy/http2_rpc_protocol.h
b/src/brpc/policy/http2_rpc_protocol.h
index b4422ee0..27055ae9 100644
--- a/src/brpc/policy/http2_rpc_protocol.h
+++ b/src/brpc/policy/http2_rpc_protocol.h
@@ -258,8 +258,6 @@ public:
return _deferred_window_update.exchange(0,
butil::memory_order_relaxed);
}
- bool ConsumeWindowSize(int64_t size);
-
#if defined(BRPC_H2_STREAM_STATE)
H2StreamState state() const { return _state; }
void SetState(H2StreamState state);
@@ -276,6 +274,9 @@ friend class H2Context;
butil::atomic<int64_t> _deferred_window_update;
uint64_t _correlation_id;
butil::IOBuf _remaining_header_fragment;
+ // Request body which cannot be sent yet due to remote flow control.
+ // Accessed under H2Context::_stream_mutex.
+ butil::IOBuf _pending_data;
};
StreamCreator* get_h2_global_stream_creator();
@@ -319,7 +320,7 @@ public:
// main_socket: the socket owns this object as parsing_context
// server: NULL means client-side
H2Context(Socket* main_socket, const Server* server);
- ~H2Context();
+ ~H2Context() override;
// Must be called before usage.
int Init();
@@ -337,10 +338,13 @@ public:
// Try to map stream_id to ctx if stream_id does not exist before
// Returns 0 on success, -1 on exist, 1 on goaway.
int TryToInsertStream(int stream_id, H2StreamContext* ctx);
- size_t VolatilePendingStreamSize() const { return _pending_streams.size();
}
+ size_t VolatilePendingStreamSize() const;
+ bool PendingDataOvercrowded() const;
HPacker& hpacker() { return _hpacker; }
- const H2Settings& remote_settings() const { return _remote_settings; }
+ // Return a consistent snapshot because SETTINGS may be processed by the
+ // socket reader while a request is being packed by a writer.
+ H2Settings remote_settings() const;
const H2Settings& local_settings() const { return _local_settings; }
bool is_client_side() const { return _socket->CreatedByConnect(); }
@@ -374,6 +378,15 @@ friend void InitFrameHandlers();
void RemoveGoAwayStreams(int goaway_stream_id,
std::vector<H2StreamContext*>* out_streams);
H2StreamContext* FindStream(int stream_id);
+ // Atomically checks stream and pending-DATA limits, inserts the client
+ // stream, and appends DATA allowed by the current remote windows. On
+ // success, ownership of sctx is transferred to this context. On failure,
+ // no stream, window, or pending-DATA state is changed.
+ butil::Status TryToInsertClientStream(
+ int stream_id, H2StreamContext*, const butil::IOBuf&, butil::IOBuf*);
+ void AppendPendingDataLocked(H2StreamContext*, butil::IOBuf*);
+ void ClearPendingData(int stream_id);
+ bool FlushPendingData(int stream_id);
// True if the connection is established by client, otherwise it's
// accepted by server.
@@ -393,6 +406,9 @@ friend void InitFrameHandlers();
typedef butil::FlatMap<int, H2StreamContext*> StreamMap;
mutable butil::Mutex _stream_mutex;
StreamMap _pending_streams;
+ // Total bytes retained in H2StreamContext::_pending_data on this
+ // connection. Accessed under _stream_mutex.
+ size_t _pending_data_size;
butil::atomic<int64_t> _deferred_window_update;
};
diff --git a/test/brpc_grpc_protocol_unittest.cpp
b/test/brpc_grpc_protocol_unittest.cpp
index f170639d..5a9752ea 100644
--- a/test/brpc_grpc_protocol_unittest.cpp
+++ b/test/brpc_grpc_protocol_unittest.cpp
@@ -88,6 +88,17 @@ public:
}
};
+class WindowGrpcService : public ::test::GrpcService {
+public:
+ void Method(::google::protobuf::RpcController*,
+ const ::test::GrpcRequest* req,
+ ::test::GrpcResponse* res,
+ ::google::protobuf::Closure* done) override {
+ brpc::ClosureGuard done_guard(done);
+ res->set_message(req->message());
+ }
+};
+
class GrpcTest : public ::testing::Test {
protected:
GrpcTest() {
@@ -272,4 +283,43 @@ TEST_F(GrpcTest, GrpcTimeOut) {
}
}
+TEST(GrpcProtocol, client_sends_large_request_with_small_remote_window) {
+ WindowGrpcService service;
+ brpc::Server server;
+ ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions server_options;
+ server_options.h2_settings.stream_window_size = 32;
+ ASSERT_EQ(0, server.Start("127.0.0.1:0", &server_options));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions channel_options;
+ channel_options.protocol = g_protocol;
+ channel_options.timeout_ms = 10000;
+ ASSERT_EQ(0, channel.Init(server.listen_address(), &channel_options));
+ test::GrpcService_Stub stub(&channel);
+
+ // Establish the H2 connection and receive the server SETTINGS first.
+ {
+ test::GrpcRequest request;
+ test::GrpcResponse response;
+ brpc::Controller cntl;
+ request.set_message("warmup");
+ request.set_gzip(false);
+ request.set_return_error(false);
+ stub.Method(&cntl, &request, &response, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ(request.message(), response.message());
+ }
+
+ test::GrpcRequest request;
+ test::GrpcResponse response;
+ brpc::Controller cntl;
+ request.set_message(std::string(128 * 1024, 'x'));
+ request.set_gzip(false);
+ request.set_return_error(false);
+ stub.Method(&cntl, &request, &response, nullptr);
+ EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ EXPECT_EQ(request.message(), response.message());
+}
+
} // namespace
diff --git a/test/brpc_h2_unsent_message_unittest.cpp
b/test/brpc_h2_unsent_message_unittest.cpp
index 5e3b266d..1c7b985a 100644
--- a/test/brpc_h2_unsent_message_unittest.cpp
+++ b/test/brpc_h2_unsent_message_unittest.cpp
@@ -27,11 +27,223 @@
#include "brpc/policy/http2_rpc_protocol.h"
#include "gperftools_helper.h"
+namespace brpc {
+DECLARE_int64(socket_max_unwritten_bytes);
+}
+
int main(int argc, char* argv[]) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
+namespace {
+
+brpc::policy::H2FrameHead PopFrame(butil::IOBuf* buf, std::string* payload) {
+ char head[brpc::policy::FRAME_HEAD_SIZE];
+ CHECK_EQ(sizeof(head), buf->cutn(head, sizeof(head)));
+ brpc::policy::H2FrameHead frame;
+ frame.payload_size =
+ (static_cast<uint8_t>(head[0]) << 16) |
+ (static_cast<uint8_t>(head[1]) << 8) |
+ static_cast<uint8_t>(head[2]);
+ frame.type = static_cast<brpc::policy::H2FrameType>(head[3]);
+ frame.flags = head[4];
+ frame.stream_id =
+ (static_cast<uint8_t>(head[5]) << 24) |
+ (static_cast<uint8_t>(head[6]) << 16) |
+ (static_cast<uint8_t>(head[7]) << 8) |
+ static_cast<uint8_t>(head[8]);
+ payload->resize(frame.payload_size);
+ if (frame.payload_size != 0) {
+ CHECK_EQ(frame.payload_size,
+ buf->cutn(&(*payload)[0], frame.payload_size));
+ }
+ return frame;
+}
+
+} // namespace
+
+TEST(H2UnsentMessage, split_request_data_by_remote_window) {
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ options.user = brpc::get_client_side_messenger();
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), nullptr);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_remote_settings.max_frame_size = 4;
+ ctx->_remote_settings.stream_window_size = 5;
+ ctx->_remote_window_left = 6;
+
+ brpc::policy::H2StreamContext* sctx =
+ new brpc::policy::H2StreamContext(false);
+ sctx->Init(ctx, 1);
+
+ butil::IOBuf body;
+ body.append("abcdefghij");
+ butil::IOBuf out;
+ ASSERT_TRUE(ctx->TryToInsertClientStream(1, sctx, body, &out).ok());
+
+ std::string payload;
+ brpc::policy::H2FrameHead frame = PopFrame(&out, &payload);
+ EXPECT_EQ(4u, frame.payload_size);
+ EXPECT_EQ(brpc::policy::H2_FRAME_DATA, frame.type);
+ EXPECT_EQ(0, frame.flags & 0x1);
+ EXPECT_EQ("abcd", payload);
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(1u, frame.payload_size);
+ EXPECT_EQ(0, frame.flags & 0x1);
+ EXPECT_EQ("e", payload);
+ EXPECT_TRUE(out.empty());
+ EXPECT_EQ(5u, sctx->_pending_data.size());
+ EXPECT_EQ(5u, ctx->_pending_data_size);
+ EXPECT_EQ(1, ctx->_remote_window_left);
+ EXPECT_EQ(0, sctx->_remote_window_left);
+
+ ctx->_remote_window_left.fetch_add(3, butil::memory_order_relaxed);
+ sctx->_remote_window_left.fetch_add(3, butil::memory_order_relaxed);
+ {
+ std::unique_lock<butil::Mutex> mu(ctx->_stream_mutex);
+ ctx->AppendPendingDataLocked(sctx, &out);
+ }
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(3u, frame.payload_size);
+ EXPECT_EQ(0, frame.flags & 0x1);
+ EXPECT_EQ("fgh", payload);
+ EXPECT_TRUE(out.empty());
+ EXPECT_EQ(2u, ctx->_pending_data_size);
+
+ ctx->_remote_window_left.fetch_add(2, butil::memory_order_relaxed);
+ sctx->_remote_window_left.fetch_add(2, butil::memory_order_relaxed);
+ {
+ std::unique_lock<butil::Mutex> mu(ctx->_stream_mutex);
+ ctx->AppendPendingDataLocked(sctx, &out);
+ }
+ frame = PopFrame(&out, &payload);
+ EXPECT_EQ(2u, frame.payload_size);
+ EXPECT_NE(0, frame.flags & 0x1);
+ EXPECT_EQ("ij", payload);
+ EXPECT_TRUE(out.empty());
+ EXPECT_TRUE(sctx->_pending_data.empty());
+ EXPECT_EQ(0u, ctx->_pending_data_size);
+}
+
+TEST(H2UnsentMessage, request_does_not_fail_when_body_exceeds_window) {
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ options.user = brpc::get_client_side_messenger();
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), nullptr);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_last_sent_stream_id = 1;
+ ctx->_remote_settings.max_frame_size = 4;
+ ctx->_remote_settings.stream_window_size = 3;
+ ctx->_remote_window_left = 3;
+
+ brpc::Controller cntl;
+ cntl.http_request().uri() = "http://example.com/echo";;
+ cntl.request_attachment().append("abcdefghij");
+ brpc::policy::H2UnsentRequest* request =
+ brpc::policy::H2UnsentRequest::New(&cntl);
+ ASSERT_TRUE(request != nullptr);
+
+ butil::IOBuf out;
+ const butil::Status status =
+ request->AppendAndDestroySelf(&out, sock.get());
+ EXPECT_TRUE(status.ok()) << status;
+ brpc::policy::H2StreamContext* sctx = ctx->FindStream(1);
+ ASSERT_TRUE(sctx != nullptr);
+ EXPECT_EQ(7u, sctx->_pending_data.size());
+ EXPECT_EQ(7u, ctx->_pending_data_size);
+ EXPECT_EQ(0, ctx->_remote_window_left);
+ EXPECT_EQ(0, sctx->_remote_window_left);
+}
+
+TEST(H2UnsentMessage, invalid_window_update_does_not_change_window) {
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ options.user = brpc::get_client_side_messenger();
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), nullptr);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ const int64_t max_window_size = std::numeric_limits<int32_t>::max();
+ ctx->_remote_window_left = max_window_size;
+
+ const char increment[] = {0, 0, 0, 1};
+ butil::IOBuf payload;
+ payload.append(increment, sizeof(increment));
+ butil::IOBufBytesIterator it(payload);
+ const brpc::policy::H2FrameHead frame = {
+ 4, brpc::policy::H2_FRAME_WINDOW_UPDATE, 0, 0};
+ const brpc::policy::H2ParseResult result = ctx->OnWindowUpdate(it, frame);
+
+ EXPECT_EQ(brpc::H2_FLOW_CONTROL_ERROR, result.error());
+ EXPECT_EQ(max_window_size,
+ ctx->_remote_window_left.load(butil::memory_order_relaxed));
+}
+
+TEST(H2UnsentMessage, clear_pending_data_releases_overcrowded_buffer) {
+ brpc::SocketId id;
+ brpc::SocketUniquePtr sock;
+ brpc::SocketOptions options;
+ options.user = brpc::get_client_side_messenger();
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ brpc::policy::H2Context* ctx =
+ new brpc::policy::H2Context(sock.get(), nullptr);
+ ASSERT_EQ(0, ctx->Init());
+ sock->initialize_parsing_context(&ctx);
+ ctx->_remote_window_left = 0;
+
+ butil::IOBuf body;
+ body.append("pending");
+ butil::IOBuf out;
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ brpc::FLAGS_socket_max_unwritten_bytes = body.size();
+
+ std::unique_ptr<brpc::policy::H2StreamContext> sctx(
+ new brpc::policy::H2StreamContext(false));
+ sctx->Init(ctx, 1);
+ ASSERT_TRUE(
+ ctx->TryToInsertClientStream(1, sctx.get(), body, &out).ok());
+ brpc::policy::H2StreamContext* inserted_sctx = sctx.release();
+ ASSERT_TRUE(out.empty());
+ ASSERT_EQ(body.size(), ctx->_pending_data_size);
+ EXPECT_TRUE(ctx->PendingDataOvercrowded());
+
+ std::unique_ptr<brpc::policy::H2StreamContext> rejected_sctx(
+ new brpc::policy::H2StreamContext(false));
+ rejected_sctx->Init(ctx, 3);
+ butil::IOBuf rejected_body;
+ rejected_body.append("x");
+ const butil::Status rejected = ctx->TryToInsertClientStream(
+ 3, rejected_sctx.get(), rejected_body, &out);
+ EXPECT_EQ(brpc::EOVERCROWDED, rejected.error_code());
+ EXPECT_EQ(nullptr, ctx->FindStream(3));
+ EXPECT_EQ(body.size(), ctx->_pending_data_size);
+
+ ctx->ClearPendingData(1);
+ EXPECT_FALSE(ctx->PendingDataOvercrowded());
+
+ EXPECT_TRUE(inserted_sctx->_pending_data.empty());
+ EXPECT_EQ(0u, ctx->_pending_data_size);
+}
+
TEST(H2UnsentMessage, request_throughput) {
brpc::Controller cntl;
butil::IOBuf request_buf;
diff --git a/test/brpc_http_rpc_protocol_unittest.cpp
b/test/brpc_http_rpc_protocol_unittest.cpp
index 6735a171..70691314 100644
--- a/test/brpc_http_rpc_protocol_unittest.cpp
+++ b/test/brpc_http_rpc_protocol_unittest.cpp
@@ -1643,12 +1643,12 @@ TEST_F(HttpTest, http2_sanity) {
options.protocol = "h2";
ASSERT_EQ(0, channel.Init(butil::EndPoint(butil::my_ip(), port),
&options));
- // Check that the first request with size larger than the default window
can
- // be sent out, when remote settings are not received.
+ // Check that the first request larger than the default window completes
+ // after SETTINGS and WINDOW_UPDATE make more capacity available.
brpc::Controller cntl;
test::EchoRequest big_req;
test::EchoResponse res;
- std::string message(2 * 1024 * 1024 /* 2M */, 'x');
+ std::string message(128 * 1024, 'x');
big_req.set_message(message);
cntl.http_request().set_method(brpc::HTTP_METHOD_POST);
cntl.http_request().uri() = "/EchoService/Echo";
@@ -1762,7 +1762,7 @@ TEST_F(HttpTest, http2_rst_after_header_and_data) {
ASSERT_TRUE(cntl.http_response().status_code() == brpc::HTTP_STATUS_OK);
}
-TEST_F(HttpTest, http2_window_used_up) {
+TEST_F(HttpTest, http2_window_used_up_buffers_request) {
brpc::Controller cntl;
butil::IOBuf request_buf;
test::EchoRequest req;
@@ -1780,6 +1780,8 @@ TEST_F(HttpTest, http2_window_used_up) {
buf.append(settingsbuf, brpc::policy::FRAME_HEAD_SIZE + nb);
brpc::policy::ParseH2Message(&buf, _h2_client_sock.get(), false, NULL);
+ brpc::policy::H2Context* ctx = static_cast<brpc::policy::H2Context*>(
+ _h2_client_sock->parsing_context());
int nsuc = brpc::H2Settings::DEFAULT_INITIAL_WINDOW_SIZE /
cntl.request_attachment().size();
for (int i = 0; i <= nsuc; i++) {
brpc::policy::H2UnsentRequest* h2_req =
brpc::policy::H2UnsentRequest::New(&cntl);
@@ -1789,15 +1791,15 @@ TEST_F(HttpTest, http2_window_used_up) {
NULL, &cntl, request_buf, NULL);
butil::IOBuf dummy;
butil::Status st = socket_message->AppendAndDestroySelf(&dummy,
_h2_client_sock.get());
+ ASSERT_TRUE(st.ok());
if (i == nsuc) {
- // the last message should fail according to flow control policy.
- ASSERT_FALSE(st.ok());
- ASSERT_TRUE(st.error_code() == brpc::ELIMIT);
-
ASSERT_TRUE(butil::StringPiece(st.error_str()).starts_with("remote_window_left
is not enough"));
+ ASSERT_GT(ctx->_pending_data_size, 0u);
+ h2_req->DestroyStreamUserData(
+ _h2_client_sock, &cntl, ECANCELED, false);
+ ASSERT_EQ(0u, ctx->_pending_data_size);
} else {
- ASSERT_TRUE(st.ok());
+ h2_req->DestroyStreamUserData(_h2_client_sock, &cntl, 0, false);
}
- h2_req->DestroyStreamUserData(_h2_client_sock, &cntl, 0, false);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
