SocketMonitor.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 "SocketMonitor.h"
#include "Utility.h"
#include <exception>
#include <set>
#include <algorithm>
#include <iostream>
namespace FIX
{
SocketMonitor::SocketMonitor( int timeout )
: m_timeout( timeout )
{
socket_init();
std::pair<int, int> sockets = socket_createpair();
m_signal = sockets.first;
m_interrupt = sockets.second;
socket_setnonblock( m_signal );
socket_setnonblock( m_interrupt );
m_readSockets.insert( m_interrupt );
m_timeval.tv_sec = 0;
m_timeval.tv_usec = 0;
#ifndef SELECT_DECREMENTS_TIME
m_ticks = clock();
#endif
}
SocketMonitor::~SocketMonitor()
{
Sockets::iterator i;
for ( i = m_readSockets.begin(); i != m_readSockets.end(); ++i ) {
socket_close( *i );
}
socket_close( m_signal );
socket_term();
}
bool SocketMonitor::addConnect( int s )
{
socket_setnonblock( s );
Sockets::iterator i = m_connectSockets.find( s );
if( i != m_connectSockets.end() ) return false;
m_connectSockets.insert( s );
return true;
}
bool SocketMonitor::addRead( int s )
{
socket_setnonblock( s );
Sockets::iterator i = m_readSockets.find( s );
if( i != m_readSockets.end() ) return false;
m_readSockets.insert( s );
return true;
}
bool SocketMonitor::addWrite( int s )
{
if( m_readSockets.find(s) == m_readSockets.end() )
return false;
socket_setnonblock( s );
Sockets::iterator i = m_writeSockets.find( s );
if( i != m_writeSockets.end() ) return false;
m_writeSockets.insert( s );
return true;
}
bool SocketMonitor::drop( int s )
{
Sockets::iterator i = m_readSockets.find( s );
Sockets::iterator j = m_writeSockets.find( s );
Sockets::iterator k = m_connectSockets.find( s );
if ( i != m_readSockets.end() ||
j != m_writeSockets.end() ||
k != m_connectSockets.end() )
{
socket_close( s );
m_readSockets.erase( s );
m_writeSockets.erase( s );
m_connectSockets.erase( s );
m_dropped.push( s );
return true;
}
return false;
}
inline timeval* SocketMonitor::getTimeval( bool poll, double timeout )
{
if ( poll )
{
m_timeval.tv_sec = 0;
m_timeval.tv_usec = 0;
return &m_timeval;
}
timeout = m_timeout;
if ( !timeout )
return 0;
#ifdef SELECT_MODIFIES_TIMEVAL
if ( !m_timeval.tv_sec && !m_timeval.tv_usec && timeout )
m_timeval.tv_sec = timeout;
return &m_timeval;
#else
double elapsed = ( double ) ( clock() - m_ticks ) / ( double ) CLOCKS_PER_SEC;
if ( elapsed >= timeout || elapsed == 0.0 )
{
m_ticks = clock();
m_timeval.tv_sec = 0;
m_timeval.tv_usec = (long)(timeout * 1000000);
}
else
{
m_timeval.tv_sec = 0;
m_timeval.tv_usec = (long)( ( timeout - elapsed ) * 1000000 );
}
return &m_timeval;
#endif
}
bool SocketMonitor::sleepIfEmpty( bool poll )
{
if( poll )
return false;
if ( m_readSockets.empty() &&
m_writeSockets.empty() &&
m_connectSockets.empty() )
{
process_sleep( m_timeout );
return true;
}
else
return false;
}
void SocketMonitor::signal( int socket )
{
socket_send( m_signal, (char*)&socket, sizeof(socket) );
}
void SocketMonitor::unsignal( int s )
{
Sockets::iterator i = m_writeSockets.find( s );
if( i == m_writeSockets.end() ) return;
m_writeSockets.erase( s );
}
void SocketMonitor::block( Strategy& strategy, bool poll, double timeout )
{
while ( m_dropped.size() )
{
strategy.onError( *this, m_dropped.front() );
m_dropped.pop();
if ( m_dropped.size() == 0 )
return ;
}
fd_set readSet;
FD_ZERO( &readSet );
buildSet( m_readSockets, readSet );
fd_set writeSet;
FD_ZERO( &writeSet );
buildSet( m_connectSockets, writeSet );
buildSet( m_writeSockets, writeSet );
fd_set exceptSet;
FD_ZERO( &exceptSet );
buildSet( m_connectSockets, exceptSet );
if ( sleepIfEmpty(poll) )
{
strategy.onTimeout( *this );
return;
}
int result = select( FD_SETSIZE, &readSet, &writeSet, &exceptSet, getTimeval(poll, timeout) );
if ( result == 0 )
{
strategy.onTimeout( *this );
return;
}
else if ( result > 0 )
{
processExceptSet( strategy, exceptSet );
processWriteSet( strategy, writeSet );
processReadSet( strategy, readSet );
}
else
{
strategy.onError( *this );
}
}
void SocketMonitor::processReadSet( Strategy& strategy, fd_set& readSet )
{
#ifdef _MSC_VER
for ( unsigned i = 0; i < readSet.fd_count; ++i )
{
int s = readSet.fd_array[ i ];
if( s == m_interrupt )
{
int socket = 0;
socket_recv( s, (char*)&socket, sizeof(socket) );
addWrite( socket );
}
else
{
strategy.onEvent( *this, s );
}
}
#else
Sockets::iterator i;
Sockets sockets = m_readSockets;
for ( i = sockets.begin(); i != sockets.end(); ++i )
{
int s = *i;
if ( !FD_ISSET( *i, &readSet ) )
continue;
if( s == m_interrupt )
{
int socket = 0;
socket_recv( s, (char*)&socket, sizeof(socket) );
addWrite( socket );
}
else
{
strategy.onEvent( *this, s );
}
}
#endif
}
void SocketMonitor::processWriteSet( Strategy& strategy, fd_set& writeSet )
{
#ifdef _MSC_VER
for ( unsigned i = 0; i < writeSet.fd_count; ++i )
{
int s = writeSet.fd_array[ i ];
if( m_connectSockets.find(s) != m_connectSockets.end() )
{
m_connectSockets.erase( s );
m_readSockets.insert( s );
strategy.onConnect( *this, s );
}
else
{
strategy.onWrite( *this, s );
}
}
#else
Sockets::iterator i;
Sockets sockets = m_connectSockets;
for( i = sockets.begin(); i != sockets.end(); ++i )
{
int s = *i;
if ( !FD_ISSET( *i, &writeSet ) )
continue;
m_connectSockets.erase( s );
m_readSockets.insert( s );
strategy.onConnect( *this, s );
}
sockets = m_writeSockets;
for( i = sockets.begin(); i != sockets.end(); ++i )
{
int s = *i;
if ( !FD_ISSET( *i, &writeSet ) )
continue;
strategy.onWrite( *this, s );
}
#endif
}
void SocketMonitor::processExceptSet( Strategy& strategy, fd_set& exceptSet )
{
#ifdef _MSC_VER
for ( unsigned i = 0; i < exceptSet.fd_count; ++i )
{
int s = exceptSet.fd_array[ i ];
strategy.onError( *this, s );
}
#else
Sockets::iterator i;
Sockets sockets = m_connectSockets;
for ( i = sockets.begin(); i != sockets.end(); ++i )
{
int s = *i;
if ( !FD_ISSET( *i, &exceptSet ) )
continue;
strategy.onError( *this, s );
}
#endif
}
void SocketMonitor::buildSet( const Sockets& sockets, fd_set& watchSet )
{
Sockets::const_iterator iter;
for ( iter = sockets.begin(); iter != sockets.end(); ++iter ) {
FD_SET( *iter, &watchSet );
}
}
}
SocketMonitor.h
Utility.h
FIX::SocketMonitor::Strategy
Definition SocketMonitor.h:96
FIX::SocketMonitor::Strategy::onWrite
virtual void onWrite(SocketMonitor &, int socket)=0
FIX::SocketMonitor::Strategy::onError
virtual void onError(SocketMonitor &, int socket)=0
FIX::SocketMonitor::Strategy::onTimeout
virtual void onTimeout(SocketMonitor &)
Definition SocketMonitor.h:105
FIX::SocketMonitor::Strategy::onEvent
virtual void onEvent(SocketMonitor &, int socket)=0
FIX::SocketMonitor::Strategy::onConnect
virtual void onConnect(SocketMonitor &, int socket)=0
FIX::SocketMonitor::unsignal
void unsignal(int socket)
Definition SocketMonitor.cpp:173
FIX::SocketMonitor::drop
bool drop(int socket)
Definition SocketMonitor.cpp:98
FIX::SocketMonitor::block
void block(Strategy &strategy, bool poll=0, double timeout=0.0)
Definition SocketMonitor.cpp:181
FIX::SocketMonitor::signal
void signal(int socket)
Definition SocketMonitor.cpp:168
FIX::SocketMonitor::m_readSockets
Sockets m_readSockets
Definition SocketMonitor.h:90
FIX::SocketMonitor::m_dropped
Queue m_dropped
Definition SocketMonitor.h:92
FIX::SocketMonitor::m_signal
int m_signal
Definition SocketMonitor.h:87
FIX::SocketMonitor::buildSet
void buildSet(const Sockets &, fd_set &)
Definition SocketMonitor.cpp:328
FIX::SocketMonitor::m_ticks
clock_t m_ticks
Definition SocketMonitor.h:84
FIX::SocketMonitor::Sockets
std::set< int > Sockets
Definition SocketMonitor.h:67
FIX::SocketMonitor::getTimeval
timeval * getTimeval(bool poll, double timeout)
Definition SocketMonitor.cpp:118
FIX::SocketMonitor::addRead
bool addRead(int socket)
Definition SocketMonitor.cpp:75
FIX::SocketMonitor::addConnect
bool addConnect(int socket)
Definition SocketMonitor.cpp:65
FIX::SocketMonitor::m_connectSockets
Sockets m_connectSockets
Definition SocketMonitor.h:89
FIX::SocketMonitor::sleepIfEmpty
bool sleepIfEmpty(bool poll)
Definition SocketMonitor.cpp:152
FIX::SocketMonitor::processWriteSet
void processWriteSet(Strategy &, fd_set &)
Definition SocketMonitor.cpp:266
FIX::SocketMonitor::processReadSet
void processReadSet(Strategy &, fd_set &)
Definition SocketMonitor.cpp:227
FIX::SocketMonitor::~SocketMonitor
virtual ~SocketMonitor()
Definition SocketMonitor.cpp:54
FIX::SocketMonitor::m_timeout
int m_timeout
Definition SocketMonitor.h:81
FIX::SocketMonitor::m_interrupt
int m_interrupt
Definition SocketMonitor.h:88
FIX::SocketMonitor::processExceptSet
void processExceptSet(Strategy &, fd_set &)
Definition SocketMonitor.cpp:307
FIX::SocketMonitor::SocketMonitor
SocketMonitor(int timeout=0)
Definition SocketMonitor.cpp:35
FIX::SocketMonitor::addWrite
bool addWrite(int socket)
Definition SocketMonitor.cpp:85
FIX::SocketMonitor::m_writeSockets
Sockets m_writeSockets
Definition SocketMonitor.h:91
FIX::SocketMonitor::m_timeval
timeval m_timeval
Definition SocketMonitor.h:82
FIX
Definition Acceptor.cpp:35
FIX::socket_setnonblock
void socket_setnonblock(int socket)
Definition Utility.cpp:268
FIX::socket_recv
ssize_t socket_recv(int s, char *buf, size_t length)
Definition Utility.cpp:170
FIX::socket_close
void socket_close(int s)
Definition Utility.cpp:180
FIX::process_sleep
void process_sleep(double s)
Definition Utility.cpp:466
FIX::socket_createpair
std::pair< int, int > socket_createpair()
Definition Utility.cpp:366
FIX::socket_init
void socket_init()
Definition Utility.cpp:81
FIX::socket_send
ssize_t socket_send(int s, const char *msg, size_t length)
Definition Utility.cpp:175
FIX::socket_term
void socket_term()
Definition Utility.cpp:96