[Feature] Server-end progressive reader for http protocol
- 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, .
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
Assessment
This issue has not been assessed yet.