.NET et dimensionnement d'applications serveur

Cyrille "cns" Szymanski wrote:

Je me suis dit que j'allais profiter de .NET pour voir si
l'utilisation des "io completion ports" (IOCP) et "asynchronous io"
(AIO) est facilitée. J'ai été agréablement surpris pas la simplicité
de la chose, ThreadPool et autres nombreuses BeginXXX/EndXXX...

Alors il est temps pour un petit "benchmarking".

Un truc m'échappe : tu parles des IOCP mais dans ton test "natif" tu ne les
utilises pas.
Quand à .NET, les IOCP ne sont utilisés dans ton exemple que pour les
BeginReceive et BeginSend.

<snip>

Annexe :
========= > Voici quelques détails sur le déroulement des tests. Si le code source
vous intéresse, faites-moi signe.

Signe!!

Serveur .NET :
----------------
Un thread boucle sur des accept au moyen de la classe TCPListener.
A chaque connexion, ThreadPool.QueueUserWorkItem().

Pourquoi faire un QueueUserWorkItem ici? Il faudrait appeler BeginReceive
directement depuis ton thread qui boucle sur les accept. Ici tu rajoutes un
niveau d'indirection de plus, ce qui pourrait expliquer les écarts obtenus.

Le traitement des données est fait come suit :
Un buffer de 1024 octets par client est alloué,
BeginReceive/EndReceive le remplit, BeginSend/EndSend est ensuite
appelé sur ce buffer et ainsi de suite.


Serveur Win32 :
-----------------
Le thread principal boucle sur accept().
A chaque connexion un thread est créé et la SOCKET est passée en
paramètre (CreateThread())
Le traitement des données est fait comme suit :
Un buffer de 1024 octets est alloué, puis il est rempli par un recv()
bloquant, et vidé par un send() bloquant et ainsi de suite.


Programme de tests :
----------------------
Ce programme crée NB_THREADS threads. Chaque thread répète PASSES fois
les opérations suivantes (dans l'ordre) :
-> crée NB_SOCKETS SOCKETs
-> connecte toutes les SOCKETs
-> NB_QUOTES fois :
-> envoie un message sur tous les SOCKETs
-> reçoit le message sur tous les SOCKETs


J'ai choisi NB_THREADS=5, NB_SOCKETS€ ce qui fait typiquement 400
clients simultanés, NB_QUOTES ce qui fait environ 1000 octets
envoyés par client, et PASSESP (chaque thread transfère 4 Mo).
Au total 40 Mo sont envoyés, ce qui sur une connexion 100 Mb/s ne
devrait pas prendre plus de 10 sec (4 sec étant le minimum théorique).

Ca ca ne veut rien dire, le débit sur un lien ethernet est tout sauf
prévisible ;-)

Quelques remarques en passant :
- je ferais le test en local pour commencer, pour essayer de m'abstraire le
plus possible des aléas du réseau.
- Est-ce que tu as fais tourner plusieurs fois le serveur .NET, ou à défaut
est-ce que tu l'as passé à ngen? Sinon, la JIT-compilation fausse
complètement ton test.

Arnaud

Arnaud Debaene wrote:

> Annexe :
> ========= > > Voici quelques détails sur le déroulement des tests. Si le code
> source vous intéresse, faites-moi signe.
Signe!!

Signe aussi! (mon adresse e-mail est valide)

Merci!


--
Quentin Pouplard (Tene/MyOE)
http://www.myoe.org | http://graff.alrj.org

> Signe aussi! (mon adresse e-mail est valide)

Merci!

Merci pour vos réponses.
Le code n'étant pas très long, je vais le poster ici. Si vous pouvez
m'aider à le rendre plus performant je vous en serai reconnaissant. Par
contre je ne cherche pas à rentrer dans de l'optimisation fine, ce n'est
pas le but du test (et on peut y passer des années).

J'ai maintenant écrit le serveur natif IOCP pour le comparer (il a aussi
l'avantage d'être plus stable que le serveur naïf), ainsi que deux
serveurs Java, l'un utilisant "java.io" et l'autre les nouvelles fonction
asynchrones "java.nio".


Voici le code tu programme de tests :

#include <stdio.h>
#include <windows.h>
#include <winsock2.h>

void die( char *msg )
{
printf( "ERROR: %sn", msg );
WSACleanup();
exit(-1);
}

#define NB_THREADS 2

#define NB_SOCKETS 500

#define PASSES 5

#define NB_QUOTES 50

#define QUOTE "If I were a bird, I'd fly into a ceiling fan."

DWORD WINAPI tester( LPVOID lpParam )
{
u_long ip = (u_long)lpParam;

int err;

// socket pool
SOCKET sd[NB_SOCKETS];

// message buffer
int quotelen = strlen(QUOTE)+1;
char *buf = (char*)malloc( quotelen );
if( buf==0 ) {
die(0);
}

for( int nb=0; nb<PASSES; nb++ )
{
// create sockets
for( int i=0; i<NB_SOCKETS; ++i ) {
sd[i] = socket( AF_INET, SOCK_STREAM, 0 );
if( sd[i]==INVALID_SOCKET ) {
die("socket()n");
}
}

// connect
SOCKADDR_IN sa;

for( int i=0; i<NB_SOCKETS; ++i ) {
sa.sin_family = AF_INET;
sa.sin_port = htons( 1234 );
sa.sin_addr.S_un.S_addr = ip;
err = connect( sd[i], (SOCKADDR*)&sa, sizeof(sa) );
if( err!=0 ) {
printf("!");
closesocket( sd[i] );
sd[i] = INVALID_SOCKET;
}
}

for( int j=0; j<NB_QUOTES; j++ )
{
// send
for( int i=0; i<NB_SOCKETS; ++i ) {
if( sd[i]==INVALID_SOCKET ) continue;
int sent = send( sd[i], QUOTE, quotelen, 0 );
if( sent<=0 ) {
printf("#");
closesocket( sd[i] );
sd[i] = INVALID_SOCKET;
}
}

// recv
for( int i=0; i<NB_SOCKETS; ++i ) {
if( sd[i]==INVALID_SOCKET ) continue;
int read = recv( sd[i], buf, quotelen, 0 );
if( read<=0 ) {
printf("$");
closesocket( sd[i] );
sd[i] = INVALID_SOCKET;
}
}
}

// close
for( int i=0; i<NB_SOCKETS; ++i ) {
if( sd[i]==INVALID_SOCKET ) continue;
closesocket( sd[i] );
}

printf(".");
Sleep(0);
}

return 0;
}


int main( int argc, char *argv[] )
{
int err;

WSAData data;
err = WSAStartup( MAKEWORD(1,1), &data );
if( err!=0 ) {
die("WSAStartup()n");
}

if( argc<2 ) {
die("not enough parameters");
}

u_long ip = inet_addr( argv[1] );

for( int i=0; i<NB_THREADS; ++i )
{
DWORD threadid;
HANDLE hthread = CreateThread( NULL, 4096, tester, (LPVOID)ip, 0,
&threadid );
if( hthread==NULL ) {
die("CreateThread()");
}
CloseHandle( hthread );
Sleep(1000);
}

printf("nTesting %d clients, %d passes : %d requestsn",
NB_THREADS*NB_SOCKETS, PASSES, NB_THREADS*NB_SOCKETS*PASSES );

getchar();

printf("nTested %d clients, %d passes : %d requestsn",
NB_THREADS*NB_SOCKETS, PASSES, NB_THREADS*NB_SOCKETS*PASSES );
}


--
_|_|_| CnS
_|_| for(n=0;b;n++)
_| b&=b-1; /*pp.47 K&R*/

> Un truc m'échappe : tu parles des IOCP mais dans ton test "natif" tu
ne les utilises pas.

Oui, le but était juste d'avoir un serveur de référence et pas de
comparer IOCP natif et IOCP .NET.

Quand à .NET, les IOCP ne sont utilisés dans ton exemple que pour les
BeginReceive et BeginSend.

Tu veux dire que je pourrais aussi y ajouter BeginAccept ? Quels
avantages y a-t-il par rapport à avoir un thread qui boucle sur accept()
plus les worker threads ?

Pourquoi faire un QueueUserWorkItem ici? Il faudrait appeler
BeginReceive directement depuis ton thread qui boucle sur les accept.
Ici tu rajoutes un niveau d'indirection de plus, ce qui pourrait
expliquer les écarts obtenus.

Oui, je vais voir ce que ça change.

Au total 40 Mo sont envoyés, ce qui sur une connexion 100 Mb/s ne
devrait pas prendre plus de 10 sec (4 sec étant le minimum
théorique).

Ca ca ne veut rien dire, le débit sur un lien ethernet est tout sauf
prévisible ;-)
Quelques remarques en passant :
- je ferais le test en local pour commencer, pour essayer de
m'abstraire le plus possible des aléas du réseau.

Alors là c'est justement le but. Je ne veux pas faire le test en local
pour deux raisons : ce n'est pas un test en condition et le programme de
test risque de perturber le serveur. Et le lien ethernet ajoute de
l'aléatoire au test ce qui évite que tous les packets arrivent bien
rangés dans le bon ordre etc.

- Est-ce que tu as fais tourner plusieurs fois le serveur .NET, ou à
défaut est-ce que tu l'as passé à ngen? Sinon, la JIT-compilation
fausse complètement ton test.

J'ai répété l'opération une dizaine de fois.

--
_|_|_| CnS
_|_| for(n=0;b;n++)
_| b&=b-1; /*pp.47 K&R*/

Voici le code du serveur IOCP natif :


#include <stdio.h>
#include <windows.h>
#include <winsock2.h>

void die( char *msg )
{
printf( "ERROR: %sn", msg );
WSACleanup();
exit(-1);
}

#define PACKET_MAX 1024
#define CPU 1
#define NB_WORKER_THREADS (2*CPU)

struct Client
{
WSAOVERLAPPED overlapped;
SOCKET client;
int state;

char buffer[PACKET_MAX];

Client( SOCKET sd ) {
state = 0;
client = sd;
}
};

bool iocpRecv( Client *context )
{
int err;
WSABUF buf = { PACKET_MAX, context->buffer };
DWORD bytesrecv;
DWORD flags=0;
bool ret = false;

err = WSARecv( context->client, &buf, 1, &bytesrecv, &flags, &
(context->overlapped), NULL );
if( err!=0 ) {
int error = WSAGetLastError();
if( error==ERROR_IO_PENDING ) {
ret = true;
} else {
printf("Unhandled recv error %dn", error );
//printf("!");
}
} else {
ret = true;
}

return ret;
}


bool iocpSend( Client *context, int to_send )
{
int err;
WSABUF buf = { to_send, context->buffer };
DWORD bytessent;
bool ret = false;

err = WSASend( context->client, &buf, 1, &bytessent, 0, &(context->
overlapped), NULL );
if( err!=0 ) {
int error = WSAGetLastError();
if( error==ERROR_IO_PENDING ) {
ret = true;
} else {
printf("Unhandled send error %dn", error );
//printf("#");
}
} else {
ret = true;
}

return ret;
}

DWORD WINAPI echo_worker_thread( LPVOID lpParam )
{
HANDLE iocp = (HANDLE)lpParam;
DWORD nb_bytes;
Client *context;
OVERLAPPED *overlapped;

//printf("Waiting for job...n");
while( GetQueuedCompletionStatus( iocp, &nb_bytes, (PDWORD)&context,
&overlapped, INFINITE ) )
{
int err;
int state = context->state;
//printf("Got job (%d) !n", state);

// state 0 is all data sent, get new one
if( state==0 )
{
if( iocpRecv( context )=úlse ) {
context->state = -1;
} else {
if( nb_bytes==0 ) {
context->state = -1;
} else {
context->state = 1;
}
}
}
// state 1 is client sent data
else if( state==1 )
{
if( iocpSend( context, nb_bytes )=úlse ) {
context->state = -1;
} else {
if( nb_bytes==0 ) {
context->state = -1;
} else {
context->state = 0;
}
}
}
// state -1 is error
else if( state==-1 )
{
closesocket( context->client );
free( context );
}
// unknown state !
else
{
printf( "Unknown state %dn", state );
}

//printf("Waiting for job...n");
}
return 0;
}


BOOL socket_iocp( SOCKET socket, HANDLE hCompletionPort, DWORD
dwCompletionKey )
{
HANDLE h = CreateIoCompletionPort((HANDLE) socket, hCompletionPort,
dwCompletionKey, 0);
return h==hCompletionPort;
}

int main( void )
{
int err;

// start winsock
WSAData data;
err = WSAStartup( MAKEWORD(1,1), &data );
if( err!=0 ) {
die("WSAStartup()n");
}

// Create an IOCP
HANDLE iocp = CreateIoCompletionPort(
INVALID_HANDLE_VALUE,
NULL,
NULL,
2);
if( iocp==INVALID_HANDLE_VALUE ) {
die("CreateIoCompletionPort()");
}

// create worker threads
for( int i=0; i<NB_WORKER_THREADS; ++i )
{
DWORD threadid;
HANDLE hthread = CreateThread( NULL, 1024, echo_worker_thread,
(LPVOID)iocp, 0, &threadid );
if( hthread==NULL ) {
die("CreateThread()");
}
Sleep(0);
CloseHandle( hthread );
}

// create listening socket
SOCKET sd = socket( AF_INET, SOCK_STREAM, 0 );
if( sd==INVALID_SOCKET ) {
die("socket()n");
}

// listen
SOCKADDR_IN sa;
sa.sin_family = AF_INET;
sa.sin_port = htons( 1234 );
sa.sin_addr.S_un.S_addr = INADDR_ANY;

err = bind( sd, (SOCKADDR*)&sa, sizeof(sa) );
if( err!=0 ) {
die("bind()");
}

err = listen( sd, SOMAXCONN );
if( err!=0 ) {
die("listen()");
}

printf("Readyn");

while(1) {
// accept incoming connexion
SOCKADDR sac;
int saclen = sizeof(sac);
SOCKET client = accept( sd, &sac, &saclen );
if( client==INVALID_SOCKET ) {
die("accept()");
}

// create client context
//printf("*");
Client *client_context = new Client( client );
if( client_context==NULL ) {
die( "malloc()" );
}

// add socket to the iocp
if( socket_iocp( client, iocp, (DWORD)client_context )=úlse ) {
die("socket_iocp()");
}

// queue job
if( PostQueuedCompletionStatus( iocp, -1, (DWORD)client_context,
NULL )=úlse ) {
die( "PostQueuedCompletionStatus()" );
}

err = listen( sd, SOMAXCONN );
if( err!=0 ) {
die("listen()");
}
}
}

--
_|_|_| CnS
_|_| for(n=0;b;n++)
_| b&=b-1; /*pp.47 K&R*/

Voici le code du serveur IOCP C# :

/* main.cs */

using System;
using System.Threading;

////
// main()
class main
{
public static int Main( string[] args )
{
Echo e = new Echo();
e.load( );

Console.WriteLine( "Press ^C to end Echo Server" );

while( true ) {
Console.Read();
}
}
}


/* echo.cs */

using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;

class Echo_client
{
// intrinsic client members
Socket client = null;

// client parameters

// client working members
int state = 0;

// io operations
AsyncCallback work_callback = null;
byte[] buffer;
int packet_max = 1024;

enum type { recv, send };

public Echo_client( Socket cli )
{
state = 0;
client = cli;
work_callback = new AsyncCallback(work);
}

public void init_work( object o )
{
buffer = new byte[packet_max];

// Asynchronous operation
client.BeginReceive( buffer, 0, packet_max, SocketFlags.None,
work_callback, type.recv );
state = 0;
}

public void work( IAsyncResult ar )
{
// state 0 is pending recv or send
if( state==0 )
{
if( !client.Connected )
{
// one end has disconected
state = 1;
}
if( (type)ar.AsyncState == type.recv )
{
// the client sent data
int read = client.EndReceive( ar );
if( read==0 ) {
state=1;
}
// send it back
client.BeginSend( buffer, 0, read, SocketFlags.None,
work_callback, type.send );
}
else if( (type)ar.AsyncState == type.send )
{
// finished sendind data to the server
int sent = client.EndSend( ar );
if( sent==0 ) {
state=1;
}
// get new bytes from the client
client.BeginReceive( buffer, 0, packet_max,
SocketFlags.None, work_callback, type.recv );
}
}
// state 1 is one end has closed connection
else if( state==1 )
{
client.Close();
}
}
}

class Echo
{
Thread main_thread = null;
ThreadStart main_threadstart = null;

public void run()
{
// loop on accept connexions
TcpListener listener = new TcpListener( IPAddress.Any, 1234 );

listener.Start();
while( true )
{
// connexion accepted
//Console.Write("Waiting for connection...");
Socket client_socket = listener.AcceptSocket();

// create a client
//Console.WriteLine("Connection accepted.");
Echo_client client = new Echo_client( client_socket );

// queue worker thread
//ThreadPool.QueueUserWorkItem(new WaitCallback
(client.init_work));
client.init_work( 0 );

// loop
}
}

////
// loads the module
public bool load( )
{
// launch the module in a separate thread
main_threadstart = new ThreadStart( run );
main_thread = new Thread( main_threadstart );
main_thread.Start();

// give the thread a chance to start
Thread.Sleep( 0 );

return true;
}

////
// cleanly unloads the module
public void unload()
{
main_thread.Abort();
}

}

--
_|_|_| CnS
_|_| for(n=0;b;n++)
_| b&=b-1; /*pp.47 K&R*/

Et enfin le code Java NIO :

import java.io.*;
import java.lang.*;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import java.util.*;

public class javaenh2
{
static class echo_worker extends Thread
{
Selector selector = null;

echo_worker() throws Exception {
selector = Selector.open();
}

public void queue_channel( SocketChannel channel, ByteBuffer
buffer ) throws Exception {
//System.out.println("reg'd");
channel.register( selector, SelectionKey.OP_READ, buffer );
}

public void new_alert() {
selector.wakeup();
}

public void run()
{
//System.out.println("runing");
try {
while( selector.select(1)>=0 )
{
// Get set of ready objects
Iterator readyItor = selector.selectedKeys().iterator();

// Walk through set
while( readyItor.hasNext() )
{
// Get key from set
SelectionKey key = (SelectionKey)readyItor.next();
readyItor.remove();

if( key.isReadable() )
{
//System.out.println("Got data");

// Get channel and context
SocketChannel keyChannel = (SocketChannel)
key.channel();
ByteBuffer buffer = (ByteBuffer)key.attachment();
buffer.clear();

// Get the data
if( keyChannel.read( buffer )==-1 ) {
//System.out.println("closed");
keyChannel.socket().close();
} else {
buffer.flip();

// Send the data
keyChannel.write( buffer );

// wait for data to be sent
keyChannel.register( selector,
SelectionKey.OP_WRITE, buffer );
}
}
else if( key.isWritable() )
{
//System.out.println("Data sent");

// Get channel and the context
SocketChannel keyChannel = (SocketChannel)
key.channel();
ByteBuffer buffer = (ByteBuffer)key.attachment();

// data sent, read again
keyChannel.register( selector,
SelectionKey.OP_READ, buffer );
}
else
{
System.err.println("Ooops");
}
}
}
} catch( Exception e ) {
System.out.println( e );
}
//System.out.println("finised");
}
}

public static void main(String args[]) throws Exception
{
// incoming connection channel
ServerSocketChannel channel = ServerSocketChannel.open();
InetSocketAddress isa = new InetSocketAddress( 1234 );
channel.socket().bind( isa );

echo_worker echo_thread = new echo_worker();
echo_thread.start();

// Wait for something of interest to happen
while( true )
{
SocketChannel clientChannel = channel.accept();
//System.out.println( "accpt" );

ByteBuffer buffer = ByteBuffer.allocateDirect( 1024 );

clientChannel.configureBlocking(false);
echo_thread.queue_channel( clientChannel, buffer );
//echo_thread.new_alert();
}

}
}

--
_|_|_| CnS
_|_| for(n=0;b;n++)
_| b&=b-1; /*pp.47 K&R*/

Le code Java io 1 thread/client :

import java.io.*;
import java.lang.*;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import java.util.*;

public class javaenh
{
//class echo_worker extends Thread
//{

public static void main(String args[]) throws Exception
{
// incoming connection channel
ServerSocketChannel channel = ServerSocketChannel.open();
channel.configureBlocking(false);
InetSocketAddress isa = new InetSocketAddress( 1234 );
channel.socket().bind( isa );

// Register interest in when connection
Selector selector = Selector.open();
channel.register( selector, SelectionKey.OP_ACCEPT );

// Wait for something of interest to happen
while( selector.select()>0 )
{
// Get set of ready objects
Iterator readyItor = selector.selectedKeys().iterator();

// Walk through set
while( readyItor.hasNext() )
{
// Get key from set
SelectionKey key = (SelectionKey)readyItor.next();
readyItor.remove();

if( key.isAcceptable() )
{
// Get channel
ServerSocketChannel keyChannel =
(ServerSocketChannel)key.channel();

// accept incoming connection
SocketChannel clientChannel = keyChannel.accept();

// create a client context
ByteBuffer buffer = ByteBuffer.allocateDirect( 1024
);

// Send hello message
//PrintWriter out = new PrintWriter
(clientChannel.socket().getOutputStream(), true);
//out.println("ECHO server ready");

// register it in the selector
clientChannel.configureBlocking(false);
clientChannel.register( selector,
SelectionKey.OP_READ, buffer );
}
else if( key.isReadable() )
{
//System.out.println("Got data");

// Get channel and context
SocketChannel keyChannel = (SocketChannel)key.channel
();
ByteBuffer buffer = (ByteBuffer)key.attachment();

// Get the data
if( keyChannel.read( buffer )==-1 ) {
//System.out.println("closed");
keyChannel.socket().close();
} else {
buffer.flip();

// Send the data
keyChannel.write( buffer );

// wait for data to be sent
keyChannel.register( selector,
SelectionKey.OP_WRITE, buffer );
}
}
else if( key.isWritable() )
{
//System.out.println("Data sent");

// Get channel and context
SocketChannel keyChannel = (SocketChannel)key.channel
();
ByteBuffer buffer = (ByteBuffer)key.attachment();

buffer.clear();

// data sent, read again
keyChannel.register( selector, SelectionKey.OP_READ,
buffer );
}
else
{
System.err.println("Ooops");
}
}
}
}
}

--
_|_|_| CnS
_|_| for(n=0;b;n++)
_| b&=b-1; /*pp.47 K&R*/