The author Herman

Introduction:

This article is from one of Herman’s series, “Building Blocks for The Goose Factory Open Source Framework TARS.” The code has been updated with the latest version of the TARS open source community.

TARS open source framework library uses C++ to achieve a lot of common components, these components are generally unified in the util folder, can also be used freely in the application layer, to do a good job must be a good tool, so it is necessary to understand these tool components, better use, improve efficiency. Next, this article will analyze the following TarsCpp components:

  • Thread operation
    • Thread-safe queue: TC_ThreadQueue
    • Normal thread lock: TC_ThreadLock
    • Thread base class: TC_Thread
  • Smart Pointers
    • Smart pointer class: TC_AutoPtr
  • The DB operation
    • MySQL operation class: TC_Mysql
  • Network operating
    • Network components
  • Service configuration
    • Command parsing class: TC_Option
    • Config file class: TC_Config
  • Copy function
    • Generic functor class: TC_Functor
  • Hash
    • The hash algorithm
  • Exception handling
    • Exception class: TC_Exception

Thread-safe queue: TC_ThreadQueue

The framework uses the TC_ThreadQueue class as follows:

typedef TC_ThreadQueue<tagRecvData*, deque<tagRecvData*> > recv_queue; // Receive queue
typedef TC_ThreadQueue<tagSendData*, deque<tagSendData*> > send_queue; // Send queue
Copy the code

The implementation of TC_ThreadQueue is relatively simple. It is important in the network layer implementation of TARS because network packets received from the framework are added to the cache queue. The ServantHandle will then call waitForRecvQueue to fetch network packets from the queue and dispatch to call the corresponding handler of the protocol message.

/** * @brief thread safe queue */
template<typename T, typename D = deque<T> >
class TC_ThreadQueue
{
public:
    TC_ThreadQueue():_size(0) {};public:

    typedef D queue_type;

	/** * @brief Obtains data from the header, no data throw exception ** @param t * @return bool: true, obtains data, false, no data */
	T front(a);

    /** * @brief gets data from the header, * * @param t * @param millSecond (wait = true) Wait * @return bool: true, data obtained, false, no data */
    bool pop_front(T& t, size_t millsecond = 0.bool wait = true); . . }Copy the code

TC_ThreadQueue uses

and

from C++11 to implement a thread lock and wait.

template<typename T, typename D> void TC_ThreadQueue<T, D>::push_front(const T& t, bool notify)
{
    if(notify) {
        std::unique_lock<std::mutex> lock(_mutex);

        _cond.notify_one(a); _queue.push_front(t);

        ++_size;
    }
    else
    {
        std::lock_guard<std::mutex> lock (_mutex);

        _queue.push_front(t); ++_size; }}Copy the code

STD ::unique_lock< STD ::mutex> lock(_mutex) to prevent data conflict between the network layer and the service layer from accessing the same queue. Notify_one () notify_one() notify_one() notify_one() notify_one() notify_one() notify_one() notify_one() notify_one() when a thread on the lock wakes up.

Another member function, pop_front, gets data from the header and waits if there is no data. Millisecond Time (ms)

  • 0Does not block
  • - 1Permanent wait
template<typename T, typename D> bool TC_ThreadQueue<T, D>::pop_front(T& t, size_t millsecond, bool wait)
{
    if(wait) {

        std::unique_lock<std::mutex> lock(_mutex);

        if (_queue.empty()) {
            if (millsecond == 0) {
                return false;
            }
            if (millsecond == (size_t) - 1) {
                _cond.wait(lock);
            }
            else {
                / / the overtime
                if (_cond.wait_for(lock, std::chrono::milliseconds(millsecond)) == std::cv_status::timeout) {
                    return false; }}}if (_queue.empty()) {
            return false;
        }

        t = _queue.front(a); _queue.pop_front(a);assert(_size > 0);
        --_size;

        return true;
    }
    else
    {
        std::lock_guard<std::mutex> lock (_mutex);
        if (_queue.empty())
        {
            return false;
        }

        t = _queue.front(a); _queue.pop_front(a);assert(_size > 0);
        --_size;

        return true; }}Copy the code

BindAdapter: : waitForRecvQueue function is called the pop_front function, to wait for receive queue, function prototype is as follows:

bool TC_EpollServer::BindAdapter::waitForRecvQueue(uint32_t handleIndex, shared_ptr<RecvContext> &data)
{
	bool bRet = getRecvQueue(handleIndex).pop_front(data);

    if(! bRet) {return bRet;
    }

    --_iRecvBufferSize;

    return bRet;
}

Copy the code

Here BindAdapter: : waitForRecvQueue for business thread is waiting for the server to monitor adapter business package after receiving the network packet processing, here the incoming handleIndex said receive queue index, to obtain the corresponding _rbuffer.

Normal thread lock: TC_ThreadLock

The TC_ThreadLock class is defined as follows

typedef TC_Monitor<TC_ThreadMutex, TC_ThreadCond> TC_ThreadLock;
Copy the code

TC_Monitor Thread lock monitor template class. In general, thread locks are used through this class instead of using TC_ThreadMutex or TC_ThreadRecMutex directly.

The definition of the class template

class TC_Monitor requires passing in two template parameters. TC_Monitor includes the following member variables:

mutable int     _nnotify;   // The number of locks
mutable P       _cond;      // Condition variable
T               _mutex;     / / the mutex
/** * @brief defines the lock control object */
typedef TC_LockT<TC_Monitor<T, P> > Lock;
typedef TC_TryLockT<TC_Monitor<T, P> > TryLock;
Copy the code

The first argument, TC_ThreadMutex, represents a thread lock: the same thread cannot be locked twice, and contains member variables

mutable std::mutex _mutex
Copy the code

TC_ThreadRecMutex class TC_ThreadRecMutex class TC_ThreadRecMutex class TC_ThreadRecMutex class

// defined in tc_monitor.h
typedef TC_Monitor<TC_ThreadRecMutex, TC_ThreadCond> TC_ThreadRecLock;
Copy the code

The second argument, TC_ThreadCond, represents the thread-signal condition class: all locks can wait for a signal to occur, and contains thread-condition member variables:

mutable std::condition_variable_any _cond
Copy the code

TC_Monitor::timedWait() calls the timedWait function of the TC_ThreadCond object, and then calls the milliseconds of the chrono library. TC_ThreadCond::signal() implements sending a signal that a thread waiting on this condition will wake up.

TC_LockT class definition: template

class TC_LockT lock template class, used in conjunction with other specific locks, to construct the lock and unlock when sufficient.

The TC_LockT constructor is used to initialize the member variable _mutex by passing in a mutex:

TC_LockT(const T& mutex) : _mutex(mutex) {
        _mutex.lock(a); _acquired =true;
}
Copy the code

TC_Monitor typedef TC_LockT

> Lock

The actual application scenarios are as follows:

Lock lock(*this);
Copy the code

The TC_LockT constructor calls _mutex.lock(); the TC_LockT constructor calls _mutex.lock(); TC_Monitor lock function; TC_Monitor lock function;

void lock(a) const
{
    _mutex.lock(a); _nnotify =0;
}
Copy the code

The TC_ThreadRecMutex::lock() member function is called as follows:

void TC_ThreadMutex::lock(a) const
{
    _mutex.lock(a); }Copy the code

Then the destructor of TC_LockT is called when the above defined lock stack variable exits the function:

virtual ~TC_LockT()
{
    if (_acquired)
    {
        _mutex.unlock(a);// TC_Monitor's unlock function is called}}Copy the code

Unlock function TC_Monitor

void unlock(a) const
{
    notifyImpl(_nnotify);
    _mutex.unlock(a);// this calls unlock from the C++ standard library 
      
}
Copy the code

NotifyImpl is called here because the TC_Monitor class can not only implement the simple mutex function, but also implement the Condition variable function, where notifyImpl is implemented as

void notifyImpl(int nnotify) const
{
    if(nnotify ! =0)
    {
        if(nnotify == - 1)
        {
            _cond.broadcast(a);return;
        }
        else
        {
            while(nnotify > 0)
            {
                _cond.signal(a); --nnotify; }}}}Copy the code

Thread base class: TC_Thread

As always, let’s take a look at the project’s actual use of thread-base classes. BasicThread class TC_Thread class BasicThread class TC_Thread class BasicThread class TC_Thread class BasicThread class

class BasicThread : public tars::TC_Thread, public tars::TC_ThreadLock
{

   ...

    void terminate(a)
    {
        _bTerm = true;
        {
            Lock lock(*this);
            notifyAll(a); }getThreadControl().join();
    }
}
Copy the code

The BasicThread class inherits TC_Thread and TC_ThreadLock. The TC_ThreadLock class is defined as TC_Thread

class TC_Thread : public TC_Runable
{

    ...
    
    /** * use the C++11 standard thread library STD ::thread, the constructor passes the threadEntry thread function, * returns TC_ThreadControl(_th), where _th is the STD ::thread object */
    TC_ThreadControl start(a);

    static void threadEntry(TC_Thread *pThread); // static function, thread entry

    virtual void run(a) = 0; . }Copy the code

Next look at the definition of the thread control class TC_ThreadControl:

class TC_ThreadControl 
{.explicit TC_ThreadControl(std::thread *th); // Construct, passing in the STD :: Thread object

void join(a); // Calling STD ::thread's join() blocks the current thread until another thread finishes running

static void sleep(a); // Calling STD ::this_thread::sleep suspends execution. }Copy the code

Let’s look at the TC_Runable definition:

class TC_Runable
{
public:
    virtual ~TC_Runable() {};virtual void run(a) = 0; // Define the run pure virtual function
};
Copy the code

Finally, take a look at the use of thread classes in a real project

class AntiSdkSyncThread : public BasicThread // The TC_Thread and TC_ThreadLock classes are inherited
{
    void run(a)  // Implement pure virtual functions of the base class
    {
        Lock lock(*this);

        timedWait(10 * 1000); (Interval execution time, to achieve the timing of the thread execution function)if(NULL! = g_busi_interf) { Int32 ret = g_busi_interf->proc_(a);// A function that needs to be executed periodically}}}Copy the code

We’ve defined AntiSdkSyncThread g_antiSdkSyncThread; Class, then g_antisdkSyncThread.start () is executed when the thread needs to be started; Will create a thread, nature and threadEntry thread function will be called the pThread – > run () polymorphic functions, process called when g_antiSdkSyncThread. The terminate (); .

Smart pointer class: TC_AutoPtr

Thread-safe smart Pointers can be placed in containers. The CPP11 library’s auto_ptr cannot be placed in containers. It seems to have been deprecated.

The TC_HandleBase class for smart Pointers is defined as follows. All classes that need smart Pointers inherit from this object, using

from the C++11 library for atomic counting.

class UTIL_DLL_API TC_HandleBase
{
public:

    /** * @brief copy ** @return TC_HandleBase& */
    TC_HandleBase& operator= (const TC_HandleBase&)
    {
        return *this;
    }

    /** * @brief increment count */
    void incRef(a) { ++_atomic; }

    /** * @brief reduces the count */
    void decRef(a)
    {
        if((--_atomic) == 0 && !_bNoDelete)
        {
            _bNoDelete = true;
            delete this; }}/** * @brief Gets the count. ** @return int Counts the value */
    int getRef(a) const        { return _atomic; }

    /** * @brief Specifies whether to automatically delete the param b file,true or false
    void setNoDelete(bool b)  { _bNoDelete = b; }

protected:

    /** * @brief constructor */
    TC_HandleBase() : _atomic(0), _bNoDelete(false) {}/** * @brief copy construction */
    TC_HandleBase(const TC_HandleBase&) : _atomic(0), _bNoDelete(false) {}/** * @brief destructor */
    virtual ~TC_HandleBase() {}protected:

    std::atomic<int>	_atomic;    // Reference count
    bool                _bNoDelete; // Whether to delete it automatically
};

Copy the code

The TC_AutoPtr smart pointer template class is a thread-safe smart pointer that can be placed in a container. This smart pointer is implemented by reference counting, and its constructor and destructor are defined as follows:

template<typename T> 
class TC_AutoPtr
{ 
    TC_AutoPtr(T* p = 0)
    {
        _ptr = p;

        if(_ptr)
        {
            _ptr->incRef(a);// Constructor reference evaluates plus 1}}... ~TC_AutoPtr()
    {
        if(_ptr)
        {
            _ptr->decRef(a);// Destructor reference evaluates minus 1}}}Copy the code

Example: Actual project use

struct ConnStruct : publicTC_HandleBase{... }typedef TC_AutoPtr<ConnStruct> ConnStructPtr;
Copy the code

TC_AutoPtr copy constructor call _ptr->incRef(); TC_HandleBaseT

::incRef() {++_atomic; }

The reference count atom operation is increased by 1, and the destructor reference count atom operation is decreased by 1. When the reference count is reduced to 0, the delete is triggered according to whether the switch is to be deleted.

Example: This is a typical example of TARS using asynchronous RPC callbacks, where the callback class uses smart Pointers

// Define a smart pointer to the callback function, where the SessionCallback parent inherits from TC_HandleBase
typedef TC_AutoPtr<SessionCallback> SessionCallbackPtr;  

// Create the callback class SessionCallbackPtr and pass in the initialization parameters uin gameID etc.
SessionCallbackPtr cb = new SessionCallback(iUin, iGameId, iSeqID, iCmd,sSessionID, theServant, current, cs, this);
// Asynchronously invoke the sessionServer remote interface
getSessionPrx() - >async_getSession(cb, iUin, iGameId); 
Copy the code

Interface to return is completed, the callback SessionCallback: : callback_getSession (tars: : Int32 ret, const MGComm: : SessionValue & retValue) function, The SessionValue structure that receives the return from the sessionServer interface.

Since SessionCallbackPtr uses a smart pointer, it is convenient for the business not to manually release the new SessionCallbackPtr.

MySQL operation class: TC_Mysql

TC_Mysql encapsulates the mysql operation class, which is not thread-safe, and can have better function encapsulation for INSERT /update, preventing SQL injection

Usage:

TC_Mysql mysql;
// initialize mysql, init does not link, request automatically link;
// The database can be empty;
// The default port is 3306
mysql.init("192.168.1.2 instead"."pc"."pc@sn"."db_tars_demo");
Copy the code

Void init(const TC_DBConf& tcDBConf); Initialize the database directly. For example: stDirectMysql init (_stZoneDirectDBConf);

Take a look at the TC_DBConf definition

struct TC_DBConf
{
    string _host;
    string _user;
    string _password;
    string _database;
    string _charset;
    int _port;
    int _flag;   // Client id

    TC_DBConf()
        : _port(0)
        , _flag(0)
    {}

    /** * @brief read the configuration of the database. ** @param mpParam: Host address * dbuser: user name * dbpass: password * dbname: database name * dbport: port */
    void loadFromMap(const map<string, string> &mpParam)
    {
        map<string, string> mpTmp = mpParam;

        _host        = mpTmp["dbhost"];
        _user        = mpTmp["dbuser"];
        _password    = mpTmp["dbpass"];
        _database    = mpTmp["dbname"];
        _charset     = mpTmp["charset"];
        _port        = atoi(mpTmp["dbport"].c_str());
        _flag        = 0;

        if(mpTmp["dbport"] = ="")
        {
            _port = 3306; }}};Copy the code
// Take a closer look at the use of fetching data
TC_Mysql::MysqlData data;
data = mysql.queryRecord("select * from t_app_users");
for(size_t i = 0; i < data.size(a); i++) {// If the ID field does not exist, an exception is thrown
    cout << data[i]["ID"] << endl;
}
Copy the code

The queried mysql data is encapsulated with MysqlData

class MysqlData
{. vector<map<string, string> >&data(a); . }Copy the code
// Insert data, specifying the type of the data: numeric or string, which is automatically escaped
map<string, pair<TC_Mysql::FT, string> > m;
m["ID"]     = make_pair(TC_Mysql::DB_INT, "2334");
m["USERID"] = make_pair(TC_Mysql::DB_STR, "abcttt");
m["APP"]    = make_pair(TC_Mysql::DB_STR, "abcapbbp");
m["LASTTIME"]    = make_pair(TC_Mysql::DB_INT, "now()");

mysql.replaceRecord("t_user_logs", m);
Copy the code

Network components

The entire TARS core provides a complete network framework, including RPC functionality. Here are just a few common network components.

TC_Socket: encapsulates the basic methods of sockets

Provide the operation class of socket; Supports TCP/UDP sockets. Local sockets are supported.

The next layer of TARS encapsulates the TC_TCPClient and TC_UDPClient classes for actual TCP and UDP applications.

Usage:

For example, TCP client

TC_TCPClient stRouterClient;

stRouterClient.init(sIP, iPort, iTimeOut); // Here the IP and port are passed in and sendRecv is called to send and receive the message

Int32 ret = stRouterClient.sendRecv(request.c_str(), request.length(), recvBuf, iRecvLen);
Copy the code

When using multiple threads, do not send/ recV at the same time.

TC_Epoller

Provides network epoll operation class. The default mode is ET. It is notified when the status changes and provides basic operations such as Add, mod, del, and wait.

TC_ClientSocket: base class for socket-related operations on clients

Provides key member functions init(const String &sIp, int iPort, int iTimeout), passing in the IP port and timeout

TC_TCPClient inherits TC_ClientSocket to provide member functions:

  • sendRecv(send to the server, return no more than iRecvLen bytes from the server)
  • sendRecvBySep(Send to the server and wait for the server until the end character, including the end character)

Example:

stRouterClient.init(sIP, iPort, iTimeOut);

size_t iRecvLen = sizeof(recvBuf)- 1;
Int32 ret = stRouterClient.sendRecv(request.c_str(), request.length(), recvBuf, iRecvLen);
Copy the code

Similarly, TC_UDPClient implements UDP client.

Command parsing class: TC_Option

  1. Command parsing class;
  2. Typically used to parse command-line arguments;
  3. Only the double – parameter is supported
  4. Analysis of themainThe following types of parameters are supported:
./main.exe --name=value --param1 param2 param3
Copy the code
TC_Option op;
// Parse the command line
op.decode(argc, argv);
// Get pairs of arguments, that is, get all pairs of arguments represented by --
map<string, string> mp = op.getMulti(a);// Non-- parameters: param2, param3
vector<string> d = op.getSingle(a);Copy the code

If value param has a space or — you can put quotes around it.

Config file class: TC_Config

  1. Configuration file parsing class (compatible with WBL mode);
  2. Support parsing configuration files from string;
  3. Support configuration file generation;
  4. Parsing error throws an exception;
  5. [] is used to obtain the configuration. If there is no configuration, an exception is thrown.
  6. Get is used to obtain the configuration. If the configuration does not exist, null is returned.
  7. Reading configuration files is thread-safe; functions such as the INSERT domain are not thread-safe

Example:

TC_Config config;
config.parseFile(ServerConfig::BasePath + ServerConfig::ServerName + ".conf");
stTmpGameServerConfig.iGameId = TC_Common::strto<UInt32>(config["/Main/<GameId>"]);
Copy the code

Example Configuration file

<Main>
    GameId = 3001
    ZoneId = 102
    AsyncThreadCheckInterval = 1000
    ...
</Main>
Copy the code

Example using the get method: If the configuration cannot be read, return the default value sDefault, which is 20000000 in the following example

stTmpGameServerConfig.iMaxRegNum = TC_Common::strto<Int32>(config.get("/Main/<MaxRegNum>"."20000000"));
Copy the code

Generic functor class: TC_Functor

TC_Functor references Loki library design

  1. The method of calling a function object, that is, you can add a pair of parentheses to the right of the above methods, and put a set of appropriate parameters inside the parentheses to call, such as a(P1,p2);

  2. Encapsulate the entire call (including parameters) into A function object. The call object is created with parameters, but is called without parameters, such as A, A (p1, p2). a();

Simple and easy to use package, see the following use examples naturally understand:

C function call

void TestFunction3(const string &s, int i){
    cout << "TestFunction3('" << s << "', '" << i << "')" << endl;   
}
// Use function Pointers to construct objects
TC_Functor<void, TL::TLMaker<const string&, int>::Result > cmd3(TestFunction3);
string s3("s3");
cmd3(s3, 10);
Copy the code

C function calls are wrapped in a wrapper:

// Call encapsulation, passing in arguments when constructing

TC_Functor<void,TL::TLMaker<const string&, int>::Result>::wrapper_type fwrapper3(cmd3, s3, 10);
fwrapper3(a);// The parameters are already passed in the constructor
Copy the code

Description:

  • void: The return value of the function
  • TL::TLMaker<const string&, int>::Result: indicates the parameter type

For call encapsulation, note that for pass reference types, specific calls must ensure that the referenced object exists.

C++ calls to class member functions

struct TestMember
{
    void mem3(const string &s, int i)
    {
        cout << "TestMember::mem3(" << s << "," << i << ") called" << endl;
    }
}
TC_Functor<void, TL::TLMaker<const string&, int>::Result > cmd3(&tm, &TestMember::mem3);
cmd3("a".33);
Copy the code

Calls to class member functions are wrapped in a Wrapper:

TC_Functor<void, TL::TLMaker<const string&, int>::Result >::wrapper_type fwrapper3(cmd3, "a".10);
fwrapper3(a);Copy the code

Practical example: Register a protocol parser

When the service is initialized, it is called

addServantProtocol(sRouterObj, AppProtocol::parseStream<0.uint16_t.false>,iHeaderLen);
Copy the code

Set BindAdapter protocol_functor _pf to parseStream as follows:

/** * @param T * @param offset * @param netorder * @param in * @param out * @return int */
template<size_t offset, typename T, bool netorder>
static TC_NetWorkBuffer::PACKET_TYPE parseStream(TC_NetWorkBuffer& in,vector<char>& out)
{
    size_t len = offset + sizeof(T);

    if (in.getBufferLength() < len)
    {
        return TC_NetWorkBuffer::PACKET_LESS;
    }

    string header;
    in.getHeader(len, header);

    assert(header.size() == len);

    T iHeaderLen = 0; : :memcpy(&iHeaderLen, header.c_str() + offset, sizeof(T));

    if (netorder)
    {
        iHeaderLen = net2host<T>(iHeaderLen);
    }

    // The length is protected
    if (iHeaderLen < (T)(len) || (uint32_t)iHeaderLen > TARS_NET_MAX_PACKAGE_SIZE)
    {
        return TC_NetWorkBuffer::PACKET_ERR;
    }

    if (in.getBufferLength() < (uint32_t)iHeaderLen)
    {
        return TC_NetWorkBuffer::PACKET_LESS;
    }

    in.getHeader(iHeaderLen, out);

    assert(out.size() == iHeaderLen);

    in.moveHeader(iHeaderLen);

    return TC_NetWorkBuffer::PACKET_FULL;
}
Copy the code

After the parseProtocol function is registered, the network layer invokes the parseProtocol function

int TC_EpollServer::Connection::parseProtocol(TC_NetWorkBuffer &rbuf)
{
    ...
    TC_NetWorkBuffer::PACKET_TYPE b = _pBindAdapter->getProtocol()(rbuf, ro); // This calls back the protocol parsing function set earlier to implement protocol parsing. }Copy the code

The hash algorithm

Util /tc_hash_fun.h contains an implementation of the hash algorithm. Using hash_new, you can hash the input byte stream to get a fairly uniform hash value as follows

#include "util/tc_hash_fun.h"
#include <iterator>
#include <iostream>
#include <sys/time.h>

using namespace tars;
using namespace std;

int main(int argc, char* *argv[])
{
    unsigned int i = tars::hash_new<string>()("abcd");
    cout << i << endl;
    return 0;
}
Copy the code

Exception class: TC_Exception

class TC_Exception : public exception
{  
    /** * @brief constructor, which provides a constructor that can be passed to errno, ** @param buffer error message ** @param err error code, strerror message */
    TC_Exception(const string &buffer, int err);  
} 
Copy the code

conclusion

This paper introduces and analyzes the common basic components of TARS framework implemented in C++, deepens the understanding of these basic components of tool classes, reduces the problems in the process of using these components, and improves the development efficiency.

TARS can quickly build systems and automatically generate code with ease of use and high performance in mind, helping developers and enterprises quickly build their own stable and reliable distributed applications in a microservice manner, allowing developers to focus only on business logic and improve operational efficiency. Multi-language, agile development, high availability, and efficient operations make TARS an enterprise-class product.

TARS micro services to help you digital transformation, welcome to visit:

TARS website: TarsCloud.org

TARS source: github.com/TarsCloud

Get the TARS Official Training Ebook: wj.qq.com/s2/6570357/…

Or scan code to obtain: