CkptContainer.cc

Go to the documentation of this file.

#include "CkptContainer.h"

CkptContainer::CkptContainer(std::string containerName,
                unsigned long maxObjects,
                unsigned long maxObjectSize,
                CkptRetentionPolicy rPolicy)
{
    // Do some parm validation first
    if (maxObjects <= 1) maxObjects = 2;       // Always have at least 2 sections
    if (maxObjectSize <= 0) maxObjectSize = 4; // Minimum object size of 4 bytes

    _maxObjectSize = maxObjectSize;
    _containerName = containerName;
    _checkpointName.length = _containerName.length();
    memcpy(_checkpointName.value, _containerName.data(), _checkpointName.length);
    SaCkptCallbacksT callbacks = { NULL, NULL };
    SaVersionT version = { 'B', 1, 1 };

    SaAisErrorT error = saCkptInitialize (&_serviceHandle, &callbacks, &version);

    SaCkptCheckpointCreationAttributesT ckptAttributes;
    ckptAttributes.creationFlags = (SA_CKPT_WR_ALL_REPLICAS | SA_CKPT_CHECKPOINT_COLLOCATED);
    ckptAttributes.checkpointSize = maxObjects*maxObjectSize;
    if (rPolicy == CKPT_RETAIN) ckptAttributes.retentionDuration = SA_TIME_END;
    else ckptAttributes.retentionDuration = 0;
    ckptAttributes.maxSections = maxObjects;
    ckptAttributes.maxSectionSize = maxObjectSize;
    ckptAttributes.maxSectionIdSize = 64;

    error = saCkptCheckpointOpen (_serviceHandle,
                   &_checkpointName,
                   &ckptAttributes,
                   SA_CKPT_CHECKPOINT_WRITE | SA_CKPT_CHECKPOINT_READ | SA_CKPT_CHECKPOINT_CREATE,
                   SA_TIME_END,
                   &_checkpointHandle);

    if (error == SA_AIS_OK) _valid=true;
    else _valid = false;
    
    _daoFactory = new RawDAOFactory();
}

CkptContainer::~CkptContainer() {
    SaAisErrorT error;
    if (_valid) {
      error = saCkptCheckpointClose(_checkpointHandle);
    }
    error = saCkptFinalize(_serviceHandle);
    delete _daoFactory;
}

unsigned long CkptContainer::size() {
    SaAisErrorT error;
    SaCkptCheckpointDescriptorT status;

    error = saCkptCheckpointStatusGet( _checkpointHandle, &status );
    if (error != SA_AIS_OK) return 0;

    return (unsigned long)status.numberOfSections;
}

CkptObject* CkptContainer::get(CkptObjectKey* key) {
    if (!_valid) return NULL;

  // Create a ioVectorElement to retrieve the data from CKPT
  SaCkptIOVectorElementT ioElement;
  ioElement.sectionId  = *(key->sectionIdValue());
  // We SHOULDN'T need a local buffer for this. CKPT is supposed to provide it if
  // one isn't supplied by the caller. This is a hack, but later I'll pass in a
  // pointer into the CkptByteArray's data buffer for speed. That WILL violate
  // encapsulation. [TODO]
  ioElement.dataBuffer = (char*)malloc(_maxObjectSize);
  ioElement.dataSize   = _maxObjectSize; // Read up to our max
  ioElement.dataOffset = 0;              // No offset

  // Read data from section [key] into a byte array
  SaAisErrorT error = saCkptCheckpointRead(_checkpointHandle, &ioElement, 1, NULL);

  if ((error != SA_AIS_OK) || (ioElement.readSize==0)) {
    free(ioElement.dataBuffer);
    return NULL;
  }

  // Convert the buffer to a byte array
  CkptByteArray* bArray = new CkptByteArray((char*)ioElement.dataBuffer,
                          (unsigned int)ioElement.readSize);
  // Free our local buffer
  free(ioElement.dataBuffer);

  CkptObject* obj_ptr;
  DAOInterface* input = ((RawDAOFactory*)_daoFactory)->createInputDAO(bArray);
  *input & obj_ptr;
  delete input;
  delete bArray;

  // Set the object internals to reference this container/key
  obj_ptr->setContainer(this);
  obj_ptr->setKey(key);
  return obj_ptr;
}

CkptObject* CkptContainer::get(CkptObjectKey &key) {
    CkptObjectKey *ownedKey = new CkptObjectKey(key);
    return this->get(ownedKey);
}

CkptRC CkptContainer::persist(CkptObject* obj, bool overwriteExisting) {
  // This associates obj with the container and creates a section for
  // it in the underlying checkpoint.

  if (!_valid) return CkptFailure;

  // Associate this object with this container
  obj->setContainer(this);

  // Create the attribute structure for the underlying section
  SaCkptSectionCreationAttributesT cAttrs;
  cAttrs.sectionId = obj->getKey()->sectionIdValue();
  cAttrs.expirationTime = SA_TIME_END;

  // Serialize the data in the object
  //std::auto_ptr<CkptByteArray> bArray(obj->serialize());
  DAOInterface* output = _daoFactory->createOutputDAO();
  *output & obj;
  byte_array* bArray = output->getOutputData();

  // Create a section for the object in the checkpoint and
  // commit the data to the section
  SaAisErrorT error = saCkptSectionCreate(_checkpointHandle,
                      &cAttrs,
                      bArray->getDataPtr(),
                      bArray->getDataSize());

  if ( (error == SA_AIS_ERR_EXIST) && (overwriteExisting) ) {
    // There's already an object with this key in the container
    // Overwrite it if the overwriteExisting flag is set.
    error = saCkptSectionOverwrite(_checkpointHandle,
              obj->getKey()->sectionIdValue(),
              bArray->getDataPtr(),
              bArray->getDataSize());
  }

  delete output;
  return (error == SA_AIS_OK) ? CkptSuccess : CkptFailure;
}

CkptRC CkptContainer::remove(CkptObject* obj) {
  if (!_valid) return CkptFailure;

  // Disassociate this object with this container
  obj->setContainer(NULL);

  // Delete the section for the object in the checkpoint
  saCkptSectionDelete(_checkpointHandle,
                      obj->getKey()->sectionIdValue());
  return CkptSuccess;
}

CkptRC CkptContainer::commit(CkptObject* obj) {
  // Make sure this is a persisted object, fail if not.
  if (!_valid || (obj == NULL) || (obj->getState() != CKPT_OBJ_MANAGED)) {
    return CkptFailure;
  }

  DAOInterface* output = _daoFactory->createOutputDAO();
  *output & obj;
  byte_array* bArray = output->getOutputData();

  // Overwrite the data from the byte array into the section
  // corresponding to the object.
  SaAisErrorT error = saCkptSectionOverwrite(_checkpointHandle,
                        obj->getKey()->sectionIdValue(),
                        bArray->getDataPtr(),
                        bArray->getDataSize());

  delete output;
  return (error == SA_AIS_OK) ? CkptSuccess : CkptFailure;
}

CkptRC CkptContainer::setRetentionPolicy(CkptRetentionPolicy rPolicy)
{
    if (!_valid) return CkptFailure;
    SaTimeT rTime = (rPolicy==CKPT_RETAIN) ? SA_TIME_END : 0;
    SaAisErrorT err = saCkptCheckpointRetentionDurationSet (_checkpointHandle, rTime);
    if (err != SA_AIS_OK) {
        return CkptFailure;
    }
    return CkptSuccess;
}

---