Logo Search packages:      
Sourcecode: zeroc-ice version File versions  Download package

MapI.cpp

// **********************************************************************
//
// Copyright (c) 2003-2006 ZeroC, Inc. All rights reserved.
//
// This copy of Ice is licensed to you under the terms described in the
// ICE_LICENSE file included in this distribution.
//
// **********************************************************************

#include <Freeze/MapI.h>
#include <Freeze/Exception.h>
#include <Freeze/SharedDb.h>
#include <Freeze/Util.h>
#include <Freeze/Catalog.h>
#include <stdlib.h>

using namespace std;
using namespace Ice;
using namespace Freeze;


//
// MapIndexBase (from Map.h)
//

Freeze::MapIndexBase::~MapIndexBase()
{
}

Freeze::MapIndexBase::MapIndexBase(
    const string& name,
    bool enabled) :
    KeyCompareBase(enabled),
    _name(name),
    _impl(0),
    _map(0)
{
}

const string&
MapIndexBase::name() const
{
    return _name;
}

IteratorHelper*
Freeze::MapIndexBase::untypedFind(const Key& k, bool ro, bool onlyDups) const
{
    return _impl->untypedFind(k, ro, *_map, onlyDups);
}

IteratorHelper*
Freeze::MapIndexBase::untypedLowerBound(const Key& k, bool ro) const
{
    return _impl->untypedLowerBound(k, ro, *_map);
}

IteratorHelper*
Freeze::MapIndexBase::untypedUpperBound(const Key& k, bool ro) const
{
    return _impl->untypedUpperBound(k, ro, *_map);
}

int 
Freeze::MapIndexBase::untypedCount(const Key& k) const
{
    return _impl->untypedCount(k, _map->connection());
}

//
// KeyCompareBase
//
Freeze::KeyCompareBase::KeyCompareBase(bool enabled) :
    _enabled(enabled)
{}

bool
Freeze::KeyCompareBase::compareEnabled() const
{
    return _enabled;
}

//
// MapHelper (from Map.h)
//

Freeze::MapHelper*
Freeze::MapHelper::create(const Freeze::ConnectionPtr& connection, 
                    const string& dbName,
                    const string& key,
                    const string& value,
                    const Freeze::KeyCompareBasePtr& keyCompare,
                    const std::vector<MapIndexBasePtr>& indices,
                    bool createDb)
{
    Freeze::ConnectionIPtr connectionI = Freeze::ConnectionIPtr::dynamicCast(connection.get());
    return new MapHelperI(connectionI, dbName, key, value, keyCompare, indices, createDb);
}

Freeze::MapHelper::~MapHelper()
{
}


//
// IteratorHelper (from Map.h)
//

Freeze::IteratorHelper* 
Freeze::IteratorHelper::create(const MapHelper& m, bool readOnly)
{
    const MapHelperI& actualMap = dynamic_cast<const MapHelperI&>(m);

    auto_ptr<IteratorHelperI> r(new IteratorHelperI(actualMap, readOnly, 
                                        0, false));
    if(r->next())
    {
      return r.release();
    }
    else
    {
      return 0;
    }
}


Freeze::IteratorHelper::~IteratorHelper()
{
}


//
// IteratorHelperI
//

Freeze::IteratorHelperI::IteratorHelperI(const MapHelperI& m, bool readOnly, 
                               const MapIndexBasePtr& index,
                               bool onlyDups) :
    _map(m),
    _dbc(0),
    _indexed(index != 0),
    _onlyDups(onlyDups),
    _tx(0)
{
    if(_map._trace >= 2)
    {
      Trace out(_map._connection->communicator()->getLogger(), "Freeze.Map");
      out << "opening iterator on Db \"" << _map._dbName << "\"";
      if(index != 0)
      {
          out << " with index \"" << index->name() << "\"";
      }
    }

    DbTxn* txn = _map._connection->dbTxn();
    
    if(txn == 0 && !readOnly)
    {
      //
      // Need to start a transaction
      //
      _tx = new Tx(_map);
      txn = _tx->getTxn();
    }

    try
    {
      if(index != 0)
      {
          index->_impl->db()->cursor(txn, &_dbc, 0);
      }
      else
      {
          _map._db->cursor(txn, &_dbc, 0);
      }
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }

    _map._iteratorList.push_back(this);
}


Freeze::IteratorHelperI::IteratorHelperI(const IteratorHelperI& it) :
    _map(it._map),
    _dbc(0),
    _indexed(it._indexed),
    _onlyDups(it._onlyDups),
    _tx(0)
{
    if(_map._trace >= 2)
    {
      Trace out(_map._connection->communicator()->getLogger(), "Freeze.Map");
      out << "duplicating iterator on Db \"" << _map._dbName << "\"";
    }

    try
    {
      it._dbc->dup(&_dbc, DB_POSITION);
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
   
    _tx = it._tx;
    _map._iteratorList.push_back(this);
}

Freeze::IteratorHelperI::~IteratorHelperI()
{
    close();
}


bool 
Freeze::IteratorHelperI::find(const Key& key) const
{
    Dbt dbKey;
    initializeInDbt(key, dbKey);

    //
    // Keep 0 length since we're not interested in the data
    //
    Dbt dbValue;
    dbValue.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);

    try
    {
      if(_dbc->get(&dbKey, &dbValue, DB_SET) == 0)
      {
          return true;
      }
      else
      {
          return false;
      }
    }
    catch(const ::DbDeadlockException& dx)
    {
      if(_tx != 0)
      {
          _tx->dead();
      }

      DeadlockException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
}

bool 
Freeze::IteratorHelperI::lowerBound(const Key& key) const
{
    //
    // We retrieve the action key for upperBound
    //
    Dbt dbKey;
    _key = key;
    initializeOutDbt(_key, dbKey);
    dbKey.set_size(static_cast<u_int32_t>(_key.size()));

    //
    // Keep 0 length since we're not interested in the data
    //
    Dbt dbValue;
    dbValue.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);

    for(;;)
    {
      try
      {
          int err = _dbc->get(&dbKey, &dbValue, DB_SET_RANGE);
          if(err == 0)
          {
            _key.resize(dbKey.get_size());
            return true;
          }
          else if(err == DB_NOTFOUND)
          {
            return false;
          }
          else
          {
            //
            // Bug in Freeze
            //
            assert(0);
            throw DatabaseException(__FILE__, __LINE__);
          }
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(_tx != 0)
          {
            _tx->dead();
          }
          
          DeadlockException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
      catch(const ::DbException& dx)
      {
          handleDbException(dx, _key, dbKey, __FILE__, __LINE__);
      }
    }
}

bool 
Freeze::IteratorHelperI::upperBound(const Key& key) const
{
    if(lowerBound(key))
    {
      if(_key == key)
      {
          return next(true);
      }
      else
      {
          return true;
      }
    }
    else
    {
      return false;
    }
}


Freeze::IteratorHelper*
Freeze::IteratorHelperI::clone() const
{
    return new IteratorHelperI(*this);
}
    
void
Freeze::IteratorHelperI::get(const Key*& key, const Value*& value) const
{
    key = &_key;
    value = &_value;

    size_t keySize = _key.capacity();
    if(keySize < 1024)
    {
      keySize = 1024;
    }
    _key.resize(keySize);

    Dbt dbKey;
    initializeOutDbt(_key, dbKey);
    
    size_t valueSize = _value.capacity();
    if(valueSize < 1024)
    {
      valueSize = 1024;
    }
    _value.resize(valueSize);
    Dbt dbValue;
    initializeOutDbt(_value, dbValue);

    for(;;)
    {
      try
      {
          int err;

          if(_indexed)
          {
            //
            // Not interested in getting the index's key
            //
            Dbt iKey;
            iKey.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
          
            err = _dbc->pget(&iKey, &dbKey, &dbValue, DB_CURRENT);
          }
          else
          {
            err = _dbc->get(&dbKey, &dbValue, DB_CURRENT);
          }

          if(err == 0)
          {
            _key.resize(dbKey.get_size());
            _value.resize(dbValue.get_size());
            break; // for(;;)
          }
          else if(err == DB_KEYEMPTY)
          {
            throw InvalidPositionException(__FILE__, __LINE__);
          }
          else 
          {
            //
            // Bug in Freeze
            //
            assert(0);
            throw DatabaseException(__FILE__, __LINE__);
          }
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(_tx != 0)
          {
            _tx->dead();
          }

          DeadlockException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
      catch(const ::DbException& dx)
      {
          handleDbException(dx, _key, dbKey, _value, dbValue,
                        __FILE__, __LINE__);
      }
    }
}
    
const Freeze::Key*
Freeze::IteratorHelperI::get() const
{
    size_t keySize = _key.capacity();
    if(keySize < 1024)
    {
      keySize = 1024;
    }
    _key.resize(keySize);

    Dbt dbKey;
    initializeOutDbt(_key, dbKey);
    
    //
    // Keep 0 length since we're not interested in the data
    //
    Dbt dbValue;
    dbValue.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
    
    for(;;)
    {
      try
      {
          int err;
          if(_indexed)
          {
            //
            // Not interested in getting the index's key
            //
            Dbt iKey;
            iKey.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
          
            err = _dbc->pget(&iKey, &dbKey, &dbValue, DB_CURRENT);
          }
          else
          {
            err = _dbc->get(&dbKey, &dbValue, DB_CURRENT);
          }

          if(err == 0)
          {
            _key.resize(dbKey.get_size());
            return &_key;
          }
          else if(err == DB_KEYEMPTY)
          {
            throw InvalidPositionException(__FILE__, __LINE__);
          }
          else 
          {
            //
            // Bug in Freeze
            //
            assert(0);
            throw DatabaseException(__FILE__, __LINE__);
          }
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(_tx != 0)
          {
            _tx->dead();
          }

          DeadlockException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
      catch(const ::DbException& dx)
      {
          handleDbException(dx, _key, dbKey, __FILE__, __LINE__);
      }
    }
}

void 
Freeze::IteratorHelperI::set(const Value& value)
{
    if(_indexed)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = "Cannot set an iterator retrieved through an index";
      throw ex;
    }
    
    //
    // key ignored
    //
    Dbt dbKey;
    dbKey.set_flags(DB_DBT_USERMEM);

    Dbt dbValue;
    initializeInDbt(value, dbValue);

    if(_tx != 0)
    {
      _map.closeAllIteratorsExcept(_tx);
    }

    try
    {
      int err;
      err = _dbc->put(&dbKey, &dbValue, DB_CURRENT);
      assert(err == 0);
    }
    catch(const ::DbDeadlockException& dx)
    {
      if(_tx != 0)
      {
          _tx->dead();
      }

      DeadlockException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
}

void
Freeze::IteratorHelperI::erase()
{
    if(_tx != 0)
    {
      _map.closeAllIteratorsExcept(_tx);
    }

    try
    {
      int err = _dbc->del(0);
      if(err == DB_KEYEMPTY)
      {
          throw InvalidPositionException(__FILE__, __LINE__);
      }
      assert(err == 0);
    }
    catch(const ::DbDeadlockException& dx)
    {
      if(_tx != 0)
      {
          _tx->dead();
      }

      DeadlockException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
}

bool 
Freeze::IteratorHelperI::next() const
{
    return next(false);
}

bool
Freeze::IteratorHelperI::next(bool skipDups) const
{
    //
    // Keep 0 length since we're not interested in the data
    //
    Dbt dbKey;
    dbKey.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
    Dbt dbValue;
    dbValue.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);

    int flags = DB_NEXT;
    if(skipDups)
    {
      flags = DB_NEXT_NODUP;
    }
    else if(_indexed && _onlyDups)
    {
      flags = DB_NEXT_DUP;
    }

    try
    {
      if(_dbc->get(&dbKey, &dbValue, flags) == 0)
      {
          return true;
      }
      else
      {
          return false;
      }
    }
    catch(const ::DbDeadlockException& dx)
    {
      if(_tx != 0)
      {
          _tx->dead();
      }

      DeadlockException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
}

void
Freeze::IteratorHelperI::close()
{
    if(_dbc != 0)
    {
      if(_map._trace >= 2)
      {
          Trace out(_map._connection->communicator()->getLogger(), "Freeze.Map");
          out << "closing iterator on Db \"" << _map._dbName << "\"";
      }
      
      try
      {
          _dbc->close();
      }
      catch(const ::DbDeadlockException& dx)
      {
          bool raiseException = (_tx == 0);
          cleanup();
          if(raiseException)
          {
            DeadlockException ex(__FILE__, __LINE__);
            ex.message = dx.what();
            throw ex;
          }
      }
      catch(const ::DbException& dx)
      {
          cleanup();
          DatabaseException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
      cleanup();
    }
}

void
Freeze::IteratorHelperI::cleanup()
{
    _dbc = 0;
    _map._iteratorList.remove(this);
    _tx = 0;
}


//
// IteratorHelperI::Tx
//

Freeze::IteratorHelperI::Tx::Tx(const MapHelperI& m) :
    _map(m),
    _txn(0),
    _dead(false)
{
    if(_map._trace >= 2)
    {
      Trace out(_map._connection->communicator()->getLogger(), "Freeze.Map");
      out << "starting transaction for Db \"" << _map._dbName << "\"";
    }

    try
    {
      _map._connection->dbEnv()->getEnv()->txn_begin(0, &_txn, 0);
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
}
      

Freeze::IteratorHelperI::Tx::~Tx()
{
    if(_dead)
    {
      if(_map._trace >= 2)
      {
          Trace out(_map._connection->communicator()->getLogger(), "");
          out << "aborting transaction for Db \"" << _map._dbName << "\"";
      }

      try
      {
          _txn->abort();
      }
      catch(...)
      {
          //
          // Ignore exceptions to avoid crash during stack unwinding
          //
      }
    }
    else
    {
      if(_map._trace >= 2)
      {
          Trace out(_map._connection->communicator()->getLogger(), "Freeze.Map");
          out << "committing transaction for Db \"" << _map._dbName.c_str() << "\"";
      }

      try
      {
          _txn->commit(0);
      }
      catch(const ::DbDeadlockException& dx)
      {
          DeadlockException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
      catch(const ::DbException& dx)
      {
          DatabaseException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
    }
}

void
Freeze::IteratorHelperI::Tx::dead()
{
    _dead = true;
}



//
// MapHelperI
//


Freeze::MapHelperI::MapHelperI(const ConnectionIPtr& connection, 
                         const string& dbName,
                         const string& key,
                         const string& value,
                         const KeyCompareBasePtr& keyCompare,
                         const vector<MapIndexBasePtr>& indices,
                         bool createDb) :
    _connection(connection),
    _db(SharedDb::get(connection, dbName, key, value, keyCompare, indices, createDb)),
    _dbName(dbName),
    _trace(connection->trace())
{ 
    for(vector<MapIndexBasePtr>::const_iterator p = indices.begin(); 
      p != indices.end(); ++p)
    {
      const MapIndexBasePtr& indexBase = *p;
      assert(indexBase->_impl != 0);
      assert(indexBase->_communicator == _connection->communicator());
      assert(indexBase->_map == 0);
      
#ifndef NDEBUG
      bool inserted =
#endif 
          _indices.insert(IndexMap::value_type(indexBase->name(), indexBase)).second;
      assert(inserted);
      indexBase->_map = this;
    }
    
    _connection->registerMap(this);
}

Freeze::MapHelperI::~MapHelperI()
{
    close();
}

Freeze::IteratorHelper*
Freeze::MapHelperI::find(const Key& k, bool readOnly) const
{
    for(;;)
    {
      try
      {  
          auto_ptr<IteratorHelperI> r(new IteratorHelperI(*this, readOnly, 0, false));
          if(r->find(k))
          {
            return r.release();
          }
          else
          {
            return 0;
          }
      }
      catch(const DeadlockException&)
      {
          if(_connection->dbTxn() != 0)
          {
            throw;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::find on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
    }
}

Freeze::IteratorHelper*
Freeze::MapHelperI::lowerBound(const Key& k, bool readOnly) const
{
    for(;;)
    {
      try
      {  
          auto_ptr<IteratorHelperI> r(new IteratorHelperI(*this, readOnly, 0, false));
          if(r->lowerBound(k))
          {
            return r.release();
          }
          else
          {
            return 0;
          }
      }
      catch(const DeadlockException&)
      {
          if(_connection->dbTxn() != 0)
          {
            throw;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::lowerBound on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
    }
}

Freeze::IteratorHelper*
Freeze::MapHelperI::upperBound(const Key& k, bool readOnly) const
{
    for(;;)
    {
      try
      {  
          auto_ptr<IteratorHelperI> r(new IteratorHelperI(*this, readOnly, 0, false));
          if(r->upperBound(k))
          {
            return r.release();
          }
          else
          {
            return 0;
          }
      }
      catch(const DeadlockException&)
      {
          if(_connection->dbTxn() != 0)
          {
            throw;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::upperBound on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
    }
}

void
Freeze::MapHelperI::put(const Key& key, const Value& value)
{
    Dbt dbKey;
    Dbt dbValue;
    initializeInDbt(key, dbKey);
    initializeInDbt(value, dbValue);
 
    DbTxn* txn = _connection->dbTxn();
    if(txn == 0)
    {
      closeAllIterators();
    }

    for(;;)
    {
      try
      {
          int err = _db->put(txn, &dbKey, &dbValue, 
                         txn != 0 ? 0 : DB_AUTO_COMMIT);
          
          if(err == 0)
          {
            break;
          }
          else
          {
            //
            // Bug in Freeze
            //
            throw DatabaseException(__FILE__, __LINE__);
          }
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(txn != 0)
          {
            DeadlockException ex(__FILE__, __LINE__);
            ex.message = dx.what();
            throw ex;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::put on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
      catch(const ::DbException& dx)
      {
          DatabaseException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
    }
}

size_t
Freeze::MapHelperI::erase(const Key& key)
{
    Dbt dbKey;
    initializeInDbt(key, dbKey);

    DbTxn* txn = _connection->dbTxn();
    if(txn == 0)
    {
      closeAllIterators();
    }

    for(;;)
    {
      try
      {
          int err = _db->del(txn, &dbKey, txn != 0 ? 0 : DB_AUTO_COMMIT);

          if(err == 0)
          {
            return true;
          }
          else if(err == DB_NOTFOUND)
          {
            return false;
          }
          else
          {
            assert(0);
            throw DatabaseException(__FILE__, __LINE__);
          }
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(txn != 0)
          {
            DeadlockException ex(__FILE__, __LINE__);
            ex.message = dx.what();
            throw ex;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::erase on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
      catch(const ::DbException& dx)
      {
          DatabaseException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
    }
}

size_t
Freeze::MapHelperI::count(const Key& key) const
{
    Dbt dbKey;
    initializeInDbt(key, dbKey);
    
    //
    // Keep 0 length since we're not interested in the data
    //
    Dbt dbValue;
    dbValue.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
    
    for(;;)
    {
      try
      {
          int err = _db->get(_connection->dbTxn(), &dbKey, &dbValue, 0);
          
          if(err == 0)
          {
            return 1;
          }
          else if(err == DB_NOTFOUND)
          {
            return 0;
          }
          else
          {
            assert(0);
            throw DatabaseException(__FILE__, __LINE__);
          }
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(_connection->dbTxn() != 0)
          {
            DeadlockException ex(__FILE__, __LINE__);
            ex.message = dx.what();
            throw ex;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::count on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
      catch(const ::DbException& dx)
      {
          DatabaseException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
    }
}
    
void
Freeze::MapHelperI::clear()
{
    DbTxn* txn = _connection->dbTxn();
    if(txn == 0)
    {
      closeAllIterators();
    }

    for(;;)
    {
      try
      {
          u_int32_t count;
          int err;
          err = _db->truncate(txn, &count, txn != 0 ? 0 : DB_AUTO_COMMIT);
          assert(err == 0);
          break;
      }
      catch(const ::DbDeadlockException& dx)
      {
          if(txn != 0)
          {
            DeadlockException ex(__FILE__, __LINE__);
            ex.message = dx.what();
            throw ex;
          }
          else
          {
            if(_connection->deadlockWarning())
            {
                Warning out(_connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapHelperI::clear on Map \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Ignored, try again
            //
          }
      }
      catch(const ::DbException& dx)
      {
          DatabaseException ex(__FILE__, __LINE__);
          ex.message = dx.what();
          throw ex;
      }
    }
}

void
Freeze::MapHelperI::destroy()
{
    if(_dbName == catalogName())
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = "You cannot destroy the " + catalogName() + " database";
      throw ex;
    }

    TransactionPtr tx = _connection->currentTransaction();
    bool ownTx = (tx == 0);
    if(ownTx)
    { 
      tx = _connection->beginTransaction();
    }
   
    DbTxn* txn = _connection->dbTxn();

    try
    {
      close();

      Catalog catalog(_connection, catalogName());
      catalog.erase(_dbName);
      _connection->dbEnv()->getEnv()->dbremove(txn, _dbName.c_str(), 0, 0);

      if(ownTx)
      {
          tx->commit();
      }
    }
    catch(const ::DbException& dx)
    {
      if(ownTx)
      {
          tx = _connection->currentTransaction();
          if(tx != 0)
          {
            try
            {
                tx->rollback();
            }
            catch(...)
            {
            }
          }
      }

      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
    catch(...)
    {
      if(ownTx)
      {
          tx = _connection->currentTransaction();
          if(tx != 0)
          {
            try
            {
                tx->rollback();
            }
            catch(...)
            {
            }
          }
      }
      throw;
    }
}


size_t
Freeze::MapHelperI::size() const
{
    //
    // TODO: DB_FAST_STAT doesn't seem to do what the documentation says...
    //
    DB_BTREE_STAT* s;

    try
    {
#if DB_VERSION_MAJOR != 4
   #error Freeze requires DB 4.x
#endif
#if DB_VERSION_MINOR < 3
       _db->stat(&s, 0);
#else
       _db->stat(_connection->dbTxn(), &s, 0);
#endif
    }
    catch(const ::DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }

    size_t num = s->bt_ndata;
    free(s);
    return num;
}


void
Freeze::MapHelperI::closeAllIterators()
{
    while(!_iteratorList.empty())
    {
      (*_iteratorList.begin())->close();
    }
}

const MapIndexBasePtr&
Freeze::MapHelperI::index(const string& name) const
{
    IndexMap::const_iterator p = _indices.find(name);
    if(p == _indices.end())
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = "Cannot find index \"" + name + "\"";
      throw ex;
    }
    return p->second;
}

void
Freeze::MapHelperI::close()
{
    if(_db != 0)
    {
      _connection->unregisterMap(this);
    }
    _db = 0;
    
    for(IndexMap::iterator p = _indices.begin(); p != _indices.end(); ++p)
    {
      MapIndexBasePtr& indexBase = p->second;

      indexBase->_impl = 0;
      indexBase->_map = 0;
    }
    _indices.clear();
}

void
Freeze::MapHelperI::closeAllIteratorsExcept(const IteratorHelperI::TxPtr& tx) const
{
    assert(tx != 0);

    list<IteratorHelperI*>::iterator q = _iteratorList.begin();

    while(q != _iteratorList.end())
    {
      if((*q)->tx().get() == tx.get())
      {
          ++q;
      }
      else
      {
          (*q)->close();
          q = _iteratorList.begin();
      }
    }
}


//
// MapIndexI
//

extern "C" 
{
static int customIndexCompare(DB* db, const DBT* dbt1, const DBT* dbt2)
{
    MapIndexI* me = static_cast<MapIndexI*>(db->app_private);
    Byte* first = static_cast<Byte*>(dbt1->data);
    Key k1(first, first + dbt1->size);
    first = static_cast<Byte*>(dbt2->data);
    Key k2(first, first + dbt2->size);

    return me->getKeyCompare()->compare(k1, k2);
}
}

static int 
callback(Db* secondary, const Dbt* key, const Dbt* value, Dbt* result)
{
    void* indexObj = secondary->get_app_private();
    MapIndexI* index = static_cast<MapIndexI*>(indexObj);
    assert(index != 0);
    return index->secondaryKeyCreate(secondary, key, value, result);
}


Freeze::MapIndexI::MapIndexI(const ConnectionIPtr& connection, SharedDb& db,
                       DbTxn* txn, bool createDb, const MapIndexBasePtr& index) :
    _index(index)
{
    assert(txn != 0);
    
    _db.reset(new Db(connection->dbEnv()->getEnv(), 0));
    _db->set_flags(DB_DUP | DB_DUPSORT);
    _db->set_app_private(this);

    u_int32_t flags = 0;
    if(createDb)
    {
      flags = DB_CREATE;
    }

    _dbName = db.dbName() + "." + _index->name();

    _db->set_app_private(this);

    if(index->compareEnabled())
    {
      _db->set_bt_compare(&customIndexCompare);
    }

    _db->open(txn, _dbName.c_str(), 0, DB_BTREE, flags, FREEZE_DB_MODE);

    //
    // To populate empty indices
    //
    flags = DB_CREATE;
    db.associate(txn, _db.get(), callback, flags);

    //
    // Note: caller catch and translates exceptions
    //
}
   
Freeze::MapIndexI::~MapIndexI()
{
    _db->close(0);
}

IteratorHelper* 
Freeze::MapIndexI::untypedFind(const Key& k, bool ro, const MapHelperI& m, 
                         bool onlyDups) const
{
    auto_ptr<IteratorHelperI> r(new IteratorHelperI(m, ro, _index, onlyDups));

    if(r->find(k))
    {
      return r.release();
    }
    else
    {
      return 0;
    }
}

IteratorHelper* 
Freeze::MapIndexI::untypedLowerBound(const Key& k, bool ro, const MapHelperI& m) const
{
    auto_ptr<IteratorHelperI> r(new IteratorHelperI(m, ro, _index, false));

    if(r->lowerBound(k))
    {
      return r.release();
    }
    else
    {
      return 0;
    }
}

IteratorHelper* 
Freeze::MapIndexI::untypedUpperBound(const Key& k, bool ro, const MapHelperI& m) const
{
    auto_ptr<IteratorHelperI> r(new IteratorHelperI(m, ro, _index, false));

    if(r->upperBound(k))
    {
      return r.release();
    }
    else
    {
      return 0;
    }
}

int 
Freeze::MapIndexI::untypedCount(const Key& k, const ConnectionIPtr& connection) const
{
    Dbt dbKey;
    initializeInDbt(k, dbKey);
    
    Dbt dbValue;
    dbValue.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);

    int result = 0;
    
    try
    {
      for(;;)
      {
          Dbc* dbc = 0;
          
          try
          {
            //
            // Move to the first record
            // 
            _db->cursor(0, &dbc, 0);
            bool found = (dbc->get(&dbKey, &dbValue, DB_SET) == 0);
            
            if(found)
            {
                db_recno_t count = 0;
                dbc->count(&count, 0);
                result = static_cast<int>(count);
            }
            
            Dbc* toClose = dbc;
            dbc = 0;
            toClose->close();
            break; // for (;;)
          }
          catch(const DbDeadlockException&)
          {
            if(dbc != 0)
            {
                try
                {
                  dbc->close();
                }
                catch(const DbDeadlockException&)
                {
                  //
                  // Ignored
                  //
                }
            }

            if(connection->deadlockWarning())
            {
                Warning out(connection->communicator()->getLogger());
                out << "Deadlock in Freeze::MapIndexI::untypedCount while searching \"" 
                  << _dbName << "\"; retrying ...";
            }

            //
            // Retry
            //
          }
          catch(...)
          {
            if(dbc != 0)
            {
                try
                {
                  dbc->close();
                }
                catch(const DbDeadlockException&)
                {
                  //
                  // Ignored
                  //
                }
            }
            throw;
          }
      }
    }
    catch(const DbException& dx)
    {
      DatabaseException ex(__FILE__, __LINE__);
      ex.message = dx.what();
      throw ex;
    }
    
    return result;
}
    
int
Freeze::MapIndexI::secondaryKeyCreate(Db* secondary, const Dbt* dbKey, 
                              const Dbt* dbValue, Dbt* result)
{
    Byte* first = static_cast<Byte*>(dbValue->get_data());
    Value value(first, first + dbValue->get_size());
    
    Key bytes;
    _index->marshalKey(value, bytes);

    result->set_flags(DB_DBT_APPMALLOC);
    void* data = malloc(bytes.size());
    memcpy(data, &bytes[0], bytes.size());
    result->set_data(data);
    result->set_size(static_cast<u_int32_t>(bytes.size()));
    return 0;
}

//
// Print for the various exception types.
//
void
Freeze::DeadlockException::ice_print(ostream& out) const
{
    Exception::ice_print(out);
    out << ":\ndatabase deadlock:\n" << message;
}

void
Freeze::NotFoundException::ice_print(ostream& out) const
{
    Exception::ice_print(out);
    out << ":\ndatabase record not found:\n" << message;
}

void
Freeze::DatabaseException::ice_print(ostream& out) const
{
    Exception::ice_print(out);
    out << ":\n" << message;
}


void
Freeze::InvalidPositionException::ice_print(ostream& out) const
{
    Exception::ice_print(out);
    out << ":\ninvalid position";
}

Generated by  Doxygen 1.6.0   Back to index