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.97 and 1.141

version 1.97, 2005/11/15 14:40:29

version 1.141, 2012/12/13 14:38:55

Line 1

//%2005////////////////////////////////////////////////////////////////////////

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

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

// Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.

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

// this work for additional information regarding copyright ownership.

// IBM Corp.; EMC Corporation, The Open Group.

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

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

// IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.

// License.

// EMC Corporation; VERITAS Software Corporation; The Open Group.

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

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

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

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

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

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

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

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

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

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

// 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 ABOVE COPYRIGHT NOTICE AND THIS PERMISSION NOTICE SHALL BE INCLUDED IN

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

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

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

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

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

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

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

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

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

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

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

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

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

// 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

// Amit K Arora (Bug#1153) amita@in.ibm.com

// Alagaraja Ramasubramanian (alags_raj@in.ibm.com) for Bug#1090

// Sushma Fernandes (sushma@hp.com) for Bug#2057

// Josephine Eskaline Joyce (jojustin@in.ibm.com) for PEP#101

// Roger Kumpf, Hewlett-Packard Company (roger_kumpf@hp.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>

#include <Pegasus/Common/MessageQueueService.h>

#include <Pegasus/Common/Exception.h>

#include "ArrayIterator.h"

#ifdef PEGASUS_OS_TYPE_WINDOWS

#include "HostAddress.h"

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

#include <errno.h>

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

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

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

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

// Define a platform-neutral socket length type

#if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM) || defined(PEGASUS_OS_VMS)

typedef size_t PEGASUS_SOCKLEN_T;

#elif defined(PEGASUS_PLATFORM_AIX_RS_IBMCXX) || defined(PEGASUS_OS_LINUX) || (defined(PEGASUS_OS_SOLARIS) && !defined(SUNOS_5_6))

typedef socklen_t PEGASUS_SOCKLEN_T;

#else

typedef int PEGASUS_SOCKLEN_T;

#endif

static AtomicInt _connections(0);

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

// Monitor

// Tickler

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

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

Tickler::Tickler()

Monitor::Monitor()

: _listenSocket(PEGASUS_INVALID_SOCKET),

: _stopConnections(0),

_clientSocket(PEGASUS_INVALID_SOCKET),

_stopConnectionsSem(0),

_serverSocket(PEGASUS_INVALID_SOCKET)

_solicitSocketCount(0),

_tickle_client_socket(-1),

_tickle_server_socket(-1),

_tickle_peer_socket(-1)

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

try

Socket::initializeInterface();

{

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

_initialize();

}

// setup the tickler

catch (...)

initializeTickler();

// Start the count at 1 because initilizeTickler()

// has added an entry in the first position of the

// _entries array

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

{

_MonitorEntry entry(0, 0, 0);

_uninitialize();

_entries.append(entry);

throw;

}

Monitor::~Monitor()

Tickler::~Tickler()

{

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

_uninitialize();

}

try{

void Tickler::notify()

if(_tickle_peer_socket >= 0)

{

Socket::close(_tickle_peer_socket);

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

}

if(_tickle_client_socket >= 0)

void Tickler::reset()

{

Socket::close(_tickle_client_socket);

// Clear all bytes from the tickle socket

}

char buffer[32];

if(_tickle_server_socket >= 0)

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

{

Socket::close(_tickle_server_socket);

}

catch(...)

#if defined(PEGASUS_OS_TYPE_UNIX)

// Use an anonymous pipe for the tickle connection.

void Tickler::_initialize()

{

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

int fds[2];

"Failed to close tickle sockets");

}

Socket::uninitializeInterface();

if (pipe(fds) == -1)

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

{

"returning from monitor destructor");

MessageLoaderParms parms(

"Common.Monitor.TICKLE_CREATE",

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

getSocketError());

throw Exception(parms);

}

void Monitor::initializeTickler(){

_serverSocket = fds[0];

_clientSocket = fds[1];

NOTE: On any errors trying to

setup out tickle connection,

throw an exception/end the server

/* setup the tickle server/listener */

Socket::disableBlocking(_serverSocket);

}

// get a socket for the server side

if((_tickle_server_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){

//handle error

MessageLoaderParms parms("Common.Monitor.TICKLE_CREATE",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

errno);

#else

WSAGetLastError());

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

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

void Tickler::_initialize()

{

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

// based on whether IPv6 is enabled.

Socket::initializeInterface();

# ifdef PEGASUS_ENABLE_IPV6

struct sockaddr_storage listenAddress;

struct sockaddr_storage clientAddress;

struct sockaddr_storage serverAddress;

# else

struct sockaddr_in listenAddress;

struct sockaddr_in clientAddress;

struct sockaddr_in serverAddress;

# endif

int addressFamily;

SocketLength addressLength;

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

# ifdef PEGASUS_ENABLE_IPV6

if (System::isIPv6StackActive())

{

// Use the IPv6 loopback address for the listen sockets

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

throw Exception(parms);

{

// Use the IPv4 loopback address for the listen sockets

HostAddress::convertTextToBinary(

HostAddress::AT_IPV4,

"127.0.0.1",

&reinterpret_cast<struct sockaddr_in*>(

&listenAddress)->sin_addr.s_addr);

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

AF_INET;

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

addressFamily = AF_INET;

addressLength = sizeof(struct sockaddr_in);

}

// initialize the address

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

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

clientAddress = listenAddress;

#ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM

#pragma convert(37)

#endif

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

#ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM

#pragma convert(0)

#endif

_tickle_server_addr.sin_family = PF_INET;

_tickle_server_addr.sin_port = 0;

PEGASUS_SOCKLEN_T _addr_size = sizeof(_tickle_server_addr);

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

// bind server side to socket

// Create the listen socket

if((::bind(_tickle_server_socket,

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

reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),

PEGASUS_INVALID_SOCKET)

sizeof(_tickle_server_addr))) < 0){

{

// handle error

MessageLoaderParms parms(

#ifdef PEGASUS_OS_ZOS

"Common.Monitor.TICKLE_CREATE",

MessageLoaderParms parms("Common.Monitor.TICKLE_BIND_LONG",

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

"Received error:$0 while binding the internal socket.",strerror(errno));

getSocketError());

#else

throw Exception(parms);

MessageLoaderParms parms("Common.Monitor.TICKLE_BIND",

}

// Bind the listen socket to the loopback address

if (::bind(

_listenSocket,

reinterpret_cast<struct sockaddr*>(&listenAddress),

addressLength) < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_BIND",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

// tell the kernel we are a server

// Listen for a connection from the tickle client

if((::listen(_tickle_server_socket,3)) < 0){

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

// handle error

{

MessageLoaderParms parms("Common.Monitor.TICKLE_LISTEN",

MessageLoaderParms parms(

"Common.Monitor.TICKLE_LISTEN",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

// make sure we have the correct socket for our server

// Verify we have the correct listen socket

int sock = ::getsockname(_tickle_server_socket,

SocketLength tmpAddressLength = addressLength;

reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),

int sock = ::getsockname(

&_addr_size);

_listenSocket,

if(sock < 0){

reinterpret_cast<struct sockaddr*>(&listenAddress),

// handle error

&tmpAddressLength);

MessageLoaderParms parms("Common.Monitor.TICKLE_SOCKNAME",

if (sock < 0)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_SOCKNAME",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

/* set up the tickle client/connector */

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

// get a socket for our tickle client

// Create the client socket

if((_tickle_client_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){

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

// handle error

PEGASUS_INVALID_SOCKET)

MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CREATE",

{

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

MessageLoaderParms parms(

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

"Common.Monitor.TICKLE_CLIENT_CREATE",

errno);

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

#else

"socket.",

WSAGetLastError());

getSocketError());

#endif

throw Exception(parms);

}

// setup the address of the client

// Bind the client socket to the loopback address

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

if (::bind(

#ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM

_clientSocket,

#pragma convert(37)

reinterpret_cast<struct sockaddr*>(&clientAddress),

#endif

addressLength) < 0)

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

{

#ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM

MessageLoaderParms parms(

#pragma convert(0)

"Common.Monitor.TICKLE_CLIENT_BIND",

#endif

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

_tickle_client_addr.sin_family = PF_INET;

"socket.",

_tickle_client_addr.sin_port = 0;

getSocketError());

throw Exception(parms);

}

// bind socket to client side

// Connect the client socket to the listen socket address

if((::bind(_tickle_client_socket,

if (::connect(

reinterpret_cast<struct sockaddr*>(&_tickle_client_addr),

_clientSocket,

sizeof(_tickle_client_addr))) < 0){

reinterpret_cast<struct sockaddr*>(&listenAddress),

// handle error

addressLength) < 0)

MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_BIND",

{

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

MessageLoaderParms parms(

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

"Common.Monitor.TICKLE_CLIENT_CONNECT",

errno);

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

#else

"socket.",

WSAGetLastError());

getSocketError());

#endif

throw Exception(parms);

}

// connect to server side

if((::connect(_tickle_client_socket,

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

reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),

sizeof(_tickle_server_addr))) < 0){

// handle error

tmpAddressLength = addressLength;

MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CONNECT",

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

// Accept the client socket connection.

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

_serverSocket = ::accept(

errno);

_listenSocket,

#else

reinterpret_cast<struct sockaddr*>(&serverAddress),

WSAGetLastError());

&tmpAddressLength);

#endif

if (_serverSocket == PEGASUS_SOCKET_ERROR)

{

MessageLoaderParms parms(

"Common.Monitor.TICKLE_ACCEPT",

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

"connection.",

getSocketError());

throw Exception(parms);

}

/* set up the slave connection */

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

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

PEGASUS_SOCKLEN_T peer_size = sizeof(_tickle_peer_addr);

pegasus_sleep(1);

Socket::close(_listenSocket);

// this call may fail, we will try a max of 20 times to establish this peer connection

Socket::disableBlocking(_serverSocket);

if((_tickle_peer_socket = ::accept(_tickle_server_socket,

Socket::disableBlocking(_clientSocket);

reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),

&peer_size)) < 0){

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

// Only retry on non-windows platforms.

if(_tickle_peer_socket == -1 && errno == EAGAIN)

{

int retries = 0;

{

pegasus_sleep(1);

_tickle_peer_socket = ::accept(_tickle_server_socket,

reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),

&peer_size);

retries++;

} while(_tickle_peer_socket == -1 && errno == EAGAIN && retries < 20);

}

#endif

void Tickler::_uninitialize()

{

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

try

{

Socket::close(_serverSocket);

Socket::close(_clientSocket);

Socket::close(_listenSocket);

}

if(_tickle_peer_socket == -1){

catch (...)

// handle error

{

MessageLoaderParms parms("Common.Monitor.TICKLE_ACCEPT",

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL2,

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

"Failed to close tickle sockets");

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

// add the tickler to the list of entries to be monitored and set to IDLE because Monitor only

Socket::uninitializeInterface();

// checks entries with IDLE state for events

_MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);

entry._status = _MonitorEntry::IDLE;

_entries.append(entry);

}

void Monitor::tickle(void)

{

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

static char _buffer[] =

// Monitor

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

#define MAX_NUMBER_OF_MONITOR_ENTRIES 32

Monitor::Monitor()

: _stopConnections(0),

_stopConnectionsSem(0),

_solicitSocketCount(0)

{

'0','0'

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));

AutoMutex autoMutex(_tickle_mutex);

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

Socket::disableBlocking(_tickle_client_socket);

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

Socket::write(_tickle_client_socket,&_buffer, 2);

{

Socket::enableBlocking(_tickle_client_socket);

_entries.append(MonitorEntry());

}

void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )

Monitor::~Monitor()

{

// Set the state to requested state

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

_entries[index]._status = status;

"returning from monitor destructor");

}

Boolean Monitor::run(Uint32 milliseconds)

void Monitor::tickle()

{

_tickler.notify();

}

Boolean handled_events = false;

void Monitor::setState(

int i = 0;

Uint32 index,

MonitorEntry::Status status)

{

AutoMutex autoEntryMutex(_entriesMutex);

// Set the state to requested state

_entries[index].status = status;

}

void Monitor::run(Uint32 milliseconds)

{

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

fd_set fdread;

FD_ZERO(&fdread);

AutoMutex autoEntryMutex(_entry_mut);

AutoMutex autoEntryMutex(_entriesMutex);

ArrayIterator<MonitorEntry> entries(_entries);

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

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

// entries

if (_stopConnections.get() == 1)

{

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

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

{

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

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

{

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

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

{

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

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

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

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

{

// remove the entry

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

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

}

else

{

// set status to DYING

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

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

}

Line 390

Line 413

_stopConnectionsSem.signal();

}

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

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

{

const _MonitorEntry &entry = _entries[indx];

const MonitorEntry& entry = entries[indx];

if ((entry._status.get() == _MonitorEntry::DYING) &&

(entry._type == Monitor::CONNECTION))

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

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

{

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

PEGASUS_ASSERT(q != 0);

Line 411

Line 435

if (h._responsePending == true)

{

Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "Monitor::run - "

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

"Ignoring connection delete request because "

"Monitor::run - Ignoring connection delete request "

"responses are still pending. "

"because responses are still pending. "

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

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

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

continue;

}

h._connectionClosePending = false;

MessageQueue &o = h.get_owner();

HTTPAcceptor &o = h.getOwningAcceptor();

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

message->dest = o.getQueueId();

Line 429

Line 453

// 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

// implementation of the entries object. Note that the

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

// 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.

autoEntryMutex.unlock();

_entriesMutex.unlock();

o.enqueue(message);

autoEntryMutex.lock();

_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);

}

Line 449

Line 479

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

because we have to traverse the entire array.

PEGASUS_SOCKET maxSocketCurrentPass = 0;

SocketHandle maxSocketCurrentPass = 0;

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

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

{

if(maxSocketCurrentPass < _entries[indx].socket)

if (maxSocketCurrentPass < entries[indx].socket)

maxSocketCurrentPass = _entries[indx].socket;

maxSocketCurrentPass = entries[indx].socket;

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

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

{

_idleEntries++;

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

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

}

Line 468

Line 498

maxSocketCurrentPass++;

autoEntryMutex.unlock();

_entriesMutex.unlock();

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

Line 480

Line 510

#else

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

#endif

autoEntryMutex.lock();

int selectErrno = getSocketError();

#ifdef PEGASUS_OS_TYPE_WINDOWS

_entriesMutex.lock();

if(events == SOCKET_ERROR)

#else

struct timeval timeNow;

if(events == -1)

Time::gettimeofday(&timeNow);

#endif

// 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);

if (events == PEGASUS_SOCKET_ERROR)

{

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

PEG_TRACE((TRC_HTTP, Tracer::LEVEL1,

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

"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

// the entries structure has been corrupted or that

// we have a synchronization error.

PEGASUS_ASSERT(errno != EBADF);

PEGASUS_ASSERT(selectErrno != EBADF);

}

else if (events)

{

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

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

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

events, _idleEntries);

"idle entries",

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

events, _idleEntries));

{

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

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

{

// owned by the Monitor).

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

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

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

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

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

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

{

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

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);

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

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

try

{

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

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

{

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

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

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

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

indx));

// Do not update the entry just yet. The entry gets updated once

HTTPConnection *dst =

// the request has been read.

reinterpret_cast<HTTPConnection *>(q);

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

dst->_entry_index = indx;

// If allocate_and_awaken failure, retry on next iteration

// Update idle start time because we have received some

/* Removed for PEP 183.

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

if (!MessageQueueService::get_thread_pool()->allocate_and_awaken(

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

(void *)q, _dispatch))

// the idleConnectionTimeout, which will be checked

{

// when we call closeConnectionOnTimeout() next.

Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,

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

"Monitor::run: Insufficient resources to process request.");

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

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

return true;

// or idle connection timeouts for sockets from which

}

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

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

// Added for PEP 183

{

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

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

"Entering HTTPConnection::run() for "

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

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

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

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

try

{

dst->run(1);

dst->run();

}

catch (...)

{

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

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL1,

"Monitor::_dispatch: exception received");

"Caught exception from "

"HTTPConnection::run()");

}

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

PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,

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

"Exited HTTPConnection::run()");

}

// It is possible the entry status may not be set to busy.

}

// The following will fail in that case.

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

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

{

// Once the HTTPConnection thread has set the status value to either

_tickler.reset();

// Monitor::DYING or Monitor::IDLE, it has returned control of the connection

// to the Monitor. It is no longer permissible to access the connection

// or the entry in the _entries table.

// The following is not relevant as the worker thread or the

// reader thread will update the status of the entry.

//if (dst->_connectionClosePending)

//{

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

//}

//else

//{

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

//}

// end Added for PEP 183

}

else if( _entries[indx]._type == Monitor::INTERNAL){

// set ourself to BUSY,

// read the data

// and set ourself back to IDLE

_entries[indx]._status.get() == _MonitorEntry::BUSY;

static char buffer[2];

Socket::disableBlocking(_entries[indx].socket);

Sint32 amt = Socket::read(_entries[indx].socket,&buffer, 2);

Socket::enableBlocking(_entries[indx].socket);

_entries[indx]._status.get() == _MonitorEntry::IDLE;

}

else

{

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

PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,

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

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

int events = 0;

"received.",

events |= SocketMessage::READ;

indx));

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

Message* msg = new SocketMessage(

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

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

autoEntryMutex.unlock();

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

_entriesMutex.unlock();

q->enqueue(msg);

autoEntryMutex.lock();

_entriesMutex.lock();

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

return true;

// 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;

}

catch(...)

{

}

handled_events = true;

}

// 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)

{

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

PEGASUS_ASSERT(q != 0);

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

dst->_entry_index = indx;

dst->closeConnectionOnTimeout(&timeNow);

}

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

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

else

{

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

{

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);

}

return(handled_events);

}

void Monitor::stopListeningForConnections(Boolean wait)

Line 626

Line 670

// 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.

try

_stopConnectionsSem.wait();

{

_stopConnectionsSem.time_wait(10000);

}

catch (TimeOut &)

{

// The monitor is probably busy processng a very long request, and is

// not accepting connections. Let the caller unbind the ports.

}

PEG_METHOD_EXIT();

Line 642

Line 678

int Monitor::solicitSocketMessages(

PEGASUS_SOCKET socket,

SocketHandle socket,

Uint32 events,

Uint32 queueId,

int type)

Uint32 type)

{

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

AutoMutex autoMut(_entry_mut);

AutoMutex autoMut(_entriesMutex);

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

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

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

// current connections requested

_solicitSocketCount++; // bump the count

int size = (int)_entries.size();

if((int)_solicitSocketCount >= (size-1)){

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

for(int i = 0; i < ((int)_solicitSocketCount - (size-1)); i++){

{

_MonitorEntry entry(0, 0, 0);

_entries.append(MonitorEntry());

_entries.append(entry);

}

int index;

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

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

{

try

{

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

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

{

_entries[index].socket = socket;

_entries[index].queueId = queueId;

_entries[index]._type = type;

_entries[index].type = type;

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

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

return index;

PEG_METHOD_EXIT();

return (int)index;

}

catch(...)

{

}

_solicitSocketCount--; // decrease the count, if we are here we didnt do anything meaningful

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

_solicitSocketCount--;

PEG_METHOD_EXIT();

return -1;

}

void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket)

void Monitor::unsolicitSocketMessages(SocketHandle socket)

{

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

AutoMutex autoMut(_entry_mut);

AutoMutex autoMut(_entriesMutex);

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

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

to be EMPTY;

never needs to be reset to EMPTY;

unsigned int index;

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

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

{

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

{

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

_entries[index].reset();

_entries[index].socket = PEGASUS_INVALID_SOCKET;

_solicitSocketCount--;

break;

}

Line 710

Line 741

Dynamic Contraction:

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

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

and remove all entries with EMPTY status until we find the first NON EMPTY.

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

This prevents the positions, of the NON EMPTY entries, from being changed.

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

entries, from being changed.

index = _entries.size() - 1;

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

while(_entries[index]._status.get() == _MonitorEntry::EMPTY){

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

if(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES)

(index >= MAX_NUMBER_OF_MONITOR_ENTRIES);

_entries.remove(index);

index--)

index--;

}

PEG_METHOD_EXIT();

}

// Note: this is no longer called with PEP 183.

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)

{

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

_entries.remove(index);

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);

}

catch (...)

{

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

"Monitor::_dispatch: exception received");

}

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

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

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

// Once the HTTPConnection thread has set the status value to either

PEG_METHOD_EXIT();

// Monitor::DYING or Monitor::IDLE, it has returned control of the connection

// to the Monitor. It is no longer permissible to access the connection

// or the entry in the _entries table.

if (dst->_connectionClosePending)

{

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

}

else

{

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

}

return 0;

}

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.97
changed lines
	Added in v.1.141

No CVS admin address has been configured