[Sylar] framework -chapter14 tcpserver module

Standing on the shoulders of giants

C++ High performance distributed server framework

Rewrite from scratch sylar C++ High performance distributed server framework

summary

• TCP Encapsulation of servers .

TcpServerConf

• TcpServer Configuration structure .
• Partial specialization of two templates ：
  • template<> class LexicalCast<std::string, TcpServerConf>
  • template<> class LexicalCast<TcpServerConf, std::string>

TcpServer

• Inherited from std::enable_shared_from_this and sylar::Noncopyable.
• TcpServer Class supports binding multiple addresses to listen at the same time , Just pass in the address array when Binding . There are two heavy-duty bind Method .
• TcpServer You can also specify the coordination scheduler of the receiving client and the processing client respectively . The concrete is m_acceptWorker and m_ioWorker.

Other instructions

• TcpServer Class adopts the template pattern （Template Pattern） Design mode of , its HandleClient It is implemented by inherited classes . Use TcpServer when , Must be from TcpServer Derive a new class , And re implement the subclass handleClient operation .
• In template mode （Template Pattern） in , An abstract class publicly defines how to execute its methods / Templates . Its subclass can override method implementation as needed , But the call will take place in the way defined in the abstract class . This type of design pattern belongs to behavioral pattern .

To be improved

• Debugging support is not available yet SSL.

Some related codes

/**
 * @filename    tcp_server.h
 * @brief   Tcp  Encapsulation of servers 
 * @author  L-ge
 * @version 0.1
 * @modify  2022-07-13
 */
#ifndef __SYLAR_TCP_SERVER_H__
#define __SYLAR_TCP_SERVER_H__

#include <memory>
#include <functional>
#include "address.h"
#include "iomanager.h"
#include "socket.h"
#include "noncopyable.h"
#include "config.h"

namespace sylar
{

/**
 * @brief   TcpServer Configuration structure 
 */
struct TcpServerConf
{
    typedef std::shared_ptr<TcpServerConf> ptr;

    std::vector<std::string> address;
    int keepalive = 0;
    int timeout = 1000 * 2 * 60;
    int ssl = 0;
    std::string id;

    //  Server type ：http、ws、rock
    std::string type = "http";
    std::string name;
    std::string cert_file;
    std::string key_file;
    std::string accept_worker;
    std::string io_worker;
    std::string process_worker;
    std::map<std::string, std::string> args;

    bool isValid() const
    {
        return !address.empty();
    }

    bool operator==(const TcpServerConf& oth) const
    {
        return address == oth.address
            && keepalive == oth.keepalive
            && timeout == oth.timeout
            && name == oth.name
            && ssl == oth.ssl
            && cert_file == oth.cert_file
            && key_file == oth.key_file
            && accept_worker == oth.accept_worker
            && io_worker == oth.io_worker
            && process_worker == oth.process_worker
            && args == oth.args
            && id == oth.id
            && type == oth.type;
    }
};

template<>
class LexicalCast<std::string, TcpServerConf>
{
public:
    TcpServerConf operator()(const std::string& v)
    {
        YAML::Node node = YAML::Load(v);
        TcpServerConf conf;
        conf.id             = node["id"].as<std::string>(conf.id);
        conf.type           = node["type"].as<std::string>(conf.type);
        conf.keepalive      = node["keepalive"].as<int>(conf.keepalive);
        conf.timeout        = node["timeout"].as<int>(conf.timeout);
        conf.name           = node["name"].as<std::string>(conf.name);
        conf.ssl            = node["ssl"].as<int>(conf.ssl);
        conf.cert_file      = node["cert_file"].as<std::string>(conf.cert_file);
        conf.key_file       = node["key_file"].as<std::string>(conf.key_file);
        conf.accept_worker  = node["accept_worker"].as<std::string>();
        conf.io_worker      = node["io_worker"].as<std::string>();
        conf.process_worker = node["process_worker"].as<std::string>();
        conf.args           = LexicalCast<std::string
            , std::map<std::string, std::string> >()(node["args"].as<std::string>(""));
        if(node["address"].IsDefined())
        {
            for(size_t i = 0; i < node["address"].size(); ++i)
            {
                conf.address.push_back(node["address"][i].as<std::string>());
            }
        }
        return conf;
    }
};

template<>
class LexicalCast<TcpServerConf, std::string>
{
public:
    std::string operator()(const TcpServerConf& conf)
    {
        YAML::Node node;
        node["id"]              = conf.id;
        node["type"]            = conf.type;
        node["keepalive"]       = conf.keepalive;
        node["timeout"]         = conf.timeout;
        node["name"]            = conf.name;
        node["ssl"]             = conf.ssl;
        node["cert_file"]       = conf.cert_file;
        node["key_file"]        = conf.key_file;
        node["accept_worker"]  = conf.accept_worker;
        node["io_worker"]       = conf.io_worker;
        node["process_worker"]  = conf.process_worker;
        node["args"]            = YAML::Load(LexicalCast<
                std::map<std::string, std::string>, std::string>()(conf.args));
        for(auto& i : conf.address)
        {
            node["address"].push_back(i);
        }
        std::stringstream ss;
        ss << node;
        return ss.str();
    }
};

/**
 * @brief   Tcp Server encapsulation 
 */
class TcpServer : public std::enable_shared_from_this<TcpServer>
                  , Noncopyable
{
public:
    typedef std::shared_ptr<TcpServer> ptr;

    /**
     * @brief   
     *
     * @param   worker           It doesn't seem to work , I don't know what it is 
     * @param   io_worker       socket The cooperation scheduler of the client 
     * @param   accpet_worker    The server socket Perform reception socket Connected collaboration scheduler 
     */
    TcpServer(sylar::IOManager* worker = sylar::IOManager::GetThis()
            , sylar::IOManager* io_worker = sylar::IOManager::GetThis()
            , sylar::IOManager* accpet_worker = sylar::IOManager::GetThis());
    virtual ~TcpServer();

    virtual bool bind(sylar::Address::ptr addr, bool ssl = false);
    
    /**
     * @brief   Bind address array  
     */
    virtual bool bind(const std::vector<Address::ptr>& addrs
                    , std::vector<Address::ptr>& fails
                    , bool ssl = false);

    //bool loadCertificates(const std::string& cert_file, const std::string& key_file);

    /**
     * @brief   Start the service  
     */
    virtual bool start();
    virtual void stop();
    
    uint64_t getRecvTimeout() const { return m_recvTimeout; }
    void setRecvTimeout(uint64_t v) { m_recvTimeout = v; }

    /**
     * @brief   Return the server name  
     */
    std::string getName() const { return m_name; }
    virtual void setName(const std::string& v) { m_name = v; }

    bool isStop() const { return m_isStop; }

    TcpServerConf::ptr getConf() const { return m_conf; }
    void setConf(TcpServerConf::ptr v) { m_conf = v; }
    void setConf(const TcpServerConf& v);

    virtual std::string toString(const std::string& prefix = "");
    std::vector<Socket::ptr> getSocks() const { return m_socks; }

protected:
    /**
     * @brief   Handle new connections Socket class  
     */
    virtual void handleClient(Socket::ptr client);
    
    /**
     * @brief    Start accepting connections 
     */
    virtual void startAccept(Socket::ptr sock);

protected:
    ///  monitor Socket Array 
    std::vector<Socket::ptr> m_socks;
    ///  I don't know what's the use 
    IOManager* m_worker;
    ///  New connection Socket Working scheduler 
    IOManager* m_ioWorker;
    ///  The server Socket The scheduler that receives the connection 
    IOManager* m_acceptWorker;
    ///  Receive timeout （ millisecond ）
    uint64_t m_recvTimeout;
    ///  Server name 
    std::string m_name;
    ///  Server type 
    std::string m_type = "tcp";
    ///  Does the service stop 
    bool m_isStop;

    bool m_ssl = false;
    TcpServerConf::ptr m_conf;
};

}

#endif


#include "tcp_server.h"
#include "log.h"

namespace sylar
{
static sylar::ConfigVar<uint64_t>::ptr g_tcp_server_read_timeout = 
    sylar::Config::Lookup("tcp_server.read_timeout", (uint64_t)(60 * 1000 * 2),
            "tcp server read timeout");

static sylar::Logger::ptr g_logger = SYLAR_LOG_NAME("system");

TcpServer::TcpServer(sylar::IOManager* worker
        , sylar::IOManager* io_worker
        , sylar::IOManager* accept_worker)
    : m_worker(worker)
    , m_ioWorker(io_worker)
    , m_acceptWorker(accept_worker)
    , m_recvTimeout(g_tcp_server_read_timeout->getValue())
    , m_name("sylar/1.0.0")
    , m_isStop(true)
{
}

TcpServer::~TcpServer()
{
    for(auto& i : m_socks)
    {
        i->close();
    }
    m_socks.clear();
}

bool TcpServer::bind(sylar::Address::ptr addr, bool ssl)
{
    std::vector<Address::ptr> addrs;
    std::vector<Address::ptr> fails;
    addrs.push_back(addr);
    return bind(addrs, fails, ssl);
}

bool TcpServer::bind(const std::vector<Address::ptr>& addrs
                , std::vector<Address::ptr>& fails
                , bool ssl)
{
    m_ssl = ssl;
    for(auto& addr : addrs)
    {
        //Socket::ptr sock = ssl ? SSLSocket::CreateTCP(addr) : Socket::CreateTCP(addr);
        Socket::ptr sock = Socket::CreateTCP(addr);
        if(!sock->bind(addr))
        {
            SYLAR_LOG_ERROR(g_logger) << "bind fail errno="
                << errno << " errstr=" << strerror(errno)
                << " addr=(" << addr->toString() << "]";
            fails.push_back(addr);
            continue;
        }
        if(!sock->listen())
        {
            SYLAR_LOG_ERROR(g_logger) << "listen fail errno="
                << errno << " errstr=" << strerror(errno)
                << " addr=" << addr->toString() << "]";
            fails.push_back(addr);
            continue;
        }
        m_socks.push_back(sock);
    }

    if(!fails.empty())
    {
        m_socks.clear();
        return false;
    }

    for(auto& i : m_socks)
    {
        SYLAR_LOG_INFO(g_logger) << "type=" << m_type
            << " name=" << m_name
            << " ssl=" << m_ssl
            << " server bind success: " << *i;
    }
    return true;
}

/*bool TcpServer::loadCertificates(const std::string& cert_file, const std::string& key_file)
{
    for(auto& i : m_socks)
    {
        auto ssl_socket = std::dynamic_pointer_cast<SSLSocket>(i);
        if(ssl_socket)
        {
            if(!ssl_socket->loadCertificates(cert_file, key_file))
            {
                return false;
            }
        }
    }
    return true;
}*/

bool TcpServer::start()
{
    if(!m_isStop)
    {
        return true;
    }
    m_isStop = false;
    for(auto& sock : m_socks)
    {
        m_acceptWorker->schedule(std::bind(&TcpServer::startAccept,
                        shared_from_this(), sock));
    }
    return true;
}

void TcpServer::stop()
{
    m_isStop = true;

    //  Pass one more here self, You should add... For the reference count 1, Prevent premature destruction 
    auto self = shared_from_this();
    m_acceptWorker->schedule([this, self]() {
        for(auto& sock : m_socks)
        {
            sock->cancelAll();
            sock->close();
        }
        m_socks.clear();
    });
}

void TcpServer::setConf(const TcpServerConf& v)
{
    m_conf.reset(new TcpServerConf(v));
}

std::string TcpServer::toString(const std::string& prefix)
{
    std::stringstream ss;
    ss << prefix << "[type=" << m_type
       << " name=" << m_name << " ssl=" << m_ssl
       << " worker=" << (m_worker ? m_worker->getName() : "")
       << " accept=" << (m_acceptWorker ? m_acceptWorker->getName() : "")
       << " recv_timeout=" << m_recvTimeout << "]" << std::endl;
    std::string pfx = prefix.empty() ? "    " : prefix;
    for(auto& i : m_socks)
    {
        ss << pfx << pfx << *i << std::endl;
    }
    return ss.str();
}

void TcpServer::handleClient(Socket::ptr client)
{
    SYLAR_LOG_INFO(g_logger) << "handleClient: " << *client;    
}

void TcpServer::startAccept(Socket::ptr sock)
{
    while(!m_isStop)
    {
        Socket::ptr client = sock->accept();
        if(client)
        {
            client->setRecvTimeout(m_recvTimeout);
            m_ioWorker->schedule(std::bind(&TcpServer::handleClient,
                        shared_from_this(), client));
        }
        else
        {
            SYLAR_LOG_ERROR(g_logger) << "accept errno=" << errno
                << " errstr=" << strerror(errno);
        }
    }
}

}

Advertising time ： be based on sylar The implementation of the framework is small demo( I hope to give star)

be based on redis Parameter query service