pegasus/src/Pegasus/ProviderManager2/Attic/OOPProviderManagerRouter.cpp - diff

Return to OOPProviderManagerRouter.cpp CVS log

Up to [Pegasus] / pegasus / src / Pegasus / ProviderManager2 / Attic

Diff for /pegasus/src/Pegasus/ProviderManager2/Attic/OOPProviderManagerRouter.cpp between version 1.17 and 1.25.2.8

version 1.17, 2005/05/18 21:26:43

version 1.25.2.8, 2006/10/10 18:10:15

Line 1

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

//%2006////////////////////////////////////////////////////////////////////////

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

Line 8

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

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

// EMC Corporation; Symantec 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"), to

Line 67

Line 69

#else

# include <unistd.h> // For fork(), exec(), and _exit()

# include <errno.h>

# include <sys/types.h>

# include <sys/resource.h>

# if defined(PEGASUS_HAS_SIGNALS)

# include <sys/wait.h>

# endif

#endif

#include "OOPProviderManagerRouter.h"

Line 91

Line 98

{

public:

OutstandingRequestEntry(

String messageId_,

String originalMessageId_,

CIMRequestMessage* requestMessage_,

CIMResponseMessage*& responseMessage_,

Semaphore* responseReady_)

: messageId(messageId_),

: originalMessageId(originalMessageId_),

requestMessage(requestMessage_),

responseMessage(responseMessage_),

responseReady(responseReady_)

{

}

String messageId;

/**

A unique value is substituted as the request messageId attribute to

allow responses to be definitively correllated with requests.

The original messageId value is stored here to avoid a race condition

between the processing of a response chunk and the resetting of the

original messageId in the request message.

String originalMessageId;

CIMRequestMessage* requestMessage;

CIMResponseMessage*& responseMessage;

Semaphore* responseReady;

};

Line 119

Line 136

ProviderAgentContainer(

const String & moduleName,

const String & userName,

PEGASUS_INDICATION_CALLBACK indicationCallback,

Uint16 userContext,

PEGASUS_INDICATION_CALLBACK_T indicationCallback,

PEGASUS_RESPONSE_CHUNK_CALLBACK_T responseChunkCallback,

PEGASUS_PROVIDERMODULEFAIL_CALLBACK_T providerModuleFailCallback,

Boolean subscriptionInitComplete);

~ProviderAgentContainer();

Line 171

Line 191

with an error result.

Note: The caller must lock the _agentMutex.

@param cleanShutdown Indicates whether the provider agent process

exited cleanly. A value of true indicates that responses have been

sent for all requests that have been processed. A value of false

indicates that one or more requests may have been partially processed.

void _uninitialize();

void _uninitialize(Boolean cleanShutdown);

/**

Performs the processMessage work, but does not retry on a transient

error.

CIMResponseMessage* _processMessage(CIMRequestMessage* request);

/**

Read and process response messages from the Provider Agent until

Line 204

Line 235

String _userName;

/**

User Context setting of the provider module served by this Provider

Agent.

Uint16 _userContext;

/**

Callback function to which all generated indications are sent for

processing.

PEGASUS_INDICATION_CALLBACK _indicationCallback;

PEGASUS_INDICATION_CALLBACK_T _indicationCallback;

/**

Callback function to which response chunks are sent for processing.

PEGASUS_RESPONSE_CHUNK_CALLBACK_T _responseChunkCallback;

/**

Callback function to be called upon detection of failure of a

provider module.

PEGASUS_PROVIDERMODULEFAIL_CALLBACK_T _providerModuleFailCallback;

/**

Indicates whether the Provider Agent is active.

Line 223

Line 271

AutoPtr<AnonymousPipe> _pipeToAgent;

#if defined(PEGASUS_HAS_SIGNALS)

/**

Process ID of the active Provider Agent.

pid_t _pid;

#endif

/**

The _outstandingRequestTable holds an entry for each request that has

been sent to this Provider Agent for which no response has been

Line 264

Line 319

static Uint32 _maxProviderProcesses;

/**

A value indicating that a request message has not been processed.

A CIMResponseMessage pointer with this value indicates that the

corresponding CIMRequestMessage has not been processed. This is

used to indicate that a provider agent exited without starting to

process the request, and that the request should be retried.

static CIMResponseMessage* _REQUEST_NOT_PROCESSED;

/**

Indicates whether the Indication Service has completed initialization.

For more information, please see the description of the

Line 276

Line 340

Mutex ProviderAgentContainer::_numProviderProcessesMutex;

Uint32 ProviderAgentContainer::_maxProviderProcesses = PEG_NOT_FOUND;

// Set this to a value that no valid CIMResponseMessage* will have.

CIMResponseMessage* ProviderAgentContainer::_REQUEST_NOT_PROCESSED =

reinterpret_cast<CIMResponseMessage*>(&_REQUEST_NOT_PROCESSED);

ProviderAgentContainer::ProviderAgentContainer(

const String & moduleName,

const String & userName,

PEGASUS_INDICATION_CALLBACK indicationCallback,

Uint16 userContext,

PEGASUS_INDICATION_CALLBACK_T indicationCallback,

PEGASUS_RESPONSE_CHUNK_CALLBACK_T responseChunkCallback,

PEGASUS_PROVIDERMODULEFAIL_CALLBACK_T providerModuleFailCallback,

Boolean subscriptionInitComplete)

: _moduleName(moduleName),

_userName(userName),

_userContext(userContext),

_indicationCallback(indicationCallback),

_responseChunkCallback(responseChunkCallback),

_providerModuleFailCallback(providerModuleFailCallback),

_isInitialized(false),

_subscriptionInitComplete(subscriptionInitComplete)

{

Line 324

Line 398

PEG_METHOD_ENTER(TRC_PROVIDERMANAGER,

"ProviderAgentContainer::_startAgentProcess");

// Serialize the starting of agent processes. If two agent processes are

// started at the same time, they may get copies of each other's pipe

// descriptors. If this happens, the cimserver will not get a pipe read

// error when one of the agent processes exits, because the pipe will

// still be writable by the other process. This locking control needs to

// cover the period from where the pipes are created to where the agent

// ends of the pipes are closed by the cimserver.

static Mutex agentStartupMutex;

AutoMutex lock(agentStartupMutex);

AutoPtr<AnonymousPipe> pipeFromAgent(new AnonymousPipe());

AutoPtr<AnonymousPipe> pipeToAgent(new AnonymousPipe());

Line 460

Line 546

}

#elif defined (PEGASUS_OS_OS400)

//Out of provider support for OS400 goes here when needed.

//Out of process provider support for OS400 goes here when needed.

#else

# ifndef PEGASUS_DISABLE_PROV_USERCTXT

// Get and save the effective user name and the uid/gid for the user

// context of the agent process

String effectiveUserName = System::getEffectiveUserName();

PEGASUS_UID_T newUid = (PEGASUS_UID_T) -1;

PEGASUS_GID_T newGid = (PEGASUS_GID_T) -1;

if (_userName != effectiveUserName)

{

if (!System::lookupUserId(_userName.getCString(), newUid, newGid))

{

throw PEGASUS_CIM_EXCEPTION_L(

CIM_ERR_FAILED,

MessageLoaderParms(

"ProviderManager.OOPProviderManagerRouter."

"USER_CONTEXT_CHANGE_FAILED",

"Unable to change user context to \"$0\".", _userName));

}

# endif

pid_t pid = fork();

if (pid < 0)

{

Line 500

Line 608

#ifndef PEGASUS_DISABLE_PROV_USERCTXT

// Set the user context of the Provider Agent process

if (_userName != System::getEffectiveUserName())

if (_userName != effectiveUserName)

{

if (!System::changeUserContext(_userName.getCString()))

if (!System::changeUserContext(newUid, newGid))

{

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

"System::changeUserContext() failed. userName = %s.",

Line 517

Line 625

}

#endif

// Close all file descriptors except stdin/stdout/stderr

// and the pipe handles needed by the Provider Agent process.

Uint32 readFd = atoi(readHandle);

Uint32 writeFd = atoi(writeHandle);

struct rlimit fileLimit;

if (getrlimit(RLIMIT_NOFILE, &fileLimit) == 0)

{

Uint32 maxFd = (Uint32)fileLimit.rlim_cur;

for (Uint32 i = 3; i < maxFd - 1; i++)

{

if ((i != readFd) && (i != writeFd))

{

close(i);

}

execl(agentCommandPathCString, agentCommandPathCString,

readHandle, writeHandle,

(const char*)_moduleName.getCString(), (char*)0);

Line 538

Line 665

_exit(1);

}

# if defined(PEGASUS_HAS_SIGNALS)

_pid = pid;

# endif

#endif

Line 609

Line 739

_moduleName));

}

// Do not wait for a response from the Provider Agent. (It isn't coming.)

// Wait for a null response from the Provider Agent indicating it has

// initialized successfully.

CIMMessage* message;

AnonymousPipe::Status readStatus;

{

readStatus = _pipeFromAgent->readMessage(message);

} while (readStatus == AnonymousPipe::STATUS_INTERRUPT);

if (readStatus != AnonymousPipe::STATUS_SUCCESS)

{

PEG_METHOD_EXIT();

throw Exception(MessageLoaderParms(

"ProviderManager.OOPProviderManagerRouter."

"CIMPROVAGT_COMMUNICATION_FAILED",

"Failed to communicate with cimprovagt \"$0\".",

_moduleName));

}

PEGASUS_ASSERT(message == 0);

PEG_METHOD_EXIT();

}

Line 629

Line 779

if (_maxProviderProcesses == PEG_NOT_FOUND)

{

String maxProviderProcesses = ConfigManager::getInstance()->getCurrentValue("maxProviderProcesses");

String maxProviderProcesses = ConfigManager::getInstance()->

getCurrentValue("maxProviderProcesses");

CString maxProviderProcessesString = maxProviderProcesses.getCString();

char* end = 0;

_maxProviderProcesses = strtol(maxProviderProcessesString, &end, 10);

Line 658

Line 809

{

_startAgentProcess();

_sendInitializationData();

_isInitialized = true;

_sendInitializationData();

// Start a thread to read and process responses from the Provider Agent

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

ThreadStatus rtn = PEGASUS_THREAD_OK;

this, _responseProcessor))

while ((rtn = MessageQueueService::get_thread_pool()->

allocate_and_awaken(this, _responseProcessor)) !=

PEGASUS_THREAD_OK)

{

if (rtn == PEGASUS_THREAD_INSUFFICIENT_RESOURCES)

{

pegasus_yield();

}

else

{

Logger::put(

Logger::STANDARD_LOG, System::CIMSERVER, Logger::TRACE,

"Not enough threads to process responses from the "

"provider agent.");

Tracer::trace(TRC_PROVIDERMANAGER, Tracer::LEVEL2,

"Could not allocate thread to process responses from the "

"provider agent.");

throw Exception(MessageLoaderParms(

"ProviderManager.OOPProviderManagerRouter."

"CIMPROVAGT_THREAD_ALLOCATION_FAILED",

"Failed to allocate thread for cimprovagt \"$0\".",

_moduleName));

}

catch (...)

{

_isInitialized = false;

// Closing the connection causes the agent process to exit

_pipeToAgent.reset();

_pipeFromAgent.reset();

#if defined(PEGASUS_HAS_SIGNALS)

if (_isInitialized)

{

// Harvest the status of the agent process to prevent a zombie

pid_t status = 0;

{

status = waitpid(_pid, 0, 0);

} while ((status == -1) && (errno == EINTR));

if (status == -1)

{

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

"ProviderAgentContainer::_initialize(): "

"waitpid failed; errno = %d.", errno);

}

#endif

_isInitialized = false;

{

AutoMutex lock(_numProviderProcessesMutex);

_numProviderProcesses--;

Line 694

Line 888

}

// Note: Caller must lock _agentMutex

void ProviderAgentContainer::_uninitialize()

void ProviderAgentContainer::_uninitialize(Boolean cleanShutdown)

{

PEG_METHOD_ENTER(TRC_PROVIDERMANAGER,

"ProviderAgentContainer::_uninitialize");

Line 719

Line 913

_numProviderProcesses--;

}

#if defined(PEGASUS_HAS_SIGNALS)

// Harvest the status of the agent process to prevent a zombie

pid_t status = 0;

{

status = waitpid(_pid, 0, 0);

} while ((status == -1) && (errno == EINTR));

if (status == -1)

{

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

"ProviderAgentContainer::_uninitialize(): "

"waitpid failed; errno = %d.", errno);

}

#endif

_isInitialized = false;

Line 728

Line 938

{

AutoMutex tableLock(_outstandingRequestTableMutex);

CIMResponseMessage* response =

cleanShutdown ? _REQUEST_NOT_PROCESSED : 0;

for (OutstandingRequestTable::Iterator i =

_outstandingRequestTable.start();

i != 0; i++)

{

PEG_TRACE_STRING(TRC_PROVIDERMANAGER, Tracer::LEVEL2,

String("Completing messageId \"") + i.value()->messageId +

String("Completing messageId \"") + i.key() +

"\" with a null response.");

i.value()->responseMessage = 0;

i.value()->responseMessage = response;

i.value()->responseReady->signal();

}

_outstandingRequestTable.clear();

// If not a clean shutdown, call the provider module failure

// callback

if (!cleanShutdown)

{

// Call the provider module failure callback to

// communicate the failure to the Provider Manager Service

// Provider Manager Service will inform Indication Service

_providerModuleFailCallback (_moduleName, _userName,

_userContext);

}

catch (...)

Line 764

Line 992

"ProviderAgentContainer::processMessage");

CIMResponseMessage* response;

{

response = _processMessage(request);

if (response == _REQUEST_NOT_PROCESSED)

{

// Check for request message types that should not be retried.

if ((request->getType() ==

CIM_STOP_ALL_PROVIDERS_REQUEST_MESSAGE) ||

(request->getType() ==

CIM_NOTIFY_CONFIG_CHANGE_REQUEST_MESSAGE) ||

(request->getType() ==

CIM_SUBSCRIPTION_INIT_COMPLETE_REQUEST_MESSAGE) ||

(request->getType() ==

CIM_DELETE_SUBSCRIPTION_REQUEST_MESSAGE))

{

response = request->buildResponse();

break;

}

else if (request->getType() == CIM_DISABLE_MODULE_REQUEST_MESSAGE)

{

CIMDisableModuleResponseMessage* dmResponse =

dynamic_cast<CIMDisableModuleResponseMessage*>(response);

PEGASUS_ASSERT(dmResponse != 0);

Array<Uint16> operationalStatus;

operationalStatus.append(CIM_MSE_OPSTATUS_VALUE_STOPPED);

dmResponse->operationalStatus = operationalStatus;

break;

}

} while (response == _REQUEST_NOT_PROCESSED);

if (request->getType() == CIM_SUBSCRIPTION_INIT_COMPLETE_REQUEST_MESSAGE)

{

_subscriptionInitComplete = true;

}

PEG_METHOD_EXIT();

return response;

}

CIMResponseMessage* ProviderAgentContainer::_processMessage(

CIMRequestMessage* request)

{

PEG_METHOD_ENTER(TRC_PROVIDERMANAGER,

"ProviderAgentContainer::_processMessage");

CIMResponseMessage* response;

String originalMessageId = request->messageId;

// These three variables are used for the provider module optimization.

Line 790

Line 1068

Semaphore waitSemaphore(0);

OutstandingRequestEntry outstandingRequestEntry(

uniqueMessageId, response, &waitSemaphore);

originalMessageId, request, response, &waitSemaphore);

// Lock the Provider Agent Container while initializing the

Line 874

Line 1152

Tracer::trace(TRC_PROVIDERMANAGER, Tracer::LEVEL2,

"Failed to write message to pipe. writeStatus = %d.",

writeStatus);

throw Exception(MessageLoaderParms(

"ProviderManager.OOPProviderManagerRouter."

request->messageId = originalMessageId;

"CIMPROVAGT_COMMUNICATION_FAILED",

"Failed to communicate with cimprovagt \"$0\".",

if (doProviderModuleOptimization)

_moduleName));

{

request->operationContext.set(*origProviderId.get());

}

// Remove this OutstandingRequestTable entry

{

AutoMutex tableLock(_outstandingRequestTableMutex);

Boolean removed =

_outstandingRequestTable.remove(uniqueMessageId);

PEGASUS_ASSERT(removed);

}

// A response value of _REQUEST_NOT_PROCESSED indicates

// that the request was not processed by the provider

// agent, so it can be retried safely.

PEG_METHOD_EXIT();

return _REQUEST_NOT_PROCESSED;

}

if (updateProviderModuleCache)

Line 904

Line 1198

_outstandingRequestTable.remove(uniqueMessageId);

PEGASUS_ASSERT(removed);

}

PEG_METHOD_EXIT();

throw;

}

Line 925

Line 1220

_outstandingRequestTable.remove(uniqueMessageId);

PEGASUS_ASSERT(removed);

}

PEG_METHOD_EXIT();

throw;

}

// A response value of _REQUEST_NOT_PROCESSED indicates that the

// provider agent process was terminating when the request was sent.

// The request was not processed by the provider agent, so it can be

// retried safely.

if (response == _REQUEST_NOT_PROCESSED)

{

PEG_METHOD_EXIT();

return response;

}

// A null response is returned when an agent connection is closed

// while requests remain outstanding.

if (response == 0)

Line 1020

Line 1326

(readStatus == AnonymousPipe::STATUS_CLOSED))

{

AutoMutex lock(_agentMutex);

_uninitialize();

_uninitialize(false);

return;

}

// A null message indicates that the provider agent process has

// finished its processing and is ready to exit.

if (message == 0)

{

AutoMutex lock(_agentMutex);

_uninitialize(true);

return;

}

Line 1031

Line 1346

reinterpret_cast<CIMProcessIndicationRequestMessage*>(

message));

}

else if (!message->isComplete())

{

CIMResponseMessage* response;

response = dynamic_cast<CIMResponseMessage*>(message);

PEGASUS_ASSERT(response != 0);

// Get the OutstandingRequestEntry for this response chunk

OutstandingRequestEntry* _outstandingRequestEntry = 0;

{

AutoMutex tableLock(_outstandingRequestTableMutex);

Boolean foundEntry = _outstandingRequestTable.lookup(

response->messageId, _outstandingRequestEntry);

PEGASUS_ASSERT(foundEntry);

}

// Put the original message ID into the response

response->messageId =

_outstandingRequestEntry->originalMessageId;

// Call the response chunk callback to process the chunk

_responseChunkCallback(

_outstandingRequestEntry->requestMessage, response);

}

else

{

CIMResponseMessage* response;

Line 1085

Line 1423

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

OOPProviderManagerRouter::OOPProviderManagerRouter(

PEGASUS_INDICATION_CALLBACK indicationCallback)

PEGASUS_INDICATION_CALLBACK_T indicationCallback,

PEGASUS_RESPONSE_CHUNK_CALLBACK_T responseChunkCallback,

PEGASUS_PROVIDERMODULEFAIL_CALLBACK_T providerModuleFailCallback)

{

PEG_METHOD_ENTER(TRC_PROVIDERMANAGER,

"OOPProviderManagerRouter::OOPProviderManagerRouter");

_indicationCallback = indicationCallback;

_responseChunkCallback = responseChunkCallback;

_providerModuleFailCallback = providerModuleFailCallback;

_subscriptionInitComplete = false;

PEG_METHOD_EXIT();

Line 1116

Line 1458

PEG_METHOD_EXIT();

}

// Private, unimplemented constructor

OOPProviderManagerRouter::OOPProviderManagerRouter()

{

}

// Private, unimplemented constructor

OOPProviderManagerRouter::OOPProviderManagerRouter(

const OOPProviderManagerRouter&)

: ProviderManagerRouter(*this)

{

}

// Private, unimplemented assignment operator

OOPProviderManagerRouter& OOPProviderManagerRouter::operator=(

const OOPProviderManagerRouter&)

{

return *this;

}

Message* OOPProviderManagerRouter::processMessage(Message* message)

{

PEG_METHOD_ENTER(TRC_PROVIDERMANAGER,

Line 1367

Line 1690

}

else

{

// Look up the Provider Agent for this module instance and requesting

// user

ProviderAgentContainer* pa = _lookupProviderAgent(providerModule,

request);

PEGASUS_ASSERT(pa != 0);

// Forward the request to the provider agent

response.reset(pa->processMessage(request));

}

response->syncAttributes(request);

PEG_METHOD_EXIT();

return response.release();

}

ProviderAgentContainer* OOPProviderManagerRouter::_lookupProviderAgent(

const CIMInstance& providerModule,

CIMRequestMessage* request)

{

// Retrieve the provider module name

String moduleName;

CIMValue nameValue = providerModule.getProperty(

Line 1389

Line 1736

if (userContext == 0)

{

userContext = PG_PROVMODULE_USERCTXT_PRIVILEGED;

userContext = PEGASUS_DEFAULT_PROV_USERCTXT;

}

String userName;

Line 1403

Line 1750

request->operationContext.get(IdentityContainer::NAME);

userName = ic.getUserName();

}

catch (Exception& e)

catch (Exception&)

{

// If no IdentityContainer is present, default to the CIM

// Server's user context

Line 1417

Line 1764

}

else if (userContext == PG_PROVMODULE_USERCTXT_DESIGNATED)

{

// Retrieve the provider module name

// Retrieve the provider module designated user property value

providerModule.getProperty(providerModule.findProperty(

PEGASUS_PROPERTYNAME_MODULE_DESIGNATEDUSER)).getValue().

get(userName);

Line 1439

Line 1786

PEG_TRACE_STRING(TRC_PROVIDERMANAGER, Tracer::LEVEL4,

"User name = " + userName);

// Look up the Provider Agent for this module and user

ProviderAgentContainer* pa = _lookupProviderAgent(moduleName, userName);

PEGASUS_ASSERT(pa != 0);

// Forward the request to the provider agent

response.reset(pa->processMessage(request));

}

response->syncAttributes(request);

PEG_METHOD_EXIT();

return response.release();

}

ProviderAgentContainer* OOPProviderManagerRouter::_lookupProviderAgent(

const String& moduleName,

const String& userName)

{

ProviderAgentContainer* pa = 0;

String key = moduleName + ":" + userName;

Line 1466

Line 1793

if (!_providerAgentTable.lookup(key, pa))

{

pa = new ProviderAgentContainer(

moduleName, userName, _indicationCallback,

moduleName, userName, userContext,

_indicationCallback, _responseChunkCallback,

_providerModuleFailCallback,

_subscriptionInitComplete);

_providerAgentTable.insert(key, pa);

}

Legend:

Removed from v.1.17
changed lines
	Added in v.1.25.2.8

No CVS admin address has been configured