netpoll cpp
1.0.0
Il s'agit d'une bibliothèque de réseau C++ NIO avec des performances et une facilité d'utilisation. Il prend en charge C++14 et plus et s'étend sur trois principales plates-formes grand public.
Le modèle IO sous-jacent emprunte one loop per thread de la bibliothèque de réseau Muduo, ce qui rend l'encapsulation d'API en filetage plus efficace et plus simple.
L'interface API fournie par la couche supérieure emprunte de la bibliothèque de réseau Nio Netpoll de Language Open Source de ByTedance, et résume Listener et Dialer pour enfin fournir des services via EventLoop .
Pour la façon dont cette bibliothèque est utilisée, vous pouvez vérifier le démarrage rapide.
Pour des exemples d'utilisation spécifiques, veuillez consulter des exemples.
Déjà pris en charge :
windows/linux/macos est implémentée à l'aide de l'implémentation de multiplexage des performances la plus élevée de la plate-forme (Epoll / Epoll / Kqueue implémentée par IOCP).EventLoop pour implémenter des appels asynchrones. Si le système d'exploitation prend en charge les appels sendfile , l'appel à copie zéro sera appelé au lieu d'appeler la bibliothèque standard C.echo , vous n'avez besoin que du code suivant: struct server
{
NETPOLL_TCP_MESSAGE (conn, buffer){conn-> send (buffer-> readAll ());}
};
int main ()
{
auto loop = netpoll::NewEventLoop ();
auto listener = netpoll::tcp::Listener::New ({ 6666 });
listener. bind <server>();
loop. serve (listener);
}100^8s , qui peut nécessiter que 100*8byte ). Pour plus de détails, vous pouvez consulter le blog: introduction détaillée.one loop per threadfuture pour la synchronisation, qui peut être utile dans la programmation client. netpoll::SignalTask::Register ({SIGINT,SIGTERM});
netpoll::SignalTask::Add ([](){
// do something
});Le soutiendra à l'avenir :
Les performances sont extrêmement élevées et j'ai temporairement testé ASIO (C ++) / NetTy (Java) / NetPoll (GO Language).
J'ai testé Asio / Netty sur Windows System, et le test du graphique suivant est basé sur Linux, et je ne suis pas très bon pour déployer des programmes Java sur Linux, donc le tableau suivant n'a pas les performances de Netty.
La latence moyenne d'un seul service d'écho connecté dans différentes situations de concurrence est la suivante (AVG):
L'architecture du modèle sous-jacent du côté serveur:
L'encapsulation Listener simplifie l'utilisation de TcpServer , tous les appels sont les suivants:
namespace netpoll ::tcp{
class Listener {
template < typename T>
void bind (Args &&...args); // 用于绑定任意类的对应方法到回调
template < typename T>
std::shared_ptr<T> instance () // 返回内部帮你创建的实例
static Listener New( const InetAddress &address,
const StringView &name = " tcp_listener " ,
bool reUseAddr = true , bool reUsePort = true ); // 建立Listener实例
void enableKickoffIdle ( size_t timeout); // 用于开启剔除空闲连接
}
} L'encapsulation Dialer simplifie l'utilisation de TcpClient , et les appels utilisés sont les suivants:
namespace netpoll ::tcp{
class Dialer {
template < typename T>
void bind (Args &&...args); // 用于绑定任意类的对应方法到回调
template < typename T>
std::shared_ptr<T> instance () // 返回内部帮你创建的实例
static Dialer New( const InetAddress &address,
const StringView &name = " tcp_dialer " ); // 建立Listener实例
void enableRetry (); // 在连接失败后重试
// 以下调用方均是为了在没有C++17的if constexpr情况下的替代品,否则应该直接使用bind
template < typename T, typename ... Args>
static std::shared_ptr<T> Register (Args &&...args);
void onMessage (netpoll::RecvMessageCallback const &cb);
void onConnection (netpoll::ConnectionCallback const &cb);
void onConnectionError (netpoll::ConnectionErrorCallback const &cb);
void onWriteComplete (netpoll::WriteCompleteCallback const &cb);
}
} La stratégie d'équilibrage de charge d' EventLoop n'a pas de paramètres distincts, et ils utilisent tous Round Robin .
Les API liées à Eventloop sont les suivantes:
NewEventLoop(size_t threadNum=2,const netpoll::StringView&name="eventloop") : créez une instance EventLoop et définissez le nombre de threads pour EventLoop.
Méthode serve : Après avoir quitté le numéroteur ou l'auditeur nouvellement, vous pouvez leur fournir des services via cette méthode.
Méthode serveAsDaemon : L'effet est le même que la méthode de service, mais l'ouverture d'un nouveau thread à servir ne bloquera pas le thread actuel.
Méthode enableQuit : permet aux méthodes de sortie de l'appel actif. Par défaut, il ne peut pas quitter activement tous les threads de boucle et est utilisé conjointement avec la méthode QuitAllEventLoop .
Méthode QuitAllEventLoop : Quittez toutes les boucles.
MessageBuffer est un cache intermédiaire pour lire et écrire des données de tampon de noyau. Il s'agit en fait d'un tampon variable, et il est également très simple à implémenter. Pour différents types d'implémentations de tampons, vous pouvez consulter mon article: Implémentation de la longueur variable et du tampon de longueur immuable
Je ne décrirai pas divers appels ici. Si vous souhaitez savoir, vous pouvez dire directement le fichier d'en-tête correspondant: netpoll / util / message_buffer.h.
Permettez-moi de parler brièvement des points forts de la mise en œuvre de ce tampon:
Lors de l'expansion, en évitant les effets secondaires du redimensionnement, il peut également simplifier les opérations d'ouverture et de copie de la mémoire.
void MessageBuffer::ensureWritableBytes ( size_t len)
{
if ( writableBytes () >= len) return ;
// move readable bytes
if (m_head + writableBytes () >= (len + kBufferOffset ))
{
std::copy ( begin () + m_head, begin () + m_tail, begin () + kBufferOffset );
m_tail = kBufferOffset + (m_tail - m_head);
m_head = kBufferOffset ;
return ;
}
// create new buffer
size_t newLen;
if ((m_buffer. size () * 2 ) > ( kBufferOffset + readableBytes () + len))
newLen = m_buffer. size () * 2 ;
else newLen = kBufferOffset + readableBytes () + len;
// Avoid the inefficiency of using resize
MessageBuffer newbuffer (newLen);
newbuffer. pushBack (* this );
swap (newbuffer);
}Fournissez la méthode readfd, qui lit les données du tampon de lecture FD correspondant au message MessageBuffer. Le contenu se lit à chaque fois est assez grand. Par exemple, si la zone écrite MessageBuffer est inférieure à 8 Ko, le cache de lecture alternatif de 8KB est activé.
ssize_t MessageBuffer::readFd ( int fd, int *retErrno)
{
char extBuffer[ 8192 ];
struct iovec vec[ 2 ];
size_t writable = writableBytes ();
vec[ 0 ]. iov_base = begin () + m_tail;
vec[ 0 ]. iov_len = static_cast < int >(writable);
vec[ 1 ]. iov_base = extBuffer;
vec[ 1 ]. iov_len = sizeof (extBuffer);
const int iovcnt = (writable < sizeof extBuffer) ? 2 : 1 ;
ssize_t n = :: readv (fd, vec, iovcnt);
if (n < 0 ) { *retErrno = errno; }
else if ( static_cast < size_t >(n) <= writable) { m_tail += n; }
else
{
m_tail = m_buffer. size ();
pushBack ({extBuffer, n - writable});
}
return n;
}La classe TCPConnection est une classe abstraite utilisée via des pointeurs intelligents lorsqu'il est utilisé.
Cette interface spécifie les fonctions suivantes:
Envoyer des données (y compris String / Buffer / File / Stream).
/* *
* @brief Send some data to the peer.
*
* @param msg
* @param len
*/
virtual void send (StringView const &msg) = 0;
virtual void send ( const MessageBuffer &buffer) = 0;
virtual void send (MessageBuffer &&buffer) = 0;
virtual void send ( const std::shared_ptr<MessageBuffer> &msgPtr) = 0;
/* *
* @brief Send a file to the peer.
*
* @param fileName in UTF-8
* @param offset
* @param length
*/
virtual void sendFile (StringView const &fileName, size_t offset = 0 ,
size_t length = 0 ) = 0;
/* *
* @brief Send a stream to the peer.
*
* @param callback function to retrieve the stream data (stream ends when a
* zero size is returned) the callback will be called with nullptr when the
* send is finished/interrupted, so that it cleans up any internal data (ex:
* close file).
* @warning The buffer size should be >= 10 to allow http chunked-encoding
* data stream
*/
// (buffer, buffer size) -> size
// of data put in buffer
virtual void sendStream (
std::function<std:: size_t ( char *, std:: size_t )> callback) = 0;Obtenez des informations de connexion, telles que les informations d'adresse ou l'état de connexion ou un tampon qui reçoit des données.
/* *
* @brief New the local address of the connection.
*
* @return const InetAddress&
*/
virtual const InetAddress & localAddr () const = 0;
/* *
* @brief New the remote address of the connection.
*
* @return const InetAddress&
*/
virtual const InetAddress & peerAddr () const = 0;
/* *
* @brief Return true if the connection is established.
*
* @return true
* @return false
*/
virtual bool connected () const = 0;
/* *
* @brief Return false if the connection is established.
*
* @return true
* @return false
*/
virtual bool disconnected () const = 0;
/* *
* @brief New the buffer in which the received data stored.
*
* @return MsgBuffer*
*/
virtual MessageBuffer * getRecvBuffer () = 0;Définissez le rappel ou l'état de connexion (déconnectez ou définissez TCPNodeLay / Keepalive).
/* *
* @brief Set the high water mark callback
*
* @param cb The callback is called when the data in sending buffer is
* larger than the water mark.
* @param markLen The water mark in bytes.
*/
virtual void setHighWaterMarkCallback ( const HighWaterMarkCallback &cb,
size_t markLen) = 0;
/* *
* @brief Set the TCP_NODELAY option to the socket.
*
* @param on
*/
virtual void setTcpNoDelay ( bool on) = 0;
/* *
* @brief Shutdown the connection.
* @note This method only closes the writing direction.
*/
virtual void shutdown () = 0;
/* *
* @brief Close the connection forcefully.
*
*/
virtual void forceClose () = 0;
/* *
* @brief Call this method to avoid being kicked off by TcpServer, refer to
* the kickoffIdleConnections method in the TcpServer class.
*
*/
virtual void keepAlive () = 0;
/* *
* @brief Return true if the keepAlive() method is called.
*
* @return true
* @return false
*/
virtual bool isKeepAlive () = 0;Définit le contexte de la connexion pour gérer la logique commerciale dédiée pour la connexion.
/* *
* @brief Set the custom data on the connection.
*
* @param context
*/
void setContext ( const Any &context) { m_context = context; }
void setContext (Any &&context) { m_context = std::move (context); }
/* *
* @brief New mutable context
*
* @return Any
*/
Any & getContextRefMut () { return m_context; }
/* *
* @brief New unmutable context
*
* @return Any
*/
Any const & getContextRef () const { return m_context; }
/* *
* @brief Return true if the custom data is set by user.
*
* @return true
* @return false
*/
bool hasContext () const
{
# if __cplusplus >= 201703L
return m_context. has_value ();
# else
return m_context. empty ();
# endif
}
/* *
* @brief Clear the custom data.
*
*/
void clearContext ()
{
# if __cplusplus >= 201703L
m_context. reset ();
# else
m_context. clear ();
# endif
}La connexion accumule la quantité de données envoyées et reçues par les données.
/* *
* @brief Return the number of bytes sent
*
* @return size_t
*/
virtual size_t bytesSent () const = 0;
/* *
* @brief Return the number of bytes received.
*
* @return size_t
*/
virtual size_t bytesReceived () const = 0;Obtenez la boucle correspondante pour cette connexion.
/* *
* @brief New the event loop in which the connection I/O is handled.
*
* @return EventLoop*
*/
virtual EventLoop * getLoop () = 0;Les deux cas d'utilisation suivants sont temporairement fournis:
