apache / apache/brpc

[Feature] Server-end progressive reader for http protocol

Open
#2,145 6 comments 0 reactions 2 assignees Claimed by @Hastyshell View on GitHub
feature
Dominant language
C++
Stars
17.6k
Forks
4.1k
Avg merge
2d 12h
Merged PRs (30d)
69

Description

**Is your feature request related to a problem? (你需要的功能是否与某个问题有关?)**

Apache Doris currently supports `stream load` feature with `http`, which transmits large size (1-10G) of data progressively from user's client to doris' BE http server (currently implemented with `libevent` and is planned to be replaced by `brpc`).

According to #1233 and its related issue, brpc needs the feature of this kind.

**Describe the solution you'd like (描述你期望的解决方法)**

Demands:
1. We need a server end progressive reader just like what the current `http client` does, ![see](https://github.com/apache/brpc/blob/b569fad719334abc0d54db3fd563f4fd7b1d8abc/docs/cn/http_client.md#%E6%8C%81%E7%BB%AD%E4%B8%8B%E8%BD%BD).
2. For the reason that `stream load` requires some pre-check once all the http headers are recieved, and stops the transmition if the pre-check is failed, we need a user-defined callback interface to control the reading process of the progressive reader.

Design:
1. Share the ProgressiveReader interface.
2. User code like:
```cpp
void example(google::protobuf::RpcController* cntl_base,
const HttpRequest*,
HttpResponse*,
google::protobuf::Closure* done) {
brpc::Controller* cntl =
static_cast(cntl_base);

// for demand2, process headers in the normal way, and end for authorization failed.
const std::string* header = cntl->http_request().GetHeader("header-name");
on_header(header);

cntl->ReadProgressiveAttachmentBy(new MyProgressiveReader);
}
```

KeyPoints:
1. Design a good way to config the relation between URI(Service Method) and progressive-read option, so that user can request a specific URI for progressive read.
2. Socket::read_will_be_progressive should be called before ParseHttpMessage.

Solutions:

1. Configuration
Server-end is a bit different with client-end. For a client, we can clearly figure out it's a progressive-reaing client and config the socket to be progressive reading before starting the connection. But for a server, we won't know that the coming request body will be read progressively or not.

Therefore, we should design an option to enable the progressive reading;

Under the trade-off, we choose to use a service option rather than a method level one, because a service level one will be more natural and user-friendly, but a method level one may need a new pattern or API to describe.

```cpp
int Server::AddServiceInternal(google::protobuf::Service *service,
bool is_builtin_service,
const ServiceOptions &svc_opt) {
//...
for (int i = 0; i < sd->method_count(); ++i) {
const google::protobuf::MethodDescriptor *md = sd->method(i);
MethodProperty mp;
//...
mp.params.pb_single_repeated_to_array = svc_opt.pb_single_repeated_to_array;
// add an option for progressive_read here
mp.params.enable_progressive_read;
mp.service = service;
mp.method = md;
mp.status = new MethodStatus;
_method_map[md->full_name()] = mp;
//...
}
}
//...
}
```

2. Active progressive read in **ParseHttpMessage**
```cpp
ParseResult ParseHttpMessage(butil::IOBuf *source, Socket *socket,
bool read_eof, const void* /*arg*/) {
//...

ssize_t rc = 0;
if (read_eof) {
rc = http_imsg->ParseFromArray(NULL, 0);
} else {
rc = http_imsg->ParseFromIOBuf(*source);
}
if (http_imsg->is_stage2()) {
//...
} else if (rc >= 0) {
source->pop_front(rc);
if (http_imsg->Completed()) {
//...
} else if (http_imsg->stage() >= HTTP_ON_HEADERS_COMPLETE) {
// active progressive read here
http_imsg->CheckProgressiveRead();
if (socket->is_read_progressive()) {
http_imsg->AddOneRefForStage2(); // released when body is fully read
return MakeMessage(http_imsg);
}
return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA);
} else {
//...
}
} else {
//...
}

//...
}

void HttpContext::CheckProgressiveRead() {
if (arg() == NULL) {
// indicates not in server-end
return;
}
// omit some pre-checks, just the main logic
const Server::MethodProperty* const sp =
FindMethodPropertyByURI(header().uri().path(), (Server*)arg(), const_cast(&header().unresolved_path()));
if (sp != NULL && sp->enable_progressive_read) {
this->setReadProgressively(true);
socket()->read_will_be_progressive(CONNECTION_TYPE_SHORT);
}
}
```

3. Modify the **ProcessHttpRequest** to adapt to progressive reaing
We need a logic like below:
```cpp
const Server::MethodProperty* const sp =
FindMethodPropertyByURI(path, server, &req_header._unresolved_path);
if(imsg_guard->read_body_progressively()){
// here in this user defined method, user can call `cntl.ReadProgressiveAttachmentBy(new MyProgressiveReader);`
// cannot get completed body msg directly though.
sp->service->CallMethod(method, cntl, req, res, done);
if (imsg_guard->body_reader() == NULL) {
cntl->SetFailed(EHTTP, "progressive reader is unset for a progressive method");
}
}
```

**Describe alternatives you've considered (描述你想到的折衷方案)**

**Additional context/screenshots (更多上下文/截图)**
issue: #1233
doris issue: https://github.com/apache/doris/issues/16087

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.