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.103.10.22 and 1.115.14.1

version 1.103.10.22, 2006/07/21 18:17:51

version 1.115.14.1, 2006/11/03 04:52:41

Line 29

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

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

Line 51

Line 43

#include <Pegasus/Common/MessageQueueService.h>

#include <Pegasus/Common/Exception.h>

#include "ArrayIterator.h"

#include <errno.h>

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

// Maximum iterations of Pipe processing in Monitor::run

//const static DWORD MAX_BUFFER_SIZE = 4096; // 4 kilobytes

const Uint32 maxIterations = 2;

#ifdef PEGASUS_OS_TYPE_WINDOWS

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

# 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

static AtomicInt _connections(0);

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

Mutex Monitor::_cout_mut;

#endif

#ifdef PEGASUS_OS_TYPE_WINDOWS

Mutex Monitor::_cout_mut;

// Added for NamedPipe implementation for windows

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

#define PIPE_INCREMENT 1

#endif

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

// Monitor

Line 104

Line 73

_tickle_client_socket(-1),

_tickle_server_socket(-1),

_tickle_peer_socket(-1)

{

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

,_solicitPipeCount(0)

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

{

int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;

Socket::initializeInterface();

_entries.reserveCapacity(numberOfMonitorEntriesToAllocate);

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

_entries_pipe.reserveCapacity(numberOfMonitorEntriesToAllocate);

#endif

// setup the tickler

initializeTickler();

Line 126

Line 95

_MonitorEntry entry(0, 0, 0);

_entries.append(entry);

}

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

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

{

AutoMutex automut(Monitor::_cout_mut);

_MonitorEntry entry(0, 0, 0);

PEGASUS_STD(cout) << "Exiting: Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

_entries_pipe.append(entry);

}

#endif

}

Monitor::~Monitor()

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

uninitializeTickler();

{

Socket::uninitializeInterface();

AutoMutex automut(Monitor::_cout_mut);

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

PEGASUS_STD(cout) << "Entering: Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

"returning from monitor destructor");

}

#endif

void Monitor::uninitializeTickler(){

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

try{

Line 164

Line 135

"Failed to close tickle sockets");

}

Socket::uninitializeInterface();

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

"returning from monitor destructor");

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

}

void Monitor::initializeTickler(){

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::initializeTickler(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

NOTE: On any errors trying to

setup out tickle connection,

Line 189

Line 145

/* setup the tickle server/listener */

// try until the tcpip is restarted

{

// get a socket for the server side

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

if((_tickle_server_socket = Socket::createSocket(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)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

Line 215

Line 171

_tickle_server_addr.sin_family = PF_INET;

_tickle_server_addr.sin_port = 0;

PEGASUS_SOCKLEN_T _addr_size = sizeof(_tickle_server_addr);

SocketLength _addr_size = sizeof(_tickle_server_addr);

// bind server side to socket

if((::bind(_tickle_server_socket,

reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),

sizeof(_tickle_server_addr))) < 0){

sizeof(_tickle_server_addr))) < 0)

{

// handle error

#ifdef PEGASUS_OS_ZOS

MessageLoaderParms parms("Common.Monitor.TICKLE_BIND_LONG",

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

"Received error:$0 while binding the internal socket."

,strerror(errno));

#else

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

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

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

{

// handle error

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

}

Line 254

Line 205

int sock = ::getsockname(_tickle_server_socket,

reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),

&_addr_size);

if(sock < 0){

if(sock < 0)

{

// handle error

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

// get a socket for our tickle client

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

if((_tickle_client_socket = Socket::createSocket(PF_INET, SOCK_STREAM, 0))

== PEGASUS_INVALID_SOCKET)

{

// handle error

MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CREATE",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

Line 296

Line 242

// bind socket to client side

if((::bind(_tickle_client_socket,

reinterpret_cast<struct sockaddr*>(&_tickle_client_addr),

sizeof(_tickle_client_addr))) < 0){

sizeof(_tickle_client_addr))) < 0)

{

// handle error

MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_BIND",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

// connect to server side

if((::connect(_tickle_client_socket,

reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),

sizeof(_tickle_server_addr))) < 0){

sizeof(_tickle_server_addr))) < 0)

{

// handle error

MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CONNECT",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

}

/* set up the slave connection */

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

PEGASUS_SOCKLEN_T peer_size = sizeof(_tickle_peer_addr);

SocketLength peer_size = sizeof(_tickle_peer_addr);

pegasus_sleep(1);

Threads::sleep(1);

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

if((_tickle_peer_socket = ::accept(_tickle_server_socket,

reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),

&peer_size)) < 0){

&peer_size)) < 0)

{

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

// Only retry on non-windows platforms.

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

if(_tickle_peer_socket == PEGASUS_SOCKET_ERROR

&& getSocketError() == PEGASUS_NETWORK_TRYAGAIN)

{

int retries = 0;

{

pegasus_sleep(1);

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

} while(_tickle_peer_socket == PEGASUS_SOCKET_ERROR

&& getSocketError() == PEGASUS_NETWORK_TRYAGAIN

&& retries < 20);

}

// TCP/IP is down, destroy sockets and retry again.

if(_tickle_peer_socket == PEGASUS_SOCKET_ERROR &&

getSocketError() == PEGASUS_NETWORK_TCPIP_STOPPED )

{

// destroy everything

uninitializeTickler();

// retry again.

continue;

}

#endif

}

if(_tickle_peer_socket == -1){

if(_tickle_peer_socket == PEGASUS_SOCKET_ERROR)

{

// handle error

MessageLoaderParms parms("Common.Monitor.TICKLE_ACCEPT",

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

#if !defined(PEGASUS_OS_TYPE_WINDOWS)

getSocketError());

errno);

#else

WSAGetLastError());

#endif

throw Exception(parms);

} else

{

// socket is ok

break;

}

} while(1); // try until TCP/IP is restarted

Socket::disableBlocking(_tickle_peer_socket);

Socket::disableBlocking(_tickle_client_socket);

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

// checks entries with IDLE state for events

_MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);

entry._status = _MonitorEntry::IDLE;

// is the tickler initalized as first socket on startup ?

if (_entries.size()==0)

{

// if yes, append a new entry

_entries.append(entry);

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

}

else

{

AutoMutex automut(Monitor::_cout_mut);

// if not, overwrite the tickler entry with new socket

PEGASUS_STD(cout) << "Exiting: Monitor::initializeTickler(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

_entries[0]=entry;

}

#endif

}

void Monitor::tickle(void)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::tickle(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

static char _buffer[] =

{

'0','0'

};

AutoMutex autoMutex(_tickle_mutex);

Socket::disableBlocking(_tickle_client_socket);

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

Socket::enableBlocking(_tickle_client_socket);

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::tickle(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

}

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

Line 403

Line 355

_entries[index]._status = status;

}

Boolean Monitor::run(Uint32 milliseconds)

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

{

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

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

// Set the state to requested state

PEGASUS_STD(cout) << "Entering: Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

_entries_pipe[index]._status = status;

}

#endif

Boolean handled_events = false;

int i = 0;

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

fd_set fdread;

int Monitor::handlePipe()

FD_ZERO(&fdread);

{

AutoMutex autoEntryMutex(_entry_mut);

ArrayIterator<_MonitorEntry> entries(_entries);

ArrayIterator<_MonitorEntry> entries(_entries_pipe);

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

if (_stopConnections.get() == 1)

Line 472

Line 417

if (h._responsePending == true)

{

if (!entry.namedPipeConnection)

{

Tracer::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());

}

else

{

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

"Ignoring connection delete request because "

"responses are still pending. "

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

(void *)&h, h.getNamedPipe().getPipe());

}

continue;

}

h._connectionClosePending = false;

MessageQueue &o = h.get_owner();

Message* message;

Message* message = 0;

if (!entry.namedPipeConnection)

{

message= new CloseConnectionMessage(entry.socket);

}

else

{

message= new CloseConnectionMessage(entry.namedPipe);

}

message->dest = o.getQueueId();

// HTTPAcceptor is responsible for closing the connection.

Line 516

Line 444

// unlocked will not result in an ArrayIndexOutOfBounds

// exception.

autoEntryMutex.unlock();

_entry_mut.unlock();

o.enqueue(message);

autoEntryMutex.lock();

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

entries.reset(_entries_pipe);

}

Line 533

Line 461

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

because we have to traverse the entire array.

//Array<HANDLE> pipeEventArray;

SocketHandle maxSocketCurrentPass = 0;

PEGASUS_SOCKET maxSocketCurrentPass = 0;

int indx = 0;

int indx;

#ifdef PEGASUS_OS_TYPE_WINDOWS

//This array associates named pipe connections to their place in [indx]

//in the entries array. The value in poition zero of the array is the

//index of the fist named pipe connection in the entries array

Array <Uint32> indexPipeCountAssociator;

int pipeEntryCount=0;

int MaxPipes = PIPE_INCREMENT;

HANDLE* hEvents = new HANDLE[PIPE_INCREMENT];

// List of Pipe Handlers

HANDLE * hPipeList = new HANDLE[PIPE_INCREMENT];

#endif

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

{

if (!entries[indx].namedPipeConnection)

continue;

#ifdef PEGASUS_OS_TYPE_WINDOWS

if(entries[indx].isNamedPipeConnection())

{

//entering this clause mean that a Named Pipe connection is at entries[indx]

//cout << "In Monitor::run in clause to to create array of for WaitformultipuleObjects" << endl;

//cout << "In Monitor::run - pipe being added to array is " << entries[indx].namedPipe.getName() << endl;

entries[indx].pipeSet = false;

// We can Keep a counter in the Monitor class for the number of named pipes ...

// Which can be used here to create the array size for hEvents..( obviously before this for loop.:-) )

if (pipeEntryCount >= MaxPipes)

{

// cout << "Monitor::run 'if (pipeEntryCount >= MaxPipes)' begining - pipeEntryCount=" <<

// pipeEntryCount << " MaxPipes=" << MaxPipes << endl;

MaxPipes += PIPE_INCREMENT;

HANDLE* temp_hEvents = new HANDLE[MaxPipes];

HANDLE* temp_pList = new HANDLE[MaxPipes];

for (Uint32 i =0;i<pipeEntryCount;i++)

{

temp_hEvents[i] = hEvents[i];

temp_pList[i] = hPipeList[i];

}

delete [] hPipeList;

delete [] hEvents;

hPipeList = temp_pList;

hEvents = temp_hEvents;

// cout << "Monitor::run 'if (pipeEntryCount >= MaxPipes)' ending"<< endl;

}

hPipeList[pipeEntryCount] = entries[indx].namedPipe.getPipe();

//pipeEventArray.append((entries[indx].namedPipe.getOverlap()).hEvent);

hEvents[pipeEntryCount] = entries[indx].namedPipe.getOverlap()->hEvent;

indexPipeCountAssociator.append(indx);

pipeEntryCount++;

}

else

#endif

{

if(maxSocketCurrentPass < entries[indx].socket)

maxSocketCurrentPass = entries[indx].socket;

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

{

_idleEntries++;

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

}

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

descriptors starting at 0.

maxSocketCurrentPass++;

autoEntryMutex.unlock();

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

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

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

int events;

int pEvents = -1;

int pEvents;

int pCount = -1;

BOOL bPeekPipe = 0;

DWORD dwBytesAvail=0;

// The pipe is sniffed and check if there are any data. If available, the

// message is picked from the Queue and appropriate methods are invoked.

#ifdef PEGASUS_OS_TYPE_WINDOWS

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

// pipeProcessCount records the number of requests that are processed.

// At the end of loop this is verified against the count of request

// on local connection . If there are any pipes which needs to be

// processed we would apply delay and then proceed to iterate.

//if (events == NULL)

Uint32 pipeProcessCount =0;

//{ // This connection uses namedPipes

events = 0;

_entry_mut.unlock();

DWORD dwWait=NULL;

pEvents = 0;

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

// pipeIndex is used to index into indexPipeCountAssociator to fetch

// index of the _MonitorEntry of Monitor

for (int pipeIndex = 0; pipeIndex < pipeEntryCount; pipeIndex++)

{

AutoMutex automut(Monitor::_cout_mut);

dwBytesAvail = 0;

cout << "Monitor::run - Calling WaitForMultipleObjects\n";

}

#endif

// }

//this should be in a try block

dwWait = WaitForMultipleObjects(

bPeekPipe = ::PeekNamedPipe(hPipeList[pipeIndex],

MaxPipes,

NULL,

hEvents, //ABB:- array of event objects

NULL,

FALSE, // ABB:-does not wait for all

NULL,

milliseconds); //ABB:- timeout value //WW this may need be shorter

&dwBytesAvail,

NULL

);

if(dwWait == WAIT_TIMEOUT)

// If peek on NamedPipe was successfull and data is available

{

if (bPeekPipe && dwBytesAvail)

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

cout << "Wait WAIT_TIMEOUT\n";

cout << "Monitor::run before the select in TIMEOUT clause events = " << events << endl;

}

#endif

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

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

AutoMutex automut(Monitor::_cout_mut);

cout << "Monitor::run after the select in TIMEOUT clause events = " << events << endl;

#endif

// Tracer::trace(TRC_HTTP,Tracer::LEVEL4," PIPE_PEEKING FOUND = %u BYTES", dwBytesAvail);

// Sleep(2000);

pEvents = 1;

//continue;

Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "EVENT TRIGGERED in Pipe = %u ",entries[indexPipeCountAssociator[pipeIndex]].namedPipe.getPipe());

entries[indexPipeCountAssociator[pipeIndex]].pipeSet = true;

int pIndx = indexPipeCountAssociator[pipeIndex];

//return false; // I think we do nothing.... Mybe there is a socket connection... so

if ((entries[pIndx]._status.get() == _MonitorEntry::IDLE) &&

// cant return.

entries[pIndx].namedPipe.isConnected() &&

}

(pEvents))

else if (dwWait == WAIT_FAILED)

{

if (GetLastError() == 6) //WW this may be too specific

MessageQueue *q = 0;

try

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

AutoMutex automut(Monitor::_cout_mut);

cout << "Monitor::run about to call 'select since waitForMultipleObjects failed\n";

#endif

/********* NOTE

this time (tv) combined with the waitForMulitpleObjects timeout is

too long it will cause the client side to time out

******************/

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

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

}

else

catch (Exception e)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

e.getMessage();

AutoMutex automut(Monitor::_cout_mut);

cout << "Wait Failed returned\n";

cout << "failed with " << GetLastError() << "." << endl;

#endif

pEvents = -1;

return false;

}

catch(...)

else

{

int pCount = dwWait - WAIT_OBJECT_0; // determines which pipe

}

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

"Monitor::run indx = %d, queueId = %d,\

q = %p",pIndx, entries[pIndx].queueId, q);

try

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

if (entries[pIndx]._type == Monitor::CONNECTION)

{

AutoMutex automut(Monitor::_cout_mut);

// cout << endl << "****************************" <<

// "Monitor::run WaitForMultiPleObject returned activity on server pipe: "<<

// pCount<< endl << endl;

cout << "Monitor::run WaitForMultiPleObject returned activity pipeEntrycount is " <<

pipeEntryCount <<

" this is the type " << entries[indexPipeCountAssociator[pCount]]._type << " this is index " << indexPipeCountAssociator[pCount] << endl;

}

#endif

/* There is a timeing problem here sometimes the wite in HTTPConnection i s

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

not all the way done (has not _monitor->setState (_entry_index, _MonitorEntry::IDLE) )

"entries[indx].type for indx = \

there for that should be done here if it is not done alread*/

%d is Monitor::CONNECTION",

pIndx);

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

HTTPConnection *dst = reinterpret_cast \

<HTTPConnection *>(q);

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

if (entries[indexPipeCountAssociator[pCount]]._status.get() != _MonitorEntry::IDLE)

try

{

this->setState(indexPipeCountAssociator[pCount], _MonitorEntry::IDLE);

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

AutoMutex automut(Monitor::_cout_mut);

cout << "setting state of index " << indexPipeCountAssociator[pCount] << " to IDLE" << endl;

dst->run(1);

#endif

}

// Record that the requested data is read/Written

pipeProcessCount++;

}

catch (...)

{

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

"Monitor::_dispatch: \

exception received");

}

pEvents = 1;

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

"Monitor::_dispatch: exited \

\run() index %d",

dst->_entry_index);

//this statment gets the pipe entry that was trigered

entries[indexPipeCountAssociator[pCount]].pipeSet = true;

}

#else

else

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

#endif

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

#ifdef PEGASUS_OS_TYPE_WINDOWS

if(pEvents == -1)

{

/* The condition

entries[indx]._type == Monitor::INTERNAL can be

ignored for pipes as the tickler is of

Monitor::INTERNAL type. The tickler is

a socket.

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

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

"Non-connection entry, indx = %d,\

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

has been received.", pIndx);

// descriptions was not valid. This could indicate that

int events = 0;

// the entries structure has been corrupted or that

Message *msg = 0;

// we have a synchronization error.

// We need to generate an assert here...

pEvents |= NamedPipeMessage::READ;

PEGASUS_ASSERT(GetLastError()!= EBADF);

msg = new NamedPipeMessage(entries[pIndx].namedPipe, pEvents);

entries[pIndx]._status = _MonitorEntry::BUSY;

q->enqueue(msg);

_entry_mut.lock();

entries.reset(_entries_pipe);

entries[pIndx]._status = _MonitorEntry::IDLE;

delete [] hPipeList;

return 1;

}

catch(...)

if(events == SOCKET_ERROR)

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

}

else if ((events)||(pEvents))

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

_entry_mut.lock();

}

delete [] hPipeList;

//If no request received on pipe rest for 4 seconds and give chance to Socket Thread

if (!pipeProcessCount)

{

AutoMutex automut(Monitor::_cout_mut);

Sleep(4000);

cout << "IN Monior::run events= " << events << " pEvents= " << pEvents<< endl;

}

return 1;

}

#endif

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

void Monitor::run(Uint32 milliseconds)

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

{

events, _idleEntries);

int i = 0;

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

fd_set fdread;

FD_ZERO(&fdread);

AutoMutex autoEntryMutex(_entry_mut);

ArrayIterator<_MonitorEntry> entries(_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++)

{

//cout << "Monitor::run at start of 'for( int indx = 0; indx ' - index = " << indx << endl;

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

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

{

// owned by the Monitor).

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

// cout << endl << " status of entry " << indx << " is " << entries[indx]._status.get() << endl;

{

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

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

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

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

(entries[indx].isNamedPipeConnection() && entries[indx].pipeSet && (pEvents))))

{

// remove the entry

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

}

else

{

// set status to DYING

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

}

_stopConnections = 0;

_stopConnectionsSem.signal();

}

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

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

{

AutoMutex automut(Monitor::_cout_mut);

const _MonitorEntry &entry = entries[indx];

cout <<"Monitor::run - index " << indx << " just got into 'if' statement" << endl;

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

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

{

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

PEGASUS_ASSERT(q != 0);

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

if (h._connectionClosePending == false)

continue;

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

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

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

{

Tracer::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;

}

#endif

h._connectionClosePending = false;

MessageQueue *q;

MessageQueue &o = h.get_owner();

try{

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.

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

_entry_mut.unlock();

o.enqueue(message);

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

}

catch (Exception e)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

AutoMutex automut(Monitor::_cout_mut);

cout << " this is what lookup gives - " << e.getMessage() << endl;

#endif

exit(1);

}

catch(...)

Uint32 _idleEntries = 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( int indx = 0; indx < (int)entries.size(); indx++)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

if(maxSocketCurrentPass < entries[indx].socket)

AutoMutex automut(Monitor::_cout_mut);

maxSocketCurrentPass = entries[indx].socket;

cout << "MessageQueue::lookup gives strange exception " << endl;

#endif

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

exit(1);

{

_idleEntries++;

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

}

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

descriptors starting at 0.

maxSocketCurrentPass++;

_entry_mut.unlock();

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

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

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

#ifdef PEGASUS_OS_TYPE_WINDOWS

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

#else

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

#endif

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

if (events == PEGASUS_SOCKET_ERROR)

{

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

}

else if (events)

{

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

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

events, _idleEntries);

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

{

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

// owned by the Monitor).

if((entries[indx]._status.get() == _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);

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

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

//cout << "Monitor::run before PEGASUS_ASSerT(q !=0) " << endl;

PEGASUS_ASSERT(q !=0);

try

{

/* {

AutoMutex automut(Monitor::_cout_mut);

cout <<" this is the type " << entries[indx]._type <<

" for index " << indx << endl;

cout << "IN Monior::run right before entries[indx]._type == Monitor::CONNECTION" << endl;

}*/

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

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

cout << "In Monitor::run Monitor::CONNECTION clause" << endl;

AutoMutex automut(Monitor::_cout_mut);

}

#endif

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

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

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

Line 889

Line 856

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

/*In the case of named Pipes, the request has already been read from the pipe

therefor this section passed the request data to the HTTPConnection

NOTE: not sure if this would be better suited in a sparate private method

dst->setNamedPipe(entries[indx].namedPipe); //this step shouldn't be needd

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

cout << "In Monitor::run after dst->setNamedPipe string read is " << entries[indx].namedPipe.raw << endl;

}

#endif

try

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

cout << "In Monitor::run about to call 'dst->run(1)' " << endl;

}

#endif

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

if (entries[indx].isNamedPipeConnection())

{

entries[indx]._type = Monitor::ACCEPTOR;

}

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

// The following will fail in that case.

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

Line 950

Line 892

// set ourself to BUSY,

// read the data

// and set ourself back to IDLE

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

AutoMutex automut(Monitor::_cout_mut);

cout << endl << " in - entries[indx]._type == Monitor::INTERNAL- " << endl << endl;

#endif

if (!entries[indx].isNamedPipeConnection())

{

entries[indx]._status = _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 = _MonitorEntry::IDLE;

}

else

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

if(amt == PEGASUS_SOCKET_ERROR &&

getSocketError() == PEGASUS_NETWORK_TCPIP_STOPPED )

{

AutoMutex automut(Monitor::_cout_mut);

cout << "In Monitor::run else clause of CONNECTION if statments" << endl;

}

#endif

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

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

"Monitor::run: Tickler socket got an IO error. "

int events = 0;

"Going to re-create Socket and wait for TCP/IP restart.");

Message *msg;

uninitializeTickler();

initializeTickler();

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

cout << " In Monitor::run Just before checking if NamedPipeConnection" << "for Index "<<indx<< endl;

}

#endif

if (entries[indx].isNamedPipeConnection())

{

if(!entries[indx].namedPipe.isConnectionPipe)

{ /*if we enter this clasue it means that the named pipe that we are

looking at has recived a connection but is not the pipe we get connection requests over.

therefore we need to change the _type to CONNECTION and wait for a CIM Operations request*/

entries[indx]._type = Monitor::CONNECTION;

/* This is a test - this shows that the read file needs to be done

before we call wiatForMultipleObjects*/

/******************************************************

********************************************************/

memset(entries[indx].namedPipe.raw,'\0',NAMEDPIPE_MAX_BUFFER_SIZE);

BOOL rc = ::ReadFile(

entries[indx].namedPipe.getPipe(),

&entries[indx].namedPipe.raw,

NAMEDPIPE_MAX_BUFFER_SIZE,

&entries[indx].namedPipe.bytesRead,

entries[indx].namedPipe.getOverlap());

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

cout << "Monitor::run just called read on index " << indx << endl;

}

#endif

//&entries[indx].namedPipe.bytesRead = &size;

} else

if(!rc)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

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

AutoMutex automut(Monitor::_cout_mut);

cout << "ReadFile failed for : " << GetLastError() << "."<< endl;

#endif

}

/******************************************************

********************************************************/

continue;

}

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

cout << " In Monitor::run about to create a Pipe message" << endl;

}

#endif

events |= NamedPipeMessage::READ;

msg = new NamedPipeMessage(entries[indx].namedPipe, events);

}

else

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

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

{

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

AutoMutex automut(Monitor::_cout_mut);

int events = 0;

cout << " In Monitor::run ..its a socket message" << endl;

}

#endif

events |= SocketMessage::READ;

msg = new SocketMessage(entries[indx].socket, events);

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

}

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

autoEntryMutex.unlock();

_entry_mut.unlock();

q->enqueue(msg);

autoEntryMutex.lock();

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

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

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

return true;

}

catch(...)

{

}

handled_events = true;

}

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

return(handled_events);

}

void Monitor::stopListeningForConnections(Boolean wait)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

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

// set boolean then tickle the server to recognize _stopConnections

_stopConnections = 1;

Line 1115

Line 953

}

PEG_METHOD_EXIT();

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

}

int Monitor::solicitSocketMessages(

PEGASUS_SOCKET socket,

SocketHandle socket,

Uint32 events,

Uint32 queueId,

int type)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

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

AutoMutex autoMut(_entry_mut);

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

Line 1162

Line 989

_entries[index]._type = type;

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

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

return index;

}

Line 1177

Line 998

}

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

PEG_METHOD_EXIT();

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

return -1;

}

void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket)

void Monitor::unsolicitSocketMessages(SocketHandle socket)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

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

AutoMutex autoMut(_entry_mut);

Line 1228

Line 1037

index--;

}

PEG_METHOD_EXIT();

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Exiting: Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

}

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

PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)

ThreadReturnType PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "Entering: Monitor::_dispatch(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);

}

#endif

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

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

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

Line 1277

Line 1074

}

return 0;

}

// Added for NamedPipe implementation for windows

#if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)

//This method is anlogsu to solicitSocketMessages. It does the same thing for named Pipes

//This method is anlogus to solicitSocketMessages. It does the same thing for named Pipes

int Monitor::solicitPipeMessages(

NamedPipe namedPipe,

Uint32 events, //not sure what has to change for this enum

Line 1287

Line 1084

int type)

{

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

AutoMutex autoMut(_entry_pipe_mut);

AutoMutex autoMut(_entry_mut);

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

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

// current connections requested

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages at the begining" << PEGASUS_STD(endl);

}

#endif

_solicitPipeCount++; // bump the count

_solicitSocketCount++; // bump the count

int size = (int)_entries_pipe.size();

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

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

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

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

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

_MonitorEntry entry(0, 0, 0);

_entries.append(entry);

_entries_pipe.append(entry);

}

int index;

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

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

{

try

{

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

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

{

_entries[index].socket = NULL;

_entries_pipe[index].socket = NULL;

_entries[index].namedPipe = namedPipe;

_entries_pipe[index].namedPipe = namedPipe;

_entries[index].namedPipeConnection = true;

_entries_pipe[index].namedPipeConnection = true;

_entries[index].queueId = queueId;

_entries_pipe[index].queueId = queueId;

_entries[index]._type = type;

_entries_pipe[index]._type = type;

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

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

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

{

AutoMutex automut(Monitor::_cout_mut);

PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages after seting up _entries[index] index = " << index << PEGASUS_STD(endl);

}

#endif

return index;

}

Line 1332

Line 1132

}

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

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

PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages nothing happed - it didn't work" << PEGASUS_STD(endl);

PEG_METHOD_EXIT();

Line 1340

Line 1140

}

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

// Method Name : unsolicitPipeMessages

// Input Parameter : namedPipe - type NamedPipe

// Return Type : void

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

// This method is invoked from HTTPAcceptor::handleEnqueue for server

// when the CLOSE_CONNECTION_MESSAGE is recieved. This method is also invoked

// from HTTPAcceptor::destroyConnections method when the CIMServer is shutdown.

// For the CIMClient, this is invoked from HTTPConnector::handleEnqueue when the

// CLOSE_CONNECTION_MESSAGE is recieved. This method is also invoked from

// HTTPConnector::disconnect when CIMClient requests a disconnect request.

// The list of _MonitorEntry is searched for the matching pipe.

// The Handle of the identified is closed and _MonitorEntry for the

// requested pipe is removed.

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

void Monitor::unsolicitPipeMessages(NamedPipe namedPipe)

{

#ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG

Line 1350

Line 1166

#endif

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

AutoMutex autoMut(_entry_mut);

AutoMutex autoMut(_entry_pipe_mut);

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

to be EMPTY;

unsigned int index;

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

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

{

if(_entries[index].namedPipe.getPipe() == namedPipe.getPipe())

if (_entries_pipe[index].namedPipe.getPipe() == namedPipe.getPipe())

{

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

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

//_entries[index].namedPipe = PEGASUS_INVALID_SOCKET;

// Ensure that the client has read the data

_solicitSocketCount--;

::FlushFileBuffers (namedPipe.getPipe());

//Disconnect to release the pipe. This doesn't release Pipe Handle

::DisconnectNamedPipe (_entries_pipe[index].namedPipe.getPipe());

// Must use CloseHandle to Close Pipe

::CloseHandle(_entries_pipe[index].namedPipe.getPipe());

_entries_pipe[index].namedPipe.disconnect();

_solicitPipeCount--;

break;

}

Line 1374

Line 1196

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.

index = _entries.size() - 1;

index = _entries_pipe.size() - 1;

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

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

if((_entries[index].namedPipe.getPipe() == namedPipe.getPipe()) ||

&& index > 0)

(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES))

{

_entries.remove(index);

if ((_entries_pipe[index].namedPipe.getPipe() == namedPipe.getPipe()) ||

(_entries_pipe.size() > MAX_NUMBER_OF_MONITOR_ENTRIES))

{

_entries_pipe.remove(index);

}

index--;

}

Line 1392

Line 1216

#endif

}

#endif

PEGASUS_NAMESPACE_END

Legend:

Removed from v.1.103.10.22
changed lines
	Added in v.1.115.14.1

No CVS admin address has been configured