Page #96 (The Architecture) | COM+ Programming. A Practical Guide Using Visual C++ and ATL

< BACK NEXT >

[oR]

Compensating Resource Manager

A resource manager has to pass the ACID test; it has to guarantee atomicity, consistency, isolation, and durability. Given the intricate footwork an RM has to perform, implementing an RM is not an easy task.

Let s look at the tasks of a typical RM.

When a client accesses a transactional resource, the corresponding RM should support enlistment with the DTC. The RM may also make a temporary copy of the resource and lock access to the actual resource (so that no other client can use it).
When the primary client attempts to modify the resource, the RM has to record the change and apply the change to the copy (not the actual resource).
If the DTC asks the RM to prepare, the RM has to play back the recorded sequence, and create an internal state for the commit phase. Alternatively, an RM may delay the playback to the commit phase, if it is confident that the updates will not fail.
If the DTC asks the RM to commit, the RM may use the prepared internal state to commit the changes or play back the recorded sequence and apply the changes to the resource.
If the DTC asks the RM to abort, the RM may just discard the prepared internal state (or the recorded sequence).

Given that a large portion of functionality is common from one RM to another, a reasonable question to ask is if there is a way to share this functionality. This would certainly simplify developing an RM.

It turns out that COM+ designers had already thought of this possibility. COM+ provides a framework to develop RMs. An RM developed using this framework is referred to as a Compensating Resource Manager (CRM).

The developer of a CRM has to write two cooperating components called the CRM worker and the CRM compensator.

The CRM worker exposes necessary COM objects to the clients. When the client requests that the resource be modified, the worker simply records the change (using CRM s service).

The CRM compensator reads the recorded changes (supplied by the CRM service) and either commits or aborts the changes.

Note that there is no direct communication between the CRM worker and the CRM compensator. The only data that has to be passed from the worker to the compensator is the sequence of changes applied on the resource.

To facilitate storing the sequence of changes, COM+ provides a component called the CRM clerk. The CRM worker instantiates the CRM clerk and starts recording the changes with the clerk. When the transaction closes, COM+ launches the CRM compensator and calls prepare, commit, or abort in whatever combination that is appropriate, and plays back the sequence of records to the compensator.

The CRM clerk supports an interface, ICrmLogControl. The following is its prototype:

 ICrmLogControl : public IUnknown  {   [propget] HRESULT TransactionUOW([retval][out] BSTR *pVal);    HRESULT RegisterCompensator(     [in] LPCWSTR pwszProgId,      [in] LPCWSTR pwszDesc,      [in] LONG lCrmRegFlags);    HRESULT STDMETHODCALLTYPE WriteLogRecordVariants(     [in] VARIANT *pLogRecord);    HRESULT ForceLog();    HRESULT ForgetLogRecord();    HRESULT ForceTransactionToAbort();    HRESULT WriteLogRecord(     [size_is][in] BLOB rgBlob[],      [in] ULONG cBlob);  };

Method RegisterCompensator is used to associate a CRM worker with a specific CRM compensator. It is the responsibility of the CRM worker to call this method.

Parameter pwszProgId is the PROGID of the CRM compensator that should be associated with the CRM worker. Parameter pwszDesc describes the CRM compensator. A transaction-monitoring program can use this description string for display purposes. Parameter lCrmRegFlags specifies the possible phases (prepare, commit, or abort) that can be passed to the CRM compensator. For example, if the compensator does not do anything specific to abort a transaction, then CRMREGFLAG_ABORTPHASE need not be specified as a possible phase.

Method WriteLogRecord can be used to record a change that is being made to the resource. The data is recorded in a form called BLOB (Binary Large Object). A BLOB is a structure that can carry any opaque data as a pointer. The structure of the BLOB is defined as follows:

 struct BLOB{   ULONG cbSize;                   // the size of the data    [size_is(cbSize)] BYTE* pBlobData;    //the actual data  };

The CRM worker can record a resource change by passing one or more BLOBS to the method WriteLogRecord.

Each call to WriteLogRecord results in a single record stored with the CRM clerk. When the transaction completes, the CRM clerk instantiates the CRM compensator and plays back the records in the same sequence as they were originally received.

Associating more than one BLOB with a single record is just a convenience provided to the CRM worker. The CRM clerk internally pastes all the BLOBS together as one big BLOB.

Method ForceTransactionToAbort can be used to abort a transaction.

Let s turn our attention to the CRM compensator.

A CRM compensator has to support an interface, ICrmCompensator.

The following is its prototype:

 ICrmCompensator : public IUnknown  {   HRESULT SetLogControl( [in] ICrmLogControl *pLogControl);    // Prepare phase    HRESULT BeginPrepare( void);    HRESULT PrepareRecord( [in] CrmLogRecordRead crmLogRec,      [retval][out] BOOL *pfForget);    HRESULT EndPrepare( [retval][out] BOOL *pfOkToPrepare);    // Commit phase    HRESULT BeginCommit( [in] BOOL fRecovery);    HRESULT CommitRecord( [in] CrmLogRecordRead crmLogRec,      [retval][out] BOOL *pfForget);    HRESULT EndCommit( void);    // Abort phase    HRESULT BeginAbort( [in] BOOL fRecovery);    HRESULT AbortRecord( [in] CrmLogRecordRead crmLogRec,      [retval][out] BOOL *pfForget);    HRESULT EndAbort( void);  };

Data type CrmLogRecordRead is a C structure that contains a BLOB (that was previously recorded using WriteLogRecord) and some other fields that might be useful for debugging.

The compensator should implement code for all three phases, at least to satisfy the compiler. The DTC enters a phase by calling the BeginXXX method on that phase, followed by one or more calls to RecordXXX, and completes the phase by calling the EndXXX method.

Once a record has been digested in any phase, if the CRM compensator feels that the record serves no purpose to some other phase that it may enter later, it can inform the CRM clerk to lose the record by setting pfForget flag to TRUE.

With this brief background, let s build a CRM.

Our CRM will use a text file as a resource. To verify its functionality, we will modify the account manager component from the previous simulation program to use a text file, W:/DB/Accounts.txt, as a transactional resource (replacing the MSDE database).

The CRM worker component will support interface IMyFileDB, as defined here:

 interface IMyFileDB : IDispatch  {   HRESULT Open([in] BSTR bsFilePath);    HRESULT GetBalance([in] BSTR bsClient,      [out, retval] long* plBalance);    HRESULT UpdateBalance([in] BSTR bsClient,      [in] long lNewBalance);  };

With this component in place, the CAccountMgr::Debit logic should be modified to use the file-based resource. The revised implementation is shown below:

 STDMETHODIMP CAccountMgr::Debit(BSTR bsClient, long lAmount)  {   CComPtr<IContextState> spState;    HRESULT hr = ::CoGetObjectContext(__uuidof(IContextState),      (void**) &spState);    if (FAILED(hr)) {     return hr;    }    try {     IMyFileDBPtr spConn(__uuidof(MyFileDB));      spConn->Open("w:/DB/Accounts.txt");      long lCurrentBalance = spConn->GetBalance(bsClient);      if (lCurrentBalance < lAmount) {       spState->SetMyTransactionVote(TxAbort);        return Error(_T("Not enough balance"), GUID_NULL,          E_FAIL);      }      long lNewBalance = lCurrentBalance - lAmount;      spConn->UpdateBalance(bsClient, lNewBalance);    }    catch(_com_error& e) {     spState->SetMyTransactionVote(TxAbort);      return Error(static_cast<LPCTSTR>(e.Description()),        GUID_NULL, e.Error());    }    spState->SetMyTransactionVote(TxCommit);    return S_OK;  }

When IMyFileDB::Open is invoked, the CRM worker should first instantiate the CRM clerk and register the associated CRM compensator. The code snippet is shown below:

 HRESULT CMyFileDB::InitCRM()  {   if (ISNOTNULL(m_spCrmLC)) {     m_spCrmLC = NULL;    }  HRESULT hr = ::CoCreateInstance(   __uuidof(CRMClerk),    NULL,    CLSCTX_INPROC_SERVER,    __uuidof(ICrmLogControl),    (void**) &m_spCrmLC);    if (FAILED(hr)) {     return hr;    }    // Register the compensator.    // Try 5 times if a recovery is in progress    for(int i=0; i<5; i++) {     hr = m_spCrmLC->RegisterCompensator(       L"TextFileDB.MyFileDBCompensator",        L"My file db compensator",        CRMREGFLAG_ALLPHASES);      if (SUCCEEDED(hr)) {       return S_OK;      }      // deal with recovery in progress      if (XACT_E_RECOVERYINPROGRESS == hr) {       Sleep(1000); // sleep for a second        continue; // and try again      }    }    m_spCrmLC = NULL;    return hr;  }

Note that it is possible for the CRM worker to receive an XACT_E_RECOVERYINPROGRESS error during the call to RegisterCompensator. If this happens, the CRM worker should call the method a few more times until it succeeds.

The data file for our CRM contains clients and their respective balances. The CRM worker loads the file into memory as an STL map. Loading all the data into memory is not always efficient. However, it works for our demonstration.

 typedef std::map<CComBSTR, long> MYACCOUNTDB;  class CMyFileDB :    ...  {   ...  private:    CComPtr<ICrmLogControl> m_spCrmLC;    MYACCOUNTDB m_AccountDB;  };

I have encapsulated serializing the MYACCOUNTDB data type to a file in class CMyFile and will not discuss it further. The code can be found on the CD.

There are only two commands that the CRM worker needs to record: the command to open a file and the command to update an account. As the command to obtain the balance does not really change the resource, there is no real need to record it.

To facilitate converting the command information into a BLOB, let s define some relevant data structures:

 enum DBACTIONTYPE {dbOpen = 0x10, dbUpdate = 0x20};  #pragma warning(disable : 4200)       // do not warn on                                        // zero-sized arrays  #pragma pack(1)                       // Pack the following                                        // structures tightly  struct DBACTION {   DBACTIONTYPE actionType;  };  struct DBACTIONOPEN : public DBACTION  {   DBACTIONOPEN()    {     actionType = dbOpen;    }    WCHAR pszFileName[0];  };  struct DBACTIONUPDATE : public DBACTION  {   DBACTIONUPDATE()    {     actionType = dbUpdate;    }    long lNewBalance;    WCHAR pszClient[0];  };  #pragma pack()                       // back to default packing  #pragma warning(default : 4200)      // back to default warning

Note that packing the data on the byte boundary is important for reinterpreting a BLOB to its original structure.

Also note that I am defining a zero-sized array for a variable-sized string. I am just taking advantage of the fact that data is stored contiguously in a BLOB.

With these structures in place, the CMyFile::Open method can be implemented as follows:

 STDMETHODIMP CMyFileDB::Open(BSTR bsFilePath)  {   HRESULT hr = InitCRM();    if (FAILED(hr)) {     return hr;    }    // Open the file    USES_CONVERSION;    LPCTSTR pszFile = W2T(bsFilePath);    CMyFile file;    hr = file.Open(pszFile, CMyFile::READ);    if (FAILED(hr)) {     m_spCrmLC->ForceTransactionToAbort();      return hr;    }    // Log info with CRM that the file is being opened    DBACTIONOPEN openAction;    BLOB blobArray[2];    blobArray[0].pBlobData = (BYTE*) &openAction;    blobArray[0].cbSize = sizeof(DBACTIONOPEN);    blobArray[1].pBlobData = (BYTE*) bsFilePath;    blobArray[1].cbSize = ::SysStringByteLen(bsFilePath) +      sizeof(OLECHAR); // account for the end of string    hr = m_spCrmLC->WriteLogRecord(blobArray, 2);    if (FAILED(hr)) {     m_spCrmLC->ForceTransactionToAbort();      return hr;    }    // Now load file into memory    hr = file.Load(m_AccountDB);    if (FAILED(hr)) {     m_spCrmLC->ForceTransactionToAbort();      return hr;    }    return S_OK;  }

Method IMyFileDB::UpdateBalance records its operations similarly. The code is not shown here.

Now let s build the CRM compensator component.

The CRM component that we are building need not take any specific action in the prepare or abort phase. Consequently, we will focus on just the commit phase. Specifically, we will look at implementing two ICrmCompensator methods CommitRecord and EndCommit.

Method CommitRecord decodes the BLOB and, depending on the action type, either loads the file into memory or updates the in-memory copy with the new balances from the clients, as follows:

 STDMETHODIMP CMyFileDBCompensator::CommitRecord(     /* [in] */ CrmLogRecordRead crmLogRec,      /* [retval][out] */ BOOL __RPC_FAR *pfForget)  {   *pfForget = FALSE; // don't drop the record    BLOB& blob = crmLogRec.blobUserData;    DBACTION* pAction =      reinterpret_cast<DBACTION*>(blob.pBlobData);    if (dbOpen == pAction->actionType) {     DBACTIONOPEN* pActionOpen =        reinterpret_cast<DBACTIONOPEN*>(pAction);      m_bsFilePath = pActionOpen->pszFileName;      // load the contents of the file      USES_CONVERSION;      CMyFile file;      HRESULT hr = file.Open(W2T(m_bsFilePath), CMyFile::READ);      if (FAILED(hr)) {       return hr;      }      hr = file.Load(m_AccountDB);      if (FAILED(hr)) {       return hr;      }      return S_OK;    }    if (dbUpdate == pAction->actionType) {     DBACTIONUPDATE* pActionUpdate =        reinterpret_cast<DBACTIONUPDATE*>(pAction);      long lNewBalance = pActionUpdate->lNewBalance;      LPWSTR pwszClient = pActionUpdate->pszClient;      MYACCOUNTDB::iterator i = m_AccountDB.find(pwszClient);      if (i == m_AccountDB.end()) {       return E_INVALIDARG;      }      (*i).second = lNewBalance;      return S_OK;    }    return S_OK;  }

Method EndCommit saves the in-memory copy back into the file, as shown here:

 STDMETHODIMP CMyFileDBCompensator::EndCommit(void)  {   // Save the information back to file    USES_CONVERSION;    CMyFile file;    HRESULT hr = file.Open(W2T(m_bsFilePath), CMyFile::WRITE);    if (FAILED(hr)) {     return hr;    }    file.Save(m_AccountDB);    return S_OK;  }

Congratulations! You have just finished building your first CRM.

The CRM components can be installed as a server application. ^[1] It is recommended that both the CRM worker and the CRM compensator for a specific CRM be installed in the same application.

^[1] It can also be installed as a library application. Check the online documentation for more details.

For CRM components it is important to turn the Enable Compensating Resource Managers option on (from the Component Services snap-in). Other wise, a call to RegisterCompensator will result in a catastrophic failure error.

The CRM worker should be marked with the transaction setting as REQUIRED (which will automatically force JIT Activation = TRUE and Synchronization = REQUIRED). The CRM compensator, however, should be marked with the transaction as disabled, the synchronization as disabled, and the JIT turned off.

< BACK NEXT >