pegasus/src/Pegasus/Common/Monitor.cpp - diff

Return to Monitor.cpp CVS log

Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/Monitor.cpp between version 1.54 and 1.142

version 1.54, 2003/10/05 23:25:43

version 1.142, 2013/02/25 06:46:07

Line 1

//%/////////////////////////////////////////////////////////////////////////////

//%LICENSE////////////////////////////////////////////////////////////////

// The Open Group, Tivoli Systems

// Permission is hereby granted, free of charge, to any person obtaining a copy

// of this software and associated documentation files (the "Software"), to

// deal in the Software without restriction, including without limitation the

// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or

// sell copies of the Software, and to permit persons to whom the Software is

// furnished to do so, subject to the following conditions:

// THE ABOVE COPYRIGHT NOTICE AND THIS PERMISSION NOTICE SHALL BE INCLUDED IN

// ALL COPIES OR SUBSTANTIAL PORTIONS OF THE SOFTWARE. THE SOFTWARE IS PROVIDED

// "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT

// LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR

// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT

// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN

// ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION

// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

//==============================================================================

// Licensed to The Open Group (TOG) under one or more contributor license

// agreements. Refer to the OpenPegasusNOTICE.txt file distributed with

// this work for additional information regarding copyright ownership.

// Each contributor licenses this file to you under the OpenPegasus Open

// Source License; you may not use this file except in compliance with the

// License.

// Permission is hereby granted, free of charge, to any person obtaining a

// copy of this software and associated documentation files (the "Software"),

// to deal in the Software without restriction, including without limitation

// the rights to use, copy, modify, merge, publish, distribute, sublicense,

// and/or sell copies of the Software, and to permit persons to whom the

// Software is furnished to do so, subject to the following conditions:

// The above copyright notice and this permission notice shall be included

// in all copies or substantial portions of the Software.

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS

// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.

// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY

// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,

// TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE

// SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

// Author: Mike Brasher (mbrasher@bmc.com)

//////////////////////////////////////////////////////////////////////////

// Modified By: Mike Day (monitor_2) mdday@us.ibm.com

//%/////////////////////////////////////////////////////////////////////////////

#include "Network.h"

#include <Pegasus/Common/Config.h>

#include <cstring>

#include "Monitor.h"

#include "MessageQueue.h"

#include "Socket.h"

#include <Pegasus/Common/Tracer.h>

#include <Pegasus/Common/HTTPConnection.h>

#include <Pegasus/Common/HTTPAcceptor.h>

#ifdef PEGASUS_OS_TYPE_WINDOWS

#include <Pegasus/Common/MessageQueueService.h>

# if defined(FD_SETSIZE) && FD_SETSIZE != 1024

#include <Pegasus/Common/Exception.h>

# error "FD_SETSIZE was not set to 1024 prior to the last inclusion \

#include "ArrayIterator.h"

of <winsock.h>. It may have been indirectly included (e.g., by including \

#include "HostAddress.h"

<windows.h>). Finthe inclusion of that header which is visible to this \

#include <errno.h>

compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \

otherwise, less than 64 clients (the default) will be able to connect to the \

CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM."

# endif

# define FD_SETSIZE 1024

# include <windows.h>

#else

# include <sys/types.h>

# include <sys/socket.h>

# include <sys/time.h>

# include <netinet/in.h>

# include <netdb.h>

# include <arpa/inet.h>

#endif

PEGASUS_USING_STD;

PEGASUS_NAMESPACE_BEGIN

static AtomicInt _connections = 0;

static struct timeval create_time = {0, 1};

static struct timeval destroy_time = {300, 0};

static struct timeval deadlock_time = {0, 0};

////////////////////////////////////////////////////////////////////////////////

// MonitorRep

// Tickler

////////////////////////////////////////////////////////////////////////////////

struct MonitorRep

Tickler::Tickler()

: _listenSocket(PEGASUS_INVALID_SOCKET),

_clientSocket(PEGASUS_INVALID_SOCKET),

_serverSocket(PEGASUS_INVALID_SOCKET)

{

fd_set rd_fd_set;

try

fd_set wr_fd_set;

fd_set ex_fd_set;

fd_set active_rd_fd_set;

fd_set active_wr_fd_set;

fd_set active_ex_fd_set;

};

////////////////////////////////////////////////////////////////////////////////

// Monitor

////////////////////////////////////////////////////////////////////////////////

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

Monitor::Monitor()

: _module_handle(0), _controller(0), _async(false), _stopConnections(0)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

_initialize();

Socket::initializeInterface();

}

_rep = 0;

catch (...)

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

for( int i = 0; i < numberOfMonitorEntriesToAllocate; i++ )

{

_MonitorEntry entry(0, 0, 0);

_uninitialize();

_entries.append(entry);

throw;

}

Monitor::Monitor(Boolean async)

Tickler::~Tickler()

: _module_handle(0), _controller(0), _async(async), _stopConnections(0)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

_uninitialize();

Socket::initializeInterface();

_rep = 0;

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

for( int i = 0; i < numberOfMonitorEntriesToAllocate; i++ )

{

_MonitorEntry entry(0, 0, 0);

_entries.append(entry);

}

if( _async == true )

void Tickler::notify()

{

_thread_pool = new ThreadPool(0,

Socket::write(_clientSocket, "\0", 1);

"Monitor",

create_time,

destroy_time,

deadlock_time);

}

else

_thread_pool = 0;

}

Monitor::~Monitor()

void Tickler::reset()

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

// Clear all bytes from the tickle socket

"deregistering with module controller");

char buffer[32];

while (Socket::read(_serverSocket, buffer, 32) > 0)

if(_module_handle != NULL)

{

_controller->deregister_module(PEGASUS_MODULENAME_MONITOR);

_controller = 0;

delete _module_handle;

}

Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");

Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");

Socket::uninitializeInterface();

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"returning from monitor destructor");

if(_async == true)

delete _thread_pool;

}

#if defined(PEGASUS_OS_TYPE_UNIX)

int Monitor::kill_idle_threads()

// Use an anonymous pipe for the tickle connection.

{

static struct timeval now, last;

gettimeofday(&now, NULL);

int dead_threads = 0;

if( now.tv_sec - last.tv_sec > 120 )

void Tickler::_initialize()

{

gettimeofday(&last, NULL);

int fds[2];

try

{

if (pipe(fds) == -1)

dead_threads = _thread_pool->kill_dead_threads();

}

catch(IPCException& )

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CREATE",

"Received error number $0 while creating the internal socket.",

getSocketError());

throw Exception(parms);

}

_serverSocket = fds[0];

return dead_threads;

_clientSocket = fds[1];

Socket::disableBlocking(_serverSocket);

}

#else

// Use an external loopback socket connection to allow the tickle socket to

// be included in the select() array on non-Unix platforms.

Boolean Monitor::run(Uint32 milliseconds)

void Tickler::_initialize()

{

// Set up the addresses for the listen, client, and server sockets

// based on whether IPv6 is enabled.

Boolean handled_events = false;

Socket::initializeInterface();

int i = 0;

#if defined(PEGASUS_OS_OS400) || defined(PEGASUS_OS_HPUX)

# ifdef PEGASUS_ENABLE_IPV6

struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};

struct sockaddr_storage listenAddress;

struct sockaddr_storage clientAddress;

struct sockaddr_storage serverAddress;

#else

struct timeval tv = {0, 1};

struct sockaddr_in listenAddress;

struct sockaddr_in clientAddress;

struct sockaddr_in serverAddress;

#endif

fd_set fdread;

FD_ZERO(&fdread);

_entry_mut.lock(pegasus_thread_self());

// Check the stopConnections flag. If set, clear the Acceptor monitor entries

int addressFamily;

if (_stopConnections == 1)

SocketLength addressLength;

{

for ( int indx = 0; indx < (int)_entries.size(); indx++)

memset(&listenAddress, 0, sizeof (listenAddress));

{

if (_entries[indx]._type == Monitor::ACCEPTOR)

# ifdef PEGASUS_ENABLE_IPV6

{

if (System::isIPv6StackActive())

if ( _entries[indx]._status.value() != _MonitorEntry::EMPTY)

{

if ( _entries[indx]._status.value() == _MonitorEntry::IDLE ||

_entries[indx]._status.value() == _MonitorEntry::DYING )

{

// remove the entry

// Use the IPv6 loopback address for the listen sockets

_entries[indx]._status = _MonitorEntry::EMPTY;

HostAddress::convertTextToBinary(

HostAddress::AT_IPV6,

"::1",

&reinterpret_cast<struct sockaddr_in6*>(&listenAddress)->sin6_addr);

listenAddress.ss_family = AF_INET6;

reinterpret_cast<struct sockaddr_in6*>(&listenAddress)->sin6_port = 0;

addressFamily = AF_INET6;

addressLength = sizeof(struct sockaddr_in6);

}

else

# endif

{

// set status to DYING

// Use the IPv4 loopback address for the listen sockets

_entries[indx]._status = _MonitorEntry::DYING;

HostAddress::convertTextToBinary(

}

HostAddress::AT_IPV4,

}

"127.0.0.1",

}

&reinterpret_cast<struct sockaddr_in*>(

}

&listenAddress)->sin_addr.s_addr);

_stopConnections = 0;

reinterpret_cast<struct sockaddr_in*>(&listenAddress)->sin_family =

}

AF_INET;

reinterpret_cast<struct sockaddr_in*>(&listenAddress)->sin_port = 0;

Uint32 _idleEntries = 0;

addressFamily = AF_INET;

addressLength = sizeof(struct sockaddr_in);

for( int indx = 0; indx < (int)_entries.size(); indx++)

{

if(_entries[indx]._status.value() == _MonitorEntry::IDLE)

{

_idleEntries++;

FD_SET(_entries[indx].socket, &fdread);

}

_entry_mut.unlock();

// Use the same address for the client socket as the listen socket

int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);

clientAddress = listenAddress;

_entry_mut.lock(pegasus_thread_self());

#ifdef PEGASUS_OS_TYPE_WINDOWS

if(events == SOCKET_ERROR)

// Set up a listen socket to allow the tickle client and server to connect

#else

if(events == -1)

#endif

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::run - errorno = %d has occurred on select.", errno);

// The EBADF error indicates that one or more or the file

// descriptions was not valid. This could indicate that

// the _entries structure has been corrupted or that

// we have a synchronization error.

PEGASUS_ASSERT(errno != EBADF);

// Create the listen socket

}

if ((_listenSocket = Socket::createSocket(addressFamily, SOCK_STREAM, 0)) ==

else if (events)

PEGASUS_INVALID_SOCKET)

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

MessageLoaderParms parms(

"Monitor::run select event received events = %d, monitoring %d idle entries",

"Common.Monitor.TICKLE_CREATE",

events, _idleEntries);

"Received error number $0 while creating the internal socket.",

for( int indx = 0; indx < (int)_entries.size(); indx++)

getSocketError());

{

throw Exception(parms);

// The Monitor should only look at entries in the table that are IDLE (i.e.,

}

// owned by the Monitor).

if((_entries[indx]._status.value() == _MonitorEntry::IDLE) &&

(FD_ISSET(_entries[indx].socket, &fdread)))

{

MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::run indx = %d, queueId = %d, q = %p",

indx, _entries[indx].queueId, q);

PEGASUS_ASSERT(q !=0);

try

// Bind the listen socket to the loopback address

if (::bind(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

addressLength) < 0)

{

if(_entries[indx]._type == Monitor::CONNECTION)

MessageLoaderParms parms(

{

"Common.Monitor.TICKLE_BIND",

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Received error number $0 while binding the internal socket.",

"_entries[indx].type for indx = %d is Monitor::CONNECTION", indx);

getSocketError());

static_cast<HTTPConnection *>(q)->_entry_index = indx;

throw Exception(parms);

if(static_cast<HTTPConnection *>(q)->_dying.value() > 0 )

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::run processing dying value > 0 for indx = %d, connection being closed.",

indx);

_entries[indx]._status = _MonitorEntry::DYING;

MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();

Message* message= new CloseConnectionMessage(_entries[indx].socket);

message->dest = o.getQueueId();

_entry_mut.unlock();

o.enqueue(message);

return true;

}

_entries[indx]._status = _MonitorEntry::BUSY;

_thread_pool->allocate_and_awaken((void *)q, _dispatch);

}

else

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Non-connection entry, indx = %d, has been received.", indx);

int events = 0;

events |= SocketMessage::READ;

Message *msg = new SocketMessage(_entries[indx].socket, events);

_entries[indx]._status = _MonitorEntry::BUSY;

_entry_mut.unlock();

q->enqueue(msg);

// Listen for a connection from the tickle client

_entries[indx]._status = _MonitorEntry::IDLE;

if ((::listen(_listenSocket, 3)) < 0)

return true;

}

catch(...)

{

}

MessageLoaderParms parms(

handled_events = true;

"Common.Monitor.TICKLE_LISTEN",

}

"Received error number $0 while listening to the internal socket.",

}

getSocketError());

}

throw Exception(parms);

_entry_mut.unlock();

return(handled_events);

}

void Monitor::stopListeningForConnections()

// Verify we have the correct listen socket

SocketLength tmpAddressLength = addressLength;

int sock = ::getsockname(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

&tmpAddressLength);

if (sock < 0)

{

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");

MessageLoaderParms parms(

"Common.Monitor.TICKLE_SOCKNAME",

_stopConnections = 1;

"Received error number $0 while getting the internal socket name.",

getSocketError());

PEG_METHOD_EXIT();

throw Exception(parms);

}

// Set up the client side of the tickle connection.

int Monitor::solicitSocketMessages(

// Create the client socket

Sint32 socket,

if ((_clientSocket = Socket::createSocket(addressFamily, SOCK_STREAM, 0)) ==

Uint32 events,

PEGASUS_INVALID_SOCKET)

Uint32 queueId,

int type)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_CREATE",

"Received error number $0 while creating the internal client "

"socket.",

getSocketError());

throw Exception(parms);

}

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");

// Bind the client socket to the loopback address

if (::bind(

_entry_mut.lock(pegasus_thread_self());

_clientSocket,

reinterpret_cast<struct sockaddr*>(&clientAddress),

for(int index = 0; index < (int)_entries.size(); index++)

addressLength) < 0)

{

try

{

if(_entries[index]._status.value() == _MonitorEntry::EMPTY)

MessageLoaderParms parms(

{

"Common.Monitor.TICKLE_CLIENT_BIND",

_entries[index].socket = socket;

"Received error number $0 while binding the internal client "

_entries[index].queueId = queueId;

"socket.",

_entries[index]._type = type;

getSocketError());

_entries[index]._status = _MonitorEntry::IDLE;

throw Exception(parms);

_entry_mut.unlock();

return index;

}

catch(...)

// Connect the client socket to the listen socket address

if (::connect(

_clientSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

addressLength) < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CLIENT_CONNECT",

"Received error number $0 while connecting the internal client "

"socket.",

getSocketError());

throw Exception(parms);

}

_entry_mut.unlock();

// Set up the server side of the tickle connection.

PEG_METHOD_EXIT();

return -1;

}

void Monitor::unsolicitSocketMessages(Sint32 socket)

tmpAddressLength = addressLength;

{

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");

// Accept the client socket connection.

_entry_mut.lock(pegasus_thread_self());

_serverSocket = ::accept(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&serverAddress),

&tmpAddressLength);

for(int index = 0; index < (int)_entries.size(); index++)

if (_serverSocket == PEGASUS_SOCKET_ERROR)

{

if(_entries[index].socket == socket)

MessageLoaderParms parms(

{

"Common.Monitor.TICKLE_ACCEPT",

_entries[index]._status = _MonitorEntry::EMPTY;

"Received error number $0 while accepting the internal socket "

_entries[index].socket = -1;

"connection.",

break;

getSocketError());

throw Exception(parms);

}

// Close the listen socket and make the other sockets non-blocking

Socket::close(_listenSocket);

Socket::disableBlocking(_serverSocket);

Socket::disableBlocking(_clientSocket);

}

_entry_mut.unlock();

PEG_METHOD_EXIT();

}

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)

#endif

void Tickler::_uninitialize()

{

HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

"Monitor::_dispatch: entering run() for indx = %d, queueId = %d, q = %p",

dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);

try

{

dst->run(1);

Socket::close(_serverSocket);

Socket::close(_clientSocket);

Socket::close(_listenSocket);

}

catch (...)

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL2,

"Monitor::_dispatch: exception received");

"Failed to close tickle sockets");

}

Tracer::trace(TRC_HTTP, Tracer::LEVEL4,

Socket::uninitializeInterface();

"Monitor::_dispatch: exited run() for index %d", dst->_entry_index);

dst->_monitor->_entry_mut.lock(pegasus_thread_self());

// It shouldn't be necessary to set status = _MonitorEntry::IDLE

// if the connection is being closed. However, the current logic

// in Monitor::run requires this value to be set for the close

// to be processed.

PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.value() == _MonitorEntry::BUSY);

dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;

if (dst->_connectionClosePending)

{

dst->_dying = 1;

}

dst->_monitor->_entry_mut.unlock();

return 0;

}

////////////////////////////////////////////////////////////////////////////////

// Monitor

////////////////////////////////////////////////////////////////////////////////

////************************* monitor 2 *****************************////

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

////************************* monitor 2 *****************************////

Monitor::Monitor()

////************************* monitor 2 *****************************////

: _stopConnections(0),

////************************* monitor 2 *****************************////

_stopConnectionsSem(0),

////************************* monitor 2 *****************************////

_solicitSocketCount(0)

////************************* monitor 2 *****************************////

m2e_rep::m2e_rep(void)

:Base(), state(IDLE)

{

}

m2e_rep::m2e_rep(monitor_2_entry_type _type,

pegasus_socket _sock,

void* _accept,

void* _dispatch)

: Base(), type(_type), state(IDLE), psock(_sock),

accept_parm(_accept), dispatch_parm(_dispatch)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

}

// Create a MonitorEntry for the Tickler and set its state to IDLE so the

// Monitor will watch for its events.

_entries.append(MonitorEntry(

_tickler.getReadHandle(),

MonitorEntry::STATUS_IDLE,

MonitorEntry::TYPE_TICKLER));

m2e_rep::~m2e_rep(void)

// Start the count at 1 because _entries[0] is the Tickler

for (int i = 1; i < numberOfMonitorEntriesToAllocate; i++)

{

}

_entries.append(MonitorEntry());

m2e_rep::m2e_rep(const m2e_rep& r)

: Base()

{

if(this != &r){

type = r.type;

psock = r.psock;

accept_parm = r.accept_parm;

dispatch_parm = r.dispatch_parm;

state = IDLE;

}

Monitor::~Monitor()

m2e_rep& m2e_rep::operator =(const m2e_rep& r)

{

if(this != &r) {

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

type = r.type;

"returning from monitor destructor");

psock = r.psock;

accept_parm = r.accept_parm;

dispatch_parm = r.dispatch_parm;

state = IDLE;

}

return *this;

}

Boolean m2e_rep::operator ==(const m2e_rep& r)

void Monitor::tickle()

{

if(this == &r)

_tickler.notify();

return true;

return false;

}

Boolean m2e_rep::operator ==(void* r)

void Monitor::setState(

Uint32 index,

MonitorEntry::Status status)

{

if((void*)this == r)

AutoMutex autoEntryMutex(_entriesMutex);

return true;

// Set the state to requested state

return false;

_entries[index].status = status;

}

m2e_rep::operator pegasus_socket() const

void Monitor::run(Uint32 milliseconds)

{

return psock;

struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};

}

fd_set fdread;

FD_ZERO(&fdread);

monitor_2_entry::monitor_2_entry(void)

AutoMutex autoEntryMutex(_entriesMutex);

{

_rep = new m2e_rep();

}

monitor_2_entry::monitor_2_entry(pegasus_socket& _psock,

ArrayIterator<MonitorEntry> entries(_entries);

monitor_2_entry_type _type,

void* _accept_parm, void* _dispatch_parm)

{

_rep = new m2e_rep(_type, _psock, _accept_parm, _dispatch_parm);

}

monitor_2_entry::monitor_2_entry(const monitor_2_entry& e)

// Check the stopConnections flag. If set, clear the Acceptor monitor

// entries

if (_stopConnections.get() == 1)

{

if(this != &e){

for (Uint32 indx = 0; indx < entries.size(); indx++)

Inc(this->_rep = e._rep);

}

monitor_2_entry::~monitor_2_entry(void)

{

Dec(_rep);

if (entries[indx].type == MonitorEntry::TYPE_ACCEPTOR)

}

monitor_2_entry& monitor_2_entry::operator=(const monitor_2_entry& e)

{

if(this != &e){

if (entries[indx].status != MonitorEntry::STATUS_EMPTY)

Dec(_rep);

Inc(this->_rep = e._rep);

}

return *this;

}

Boolean monitor_2_entry::operator ==(const monitor_2_entry& me) const

{

if(this == &me)

if (entries[indx].status == MonitorEntry::STATUS_IDLE ||

return true;

entries[indx].status == MonitorEntry::STATUS_DYING)

return false;

}

Boolean monitor_2_entry::operator ==(void* k) const

{

if((void *)this == k)

// remove the entry

return true;

entries[indx].status = MonitorEntry::STATUS_EMPTY;

return false;

}

else

monitor_2_entry_type monitor_2_entry::get_type(void) const

{

return _rep->type;

// set status to DYING

entries[indx].status = MonitorEntry::STATUS_DYING;

}

void monitor_2_entry::set_type(monitor_2_entry_type t)

{

_rep->type = t;

}

monitor_2_entry_state monitor_2_entry::get_state(void) const

{

return (monitor_2_entry_state) _rep->state.value();

}

void monitor_2_entry::set_state(monitor_2_entry_state t)

{

_rep->state = t;

}

_stopConnections = 0;

void* monitor_2_entry::get_accept(void) const

_stopConnectionsSem.signal();

{

return _rep->accept_parm;

}

void monitor_2_entry::set_accept(void* a)

for (Uint32 indx = 0; indx < entries.size(); indx++)

{

_rep->accept_parm = a;

const MonitorEntry& entry = entries[indx];

}

void* monitor_2_entry::get_dispatch(void) const

if ((entry.status == MonitorEntry::STATUS_DYING) &&

(entry.type == MonitorEntry::TYPE_CONNECTION))

{

return _rep->dispatch_parm;

MessageQueue *q = MessageQueue::lookup(entry.queueId);

}

PEGASUS_ASSERT(q != 0);

HTTPConnection &h = *static_cast<HTTPConnection *>(q);

void monitor_2_entry::set_dispatch(void* a)

if (h._connectionClosePending == false)

{

continue;

_rep->dispatch_parm = a;

}

pegasus_socket monitor_2_entry::get_sock(void) const

// NOTE: do not attempt to delete while there are pending responses

{

// coming thru. The last response to come thru after a

return _rep->psock;

// _connectionClosePending will reset _responsePending to false

}

// and then cause the monitor to rerun this code and clean up.

// (see HTTPConnection.cpp)

if (h._responsePending == true)

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"Monitor::run - Ignoring connection delete request "

"because responses are still pending. "

"connection=0x%p, socket=%d\n",

(void *)&h, h.getSocket()));

continue;

}

h._connectionClosePending = false;

HTTPAcceptor &o = h.getOwningAcceptor();

Message* message= new CloseConnectionMessage(entry.socket);

message->dest = o.getQueueId();

// HTTPAcceptor is responsible for closing the connection.

// The lock is released to allow HTTPAcceptor to call

// unsolicitSocketMessages to free the entry.

// Once HTTPAcceptor completes processing of the close

// connection, the lock is re-requested and processing of

// the for loop continues. This is safe with the current

// implementation of the entries object. Note that the

// loop condition accesses the entries.size() on each

// iteration, so that a change in size while the mutex is

// unlocked will not result in an ArrayIndexOutOfBounds

// exception.

void monitor_2_entry::set_sock(pegasus_socket& s)

_entriesMutex.unlock();

{

o.enqueue(message);

_rep->psock = s;

_entriesMutex.lock();

// After enqueue a message and the autoEntryMutex has been

// released and locked again, the array of _entries can be

// changed. The ArrayIterator has be reset with the original

// _entries.

entries.reset(_entries);

}

Uint32 _idleEntries = 0;

AsyncDQueue<HTTPConnection2> monitor_2::_connections(true, 0);

We will keep track of the maximum socket number and pass this value

to the kernel as a parameter to SELECT. This loop seems like a good

place to calculate the max file descriptor (maximum socket number)

because we have to traverse the entire array.

SocketHandle maxSocketCurrentPass = 0;

for (Uint32 indx = 0; indx < entries.size(); indx++)

{

if (maxSocketCurrentPass < entries[indx].socket)

maxSocketCurrentPass = entries[indx].socket;

monitor_2::monitor_2(void)

if (entries[indx].status == MonitorEntry::STATUS_IDLE)

: _session_dispatch(0), _accept_dispatch(0), _listeners(true, 0),

_ready(true, 0), _die(0), _requestCount(0)

{

try {

_idleEntries++;

FD_SET(entries[indx].socket, &fdread);

}

bsd_socket_factory _factory;

Add 1 then assign maxSocket accordingly. We add 1 to account for

descriptors starting at 0.

maxSocketCurrentPass++;

// set up the listener/acceptor

_entriesMutex.unlock();

pegasus_socket temp = pegasus_socket(&_factory);

temp.socket(PF_INET, SOCK_STREAM, 0);

// initialize the address

// The first argument to select() is ignored on Windows and it is not

memset(&_tickle_addr, 0, sizeof(_tickle_addr));

// a socket value. The original code assumed that the number of sockets

#ifdef PEGASUS_OS_ZOS

// and a socket value have the same type. On Windows they do not.

_tickle_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");

#ifdef PEGASUS_OS_TYPE_WINDOWS

int events = select(0, &fdread, NULL, NULL, &tv);

#else

_tickle_addr.sin_addr.s_addr = inet_addr("127.0.0.1");

int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);

#endif

_tickle_addr.sin_family = PF_INET;

_tickle_addr.sin_port = 0;

PEGASUS_SOCKLEN_SIZE _addr_size = sizeof(_tickle_addr);

_entriesMutex.lock();

temp.bind((struct sockaddr *)&_tickle_addr, sizeof(_tickle_addr));

temp.listen(3);

temp.getsockname((struct sockaddr*)&_tickle_addr, &_addr_size);

// set up the connector

pegasus_socket tickler = pegasus_socket(&_factory);

tickler.socket(PF_INET, SOCK_STREAM, 0);

struct sockaddr_in _addr;

memset(&_addr, 0, sizeof(_addr));

#ifdef PEGASUS_OS_ZOS

_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");

#else

_addr.sin_addr.s_addr = inet_addr("127.0.0.1");

#endif

_addr.sin_family = PF_INET;

_addr.sin_port = 0;

tickler.bind((struct sockaddr*)&_addr, sizeof(_addr));

tickler.connect((struct sockaddr*)&_tickle_addr, sizeof(_tickle_addr));

_tickler.set_sock(tickler);

struct timeval timeNow;

_tickler.set_type(INTERNAL);

Time::gettimeofday(&timeNow);

_tickler.set_state(BUSY);

struct sockaddr_in peer;

// After enqueue a message and the autoEntryMutex has been released and

memset(&peer, 0, sizeof(peer));

// locked again, the array of _entries can be changed. The ArrayIterator

PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);

// has be reset with the original _entries

entries.reset(_entries);

pegasus_socket accepted = temp.accept((struct sockaddr*)&peer, &peer_size);

if (events == PEGASUS_SOCKET_ERROR)

monitor_2_entry* _tickle = new monitor_2_entry(accepted, INTERNAL, 0, 0);

{

_tickle->set_state(BUSY);

int selectErrno = 0;

selectErrno = getSocketError();

_listeners.insert_first(_tickle);

PEG_TRACE((TRC_HTTP, Tracer::LEVEL1,

"Monitor::run - select() returned error %d.", selectErrno));

// The EBADF error indicates that one or more or the file

// descriptions was not valid. This could indicate that

// the entries structure has been corrupted or that

// we have a synchronization error.

PEGASUS_ASSERT(selectErrno != EBADF);

}

catch(...){ }

else if (events)

}

monitor_2::~monitor_2(void)

{

try {

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

monitor_2_entry* temp = _listeners.remove_first();

"Monitor::run select event received events = %d, monitoring %d "

while(temp){

"idle entries",

delete temp;

events, _idleEntries));

temp = _listeners.remove_first();

for (Uint32 indx = 0; indx < entries.size(); indx++)

}

{

}

// The Monitor should only look at entries in the table that are

catch(...){ }

// IDLE (i.e., owned by the Monitor).

}

if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&

(FD_ISSET(entries[indx].socket, &fdread)))

void monitor_2::run(void)

{

monitor_2_entry* temp;

MessageQueue* q = MessageQueue::lookup(entries[indx].queueId);

while(_die.value() == 0) {

PEGASUS_ASSERT(q != 0);

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"Monitor::run indx = %d, queueId = %d, q = %p",

indx, entries[indx].queueId, q));

struct timeval tv = {0, 0};

try

{

if (entries[indx].type == MonitorEntry::TYPE_CONNECTION)

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"entries[%d].type is TYPE_CONNECTION",

indx));

HTTPConnection *dst =

reinterpret_cast<HTTPConnection *>(q);

dst->_entry_index = indx;

// Update idle start time because we have received some

// data. Any data is good data at this point, and we'll

// keep the connection alive, even if we've exceeded

// the idleConnectionTimeout, which will be checked

// when we call closeConnectionOnTimeout() next.

Time::gettimeofday(&dst->_idleStartTime);

// Check for accept pending (ie. SSL handshake pending)

// or idle connection timeouts for sockets from which

// we received data (avoiding extra queue lookup below).

if (!dst->closeConnectionOnTimeout(&timeNow))

{

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"Entering HTTPConnection::run() for "

"indx = %d, queueId = %d, q = %p",

indx, entries[indx].queueId, q));

// place all sockets in the select set

FD_ZERO(&rd_fd_set);

try {

_listeners.lock(pegasus_thread_self());

temp = _listeners.next(0);

while(temp != 0 ){

if(temp->get_state() == CLOSED ){

monitor_2_entry* closed = temp;

temp = _listeners.next(closed);

_listeners.remove_no_lock(closed);

HTTPConnection2 *cn = monitor_2::remove_connection((Sint32)(closed->get_sock()));

delete cn;

delete closed;

}

if(temp == 0)

break;

Sint32 fd = (Sint32) temp->get_sock();

if(fd >= 0 )

FD_SET(fd , &rd_fd_set);

temp = _listeners.next(temp);

}

_listeners.unlock();

}

catch(...){

return;

}

// important - the dispatch routine has pointers to all the

// entries that are readable. These entries can be changed but

// the pointer must not be tampered with.

int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, NULL);

try {

_listeners.lock(pegasus_thread_self());

temp = _listeners.next(0);

while(temp != 0 ){

Sint32 fd = (Sint32) temp->get_sock();

if(fd >= 0 && FD_ISSET(fd, &rd_fd_set)) {

temp->set_state(BUSY);

FD_CLR(fd, &rd_fd_set);

monitor_2_entry* ready = new monitor_2_entry(*temp);

try

{

_ready.insert_first(ready);

dst->run();

}

catch(...)

{

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL1,

"Caught exception from "

"HTTPConnection::run()");

}

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

_requestCount++;

"Exited HTTPConnection::run()");

}

temp = _listeners.next(temp);

}

_listeners.unlock();

}

catch(...){

else if (entries[indx].type == MonitorEntry::TYPE_TICKLER)

return;

}

// now handle the sockets that are ready to read

_dispatch();

} // while alive

}

void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*))

{

void* old = (void *)_session_dispatch;

_tickler.reset();

_session_dispatch = dp;

return old;

}

else

void* monitor_2::set_accept_dispatch(void (*dp)(monitor_2_entry*))

{

void* old = (void*)_accept_dispatch;

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

_accept_dispatch = dp;

"Non-connection entry, indx = %d, has been "

return old;

"received.",

indx));

Message* msg = new SocketMessage(

entries[indx].socket, SocketMessage::READ);

entries[indx].status = MonitorEntry::STATUS_BUSY;

_entriesMutex.unlock();

q->enqueue(msg);

_entriesMutex.lock();

// After enqueue a message and the autoEntryMutex has

// been released and locked again, the array of

// entries can be changed. The ArrayIterator has to be

// reset with the latest _entries.

entries.reset(_entries);

entries[indx].status = MonitorEntry::STATUS_IDLE;

}

// important - the dispatch routine has pointers to all the

// entries that are readable. These entries can be changed but

// the pointer must not be tampered with.

void monitor_2::_dispatch(void)

{

monitor_2_entry* entry;

if(_ready.count() == 0 )

return;

try

{

entry = _ready.remove_first();

}

catch(...)

{

}

// else check for accept pending (ie. SSL handshake pending) or

// idle connection timeouts for sockets from which we did not

// receive data.

else if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&

entries[indx].type == MonitorEntry::TYPE_CONNECTION)

while(entry != 0 ) {

switch(entry->get_type()) {

case INTERNAL:

static char buffer[2];

entry->get_sock().disableBlocking();

entry->get_sock().read(&buffer, 2);

entry->get_sock().enableBlocking();

break;

case LISTEN:

{

static struct sockaddr peer;

MessageQueue* q = MessageQueue::lookup(entries[indx].queueId);

static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);

PEGASUS_ASSERT(q != 0);

entry->get_sock().disableBlocking();

HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);

pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);

dst->_entry_index = indx;

entry->get_sock().enableBlocking();

dst->closeConnectionOnTimeout(&timeNow);

monitor_2_entry *temp = add_entry(connected, SESSION, entry->get_accept(), entry->get_dispatch());

if(temp && _accept_dispatch != 0)

_accept_dispatch(temp);

}

break;

case SESSION:

if(_session_dispatch != 0 )

_session_dispatch(entry);

else {

static char buffer[4096];

int bytes = entry->get_sock().read(&buffer, 4096);

}

break;

case UNTYPED:

default:

break;

}

_requestCount--;

// else if "events" is zero (ie. select timed out) then we still need

delete entry;

// to check if there are any pending SSL handshakes that have timed out.

else

if(_ready.count() == 0 )

break;

try

{

entry = _ready.remove_first();

for (Uint32 indx = 0; indx < entries.size(); indx++)

}

catch(...)

{

if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&

entries[indx].type == MonitorEntry::TYPE_CONNECTION)

{

MessageQueue* q = MessageQueue::lookup(entries[indx].queueId);

PEGASUS_ASSERT(q != 0);

HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);

dst->_entry_index = indx;

dst->closeConnectionOnTimeout(&timeNow);

}

void monitor_2::stop(void)

void Monitor::stopListeningForConnections(Boolean wait)

{

_die = 1;

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");

// set boolean then tickle the server to recognize _stopConnections

_stopConnections = 1;

tickle();

// shut down the listener list, free the list nodes

if (wait)

_tickler.get_sock().close();

_listeners.shutdown_queue();

}

void monitor_2::tickle(void)

{

static char _buffer[] =

{

'0','0'

// Wait for the monitor to notice _stopConnections. Otherwise the

};

// caller of this function may unbind the ports while the monitor

// is still accepting connections on them.

_stopConnectionsSem.wait();

}

_tickler.get_sock().write(&_buffer, 2);

PEG_METHOD_EXIT();

}

monitor_2_entry* monitor_2::add_entry(pegasus_socket& ps,

int Monitor::solicitSocketMessages(

monitor_2_entry_type type,

SocketHandle socket,

void* accept_parm,

Uint32 queueId,

void* dispatch_parm)

Uint32 type)

{

monitor_2_entry* m2e = new monitor_2_entry(ps, type, accept_parm, dispatch_parm);

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");

AutoMutex autoMut(_entriesMutex);

try{

_listeners.insert_first(m2e);

}

catch(...){

delete m2e;

return 0;

}

tickle();

return m2e;

}

Boolean monitor_2::remove_entry(Sint32 s)

// Check to see if we need to dynamically grow the _entries array

{

// We always want the _entries array to be 2 bigger than the

monitor_2_entry* temp;

// current connections requested

try {

_solicitSocketCount++; // bump the count

_listeners.try_lock(pegasus_thread_self());

temp = _listeners.next(0);

while(temp != 0){

if(s == (Sint32)temp->_rep->psock ){

temp = _listeners.remove_no_lock(temp);

delete temp;

_listeners.unlock();

return true;

}

temp = _listeners.next(temp);

}

_listeners.unlock();

}

catch(...){

}

return false;

}

Uint32 monitor_2::getOutstandingRequestCount(void)

for (Uint32 i = _entries.size(); i < _solicitSocketCount + 1; i++)

{

return _requestCount.value();

_entries.append(MonitorEntry());

}

for (Uint32 index = 1; index < _entries.size(); index++)

HTTPConnection2* monitor_2::remove_connection(Sint32 sock)

{

HTTPConnection2* temp;

try

{

monitor_2::_connections.lock(pegasus_thread_self());

if (_entries[index].status == MonitorEntry::STATUS_EMPTY)

temp = monitor_2::_connections.next(0);

while(temp != 0 )

{

if(sock == temp->getSocket())

_entries[index].socket = socket;

{

_entries[index].queueId = queueId;

temp = monitor_2::_connections.remove_no_lock(temp);

_entries[index].type = type;

monitor_2::_connections.unlock();

_entries[index].status = MonitorEntry::STATUS_IDLE;

return temp;

}

PEG_METHOD_EXIT();

temp = monitor_2::_connections.next(temp);

return (int)index;

}

monitor_2::_connections.unlock();

}

catch(...)

{

}

return 0;

}

// decrease the count, if we are here we didn't do anything meaningful

_solicitSocketCount--;

PEG_METHOD_EXIT();

return -1;

}

Boolean monitor_2::insert_connection(HTTPConnection2* connection)

void Monitor::unsolicitSocketMessages(SocketHandle socket)

{

try

PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");

AutoMutex autoMut(_entriesMutex);

Start at index = 1 because _entries[0] is the tickle entry which

never needs to be reset to EMPTY;

for (Uint32 index = 1; index < _entries.size(); index++)

{

monitor_2::_connections.insert_first(connection);

if (_entries[index].socket == socket)

}

catch(...)

{

return false;

_entries[index].reset();

_solicitSocketCount--;

break;

}

return true;

}

Dynamic Contraction:

To remove excess entries we will start from the end of the _entries

array and remove all entries with EMPTY status until we find the

first NON EMPTY. This prevents the positions, of the NON EMPTY

entries, from being changed.

for (Uint32 index = _entries.size() - 1;

(_entries[index].status == MonitorEntry::STATUS_EMPTY) &&

(index >= MAX_NUMBER_OF_MONITOR_ENTRIES);

index--)

{

_entries.remove(index);

}

PEG_METHOD_EXIT();

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.54
changed lines
	Added in v.1.142

No CVS admin address has been configured