TCP 字节流、分包、超时与背压

#include <cstdint>
#include <iostream>
#include <stdexcept>
#include <string>
#include <vector>
class Parser {
    int remaining=-1; std::string current;
public:
    std::vector<std::string> frames;
    void feed(const std::vector<std::uint8_t>& bytes) {
        for(auto b:bytes) {
            if(remaining<0) {
                if(b>8 || frames.size()>=16) throw std::runtime_error("limit");
                remaining=b; current.clear();
                if(remaining==0) {frames.emplace_back(); remaining=-1;}
            } else {
                current.push_back(static_cast<char>(b));
                if(--remaining==0) {frames.push_back(current); remaining=-1;}
            }
        }
    }
    void finish() const {if(remaining>=0) throw std::runtime_error("truncated");}
};
int main() {
    const std::vector<std::uint8_t> wire{3,'c','a','t',2,'h','i'};
    Parser one,chunk; for(auto b:wire) one.feed({b}); chunk.feed(wire);
    one.finish(); chunk.finish();
    if(one.frames!=chunk.frames || one.frames!=std::vector<std::string>{"cat","hi"})
        throw std::runtime_error("framing");
    int invalid=0;
    for(const auto& bytes:std::vector<std::vector<std::uint8_t>>{{9},{3,'x'}}) {
        try {Parser p;p.feed(bytes);p.finish();} catch(const std::runtime_error&) {++invalid;}
    }
    Parser empty;empty.feed({0});empty.finish();
    if(invalid!=2 || empty.frames!=std::vector<std::string>{""}) throw std::runtime_error("edges");
    std::cout<<"frames=cat,hi invalid="<<invalid<<" empty="<<empty.frames.size()<<'\n';
}

#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>
#include <poll.h>
#include <cerrno>
#include <chrono>
#include <iostream>
#include <stdexcept>
#include <string>

class Fd {
public:
    explicit Fd(int value):value_(value) { if(value<0) throw std::runtime_error("socket/accept"); }
    ~Fd(){::close(value_);}
    Fd(const Fd&)=delete;
    Fd& operator=(const Fd&)=delete;
    int get() const{return value_;}
private:
    int value_;
};
using Clock=std::chrono::steady_clock;
void nonblocking(int fd){
    const int flags=::fcntl(fd,F_GETFL,0);
    if(flags<0 || ::fcntl(fd,F_SETFL,flags|O_NONBLOCK)<0) throw std::runtime_error("fcntl");
}
void wait_ready(int fd,short events,Clock::time_point deadline){
    for(;;){
        const auto left=std::chrono::duration_cast<std::chrono::milliseconds>(deadline-Clock::now()).count();
        if(left<=0) throw std::runtime_error("deadline");
        pollfd event{fd,events,0};
        const int result=::poll(&event,1,static_cast<int>(left));
        if(result<0 && errno==EINTR) continue;
        if(result<=0 || (event.revents&(POLLERR|POLLNVAL))) throw std::runtime_error("poll");
        if(event.revents&(events|POLLHUP)) return;
    }
}
int main(){
    const auto deadline=Clock::now()+std::chrono::seconds(2);
    Fd listener(::socket(AF_INET,SOCK_STREAM,0));
    sockaddr_in address{};address.sin_family=AF_INET;
    address.sin_addr.s_addr=htonl(INADDR_LOOPBACK);address.sin_port=htons(0);
    if(::bind(listener.get(),reinterpret_cast<sockaddr*>(&address),sizeof address)!=0 ||
       ::listen(listener.get(),1)!=0) throw std::runtime_error("bind/listen");
    socklen_t length=sizeof address;
    if(::getsockname(listener.get(),reinterpret_cast<sockaddr*>(&address),&length)!=0)
        throw std::runtime_error("getsockname");
    nonblocking(listener.get());
    Fd client(::socket(AF_INET,SOCK_STREAM,0));nonblocking(client.get());
    if(::connect(client.get(),reinterpret_cast<sockaddr*>(&address),length)!=0 && errno!=EINPROGRESS)
        throw std::runtime_error("connect");
    wait_ready(client.get(),POLLOUT,deadline);
    int error=0;length=sizeof error;
    if(::getsockopt(client.get(),SOL_SOCKET,SO_ERROR,&error,&length)!=0 || error!=0)
        throw std::runtime_error("connect completion");
    wait_ready(listener.get(),POLLIN,deadline);
    Fd server(::accept(listener.get(),nullptr,nullptr));nonblocking(server.get());
    const std::string input="abcdef";std::size_t sent=0;
    while(sent<input.size()){
        wait_ready(client.get(),POLLOUT,deadline);
        const auto n=::send(client.get(),input.data()+sent,input.size()-sent,0);
        if(n<0 && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK))continue;
        if(n<=0)throw std::runtime_error("send");sent+=static_cast<std::size_t>(n);
    }
    if(::shutdown(client.get(),SHUT_WR)!=0)throw std::runtime_error("shutdown");
    std::string output;char buffer[2];
    for(;;){
        wait_ready(server.get(),POLLIN,deadline);
        const auto n=::recv(server.get(),buffer,sizeof buffer,0);
        if(n<0 && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK))continue;
        if(n<0)throw std::runtime_error("recv");if(n==0)break;
        output.append(buffer,static_cast<std::size_t>(n));
    }
    if(output!=input)throw std::runtime_error("stream");
    std::cout<<"data="<<output<<" eof=1\n";
}