#include <ConnectionServer.h>

Collaboration diagram for CPN::ConnectionServer:

Classes
class	PendingConnection

Public Member Functions
	ConnectionServer (SockAddrList addrs, shared_ptr< Context > ctx)

void	Poll ()

void	Wakeup ()

void	Close ()

shared_ptr< Sync::Future< int > >	ConnectWriter (Key_t writerkey)

shared_ptr< Sync::Future< int > >	ConnectReader (Key_t readerkey)

SocketAddress	GetAddress ()


void	Disable ()

void	Enable ()

void	LogState ()

Private Types
typedef std::multimap< Key_t, shared_ptr< PendingConnection > >	PendingMap

Private Member Functions
void	PendingDone (Key_t key, PendingConnection *conn)

Private Attributes
PthreadMutex	lock

shared_ptr< Context >	context

Logger	logger

ServerSocketHandle	server

WakeupHandle	wakeup

PendingMap	pendingconnections

bool	enabled

Detailed Description

The connnection server takes ownership of the listening socket and accepts new connections. It also has all the logic for creating new connections. When remote is active in the Kernel, the Kernel creates a connection server and all RemoteQueues receive a reference to it as well.

Definition at line 41 of file ConnectionServer.h.

Member Typedef Documentation

typedef std::multimap<Key_t, shared_ptr<PendingConnection> > CPN::ConnectionServer::PendingMap

private

Definition at line 106 of file ConnectionServer.h.

Constructor & Destructor Documentation

CPN::ConnectionServer::ConnectionServer	(	SockAddrList	addrs,
		shared_ptr< Context >	ctx
	)

Parameters

addrs	list of socket addresses to use to listen on
ctx	the context which contains all the data connection data.

Definition at line 37 of file ConnectionServer.cc.

References ServerSocketHandle::Listen(), and server.

         : context(ctx), logger(ctx.get(), Logger::INFO), enabled(true)
     {
         server.Listen(addrs);
     }

Here is the call graph for this function:

Member Function Documentation

void CPN::ConnectionServer::Close ( )

Close the listening socket and shutdown the connection server.

Definition at line 92 of file ConnectionServer.cc.

References FileHandle::Close(), lock, pendingconnections, and server.

                                  {
         AutoPLock al(lock);
         server.Close();
         for (PendingMap::iterator itr = pendingconnections.begin();
                 itr != pendingconnections.end(); ++itr)
         {
             itr->second->Cancel();
         }
     }

Here is the call graph for this function:

shared_ptr< Sync::Future< int > > CPN::ConnectionServer::ConnectReader ( Key_t readerkey )

Request the connection server connect the RemoteQueue which has key readerkey to its other endpoint.

Parameters

readerkey the key for the reader of this queue

Returns: A Sync::Future which will have the file descriptor.

Definition at line 134 of file ConnectionServer.cc.

References FileHandle::Closed(), lock, pendingconnections, and server.

Referenced by CPN::RemoteQueue::FileThreadEntryPoint().

                                                                                 {
         if (server.Closed()) {
             return shared_ptr<Sync::Future<int> >();
         }
         AutoPLock al(lock);
         shared_ptr<PendingConnection> conn;
         PendingMap::iterator entry = pendingconnections.find(readerkey);
         if (entry == pendingconnections.end()) {
             conn = shared_ptr<PendingConnection>(new PendingConnection(readerkey, this));
             pendingconnections.insert(std::make_pair(readerkey, conn));
         } else {
             conn = entry->second;
         }
         return conn;
     }

Here is the call graph for this function:

Here is the caller graph for this function:

shared_ptr< Sync::Future< int > > CPN::ConnectionServer::ConnectWriter ( Key_t writerkey )

Request the connectino server to connect the RemoteQueue which has key writekey to its other endpoint.

Parameters

writerkey the writekey of the endpoint.

Returns: A Sync::Future which will contain the file descriptor when finished

Definition at line 102 of file ConnectionServer.cc.

References FileHandle::Closed(), SocketHandle::Connect(), context, SocketAddress::CreateIP(), enabled, FileHandle::FD(), CPN::Packet::header, lock, CPN::PACKET_ID_WRITER, pendingconnections, FileHandle::Reset(), server, CPN::Packet::SourceKey(), AutoLock< Lockable >::Unlock(), and FileHandle::Write().

Referenced by CPN::RemoteQueue::FileThreadEntryPoint().

                                                                                 {
         if (server.Closed()) {
             return shared_ptr<Sync::Future<int> >();
         }
         AutoPLock al(lock);
         shared_ptr<PendingConnection> conn;
         if (enabled) {
             al.Unlock();
             conn = shared_ptr<PendingConnection>(new PendingConnection(writerkey, this));
             Key_t readerkey = context->GetWritersReader(writerkey);
             Key_t kernelkey = context->GetReaderKernel(readerkey);
             std::string hostname;
             std::string servname;
             context->GetKernelConnectionInfo(kernelkey, hostname, servname);
             SocketHandle sock;
             sock.Connect(SocketAddress::CreateIP(hostname, servname));
             Packet packet(PACKET_ID_WRITER);
             packet.SourceKey(writerkey).DestinationKey(readerkey);
             unsigned num = sock.Write(&packet.header, sizeof(packet.header));
             if (num != sizeof(packet.header)) {
                 conn->Cancel();
             } else {
                 conn->Set(sock.FD());
                 sock.Reset();
             }
         } else {
             conn = shared_ptr<PendingConnection>(new PendingConnection(writerkey, this));
             pendingconnections.insert(std::make_pair(writerkey, conn));
         }
         return conn;
     }

Here is the call graph for this function:

Here is the caller graph for this function:

void CPN::ConnectionServer::Disable ( )

These functions are for testing and debugging. They should only be called from the unit tests or from the debugger.

Definition at line 156 of file ConnectionServer.cc.

References enabled, and lock.

                                    {
         AutoPLock al(lock);
         enabled = false;
     }

void CPN::ConnectionServer::Enable ( )

Definition at line 161 of file ConnectionServer.cc.

References enabled, lock, and pendingconnections.

                                   {
         AutoPLock al(lock);
         enabled = true;
         PendingMap::iterator entry = pendingconnections.begin();
         while (entry != pendingconnections.end()) {
             entry->second->Cancel();
             ++entry;
         }
     }

SocketAddress CPN::ConnectionServer::GetAddress ( )

Returns: the address this connection server is listening on

Definition at line 150 of file ConnectionServer.cc.

References FileHandle::FD(), server, and SocketAddress::SetFromSockName().

                                                {
         SocketAddress addr;
         addr.SetFromSockName(server.FD());
         return addr;
     }

Here is the call graph for this function:

void CPN::ConnectionServer::LogState ( )

Definition at line 171 of file ConnectionServer.cc.

References FileHandle::Closed(), enabled, Logger::Error(), logger, pendingconnections, and server.

                                     {
         if (server.Closed()) {
             logger.Error("Server socket closed");
         }
         if (!enabled) {
             logger.Error("Connection handler disabled??");
         }
         logger.Error("%u pending connections", pendingconnections.size());
     }

Here is the call graph for this function:

void CPN::ConnectionServer::PendingDone	(	Key_t	key,
		PendingConnection *	conn
	)

private

Definition at line 181 of file ConnectionServer.cc.

References lock, and pendingconnections.

                                                                          {
         AutoPLock al(lock);
         std::pair<PendingMap::iterator, PendingMap::iterator> range;
         range = pendingconnections.equal_range(key);
         PendingMap::iterator entry = range.first;
         PendingMap::iterator end = range.second;
         while (entry != end) {
             if (entry->second.get() == conn) {
                 pendingconnections.erase(entry);
                 break;
             }
             ++entry;
         }
     }

void CPN::ConnectionServer::Poll ( void )

Poll is periodically called by the Kernel to accept new connections.

Definition at line 43 of file ConnectionServer.cc.

References ServerSocketHandle::Accept(), ErrnoException::Error(), Logger::Error(), lock, logger, CPN::PACKET_ID_READER, CPN::PACKET_ID_WRITER, pendingconnections, FileHandle::Poll(), WakeupHandle::Read(), FileHandle::Readable(), server, wakeup, and ErrnoException::what().

                                 {
         std::deque<FileHandle*> files;
         files.push_back(&server);
         files.push_back(&wakeup);
         FileHandle::Poll(files.begin(), files.end(), -1);
         wakeup.Read();
         if (server.Readable()) {
             server.Readable(false);
             try {
                 SocketHandle sock(server.Accept());
                 Packet packet;
                 unsigned num = sock.Read(&packet.header, sizeof(packet.header));
                 if (num != sizeof(packet.header)) {
                     return;
                 }
                 if (!packet.Valid()) {
                     return;
                 }
                 if ((packet.Type() == PACKET_ID_READER)
                         || (packet.Type() == PACKET_ID_WRITER)) {
                     AutoPLock al(lock);
                     Key_t key = packet.DestinationKey();
                     shared_ptr<PendingConnection> conn;
                     std::pair<PendingMap::iterator, PendingMap::iterator> range;
                     range = pendingconnections.equal_range(key);
                     PendingMap::iterator entry = range.first;
                     PendingMap::iterator end = range.second;
                     while (entry != end) {
                         if (!entry->second->Done()) {
                             conn = entry->second;
                             break;
                         }
                         ++entry;
                     }
                     if (!conn) {
                         conn = shared_ptr<PendingConnection>(new PendingConnection(key, this));
                         pendingconnections.insert(std::make_pair(key, conn));
                     }
                     conn->Set(sock.FD());
                     sock.Reset();
                 }
             } catch (const ErrnoException &e) {
                 // Ignore, if we had an error we closed the socket when we
                 // left the scope, will try again later
                 logger.Error("Exception in ConnectionServer main loop (e: %d): %s", e.Error(), e.what());
             }
         }
     }