path: root/library/cpp/http/server/http.cpp

#include "http.h"
#include "http_ex.h"
 
#include <library/cpp/threading/equeue/equeue.h>

#include <util/generic/buffer.h>
#include <util/generic/cast.h>
#include <util/generic/intrlist.h> 
#include <util/generic/yexception.h> 
#include <util/network/address.h>
#include <util/network/socket.h>
#include <util/network/poller.h>
#include <util/system/atomic.h>
#include <util/system/compat.h> // stricmp, strnicmp, strlwr, strupr, stpcpy
#include <util/system/defaults.h>
#include <util/system/event.h>
#include <util/system/mutex.h>
#include <util/system/pipe.h>
#include <util/system/thread.h>
#include <util/thread/factory.h>
 
#include <cerrno>
#include <cstring>
#include <ctime>

#include <sys/stat.h>
#include <sys/types.h>

using namespace NAddr;

namespace { 
    class IPollAble { 
    public: 
        inline IPollAble() noexcept {
        } 
 
        virtual ~IPollAble() { 
        } 
 
        virtual void OnPollEvent(TInstant now) = 0;
    }; 
 
    struct TShouldStop { 
    }; 

    struct TWakeupPollAble: public IPollAble { 
        void OnPollEvent(TInstant) override {
            throw TShouldStop(); 
        } 
    }; 
} 
 
class TClientConnection: public IPollAble, public TIntrusiveListItem<TClientConnection> { 
public:
    TClientConnection(const TSocket& s, THttpServer::TImpl* serv, NAddr::IRemoteAddrRef listenerSockAddrRef);
    ~TClientConnection() override;
 
    void OnPollEvent(TInstant now) override;
 
    inline void Activate(TInstant now) noexcept;
    inline void DeActivate();
    inline void Reject();
 
public:
    TSocket Socket_;
    NAddr::IRemoteAddrRef ListenerSockAddrRef_;
    THttpServer::TImpl* HttpServ_ = nullptr;
    bool Reject_ = false;
    TInstant LastUsed;
    TInstant AcceptMoment;
    size_t ReceivedRequests = 0;
}; 
 
class THttpServer::TImpl { 
public:
    class TConnections {
    public: 
        inline TConnections(TSocketPoller* poller, const THttpServerOptions& options)
            : Poller_(poller)
            , Options(options)
        {
        }
 
        inline ~TConnections() {
        }
 
        inline void Add(TClientConnection* c) noexcept {
            TGuard<TMutex> g(Mutex_);
 
            Conns_.PushBack(c);
            Poller_->WaitRead(c->Socket_, (void*)static_cast<const IPollAble*>(c));
        }
 
        inline void Erase(TClientConnection* c, TInstant now) noexcept {
            TGuard<TMutex> g(Mutex_);
            EraseUnsafe(c);
            if (Options.ExpirationTimeout > TDuration::Zero()) {
                TryRemovingUnsafe(now - Options.ExpirationTimeout);
            }
        }
 
        inline void Clear() noexcept {
            TGuard<TMutex> g(Mutex_);
 
            Conns_.Clear();
        }
 
        inline bool RemoveOld(TInstant border) noexcept {
            TGuard<TMutex> g(Mutex_);
            return TryRemovingUnsafe(border);
        }
 
        bool TryRemovingUnsafe(TInstant border) noexcept {
            if (Conns_.Empty()) {
                return false;
            }
            TClientConnection* c = &*(Conns_.Begin());
            if (c->LastUsed > border) {
                return false;
            }
            EraseUnsafe(c);
            delete c;
            return true;
        }
 
        void EraseUnsafe(TClientConnection* c) noexcept {
            Poller_->Unwait(c->Socket_);
            c->Unlink();
        }

    public:
        TMutex Mutex_;
        TIntrusiveListWithAutoDelete<TClientConnection, TDelete> Conns_;
        TSocketPoller* Poller_ = nullptr;
        const THttpServerOptions& Options;
    };
 
    static void* ListenSocketFunction(void* param) {
        try {
            ((TImpl*)param)->ListenSocket();
        } catch (...) {

        }

        return nullptr;
    }
 
    TAutoPtr<TClientRequest> CreateRequest(TAutoPtr<TClientConnection> c) {
        THolder<TClientRequest> obj(Cb_->CreateClient());
 
        obj->Conn_.Reset(c.Release());

        return obj;
    }

    void AddRequestFromSocket(const TSocket& s, TInstant now, NAddr::IRemoteAddrRef listenerSockAddrRef) {
        if (MaxRequestsReached()) {
            Cb_->OnMaxConn();
            bool wasRemoved = Connections->RemoveOld(TInstant::Max());
            if (!wasRemoved && Options_.RejectExcessConnections) {
                (new TClientConnection(s, this, listenerSockAddrRef))->Reject();
                return;
            }
        } 
 
        auto connection = new TClientConnection(s, this, listenerSockAddrRef);
        connection->LastUsed = now;
        connection->DeActivate();
    }
 
    void SaveErrorCode() {
        ErrorCode = WSAGetLastError();
    }

    int GetErrorCode() const {
        return ErrorCode;
    }
 
    const char* GetError() const {
        return LastSystemErrorText(ErrorCode);
    }
 
    bool Start() {
        Poller.Reset(new TSocketPoller());
        Connections.Reset(new TConnections(Poller.Get(), Options_));

        // Start the listener thread
        ListenerRunningOK = false;

        // throws on error
        TPipeHandle::Pipe(ListenWakeupReadFd, ListenWakeupWriteFd);

        SetNonBlock(ListenWakeupWriteFd, true);
        SetNonBlock(ListenWakeupReadFd, true);

        Poller->WaitRead(ListenWakeupReadFd, &WakeupPollAble);

        ListenStartEvent.Reset();
        try {
            ListenThread.Reset(new TThread(ListenSocketFunction, this));
            ListenThread->Start();
        } catch (const yexception&) {
            SaveErrorCode();
            return false;
        } 
 
        // Wait until the thread has completely started and return the success indicator
        ListenStartEvent.Wait();
 
        return ListenerRunningOK;
    }
 
    void Wait() {
        Cb_->OnWait();
        TGuard<TMutex> g(StopMutex);
        if (ListenThread) {
            ListenThread->Join();
            ListenThread.Reset(nullptr);
        }
    }

    void Stop() {
        Shutdown();

        TGuard<TMutex> g(StopMutex);
        if (ListenThread) {
            ListenThread->Join();
            ListenThread.Reset(nullptr);
        }
 
        while (ConnectionCount) {
            usleep(10000);
            Connections->Clear();
        } 
 
        Connections.Destroy();
        Poller.Destroy();
    }
 
    void Shutdown() {
        ListenWakeupWriteFd.Write("", 1);
        // ignore result
    }
 
    void AddRequest(TAutoPtr<TClientRequest> req, bool fail) {
        struct TFailRequest: public THttpClientRequestEx {
            inline TFailRequest(TAutoPtr<TClientRequest> parent) { 
                Conn_.Reset(parent->Conn_.Release()); 
                HttpConn_.Reset(parent->HttpConn_.Release()); 
            } 
 
            bool Reply(void*) override {
                if (!ProcessHeaders()) {
                    return true;
                }

                ProcessFailRequest(0); 
                return true; 
            } 
        };
 
        if (!fail && Requests->Add(req.Get())) {
            Y_UNUSED(req.Release());
        } else {
            req = new TFailRequest(req);
 
            if (FailRequests->Add(req.Get())) {
                Y_UNUSED(req.Release());
            } else { 
                Cb_->OnFailRequest(-1);
            }
        }
    }

    size_t GetRequestQueueSize() const {
        return Requests->Size();
    }

    size_t GetFailQueueSize() const {
        return FailRequests->Size();
    }

    const IThreadPool& GetRequestQueue() const {
        return *Requests;
    }

    const IThreadPool& GetFailQueue() const {
        return *FailRequests;
    }

    class TListenSocket: public IPollAble, public TIntrusiveListItem<TListenSocket> { 
    public: 
        inline TListenSocket(const TSocket& s, TImpl* parent) 
            : S_(s) 
            , Server_(parent) 
            , SockAddrRef_(GetSockAddr(S_))
        { 
        } 
 
        ~TListenSocket() override {
        } 
 
        void OnPollEvent(TInstant) override {
            SOCKET s = ::accept(S_, nullptr, nullptr);
 
            if (s == INVALID_SOCKET) { 
                ythrow yexception() << "accept: " << LastSystemErrorText(); 
            }

            Server_->AddRequestFromSocket(s, TInstant::Now(), SockAddrRef_);
        } 

        SOCKET GetSocket() const noexcept {
            return S_; 
        } 
 
    private: 
        TSocket S_; 
        TImpl* Server_ = nullptr;
        NAddr::IRemoteAddrRef SockAddrRef_;
    };
 
    void ListenSocket() {
        TThread::SetCurrentThreadName(Options_.ListenThreadName.c_str());

        ErrorCode = 0;
        TIntrusiveListWithAutoDelete<TListenSocket, TDelete> Reqs;
 
        std::function<void(TSocket)> callback = [&](TSocket socket) {
            THolder<TListenSocket> ls(new TListenSocket(socket, this));
            Poller->WaitRead(socket, static_cast<IPollAble*>(ls.Get()));
            Reqs.PushBack(ls.Release());
        };
        bool addressesBound = TryToBindAddresses(Options_, &callback);
        if (!addressesBound) {
            SaveErrorCode();
            ListenStartEvent.Signal();
 
            return;
        }

        Requests->Start(Options_.nThreads, Options_.MaxQueueSize);
        FailRequests->Start(Options_.nFThreads, Options_.MaxFQueueSize);
        Cb_->OnListenStart();
        ListenerRunningOK = true;
        ListenStartEvent.Signal();

        TVector<void*> events;
        events.resize(1);
 
        TInstant now = TInstant::Now();
        for (;;) { 
            try {
                const TInstant deadline = Options_.PollTimeout == TDuration::Zero() ? TInstant::Max() : now + Options_.PollTimeout;
                const size_t ret = Poller->WaitD(events.data(), events.size(), deadline);
 
                now = TInstant::Now();
                for (size_t i = 0; i < ret; ++i) {
                    ((IPollAble*)events[i])->OnPollEvent(now);
                }
 
                if (ret == 0 && Options_.ExpirationTimeout > TDuration::Zero()) {
                    Connections->RemoveOld(now - Options_.ExpirationTimeout);
                }

                // When MaxConnections is limited or ExpirationTimeout is set, OnPollEvent can call
                // RemoveOld and destroy other IPollAble* objects in the
                // poller. Thus in this case we can safely process only one
                // event from the poller at a time.
                if (!Options_.MaxConnections && Options_.ExpirationTimeout == TDuration::Zero()) {
                    if (ret >= events.size()) {
                        events.resize(ret * 2);
                    } 
                } 
            } catch (const TShouldStop&) { 
                break; 
            } catch (...) {
                Cb_->OnException();
            } 
        } 

        while (!Reqs.Empty()) {
            THolder<TListenSocket> ls(Reqs.PopFront());
 
            Poller->Unwait(ls->GetSocket());
        } 
 
        Requests->Stop();
        FailRequests->Stop();
        Cb_->OnListenStop();
    }
 
    void RestartRequestThreads(ui32 nTh, ui32 maxQS) {
        Requests->Stop();
        Options_.nThreads = nTh;
        Options_.MaxQueueSize = maxQS;
        Requests->Start(Options_.nThreads, Options_.MaxQueueSize);
    }

    TImpl(THttpServer* parent, ICallBack* cb, TMtpQueueRef mainWorkers, TMtpQueueRef failWorkers, const TOptions& options_)
        : Requests(mainWorkers)
        , FailRequests(failWorkers)
        , Options_(options_)
        , Cb_(cb)
        , Parent_(parent)
    {
    }

    TImpl(THttpServer* parent, ICallBack* cb, const TOptions& options, IThreadFactory* factory)
        : TImpl(
              parent, 
              cb, 
              MakeThreadPool<TSimpleThreadPool>(factory, options.UseElasticQueues, cb, options.RequestsThreadName),
              MakeThreadPool<TThreadPool>(factory, options.UseElasticQueues, nullptr, options.FailRequestsThreadName),
              options) { 
    }

    ~TImpl() {
        try {
            Stop();
        } catch (...) {
        } 
    }
 
    inline const TOptions& Options() const noexcept {
        return Options_;
    }
 
    inline void DecreaseConnections() noexcept {
        AtomicDecrement(ConnectionCount);
    }

    inline void IncreaseConnections() noexcept {
        AtomicIncrement(ConnectionCount);
    }

    inline i64 GetClientCount() const {
        return AtomicGet(ConnectionCount);
    }

    inline bool MaxRequestsReached() const {
        return Options_.MaxConnections && ((size_t)GetClientCount() >= Options_.MaxConnections); 
    }

    THolder<TThread> ListenThread;
    TPipeHandle ListenWakeupReadFd;
    TPipeHandle ListenWakeupWriteFd;
    TSystemEvent ListenStartEvent;
    TMtpQueueRef Requests;
    TMtpQueueRef FailRequests;
    TAtomic ConnectionCount = 0;
    THolder<TSocketPoller> Poller;
    THolder<TConnections> Connections;
    bool ListenerRunningOK = false;
    int ErrorCode = 0;
    TOptions Options_;
    ICallBack* Cb_ = nullptr;
    THttpServer* Parent_ = nullptr;
    TWakeupPollAble WakeupPollAble;
    TMutex StopMutex;

private:
    template <class TThreadPool_>
    static THolder<IThreadPool> MakeThreadPool(IThreadFactory* factory, bool elastic, ICallBack* callback = nullptr, const TString& threadName = {}) {
        if (!factory) {
            factory = SystemThreadFactory();
        }

        THolder<IThreadPool> pool;
        const auto params = IThreadPool::TParams().SetFactory(factory).SetThreadName(threadName);
        if (callback) {
            pool = MakeHolder<TThreadPoolBinder<TThreadPool_, THttpServer::ICallBack>>(callback, params);
        } else {
            pool = MakeHolder<TThreadPool_>(params);
        }

        if (elastic) {
            pool = MakeHolder<TElasticQueue>(std::move(pool));
        }

        return pool;
    }
}; 

THttpServer::THttpServer(ICallBack* cb, const TOptions& options, IThreadFactory* pool)
    : Impl_(new TImpl(this, cb, options, pool))
{ 
} 
 
THttpServer::THttpServer(ICallBack* cb, TMtpQueueRef mainWorkers, TMtpQueueRef failWorkers, const TOptions& options)
    : Impl_(new TImpl(this, cb, mainWorkers, failWorkers, options))
{
}

THttpServer::~THttpServer() { 
} 
 
i64 THttpServer::GetClientCount() const {
    return Impl_->GetClientCount();
}

bool THttpServer::Start() { 
    return Impl_->Start(); 
} 
 
void THttpServer::Stop() { 
    Impl_->Stop(); 
} 
 
void THttpServer::Shutdown() { 
    Impl_->Shutdown(); 
} 
 
void THttpServer::Wait() { 
    Impl_->Wait(); 
} 
 
int THttpServer::GetErrorCode() { 
    return Impl_->GetErrorCode(); 
} 
 
const char* THttpServer::GetError() { 
    return Impl_->GetError(); 
} 
 
void THttpServer::RestartRequestThreads(ui32 n, ui32 queue) { 
    Impl_->RestartRequestThreads(n, queue); 
} 
 
const THttpServer::TOptions& THttpServer::Options() const noexcept {
    return Impl_->Options(); 
} 
 
size_t THttpServer::GetRequestQueueSize() const {
    return Impl_->GetRequestQueueSize();
}

size_t THttpServer::GetFailQueueSize() const {
    return Impl_->GetFailQueueSize();
}

const IThreadPool& THttpServer::GetRequestQueue() const {
    return Impl_->GetRequestQueue();
}

const IThreadPool& THttpServer::GetFailQueue() const {
    return Impl_->GetFailQueue();
}

bool THttpServer::MaxRequestsReached() const {
    return Impl_->MaxRequestsReached();
}

TClientConnection::TClientConnection(const TSocket& s, THttpServer::TImpl* serv, NAddr::IRemoteAddrRef listenerSockAddrRef)
    : Socket_(s) 
    , ListenerSockAddrRef_(listenerSockAddrRef)
    , HttpServ_(serv) 
{ 
    SetNoDelay(Socket_, true); 
 
    const TDuration& clientTimeout = HttpServ_->Options().ClientTimeout;
    if (clientTimeout != TDuration::Zero()) {
        SetSocketTimeout(Socket_, (long)clientTimeout.Seconds(), clientTimeout.MilliSecondsOfSecond());
    }

    HttpServ_->IncreaseConnections(); 
} 
 
TClientConnection::~TClientConnection() {
    HttpServ_->DecreaseConnections(); 
} 
 
void TClientConnection::OnPollEvent(TInstant now) {
    THolder<TClientConnection> this_(this);
    Activate(now);
 
    { 
        char tmp[1]; 
 
        if (::recv(Socket_, tmp, 1, MSG_PEEK) < 1) { 
            /* 
             * We can received a FIN so our socket was moved to 
             * TCP_CLOSE_WAIT state. Check it before adding work 
             * for this socket. 
             */ 
 
            return; 
        } 
    }
 
    THolder<TClientRequest> obj(HttpServ_->CreateRequest(this_));
    AcceptMoment = now;
 
    HttpServ_->AddRequest(obj, Reject_);
}

void TClientConnection::Activate(TInstant now) noexcept {
    HttpServ_->Connections->Erase(this, now);
    LastUsed = now;
    ++ReceivedRequests;
}

void TClientConnection::DeActivate() { 
    HttpServ_->Connections->Add(this);
} 
 
void TClientConnection::Reject() {
    Reject_ = true;

    HttpServ_->Connections->Add(this);
}

TClientRequest::TClientRequest() { 
} 
 
TClientRequest::~TClientRequest() { 
} 
 
bool TClientRequest::Reply(void* /*ThreadSpecificResource*/) { 
    if (strnicmp(RequestString.data(), "GET ", 4)) {
        Output() << "HTTP/1.0 501 Not Implemented\r\n\r\n"; 
    } else { 
        Output() << "HTTP/1.0 200 OK\r\n" 
                    "Content-Type: text/html\r\n" 
                    "\r\n" 
                    "Hello World!\r\n"; 
    } 
 
    return true; 
} 
 
bool TClientRequest::IsLocal() const {
    return HasLocalAddress(Socket());
}

bool TClientRequest::CheckLoopback() {
    bool isLocal = false;

    try {
        isLocal = IsLocal();
    } catch (const yexception& e) {
        Output() << "HTTP/1.0 500 Oops\r\n\r\n" 
                 << e.what() << "\r\n"; 
        return false; 
    }

    if (!isLocal) {
        Output() << "HTTP/1.0 403 Permission denied\r\n" 
                    "Content-Type: text/html; charset=windows-1251\r\n" 
                    "Connection: close\r\n" 
                    "\r\n" 
                    "<html><head><title>Permission denied</title></head>" 
                    "<body><h1>Permission denied</h1>" 
                    "<p>This request must be sent from the localhost.</p>" 
                    "</body></html>\r\n"; 
 
        return false; 
    } 
 
    return true; 
} 
 
void TClientRequest::ReleaseConnection() {
    if (Conn_ && HttpConn_ && HttpServ()->Options().KeepAliveEnabled && HttpConn_->CanBeKeepAlive() && (!HttpServ()->Options().RejectExcessConnections || !HttpServ()->MaxRequestsReached())) { 
        Output().Finish();
        Conn_->DeActivate();
        Y_UNUSED(Conn_.Release());
    }
}

void TClientRequest::ResetConnection() {
    if (HttpConn_) {
        // send RST packet to client
        HttpConn_->Reset();
        HttpConn_.Destroy();
    }
}

void TClientRequest::Process(void* ThreadSpecificResource) { 
    THolder<TClientRequest> this_(this);
 
    auto* serverImpl = Conn_->HttpServ_;

    try { 
        if (!HttpConn_) { 
            const size_t outputBufferSize = HttpServ()->Options().OutputBufferSize;
            if (outputBufferSize) {
                HttpConn_.Reset(new THttpServerConn(Socket(), outputBufferSize));
            } else {
                HttpConn_.Reset(new THttpServerConn(Socket()));
            }
 
            auto maxRequestsPerConnection = HttpServ()->Options().MaxRequestsPerConnection;
            HttpConn_->Output()->EnableKeepAlive(HttpServ()->Options().KeepAliveEnabled && (!maxRequestsPerConnection || Conn_->ReceivedRequests < maxRequestsPerConnection));
            HttpConn_->Output()->EnableCompression(HttpServ()->Options().CompressionEnabled); 
        } 
 
        if (ParsedHeaders.empty()) {
            RequestString = Input().FirstLine();
 
            const THttpHeaders& h = Input().Headers(); 
            ParsedHeaders.reserve(h.Count());
            for (THttpHeaders::TConstIterator it = h.Begin(); it != h.End(); ++it) { 
                ParsedHeaders.emplace_back(it->Name(), it->Value());
            } 
        } 
 
        if (Reply(ThreadSpecificResource)) { 
            ReleaseConnection();
 
            /* 
             * *this will be destroyed... 
             */ 
 
            return; 
        } 
    } catch (...) { 
        serverImpl->Cb_->OnException();
 
        throw; 
    } 
 
    Y_UNUSED(this_.Release());
} 
 
void TClientRequest::ProcessFailRequest(int failstate) { 
    Output() << "HTTP/1.1 503 Service Unavailable\r\n"
                "Content-Type: text/plain\r\n"
                "Content-Length: 21\r\n"
                "\r\n"
                "Service Unavailable\r\n";
 
    TString url;
 
    if (!strnicmp(RequestString.data(), "GET ", 4)) {
        // Trying to extract url... 
        const char* str = RequestString.data();
 
        // Skipping spaces before url... 
        size_t start = 3; 
        while (str[start] == ' ') {
            ++start; 
        } 
 
        if (str[start]) { 
            // Traversing url... 
            size_t idx = start; 
 
            while (str[idx] != ' ' && str[idx]) { 
                ++idx; 
            } 
 
            url = RequestString.substr(start, idx - start);
        } 
    } 
 
    const THttpServer::ICallBack::TFailLogData d = { 
        failstate, 
        url, 
    }; 
 
    // Handling failure... 
    Conn_->HttpServ_->Cb_->OnFailRequestEx(d); 
    Output().Flush(); 
} 
 
THttpServer* TClientRequest::HttpServ() const noexcept {
    return Conn_->HttpServ_->Parent_; 
} 
 
const TSocket& TClientRequest::Socket() const noexcept {
    return Conn_->Socket_; 
} 

NAddr::IRemoteAddrRef TClientRequest::GetListenerSockAddrRef() const noexcept {
    return Conn_->ListenerSockAddrRef_;
}

TInstant TClientRequest::AcceptMoment() const noexcept {
    return Conn_->AcceptMoment;
}

/*
 * TRequestReplier
 */
TRequestReplier::TRequestReplier() {
}

TRequestReplier::~TRequestReplier() {
}

bool TRequestReplier::Reply(void* threadSpecificResource) {
    const TReplyParams params = {
        threadSpecificResource, Input(), Output()}; 
 
    return DoReply(params);
}

bool TryToBindAddresses(const THttpServerOptions& options, const std::function<void(TSocket)>* callbackOnBoundAddress) {
    THttpServerOptions::TBindAddresses addrs;
    try {
        options.BindAddresses(addrs);
    } catch (const std::exception&) {
        return false;
    }

    for (const auto& na : addrs) {
        for (TNetworkAddress::TIterator ai = na.Begin(); ai != na.End(); ++ai) {
            NAddr::TAddrInfo addr(&*ai);

            TSocket socket(::socket(addr.Addr()->sa_family, SOCK_STREAM, 0));

            if (socket == INVALID_SOCKET) {
                return false;
            }

            FixIPv6ListenSocket(socket);

            if (options.ReuseAddress) {
                int yes = 1;
                ::setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, (const char*)&yes, sizeof(yes));
            }

            if (options.ReusePort) {
                SetReusePort(socket, true);
            }

            if (::bind(socket, addr.Addr(), addr.Len()) == SOCKET_ERROR) {
                return false;
            }

            if (::listen(socket, options.ListenBacklog) == SOCKET_ERROR) {
                return false;
            }

            if (callbackOnBoundAddress != nullptr) { 
                (*callbackOnBoundAddress)(socket);
            }
        }
    }

    return true;
}