ThreadedSocketAcceptor.cpp
Go to the documentation of this file.
/****************************************************************************
** Copyright (c) 2001-2014
**
** This file is part of the QuickFIX FIX Engine
**
** This file may be distributed under the terms of the quickfixengine.org
** license as defined by quickfixengine.org and appearing in the file
** LICENSE included in the packaging of this file.
**
** This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
** WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
**
** See http://www.quickfixengine.org/LICENSE for licensing information.
**
** Contact ask@quickfixengine.org if any conditions of this licensing are
** not clear to you.
**
****************************************************************************/
#ifdef _MSC_VER
#include "stdafx.h"
#else
#include "config.h"
#endif
#include "ThreadedSocketAcceptor.h"
#include "Settings.h"
#include "Utility.h"
namespace FIX
{
ThreadedSocketAcceptor::ThreadedSocketAcceptor(
Application& application,
MessageStoreFactory& factory,
const SessionSettings& settings ) throw( ConfigError )
: Acceptor( application, factory, settings )
{ socket_init(); }
ThreadedSocketAcceptor::ThreadedSocketAcceptor(
Application& application,
MessageStoreFactory& factory,
const SessionSettings& settings,
LogFactory& logFactory ) throw( ConfigError )
: Acceptor( application, factory, settings, logFactory )
{
socket_init();
}
ThreadedSocketAcceptor::~ThreadedSocketAcceptor()
{
socket_term();
}
void ThreadedSocketAcceptor::onConfigure( const SessionSettings& s )
throw ( ConfigError )
{
std::set<SessionID> sessions = s.getSessions();
std::set<SessionID>::iterator i;
for( i = sessions.begin(); i != sessions.end(); ++i )
{
const Dictionary& settings = s.get( *i );
settings.getInt( SOCKET_ACCEPT_PORT );
if( settings.has(SOCKET_REUSE_ADDRESS) )
settings.getBool( SOCKET_REUSE_ADDRESS );
if( settings.has(SOCKET_NODELAY) )
settings.getBool( SOCKET_NODELAY );
}
}
void ThreadedSocketAcceptor::onInitialize( const SessionSettings& s )
throw ( RuntimeError )
{
short port = 0;
std::set<int> ports;
std::set<SessionID> sessions = s.getSessions();
std::set<SessionID>::iterator i = sessions.begin();
for( ; i != sessions.end(); ++i )
{
const Dictionary& settings = s.get( *i );
port = (short)settings.getInt( SOCKET_ACCEPT_PORT );
m_portToSessions[port].insert( *i );
if( ports.find(port) != ports.end() )
continue;
ports.insert( port );
const bool reuseAddress = settings.has( SOCKET_REUSE_ADDRESS ) ?
settings.getBool( SOCKET_REUSE_ADDRESS ) : true;
const bool noDelay = settings.has( SOCKET_NODELAY ) ?
settings.getBool( SOCKET_NODELAY ) : false;
const int sendBufSize = settings.has( SOCKET_SEND_BUFFER_SIZE ) ?
settings.getInt( SOCKET_SEND_BUFFER_SIZE ) : 0;
const int rcvBufSize = settings.has( SOCKET_RECEIVE_BUFFER_SIZE ) ?
settings.getInt( SOCKET_RECEIVE_BUFFER_SIZE ) : 0;
int socket = socket_createAcceptor( port, reuseAddress );
if( socket < 0 )
{
SocketException e;
socket_close( socket );
throw RuntimeError( "Unable to create, bind, or listen to port "
+ IntConvertor::convert( (unsigned short)port ) + " (" + e.what() + ")" );
}
if( noDelay )
socket_setsockopt( socket, TCP_NODELAY );
if( sendBufSize )
socket_setsockopt( socket, SO_SNDBUF, sendBufSize );
if( rcvBufSize )
socket_setsockopt( socket, SO_RCVBUF, rcvBufSize );
m_socketToPort[socket] = port;
m_sockets.insert( socket );
}
}
void ThreadedSocketAcceptor::onStart()
{
Sockets::iterator i;
for( i = m_sockets.begin(); i != m_sockets.end(); ++i )
{
Locker l( m_mutex );
int port = m_socketToPort[*i];
AcceptorThreadInfo* info = new AcceptorThreadInfo( this, *i, port );
thread_id thread;
thread_spawn( &socketAcceptorThread, info, thread );
addThread( *i, thread );
}
}
bool ThreadedSocketAcceptor::onPoll( double timeout )
{
return false;
}
void ThreadedSocketAcceptor::onStop()
{
SocketToThread threads;
SocketToThread::iterator i;
{
Locker l(m_mutex);
time_t start = 0;
time_t now = 0;
::time( &start );
while ( isLoggedOn() )
{
if( ::time(&now) -5 >= start )
break;
}
threads = m_threads;
m_threads.clear();
}
for ( i = threads.begin(); i != threads.end(); ++i )
socket_close( i->first );
for ( i = threads.begin(); i != threads.end(); ++i )
thread_join( i->second );
}
void ThreadedSocketAcceptor::addThread( int s, thread_id t )
{
Locker l(m_mutex);
m_threads[ s ] = t;
}
void ThreadedSocketAcceptor::removeThread( int s )
{
Locker l(m_mutex);
SocketToThread::iterator i = m_threads.find( s );
if ( i != m_threads.end() )
{
thread_detach( i->second );
m_threads.erase( i );
}
}
THREAD_PROC ThreadedSocketAcceptor::socketAcceptorThread( void* p )
{
AcceptorThreadInfo * info = reinterpret_cast < AcceptorThreadInfo* > ( p );
ThreadedSocketAcceptor* pAcceptor = info->m_pAcceptor;
int s = info->m_socket;
int port = info->m_port;
delete info;
int noDelay = 0;
int sendBufSize = 0;
int rcvBufSize = 0;
socket_getsockopt( s, TCP_NODELAY, noDelay );
socket_getsockopt( s, SO_SNDBUF, sendBufSize );
socket_getsockopt( s, SO_RCVBUF, rcvBufSize );
int socket = 0;
while ( ( !pAcceptor->isStopped() && ( socket = socket_accept( s ) ) >= 0 ) )
{
if( noDelay )
socket_setsockopt( socket, TCP_NODELAY );
if( sendBufSize )
socket_setsockopt( socket, SO_SNDBUF, sendBufSize );
if( rcvBufSize )
socket_setsockopt( socket, SO_RCVBUF, rcvBufSize );
Sessions sessions = pAcceptor->m_portToSessions[port];
ThreadedSocketConnection * pConnection =
new ThreadedSocketConnection
( socket, sessions, pAcceptor->getLog() );
ConnectionThreadInfo* info = new ConnectionThreadInfo( pAcceptor, pConnection );
{
Locker l( pAcceptor->m_mutex );
std::stringstream stream;
stream << "Accepted connection from " << socket_peername( socket ) << " on port " << port;
if( pAcceptor->getLog() )
pAcceptor->getLog()->onEvent( stream.str() );
thread_id thread;
if ( !thread_spawn( &socketConnectionThread, info, thread ) )
delete info;
pAcceptor->addThread( socket, thread );
}
}
if( !pAcceptor->isStopped() )
pAcceptor->removeThread( s );
return 0;
}
THREAD_PROC ThreadedSocketAcceptor::socketConnectionThread( void* p )
{
ConnectionThreadInfo * info = reinterpret_cast < ConnectionThreadInfo* > ( p );
ThreadedSocketAcceptor* pAcceptor = info->m_pAcceptor;
ThreadedSocketConnection* pConnection = info->m_pConnection;
delete info;
int socket = pConnection->getSocket();
while ( pConnection->read() ) {}
delete pConnection;
if( !pAcceptor->isStopped() )
pAcceptor->removeThread( socket );
return 0;
}
}
Settings.h
FIX::Acceptor::isLoggedOn
bool isLoggedOn()
Check to see if any sessions are currently logged on.
Definition: Acceptor.cpp:230
FIX::thread_join
void thread_join(thread_id thread)
Definition: Utility.cpp:415
FIX::socket_getsockopt
int socket_getsockopt(int s, int opt, int &optval)
Definition: Utility.cpp:211
FIX::ThreadedSocketConnection::getSocket
int getSocket() const
Definition: ThreadedSocketConnection.h:56
FIX::socket_accept
int socket_accept(int s)
Definition: Utility.cpp:147
FIX::ThreadedSocketAcceptor::onInitialize
void onInitialize(const SessionSettings &)
Implemented to initialize acceptor.
Definition: ThreadedSocketAcceptor.cpp:70
FIX::ThreadedSocketAcceptor::m_socketToPort
SocketToPort m_socketToPort
Definition: ThreadedSocketAcceptor.h:91
FIX::socket_init
void socket_init()
Definition: Utility.cpp:81
FIX::ThreadedSocketAcceptor::removeThread
void removeThread(int s)
Definition: ThreadedSocketAcceptor.cpp:175
FIX::ThreadedSocketAcceptor::onConfigure
void onConfigure(const SessionSettings &)
Implemented to configure acceptor.
Definition: ThreadedSocketAcceptor.cpp:54
FIX::ThreadedSocketAcceptor::ConnectionThreadInfo::m_pConnection
ThreadedSocketConnection * m_pConnection
Definition: ThreadedSocketAcceptor.h:66
FIX::ThreadedSocketConnection
Encapsulates a socket file descriptor (multi-threaded).
Definition: ThreadedSocketConnection.h:44
FIX::socket_term
void socket_term()
Definition: Utility.cpp:96
Utility.h
FIX::thread_detach
void thread_detach(thread_id thread)
Definition: Utility.cpp:425
FIX::SOCKET_NODELAY
const char SOCKET_NODELAY[]
Definition: SessionSettings.h:66
FIX::SOCKET_ACCEPT_PORT
const char SOCKET_ACCEPT_PORT[]
Definition: SessionSettings.h:62
FIX::ThreadedSocketAcceptor::onStop
void onStop()
Implemented to stop a running acceptor.
Definition: ThreadedSocketAcceptor.cpp:140
FIX::SOCKET_SEND_BUFFER_SIZE
const char SOCKET_SEND_BUFFER_SIZE[]
Definition: SessionSettings.h:67
FIX::thread_spawn
bool thread_spawn(THREAD_START_ROUTINE func, void *var, thread_id &thread)
Definition: Utility.cpp:394
FIX::Dictionary::has
bool has(const std::string &) const
Check if the dictionary contains a value for key.
Definition: Dictionary.cpp:150
FIX::ThreadedSocketAcceptor::socketAcceptorThread
static THREAD_PROC socketAcceptorThread(void *p)
Definition: ThreadedSocketAcceptor.cpp:186
FIX::ThreadedSocketAcceptor::m_threads
SocketToThread m_threads
Definition: ThreadedSocketAcceptor.h:92
FIX::IntConvertor::convert
static std::string convert(signed_int value)
Definition: FieldConvertors.h:154
FIX::ThreadedSocketConnection::read
bool read()
Definition: ThreadedSocketConnection.cpp:91
FIX::SOCKET_RECEIVE_BUFFER_SIZE
const char SOCKET_RECEIVE_BUFFER_SIZE[]
Definition: SessionSettings.h:68
FIX::socket_setsockopt
int socket_setsockopt(int s, int opt)
Definition: Utility.cpp:186
FIX
Definition: Acceptor.cpp:34
FIX::ThreadedSocketAcceptor::SocketToThread
std::map< int, thread_id > SocketToThread
Definition: ThreadedSocketAcceptor.h:75
FIX::Application
This interface must be implemented to define what your FIX application does.
Definition: Application.h:43
FIX::ThreadedSocketAcceptor::~ThreadedSocketAcceptor
virtual ~ThreadedSocketAcceptor()
Definition: ThreadedSocketAcceptor.cpp:49
FIX::Acceptor::start
void start()
Start acceptor.
Definition: Acceptor.cpp:158
FIX::RuntimeError
Application encountered serious error during runtime
Definition: Exceptions.h:94
FIX::ConfigError
Application is not configured correctly
Definition: Exceptions.h:87
FIX::Locker
Locks/Unlocks a mutex using RAII.
Definition: Mutex.h:95
FIX::ThreadedSocketAcceptor::AcceptorThreadInfo::m_pAcceptor
ThreadedSocketAcceptor * m_pAcceptor
Definition: ThreadedSocketAcceptor.h:54
FIX::SessionSettings
Container for setting dictionaries mapped to sessions.
Definition: SessionSettings.h:130
FIX::ThreadedSocketAcceptor::onPoll
bool onPoll(double timeout)
Implemented to connect and poll for events.
Definition: ThreadedSocketAcceptor.cpp:135
FIX::ThreadedSocketAcceptor::ConnectionThreadInfo
Definition: ThreadedSocketAcceptor.h:59
FIX::Acceptor
Base for classes which act as an acceptor for incoming connections.
Definition: Acceptor.h:49
FIX::Acceptor::Sessions
std::map< SessionID, Session *> Sessions
Definition: Acceptor.h:110
FIX::ThreadedSocketAcceptor::socketConnectionThread
static THREAD_PROC socketConnectionThread(void *p)
Definition: ThreadedSocketAcceptor.cpp:242
FIX::ThreadedSocketAcceptor::ThreadedSocketAcceptor
ThreadedSocketAcceptor(Application &, MessageStoreFactory &, const SessionSettings &)
Definition: ThreadedSocketAcceptor.cpp:32
FIX::ThreadedSocketAcceptor::m_portToSessions
PortToSessions m_portToSessions
Definition: ThreadedSocketAcceptor.h:90
FIX::LogFactory
This interface must be implemented to create a Log.
Definition: Log.h:42
FIX::MessageStoreFactory
This interface must be implemented to create a MessageStore.
Definition: MessageStore.h:41
FIX::ThreadedSocketAcceptor::addThread
void addThread(int s, thread_id t)
Definition: ThreadedSocketAcceptor.cpp:168
FIX::ThreadedSocketAcceptor
Threaded Socket implementation of Acceptor.
Definition: ThreadedSocketAcceptor.h:36
FIX::SocketException
Socket Error.
Definition: Exceptions.h:245
FIX::socket_createAcceptor
int socket_createAcceptor(int port, bool reuse)
Definition: Utility.cpp:103
FIX::ThreadedSocketAcceptor::AcceptorThreadInfo
Definition: ThreadedSocketAcceptor.h:49
FIX::thread_id
pthread_t thread_id
Definition: Utility.h:164
FIX::ThreadedSocketAcceptor::AcceptorThreadInfo::m_port
int m_port
Definition: ThreadedSocketAcceptor.h:56
FIX::Dictionary
For storage and retrieval of key/value pairs.
Definition: Dictionary.h:36
FIX::ThreadedSocketAcceptor::m_mutex
Mutex m_mutex
Definition: ThreadedSocketAcceptor.h:93
FIX::SOCKET_REUSE_ADDRESS
const char SOCKET_REUSE_ADDRESS[]
Definition: SessionSettings.h:63
FIX::ThreadedSocketAcceptor::AcceptorThreadInfo::m_socket
int m_socket
Definition: ThreadedSocketAcceptor.h:55
FIX::socket_close
void socket_close(int s)
Definition: Utility.cpp:158
FIX::Dictionary::getBool
bool getBool(const std::string &) const
Get a value as a bool.
Definition: Dictionary.cpp:71
ThreadedSocketAcceptor.h
THREAD_PROC
#define THREAD_PROC
Definition: Utility.h:158
FIX::Dictionary::getInt
int getInt(const std::string &) const
Get a value as a int.
Definition: Dictionary.cpp:45
FIX::socket_peername
const char * socket_peername(int socket)
Definition: Utility.cpp:331
FIX::ThreadedSocketAcceptor::m_sockets
Sockets m_sockets
Definition: ThreadedSocketAcceptor.h:89
FIX::ThreadedSocketAcceptor::onStart
void onStart()
Implemented to start listening for connections.
Definition: ThreadedSocketAcceptor.cpp:121
FIX::ThreadedSocketAcceptor::ConnectionThreadInfo::m_pAcceptor
ThreadedSocketAcceptor * m_pAcceptor
Definition: ThreadedSocketAcceptor.h:65