text:
/*
 * Copyright (C) 2005-2015 Christoph Rupp (chris@crupp.de).
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "0root/root.h"

#include <boost/scope_exit.hpp>

// Always verify that a file of level N does not include headers > N!
#include "3page_manager/page_manager.h"
#include "3journal/journal.h"
#include "3blob_manager/blob_manager.h"
#include "3btree/btree_index.h"
#include "3btree/btree_index_factory.h"
#include "4db/db_local.h"
#include "4context/context.h"
#include "4cursor/cursor_local.h"
#include "4txn/txn_local.h"
#include "4txn/txn_cursor.h"

#ifndef HAM_ROOT_H
#  error "root.h was not included"
#endif

namespace hamsterdb {

ham_status_t
LocalDatabase::check_insert_conflicts(Context *context, TransactionNode *node,
                    ham_key_t *key, uint32_t flags)
{
  TransactionOperation *op = 0;

  /*
   * pick the tree_node of this key, and walk through each operation
   * in reverse chronological order (from newest to oldest):
   * - is this op part of an aborted txn? then skip it
   * - is this op part of a committed txn? then look at the
   *    operation in detail
   * - is this op part of an txn which is still active? return an error
   *    because we've found a conflict
   * - if a committed txn has erased the item then there's no need
   *    to continue checking older, committed txns
   */
  op = node->get_newest_op();
  while (op) {
    LocalTransaction *optxn = op->get_txn();
    if (optxn->is_aborted())
      ; /* nop */
    else if (optxn->is_committed() || context->txn == optxn) {
      /* if key was erased then it doesn't exist and can be
       * inserted without problems */
      if (op->get_flags() & TransactionOperation::kIsFlushed)
        ; /* nop */
      else if (op->get_flags() & TransactionOperation::kErase)
        return (0);
      /* if the key already exists then we can only continue if
       * we're allowed to overwrite it or to insert a duplicate */
      else if ((op->get_flags() & TransactionOperation::kInsert)
          || (op->get_flags() & TransactionOperation::kInsertOverwrite)
          || (op->get_flags() & TransactionOperation::kInsertDuplicate)) {
        if ((flags & HAM_OVERWRITE) || (flags & HAM_DUPLICATE))
          return (0);
        else
          return (HAM_DUPLICATE_KEY);
      }
      else if (!(op->get_flags() & TransactionOperation::kNop)) {
        ham_assert(!"shouldn't be here");
        return (HAM_DUPLICATE_KEY);
      }
    }
    else { /* txn is still active */
      return (HAM_TXN_CONFLICT);
    }

    op = op->get_previous_in_node();
  }

  /*
   * we've successfully checked all un-flushed transactions and there
   * were no conflicts. Now check all transactions which are already
   * flushed - basically that's identical to a btree lookup.
   *
   * however we can skip this check if we do not care about duplicates.
   */
  if ((flags & HAM_OVERWRITE)
          || (flags & HAM_DUPLICATE)
          || (get_flags() & (HAM_RECORD_NUMBER32 | HAM_RECORD_NUMBER64)))
    return (0);

  ham_status_t st = m_btree_index->find(context, 0, key, 0, 0, 0, flags);
  switch (st) {
    case HAM_KEY_NOT_FOUND:
      return (0);
    case HAM_SUCCESS:
      return (HAM_DUPLICATE_KEY);
    default:
      return (st);
  }
}

ham_status_t
LocalDatabase::check_erase_conflicts(Context *context, TransactionNode *node,
                    ham_key_t *key, uint32_t flags)
{
  TransactionOperation *op = 0;

  /*
   * pick the tree_node of this key, and walk through each operation
   * in reverse chronological order (from newest to oldest):
   * - is this op part of an aborted txn? then skip it
   * - is this op part of a committed txn? then look at the
   *    operation in detail
   * - is this op part of an txn which is still active? return an error
   *    because we've found a conflict
   * - if a committed txn has erased the item then there's no need
   *    to continue checking older, committed txns
   */
  op = node->get_newest_op();
  while (op) {
    Transaction *optxn = op->get_txn();
    if (optxn->is_aborted())
      ; /* nop */
    else if (optxn->is_committed() || context->txn == optxn) {
      if (op->get_flags() & TransactionOperation::kIsFlushed)
        ; /* nop */
      /* if key was erased then it doesn't exist and we fail with
       * an error */
      else if (op->get_flags() & TransactionOperation::kErase)
        return (HAM_KEY_NOT_FOUND);
      /* if the key exists then we're successful */
      else if ((op->get_flags() & TransactionOperation::kInsert)
          || (op->get_flags() & TransactionOperation::kInsertOverwrite)
          || (op->get_flags() & TransactionOperation::kInsertDuplicate)) {
        return (0);
      }
      else if (!(op->get_flags() & TransactionOperation::kNop)) {
        ham_assert(!"shouldn't be here");
        return (HAM_KEY_NOT_FOUND);
      }
    }
    else { /* txn is still active */
      return (HAM_TXN_CONFLICT);
    }

    op = op->get_previous_in_node();
  }

  /*
   * we've successfully checked all un-flushed transactions and there
   * were no conflicts. Now check all transactions which are already
   * flushed - basically that's identical to a btree lookup.
   */
  return (m_btree_index->find(context, 0, key, 0, 0, 0, flags));
}

ham_status_t
LocalDatabase::insert_txn(Context *context, ham_key_t *key,
                ham_record_t *record, uint32_t flags, TransactionCursor *cursor)
{
  ham_status_t st = 0;
  TransactionOperation *op;
  bool node_created = false;

  /* get (or create) the node for this key */
  TransactionNode *node = m_txn_index->get(key, 0);
  if (!node) {
    node = new TransactionNode(this, key);
    node_created = true;
    // TODO only store when the operation is successful?
    m_txn_index->store(node);
  }

  // check for conflicts of this key
  //
  // !!
  // afterwards, clear the changeset; check_insert_conflicts()
  // checks if a key already exists, and this fills the changeset
  st = check_insert_conflicts(context, node, key, flags);
  if (st) {
    if (node_created) {
      m_txn_index->remove(node);
      delete node;
    }
    return (st);
  }

  // append a new operation to this node
  op = node->append(context->txn, flags,
                (flags & HAM_PARTIAL) |
                ((flags & HAM_DUPLICATE)
                    ? TransactionOperation::kInsertDuplicate
                    : (flags & HAM_OVERWRITE)
                        ? TransactionOperation::kInsertOverwrite
                        : TransactionOperation::kInsert),
                lenv()->next_lsn(), key, record);

  // if there's a cursor then couple it to the op; also store the
  // dupecache-index in the op (it's needed for DUPLICATE_INSERT_BEFORE/NEXT) */
  if (cursor) {
    LocalCursor *c = cursor->get_parent();
    if (c->get_dupecache_index())
      op->set_referenced_dupe(c->get_dupecache_index());

    cursor->couple_to_op(op);

    // all other cursors need to increment their dupe index, if their
    // index is > this cursor's index
    increment_dupe_index(context, node, c, c->get_dupecache_index());
  }

  // append journal entry
  if (m_env->get_flags() & HAM_ENABLE_RECOVERY
      && m_env->get_flags() & HAM_ENABLE_TRANSACTIONS) {
    Journal *j = lenv()->journal();
    j->append_insert(this, context->txn, key, record,
              flags & HAM_DUPLICATE ? flags : flags | HAM_OVERWRITE,
              op->get_lsn());
  }

  ham_assert(st == 0);
  return (0);
}

bool
LocalDatabase::is_key_erased(Context *context, ham_key_t *key)
{
  /* get the node for this key (but don't create a new one if it does
   * not yet exist) */
  TransactionNode *node = m_txn_index->get(key, 0);
  if (!node)
    return (false);

  /* now traverse the tree, check if the key was erased */
  TransactionOperation *op = node->get_newest_op();
  while (op) {
    Transaction *optxn = op->get_txn();
    if (optxn->is_aborted())
      ; /* nop */
    else if (optxn->is_committed() || context->txn == optxn) {
      if (op->get_flags() & TransactionOperation::kIsFlushed)
        ; /* continue */
      else if (op->get_flags() & TransactionOperation::kErase) {
        /* TODO does not check duplicates!! */
        return (true);
      }
      else if ((op->get_flags() & TransactionOperation::kInsert)
          || (op->get_flags() & TransactionOperation::kInsertOverwrite)
          || (op->get_flags() & TransactionOperation::kInsertDuplicate)) {
        return (false);
      }
    }

    op = op->get_previous_in_node();
  }

  return (false);
}

ham_status_t
LocalDatabase::find_txn(Context *context, LocalCursor *cursor,
                ham_key_t *key, ham_record_t *record, uint32_t flags)
{
  ham_status_t st = 0;
  TransactionOperation *op = 0;
  bool first_loop = true;
  bool exact_is_erased = false;

  ByteArray *pkey_arena = &key_arena(context->txn);
  ByteArray *precord_arena = &record_arena(context->txn);

  ham_key_set_intflags(key,
        (ham_key_get_intflags(key) & (~BtreeKey::kApproximate)));

  /* get the node for this key (but don't create a new one if it does
   * not yet exist) */
  TransactionNode *node = m_txn_index->get(key, flags);

  /*
   * pick the node of this key, and walk through each operation
   * in reverse chronological order (from newest to oldest):
   * - is this op part of an aborted txn? then skip it
   * - is this op part of a committed txn? then look at the
   *    operation in detail
   * - is this op part of an txn which is still active? return an error
   *    because we've found a conflict
   * - if a committed txn has erased the item then there's no need
   *    to continue checking older, committed txns
   */
retry:
  if (node)
    op = node->get_newest_op();
  while (op) {
    Transaction *optxn = op->get_txn();
    if (optxn->is_aborted())
      ; /* nop */
    else if (optxn->is_committed() || context->txn == optxn) {
      if (op->get_flags() & TransactionOperation::kIsFlushed)
        ; /* nop */
      /* if key was erased then it doesn't exist and we can return
       * immediately
       *
       * if an approximate match is requested then move to the next
       * or previous node
       */
      else if (op->get_flags() & TransactionOperation::kErase) {
        if (first_loop
            && !(ham_key_get_intflags(key) & BtreeKey::kApproximate))
          exact_is_erased = true;
        first_loop = false;
        if (flags & HAM_FIND_LT_MATCH) {
          node = node->get_previous_sibling();
          if (!node)
            break;
          ham_key_set_intflags(key,
              (ham_key_get_intflags(key) | BtreeKey::kApproximate));
          goto retry;
        }
        else if (flags & HAM_FIND_GT_MATCH) {
          node = node->get_next_sibling();
          if (!node)
            break;
          ham_key_set_intflags(key,
              (ham_key_get_intflags(key) | BtreeKey::kApproximate));
          goto retry;
        }
        /* if a duplicate was deleted then check if there are other duplicates
         * left */
        st = HAM_KEY_NOT_FOUND;
        // TODO merge both calls
        if (cursor) {
          cursor->get_txn_cursor()->couple_to_op(op);
          cursor->couple_to_txnop();
        }
        if (op->get_referenced_dupe() > 1) {
          // not the first dupe - there are other dupes
          st = 0;
        }
        else if (op->get_referenced_dupe() == 1) {
          // check if there are other dupes
          bool is_equal;
          (void)cursor->sync(context, LocalCursor::kSyncOnlyEqualKeys, &is_equal);
          if (!is_equal) // TODO merge w/ line above?
            cursor->set_to_nil(LocalCursor::kBtree);
          st = cursor->get_dupecache_count(context) ? 0 : HAM_KEY_NOT_FOUND;
        }
        return (st);
      }
      /* if the key already exists then return its record; do not
       * return pointers to TransactionOperation::get_record, because it may be
       * flushed and the user's pointers would be invalid */
      else if ((op->get_flags() & TransactionOperation::kInsert)
          || (op->get_flags() & TransactionOperation::kInsertOverwrite)
          || (op->get_flags() & TransactionOperation::kInsertDuplicate)) {
        if (cursor) { // TODO merge those calls
          cursor->get_txn_cursor()->couple_to_op(op);
          cursor->couple_to_txnop();
        }
        // approx match? leave the loop and continue
        // with the btree
        if (ham_key_get_intflags(key) & BtreeKey::kApproximate)
          break;
        // otherwise copy the record and return
        if (record)
          return (LocalDatabase::copy_record(this, context->txn, op, record));
        return (0);
      }
      else if (!(op->get_flags() & TransactionOperation::kNop)) {
        ham_assert(!"shouldn't be here");
        return (HAM_KEY_NOT_FOUND);
      }
    }
    else { /* txn is still active */
      return (HAM_TXN_CONFLICT);
    }

    op = op->get_previous_in_node();
  }

  /*
   * if there was an approximate match: check if the btree provides
   * a better match
   */
  if (op && ham_key_get_intflags(key) & BtreeKey::kApproximate) {
    ham_key_t *k = op->get_node()->get_key();

    ham_key_t txnkey = ham_make_key(::alloca(k->size), k->size);
    txnkey._flags = BtreeKey::kApproximate;
    ::memcpy(txnkey.data, k->data, k->size);

    ham_key_set_intflags(key, 0);

    // now lookup in the btree, but make sure that the retrieved key was
    // not deleted or overwritten in a transaction
    bool first_run = true;
    do {
      uint32_t new_flags = flags;

      // the "exact match" key was erased? then don't fetch it again
      if (!first_run || exact_is_erased) {
        first_run = false;
        new_flags = flags & (~HAM_FIND_EXACT_MATCH);
      }

      if (cursor)
        cursor->set_to_nil(LocalCursor::kBtree);
      st = m_btree_index->find(context, cursor, key, pkey_arena, record,
                      precord_arena, new_flags);
    } while (st == 0 && is_key_erased(context, key));

    // if the key was not found in the btree: return the key which was found
    // in the transaction tree
    if (st == HAM_KEY_NOT_FOUND) {
      if (!(key->flags & HAM_KEY_USER_ALLOC) && txnkey.data) {
        pkey_arena->resize(txnkey.size);
        key->data = pkey_arena->get_ptr();
      }
      if (txnkey.data)
        ::memcpy(key->data, txnkey.data, txnkey.size);
      key->size = txnkey.size;
      key->_flags = txnkey._flags;

      if (cursor) { // TODO merge those calls
        cursor->get_txn_cursor()->couple_to_op(op);
        cursor->couple_to_txnop();
      }
      if (record)
        return (LocalDatabase::copy_record(this, context->txn, op, record));
      return (0);
    }
    else if (st)
      return (st);

    // the btree key is a direct match? then return it
    if ((!(ham_key_get_intflags(key) & BtreeKey::kApproximate))
          && (flags & HAM_FIND_EXACT_MATCH)
          && !exact_is_erased) {
      if (cursor)
        cursor->couple_to_btree();
      return (0);
    }

    // if there's an approx match in the btree: compare both keys and
    // use the one that is closer. if the btree is closer: make sure
    // that it was not erased or overwritten in a transaction
    int cmp = m_btree_index->compare_keys(key, &txnkey);
    bool use_btree = false;
    if (flags & HAM_FIND_GT_MATCH) {
      if (cmp < 0)
        use_btree = true;
    }
    else if (flags & HAM_FIND_LT_MATCH) {
      if (cmp > 0)
        use_btree = true;
    }
    else
      ham_assert(!"shouldn't be here");

    if (use_btree) {
      // lookup again, with the same flags and the btree key.
      // this will check if the key was erased or overwritten
      // in a transaction
      st = find_txn(context, cursor, key, record, flags | HAM_FIND_EXACT_MATCH);
      if (st == 0)
        ham_key_set_intflags(key,
          (ham_key_get_intflags(key) | BtreeKey::kApproximate));
      return (st);
    }
    else { // use txn
      if (!(key->flags & HAM_KEY_USER_ALLOC) && txnkey.data) {
        pkey_arena->resize(txnkey.size);
        key->data = pkey_arena->get_ptr();
      }
      if (txnkey.data) {
        ::memcpy(key->data, txnkey.data, txnkey.size);
      }
      key->size = txnkey.size;
      key->_flags = txnkey._flags;

      if (cursor) { // TODO merge those calls
        cursor->get_txn_cursor()->couple_to_op(op);
        cursor->couple_to_txnop();
      }
      if (record)
        return (LocalDatabase::copy_record(this, context->txn, op, record));
      return (0);
    }
  }

  /*
   * no approximate match:
   *
   * we've successfully checked all un-flushed transactions and there
   * were no conflicts, and we have not found the key: now try to
   * lookup the key in the btree.
   */
  return (m_btree_index->find(context, cursor, key, pkey_arena, record,
                          precord_arena, flags));
}

ham_status_t
LocalDatabase::erase_txn(Context *context, ham_key_t *key, uint32_t flags,
                TransactionCursor *cursor)
{
  ham_status_t st = 0;
  TransactionOperation *op;
  bool node_created = false;
  LocalCursor *pc = 0;
  if (cursor)
    pc = cursor->get_parent();

  /* get (or create) the node for this key */
  TransactionNode *node = m_txn_index->get(key, 0);
  if (!node) {
    node = new TransactionNode(this, key);
    node_created = true;
    // TODO only store when the operation is successful?
    m_txn_index->store(node);
  }

  /* check for conflicts of this key - but only if we're not erasing a
   * duplicate key. dupes are checked for conflicts in _local_cursor_move TODO that function no longer exists */
  if (!pc || (!pc->get_dupecache_index())) {
    st = check_erase_conflicts(context, node, key, flags);
    if (st) {
      if (node_created) {
        m_txn_index->remove(node);
        delete node;
      }
      return (st);
    }
  }

  /* append a new operation to this node */
  op = node->append(context->txn, flags, TransactionOperation::kErase,
                  lenv()->next_lsn(), key, 0);

  /* is this function called through ham_cursor_erase? then add the
   * duplicate ID */
  if (cursor) {
    if (pc->get_dupecache_index())
      op->set_referenced_dupe(pc->get_dupecache_index());
  }

  /* the current op has no cursors attached; but if there are any
   * other ops in this node and in this transaction, then they have to
   * be set to nil. This only nil's txn-cursors! */
  nil_all_cursors_in_node(context->txn, pc, node);

  /* in addition we nil all btree cursors which are coupled to this key */
  nil_all_cursors_in_btree(context, pc, node->get_key());

  /* append journal entry */
  if (m_env->get_flags() & HAM_ENABLE_RECOVERY
      && m_env->get_flags() & HAM_ENABLE_TRANSACTIONS) {
    Journal *j = lenv()->journal();
    j->append_erase(this, context->txn, key, 0,
                    flags | HAM_ERASE_ALL_DUPLICATES, op->get_lsn());
  }

  ham_assert(st == 0);
  return (0);
}

ham_status_t
LocalDatabase::create(Context *context, PBtreeHeader *btree_header)
{
  /* set the flags; strip off run-time (per session) flags for the btree */
  uint32_t persistent_flags = get_flags();
  persistent_flags &= ~(HAM_CACHE_UNLIMITED
            | HAM_DISABLE_MMAP
            | HAM_ENABLE_FSYNC
            | HAM_READ_ONLY
            | HAM_ENABLE_RECOVERY
            | HAM_AUTO_RECOVERY
            | HAM_ENABLE_TRANSACTIONS);

  switch (m_config.key_type) {
    case HAM_TYPE_UINT8:
      m_config.key_size = 1;
      break;
    case HAM_TYPE_UINT16:
      m_config.key_size = 2;
      break;
    case HAM_TYPE_REAL32:
    case HAM_TYPE_UINT32:
      m_config.key_size = 4;
      break;
    case HAM_TYPE_REAL64:
    case HAM_TYPE_UINT64:
      m_config.key_size = 8;
      break;
  }

  // if we cannot fit at least 10 keys in a page then refuse to continue
  if (m_config.key_size != HAM_KEY_SIZE_UNLIMITED) {
    if (lenv()->config().page_size_bytes / (m_config.key_size + 8) < 10) {
      ham_trace(("key size too large; either increase page_size or decrease "
                "key size"));
      return (HAM_INV_KEY_SIZE);
    }
  }

  // fixed length records:
  //
  // if records are <= 8 bytes OR if we can fit at least 500 keys AND
  // records into the leaf then store the records in the leaf;
  // otherwise they're allocated as a blob
  if (m_config.record_size != HAM_RECORD_SIZE_UNLIMITED) {
    if (m_config.record_size <= 8
        || (m_config.record_size <= kInlineRecordThreshold
          && lenv()->config().page_size_bytes
                / (m_config.key_size + m_config.record_size) > 500)) {
      persistent_flags |= HAM_FORCE_RECORDS_INLINE;
      m_config.flags |= HAM_FORCE_RECORDS_INLINE;
    }
  }

  // create the btree
  m_btree_index.reset(new BtreeIndex(this, btree_header, persistent_flags,
                        m_config.key_type, m_config.key_size));

  /* initialize the btree */
  m_btree_index->create(context, m_config.key_type, m_config.key_size,
                  m_config.record_size);

  /* the header page is now dirty */
  Page *header = lenv()->page_manager()->fetch(context, 0);
  header->set_dirty(true);

  /* and the TransactionIndex */
  m_txn_index.reset(new TransactionIndex(this));

  return (0);
}

ham_status_t
LocalDatabase::open(Context *context, PBtreeHeader *btree_header)
{
  /*
   * set the database flags; strip off the persistent flags that may have been
   * set by the caller, before mixing in the persistent flags as obtained
   * from the btree.
   */
  uint32_t flags = get_flags();
  flags &= ~(HAM_CACHE_UNLIMITED
            | HAM_DISABLE_MMAP
            | HAM_ENABLE_FSYNC
            | HAM_READ_ONLY
            | HAM_ENABLE_RECOVERY
            | HAM_AUTO_RECOVERY
            | HAM_ENABLE_TRANSACTIONS);

  m_config.key_type = btree_header->key_type();
  m_config.key_size = btree_header->key_size();

  /* create the BtreeIndex */
  m_btree_index.reset(new BtreeIndex(this, btree_header,
                            flags | btree_header->flags(),
                            btree_header->key_type(),
                            btree_header->key_size()));

  ham_assert(!(m_btree_index->flags() & HAM_CACHE_UNLIMITED));
  ham_assert(!(m_btree_index->flags() & HAM_DISABLE_MMAP));
  ham_assert(!(m_btree_index->flags() & HAM_ENABLE_FSYNC));
  ham_assert(!(m_btree_index->flags() & HAM_READ_ONLY));
  ham_assert(!(m_btree_index->flags() & HAM_ENABLE_RECOVERY));
  ham_assert(!(m_btree_index->flags() & HAM_AUTO_RECOVERY));
  ham_assert(!(m_btree_index->flags() & HAM_ENABLE_TRANSACTIONS));

  /* initialize the btree */
  m_btree_index->open();

  /* create the TransactionIndex - TODO only if txn's are enabled? */
  m_txn_index.reset(new TransactionIndex(this));

  /* merge the non-persistent database flag with the persistent flags from
   * the btree index */
  m_config.flags = config().flags | m_btree_index->flags();
  m_config.key_size = m_btree_index->key_size();
  m_config.key_type = m_btree_index->key_type();
  m_config.record_size = m_btree_index->record_size();

  // fetch the current record number
  if ((get_flags() & (HAM_RECORD_NUMBER32 | HAM_RECORD_NUMBER64))) {
    ham_key_t key = {};
    LocalCursor *c = new LocalCursor(this, 0);
    ham_status_t st = cursor_move_impl(context, c, &key, 0, HAM_CURSOR_LAST);
    cursor_close(c);
    if (st)
      return (st == HAM_KEY_NOT_FOUND ? 0 : st);

    if (get_flags() & HAM_RECORD_NUMBER32)
      m_recno = *(uint32_t *)key.data;
    else
      m_recno = *(uint64_t *)key.data;
  }

  return (0);
}

struct MetricsVisitor : public BtreeVisitor {
  MetricsVisitor(ham_env_metrics_t *metrics)
    : m_metrics(metrics) {
  }

  // Specifies if the visitor modifies the node
  virtual bool is_read_only() const {
    return (true);
  }

  // called for each node
  virtual void operator()(Context *context, BtreeNodeProxy *node) {
    if (node->is_leaf())
      node->fill_metrics(&m_metrics->btree_leaf_metrics);
    else
      node->fill_metrics(&m_metrics->btree_internal_metrics);
  }

  ham_env_metrics_t *m_metrics;
};

void
LocalDatabase::fill_metrics(ham_env_metrics_t *metrics)
{
  metrics->btree_leaf_metrics.database_name = name();
  metrics->btree_internal_metrics.database_name = name();

  try {
    MetricsVisitor visitor(metrics);
    Context context(lenv(), 0, this);
    m_btree_index->visit_nodes(&context, visitor, true);

    // calculate the "avg" values
    BtreeStatistics::finalize_metrics(&metrics->btree_leaf_metrics);
    BtreeStatistics::finalize_metrics(&metrics->btree_internal_metrics);
  }
  catch (Exception &) {
  }
}

ham_status_t
LocalDatabase::get_parameters(ham_parameter_t *param)
{
  try {
    Context context(lenv(), 0, this);

    Page *page = 0;
    ham_parameter_t *p = param;

    if (p) {
      for (; p->name; p++) {
        switch (p->name) {
        case HAM_PARAM_KEY_SIZE:
          p->value = m_config.key_size;
          break;
        case HAM_PARAM_KEY_TYPE:
          p->value = m_config.key_type;
          break;
        case HAM_PARAM_RECORD_SIZE:
          p->value = m_config.record_size;
          break;
        case HAM_PARAM_FLAGS:
          p->value = (uint64_t)get_flags();
          break;
        case HAM_PARAM_DATABASE_NAME:
          p->value = (uint64_t)name();
          break;
        case HAM_PARAM_MAX_KEYS_PER_PAGE:
          p->value = 0;
          page = lenv()->page_manager()->fetch(&context,
                          m_btree_index->root_address(),
                        PageManager::kReadOnly);
          if (page) {
            BtreeNodeProxy *node = m_btree_index->get_node_from_page(page);
            p->value = node->estimate_capacity();
          }
          break;
        case HAM_PARAM_RECORD_COMPRESSION:
          p->value = 0;
          break;
        case HAM_PARAM_KEY_COMPRESSION:
          p->value = 0;
          break;
        default:
          ham_trace(("unknown parameter %d", (int)p->name));
          throw Exception(HAM_INV_PARAMETER);
        }
      }
    }
  }
  catch (Exception &ex) {
    return (ex.code);
  }
  return (0);
}

ham_status_t
LocalDatabase::check_integrity(uint32_t flags)
{
  try {
    Context context(lenv(), 0, this);

    /* purge cache if necessary */
    lenv()->page_manager()->purge_cache(&context);

    /* call the btree function */
    m_btree_index->check_integrity(&context, flags);

    /* call the txn function */
    //m_txn_index->check_integrity(flags);
  }
  catch (Exception &ex) {
    return (ex.code);
  }
  return (0);
}

ham_status_t
LocalDatabase::count(Transaction *htxn, bool distinct, uint64_t *pcount)
{
  LocalTransaction *txn = dynamic_cast<LocalTransaction *>(htxn);

  try {
    Context context(lenv(), txn, this);

    /* purge cache if necessary */
    lenv()->page_manager()->purge_cache(&context);

    /*
     * call the btree function - this will retrieve the number of keys
     * in the btree
     */
    uint64_t keycount = m_btree_index->count(&context, distinct);

    /*
     * if transactions are enabled, then also sum up the number of keys
     * from the transaction tree
     */
    if (get_flags() & HAM_ENABLE_TRANSACTIONS)
      keycount += m_txn_index->count(&context, txn, distinct);

    *pcount = keycount;
    return (0);
  }
  catch (Exception &ex) {
    return (ex.code);
  }
}

ham_status_t
LocalDatabase::scan(Transaction *txn, ScanVisitor *visitor, bool distinct)
{
  ham_status_t st = 0;
  LocalCursor *cursor = 0;

  try {
    Context context(lenv(), (LocalTransaction *)txn, this);

    Page *page;
    ham_key_t key = {0};

    /* purge cache if necessary */
    lenv()->page_manager()->purge_cache(&context);

    /* create a cursor, move it to the first key */
    LocalCursor *cursor = (LocalCursor *)cursor_create_impl(txn);

    st = cursor_move_impl(&context, cursor, &key, 0, HAM_CURSOR_FIRST);
    if (st)
      goto bail;

    /* only transaction keys? then use a regular cursor */
    if (!cursor->is_coupled_to_btree()) {
      do {
        /* process the key */
        (*visitor)(key.data, key.size, distinct
                                        ? cursor->get_duplicate_count(&context)
                                        : 1);
      } while ((st = cursor_move_impl(&context, cursor, &key,
                            0, HAM_CURSOR_NEXT)) == 0);
      goto bail;
    }

    /* only btree keys? then traverse page by page */
    if (!(get_flags() & HAM_ENABLE_TRANSACTIONS)) {
      ham_assert(cursor->is_coupled_to_btree());

      do {
        // get the coupled page
        cursor->get_btree_cursor()->get_coupled_key(&page);
        BtreeNodeProxy *node = m_btree_index->get_node_from_page(page);
        // and let the btree node perform the remaining work
        node->scan(&context, visitor, 0, distinct);
      } while (cursor->get_btree_cursor()->move_to_next_page(&context) == 0);

      goto bail;
    }

    /* mixed txn/btree load? if there are btree nodes which are NOT modified
     * in transactions then move the scan to the btree node. Otherwise use
     * a regular cursor */
    while (true) {
      if (!cursor->is_coupled_to_btree())
        break;

      int slot;
      cursor->get_btree_cursor()->get_coupled_key(&page, &slot);
      BtreeNodeProxy *node = m_btree_index->get_node_from_page(page);

      /* are transactions present? then check if the next txn key is >= btree[0]
       * and <= btree[n] */
      ham_key_t *txnkey = 0;
      if (cursor->get_txn_cursor()->get_coupled_op())
        txnkey = cursor->get_txn_cursor()->get_coupled_op()->get_node()->get_key();
      // no (more) transactional keys left - process the current key, then
      // scan the remaining keys directly in the btree
      if (!txnkey) {
        /* process the key */
        (*visitor)(key.data, key.size, distinct
                                        ? cursor->get_duplicate_count(&context)
                                        : 1);
        break;
      }

      /* if yes: use the cursor to traverse the page */
      if (node->compare(&context, txnkey, 0) >= 0
          && node->compare(&context, txnkey, node->get_count() - 1) <= 0) {
        do {
          Page *new_page = 0;
          if (cursor->is_coupled_to_btree())
            cursor->get_btree_cursor()->get_coupled_key(&new_page);
          /* break the loop if we've reached the next page */
          if (new_page && new_page != page) {
            page = new_page;
            break;
          }
          /* process the key */
          (*visitor)(key.data, key.size, distinct
                                        ? cursor->get_duplicate_count(&context)
                                        : 1);
        } while ((st = cursor_move_impl(&context, cursor, &key,
                                0, HAM_CURSOR_NEXT)) == 0);

        if (st != HAM_SUCCESS)
          goto bail;
      }
      else {
        /* Otherwise traverse directly in the btree page. This is the fastest
         * code path. */
        node->scan(&context, visitor, slot, distinct);
        /* and then move to the next page */
        if (cursor->get_btree_cursor()->move_to_next_page(&context) != 0)
          break;
      }
    }

    /* pick up the remaining transactional keys */
    while ((st = cursor_move_impl(&context, cursor, &key,
                            0, HAM_CURSOR_NEXT)) == 0) {
      (*visitor)(key.data, key.size, distinct
                                     ? cursor->get_duplicate_count(&context)
                                     : 1);
    }

bail:
    if (cursor) {
      cursor->close();
      delete cursor;
    }
    return (st == HAM_KEY_NOT_FOUND ? 0 : st);
  }
  catch (Exception &ex) {
    if (cursor) {
      cursor->close();
      delete cursor;
    }
    return (ex.code);
  }
}

ham_status_t
LocalDatabase::insert(Cursor *hcursor, Transaction *txn, ham_key_t *key,
            ham_record_t *record, uint32_t flags)
{
  LocalCursor *cursor = (LocalCursor *)hcursor;
  Context context(lenv(), (LocalTransaction *)txn, this);

  try {
    if (m_config.flags & (HAM_RECORD_NUMBER32 | HAM_RECORD_NUMBER64)) {
      if (key->size == 0 && key->data == 0) {
        // ok!
      }
      else if (key->size == 0 && key->data != 0) {
        ham_trace(("for record number keys set key size to 0, "
                               "key->data to null"));
        return (HAM_INV_PARAMETER);
      }
      else if (key->size != m_config.key_size) {
        ham_trace(("invalid key size (%u instead of %u)",
              key->size, m_config.key_size));
        return (HAM_INV_KEY_SIZE);
      }
    }
    else if (m_config.key_size != HAM_KEY_SIZE_UNLIMITED
        && key->size != m_config.key_size) {
      ham_trace(("invalid key size (%u instead of %u)",
            key->size, m_config.key_size));
      return (HAM_INV_KEY_SIZE);
    }
    if (m_config.record_size != HAM_RECORD_SIZE_UNLIMITED
        && record->size != m_config.record_size) {
      ham_trace(("invalid record size (%u instead of %u)",
            record->size, m_config.record_size));
      return (HAM_INV_RECORD_SIZE);
    }

    ByteArray *arena = &key_arena(txn);

    /*
     * record number: make sure that we have a valid key structure,
     * and lazy load the last used record number
     *
     * TODO TODO
     * too much duplicated code
     */
    uint64_t recno = 0;
    if (get_flags() & HAM_RECORD_NUMBER64) {
      if (flags & HAM_OVERWRITE) {
        ham_assert(key->size == sizeof(uint64_t));
        ham_assert(key->data != 0);
        recno = *(uint64_t *)key->data;
      }
      else {
        /* get the record number and increment it */
        recno = next_record_number();
      }

      /* allocate memory for the key */
      if (!key->data) {
        arena->resize(sizeof(uint64_t));
        key->data = arena->get_ptr();
      }
      key->size = sizeof(uint64_t);
      *(uint64_t *)key->data = recno;

      /* A recno key is always appended sequentially */
      flags |= HAM_HINT_APPEND;
    }
    else if (get_flags() & HAM_RECORD_NUMBER32) {
      if (flags & HAM_OVERWRITE) {
        ham_assert(key->size == sizeof(uint32_t));
        ham_assert(key->data != 0);
        recno = *(uint32_t *)key->data;
      }
      else {
        /* get the record number and increment it */
        recno = next_record_number();
      }

      /* allocate memory for the key */
      if (!key->data) {
        arena->resize(sizeof(uint32_t));
        key->data = arena->get_ptr();
      }
      key->size = sizeof(uint32_t);
      *(uint32_t *)key->data = (uint32_t)recno;

      /* A recno key is always appended sequentially */
      flags |= HAM_HINT_APPEND;
    }

    ham_status_t st = 0;
    LocalTransaction *local_txn = 0;

    /* purge cache if necessary */
    if (!txn && (get_flags() & HAM_ENABLE_TRANSACTIONS)) {
      local_txn = begin_temp_txn();
      context.txn = local_txn;
    }

    st = insert_impl(&context, cursor, key, record, flags);
    return (finalize(&context, st, local_txn));
  }
  catch (Exception &ex) {
    return (ex.code);
  }
}

ham_status_t
LocalDatabase::erase(Cursor *hcursor, Transaction *txn, ham_key_t *key,
                uint32_t flags)
{
  LocalCursor *cursor = (LocalCursor *)hcursor;
  Context context(lenv(), (LocalTransaction *)txn, this);

  try {
    ham_status_t st = 0;
    LocalTransaction *local_txn = 0;

    if (cursor) {
      if (cursor->is_nil())
        throw Exception(HAM_CURSOR_IS_NIL);
      if (cursor->is_coupled_to_txnop()) // TODO rewrite the next line, it's ugly
        key = cursor->get_txn_cursor()->get_coupled_op()->get_node()->get_key();
      else // cursor->is_coupled_to_btree()
        key = 0;
    }

    if (key) {
      if (m_config.key_size != HAM_KEY_SIZE_UNLIMITED
          && key->size != m_config.key_size) {
        ham_trace(("invalid key size (%u instead of %u)",
              key->size, m_config.key_size));
        return (HAM_INV_KEY_SIZE);
      }
    }

    if (!txn && (get_flags() & HAM_ENABLE_TRANSACTIONS)) {
      local_txn = begin_temp_txn();
      context.txn = local_txn;
    }

    st = erase_impl(&context, cursor, key, flags);
    return (finalize(&context, st, local_txn));
  }
  catch (Exception &ex) {
    return (ex.code);
  }
}

ham_status_t
LocalDatabase::find(Cursor *hcursor, Transaction *txn, ham_key_t *key,
            ham_record_t *record, uint32_t flags)
{
  LocalCursor *cursor = (LocalCursor *)hcursor;
  Context context(lenv(), (LocalTransaction *)txn, this);

  try {
    ham_status_t st = 0;

    /* Duplicates AND Transactions require a Cursor because only
     * Cursors can build lists of duplicates.
     * TODO not exception safe - if find() throws then the cursor is not closed
     */
    if (!cursor
          && (get_flags() & (HAM_ENABLE_DUPLICATE_KEYS|HAM_ENABLE_TRANSACTIONS))) {
      LocalCursor *c = (LocalCursor *)cursor_create_impl(txn);
      st = find(c, txn, key, record, flags);
      c->close();
      delete c;
      return (st);
    }

    if (m_config.key_size != HAM_KEY_SIZE_UNLIMITED
        && key->size != m_config.key_size) {
      ham_trace(("invalid key size (%u instead of %u)",
            key->size, m_config.key_size));
      return (HAM_INV_KEY_SIZE);
    }

    // cursor: reset the dupecache, set to nil
    if (cursor)
      cursor->set_to_nil(LocalCursor::kBoth);

    st = find_impl(&context, cursor, key, record, flags);
    if (st)
      return (finalize(&context, st, 0));

    if (cursor) {
      // make sure that txn-cursor and btree-cursor point to the same keys
      if (get_flags() & HAM_ENABLE_TRANSACTIONS) {
        bool is_equal;
        (void)cursor->sync(&context, LocalCursor::kSyncOnlyEqualKeys, &is_equal);
        if (!is_equal && cursor->is_coupled_to_txnop())
          cursor->set_to_nil(LocalCursor::kBtree);
      }

      /* if the key has duplicates: build a duplicate table, then couple to the
       * first/oldest duplicate */
      if (cursor->get_dupecache_count(&context, true)) {
        cursor->couple_to_dupe(1); // 1-based index!
        if (record) { // TODO don't copy record if it was already
                      // copied in find_impl
          if (cursor->is_coupled_to_txnop())
            cursor->get_txn_cursor()->copy_coupled_record(record);
          else {
            Transaction *txn = cursor->get_txn();
            st = cursor->get_btree_cursor()->move(&context, 0, 0, record,
                          &record_arena(txn), 0);
          }
        }
      }

      /* set a flag that the cursor just completed an Insert-or-find
       * operation; this information is needed in ham_cursor_move */
      cursor->set_last_operation(LocalCursor::kLookupOrInsert);
    }

    return (finalize(&context, st, 0));
  }
  catch (Exception &ex) {
    return (ex.code);
  }
}

Cursor *
LocalDatabase::cursor_create_impl(Transaction *txn)
{
  return (new LocalCursor(this, txn));
}

Cursor *
LocalDatabase::cursor_clone_impl(Cursor *hsrc)
{
  return (new LocalCursor(*(LocalCursor *)hsrc));
}

ham_status_t
LocalDatabase::cursor_move(Cursor *hcursor, ham_key_t *key,
                ham_record_t *record, uint32_t flags)
{
  LocalCursor *cursor = (LocalCursor *)hcursor;

  try {
    Context context(lenv(), (LocalTransaction *)cursor->get_txn(),
            this);

    return (cursor_move_impl(&context, cursor, key, record, flags));
  }
  catch (Exception &ex) {
    return (ex.code);
  }
}

ham_status_t
LocalDatabase::cursor_move_impl(Context *context, LocalCursor *cursor,
                ham_key_t *key, ham_record_t *record, uint32_t flags)
{
  /* purge cache if necessary */
  lenv()->page_manager()->purge_cache(context);

  /*
   * if the cursor was never used before and the user requests a NEXT then
   * move the cursor to FIRST; if the user requests a PREVIOUS we set it
   * to LAST, resp.
   *
   * if the cursor was already used but is nil then we've reached EOF,
   * and a NEXT actually tries to move to the LAST key (and PREVIOUS
   * moves to FIRST)
   */
  if (cursor->is_nil(0)) {
    if (flags & HAM_CURSOR_NEXT) {
      flags &= ~HAM_CURSOR_NEXT;
      if (cursor->is_first_use())
        flags |= HAM_CURSOR_FIRST;
      else
        flags |= HAM_CURSOR_LAST;
    }
    else if (flags & HAM_CURSOR_PREVIOUS) {
      flags &= ~HAM_CURSOR_PREVIOUS;
      if (cursor->is_first_use())
        flags |= HAM_CURSOR_LAST;
      else
        flags |= HAM_CURSOR_FIRST;
    }
  }

  ham_status_t st = 0;

  /* everything else is handled by the cursor function */
  st = cursor->move(context, key, record, flags);

  /* store the direction */
  if (flags & HAM_CURSOR_NEXT)
    cursor->set_last_operation(HAM_CURSOR_NEXT);
  else if (flags & HAM_CURSOR_PREVIOUS)
    cursor->set_last_operation(HAM_CURSOR_PREVIOUS);
  else
    cursor->set_last_operation(0);

  if (st) {
    if (st == HAM_KEY_ERASED_IN_TXN)
      st = HAM_KEY_NOT_FOUND;
    /* trigger a sync when the function is called again */
    cursor->set_last_operation(0);
    return (st);
  }

  return (0);
}

ham_status_t
LocalDatabase::close_impl(uint32_t flags)
{
  Context context(lenv(), 0, this);

  /* check if this database is modified by an active transaction */
  if (m_txn_index) {
    TransactionNode *node = m_txn_index->get_first();
    while (node) {
      TransactionOperation *op = node->get_newest_op();
      while (op) {
        Transaction *optxn = op->get_txn();
        if (!optxn->is_committed() && !optxn->is_aborted()) {
          ham_trace(("cannot close a Database that is modified by "
                 "a currently active Transaction"));
          return (set_error(HAM_TXN_STILL_OPEN));
        }
        op = op->get_previous_in_node();
      }
      node = node->get_next_sibling();
    }
  }

  /* in-memory-database: free all allocated blobs */
  if (m_btree_index && m_env->get_flags() & HAM_IN_MEMORY)
   m_btree_index->drop(&context);

  /*
   * flush all pages of this database (but not the header page,
   * it's still required and will be flushed below)
   */
  lenv()->page_manager()->close_database(&context, this);

  return (0);
}

void
LocalDatabase::increment_dupe_index(Context *context, TransactionNode *node,
                LocalCursor *skip, uint32_t start)
{
  LocalCursor *c = (LocalCursor *)m_cursor_list;

  while (c) {
    bool hit = false;

    if (c == skip || c->is_nil(0))
      goto next;

    /* if cursor is coupled to an op in the same node: increment
     * duplicate index (if required) */
    if (c->is_coupled_to_txnop()) {
      TransactionCursor *txnc = c->get_txn_cursor();
      TransactionNode *n = txnc->get_coupled_op()->get_node();
      if (n == node)
        hit = true;
    }
    /* if cursor is coupled to the same key in the btree: increment
     * duplicate index (if required) */
    else if (c->get_btree_cursor()->points_to(context, node->get_key())) {
      hit = true;
    }

    if (hit) {
      if (c->get_dupecache_index() > start)
        c->set_dupecache_index(c->get_dupecache_index() + 1);
    }

next:
    c = (LocalCursor *)c->get_next();
  }
}

void
LocalDatabase::nil_all_cursors_in_node(LocalTransaction *txn,
                LocalCursor *current, TransactionNode *node)
{
  TransactionOperation *op = node->get_newest_op();
  while (op) {
    TransactionCursor *cursor = op->cursor_list();
    while (cursor) {
      LocalCursor *parent = cursor->get_parent();
      // is the current cursor to a duplicate? then adjust the
      // coupled duplicate index of all cursors which point to a duplicate
      if (current) {
        if (current->get_dupecache_index()) {
          if (current->get_dupecache_index() < parent->get_dupecache_index()) {
            parent->set_dupecache_index(parent->get_dupecache_index() - 1);
            cursor = cursor->get_coupled_next();
            continue;
          }
          else if (current->get_dupecache_index() > parent->get_dupecache_index()) {
            cursor = cursor->get_coupled_next();
            continue;
          }
          // else fall through
        }
      }
      parent->couple_to_btree(); // TODO merge these two lines
      parent->set_to_nil(LocalCursor::kTxn);
      // set a flag that the cursor just completed an Insert-or-find
      // operation; this information is needed in ham_cursor_move
      // (in this aspect, an erase is the same as insert/find)
      parent->set_last_operation(LocalCursor::kLookupOrInsert);

      cursor = op->cursor_list();
    }

    op = op->get_previous_in_node();
  }
}

ham_status_t
LocalDatabase::copy_record(LocalDatabase *db, Transaction *txn,
                TransactionOperation *op, ham_record_t *record)
{
  ByteArray *arena = &db->record_arena(txn);

  if (!(record->flags & HAM_RECORD_USER_ALLOC)) {
    arena->resize(op->get_record()->size);
    record->data = arena->get_ptr();
  }
  memcpy(record->data, op->get_record()->data, op->get_record()->size);
  record->size = op->get_record()->size;
  return (0);
}

void
LocalDatabase::nil_all_cursors_in_btree(Context *context, LocalCursor *current,
                ham_key_t *key)
{
  LocalCursor *c = (LocalCursor *)m_cursor_list;

  /* foreach cursor in this database:
   *  if it's nil or coupled to the txn: skip it
   *  if it's coupled to btree AND uncoupled: compare keys; set to nil
   *    if keys are identical
   *  if it's uncoupled to btree AND coupled: compare keys; set to nil
   *    if keys are identical; (TODO - improve performance by nil'ling
   *    all other cursors from the same btree page)
   *
   *  do NOT nil the current cursor - it's coupled to the key, and the
   *  coupled key is still needed by the caller
   */
  while (c) {
    if (c->is_nil(0) || c == current)
      goto next;
    if (c->is_coupled_to_txnop())
      goto next;

    if (c->get_btree_cursor()->points_to(context, key)) {
      /* is the current cursor to a duplicate? then adjust the
       * coupled duplicate index of all cursors which point to a
       * duplicate */
      if (current) {
        if (current->get_dupecache_index()) {
          if (current->get_dupecache_index() < c->get_dupecache_index()) {
            c->set_dupecache_index(c->get_dupecache_index() - 1);
            goto next;
          }
          else if (current->get_dupecache_index() > c->get_dupecache_index()) {
            goto next;
          }
          /* else fall through */
        }
      }
      c->set_to_nil(0);
    }
next:
    c = (LocalCursor *)c->get_next();
  }
}

ham_status_t
LocalDatabase::flush_txn_operation(Context *context, LocalTransaction *txn,
                TransactionOperation *op)
{
  ham_status_t st = 0;
  TransactionNode *node = op->get_node();

  /*
   * depending on the type of the operation: actually perform the
   * operation on the btree
   *
   * if the txn-op has a cursor attached, then all (txn)cursors
   * which are coupled to this op have to be uncoupled, and their
   * parent (btree) cursor must be coupled to the btree item instead.
   */
  if ((op->get_flags() & TransactionOperation::kInsert)
      || (op->get_flags() & TransactionOperation::kInsertOverwrite)
      || (op->get_flags() & TransactionOperation::kInsertDuplicate)) {
    uint32_t additional_flag =
      (op->get_flags() & TransactionOperation::kInsertDuplicate)
          ? HAM_DUPLICATE
          : HAM_OVERWRITE;
    if (!op->cursor_list()) {
      st = m_btree_index->insert(context, 0, node->get_key(), op->get_record(),
                  op->get_orig_flags() | additional_flag);
    }
    else {
      TransactionCursor *tc1 = op->cursor_list();
      LocalCursor *c1 = tc1->get_parent();
      /* pick the first cursor, get the parent/btree cursor and
       * insert the key/record pair in the btree. The btree cursor
       * then will be coupled to this item. */
      st = m_btree_index->insert(context, c1, node->get_key(), op->get_record(),
                  op->get_orig_flags() | additional_flag);
      if (!st) {
        /* uncouple the cursor from the txn-op, and remove it */
        c1->couple_to_btree(); // TODO merge these two calls
        c1->set_to_nil(LocalCursor::kTxn);

        /* all other (btree) cursors need to be coupled to the same
         * item as the first one. */
        TransactionCursor *tc2;
        while ((tc2 = op->cursor_list())) {
          LocalCursor *c2 = tc2->get_parent();
          c2->get_btree_cursor()->clone(c1->get_btree_cursor());
          c2->couple_to_btree(); // TODO merge these two calls
          c2->set_to_nil(LocalCursor::kTxn);
        }
      }
    }
  }
  else if (op->get_flags() & TransactionOperation::kErase) {
    st = m_btree_index->erase(context, 0, node->get_key(),
                  op->get_referenced_dupe(), op->get_flags());
    if (st == HAM_KEY_NOT_FOUND)
      st = 0;
  }

  return (st);
}

ham_status_t
LocalDatabase::drop(Context *context)
{
  m_btree_index->drop(context);
  return (0);
}

ham_status_t
LocalDatabase::insert_impl(Context *context, LocalCursor *cursor,
                ham_key_t *key, ham_record_t *record, uint32_t flags)
{
  ham_status_t st = 0;

  lenv()->page_manager()->purge_cache(context);

  /*
   * if transactions are enabled: only insert the key/record pair into
   * the Transaction structure. Otherwise immediately write to the btree.
   */
  if (context->txn || m_env->get_flags() & HAM_ENABLE_TRANSACTIONS)
    st = insert_txn(context, key, record, flags, cursor
                                                ? cursor->get_txn_cursor()
                                                : 0);
  else
    st = m_btree_index->insert(context, cursor, key, record, flags);

  // couple the cursor to the inserted key
  if (st == 0 && cursor) {
    if (m_env->get_flags() & HAM_ENABLE_TRANSACTIONS) {
      DupeCache *dc = cursor->get_dupecache();
      // TODO required? should have happened in insert_txn
      cursor->couple_to_txnop();
      /* the cursor is coupled to the txn-op; nil the btree-cursor to
       * trigger a sync() call when fetching the duplicates */
      // TODO merge with the line above
      cursor->set_to_nil(LocalCursor::kBtree);

      /* if duplicate keys are enabled: set the duplicate index of
       * the new key  */
      if (st == 0 && cursor->get_dupecache_count(context, true)) {
        TransactionOperation *op = cursor->get_txn_cursor()->get_coupled_op();
        ham_assert(op != 0);

        for (uint32_t i = 0; i < dc->get_count(); i++) {
          DupeCacheLine *l = dc->get_element(i);
          if (!l->use_btree() && l->get_txn_op() == op) {
            cursor->set_dupecache_index(i + 1);
            break;
          }
        }
      }
    }
    else {
      // TODO required? should have happened in BtreeInsertAction
      cursor->couple_to_btree();
    }

    /* set a flag that the cursor just completed an Insert-or-find
     * operation; this information is needed in ham_cursor_move */
    cursor->set_last_operation(LocalCursor::kLookupOrInsert);
  }

  return (st);
}

ham_status_t
LocalDatabase::find_impl(Context *context, LocalCursor *cursor,
                ham_key_t *key, ham_record_t *record, uint32_t flags)
{
  /* purge cache if necessary */
  lenv()->page_manager()->purge_cache(context);

  /*
   * if transactions are enabled: read keys from transaction trees,
   * otherwise read immediately from disk
   */
  if (context->txn || m_env->get_flags() & HAM_ENABLE_TRANSACTIONS)
    return (find_txn(context, cursor, key, record, flags));

  return (m_btree_index->find(context, cursor, key, &key_arena(context->txn),
                          record, &record_arena(context->txn), flags));
}

ham_status_t
LocalDatabase::erase_impl(Context *context, LocalCursor *cursor, ham_key_t *key,
                uint32_t flags)
{
  ham_status_t st = 0;

  /*
   * if transactions are enabled: append a 'erase key' operation into
   * the txn tree; otherwise immediately erase the key from disk
   */
  if (context->txn || m_env->get_flags() & HAM_ENABLE_TRANSACTIONS) {
    if (cursor) {
      /*
       * !!
       * we have two cases:
       *
       * 1. the cursor is coupled to a btree item (or uncoupled, but not nil)
       *    and the txn_cursor is nil; in that case, we have to
       *    - uncouple the btree cursor
       *    - insert the erase-op for the key which is used by the btree cursor
       *
       * 2. the cursor is coupled to a txn-op; in this case, we have to
       *    - insert the erase-op for the key which is used by the txn-op
       *
       * TODO clean up this whole mess. code should be like
       *
       *   if (txn)
       *     erase_txn(txn, cursor->get_key(), 0, cursor->get_txn_cursor());
       */
      /* case 1 described above */
      if (cursor->is_coupled_to_btree()) {
        cursor->set_to_nil(LocalCursor::kTxn);
        cursor->get_btree_cursor()->uncouple_from_page(context);
        st = erase_txn(context, cursor->get_btree_cursor()->get_uncoupled_key(),
                        0, cursor->get_txn_cursor());
      }
      /* case 2 described above */
      else {
        // TODO this line is ugly
        st = erase_txn(context,
                        cursor->get_txn_cursor()->get_coupled_op()->get_key(),
                        0, cursor->get_txn_cursor());
      }
    }
    else {
      st = erase_txn(context, key, flags, 0);
    }
  }
  else {
    st = m_btree_index->erase(context, cursor, key, 0, flags);
  }

  /* on success: 'nil' the cursor */
  if (cursor && st == 0) {
    cursor->set_to_nil(0);
    ham_assert(cursor->get_txn_cursor()->is_nil());
    ham_assert(cursor->is_nil(0));
  }

  return (st);
}

ham_status_t
LocalDatabase::finalize(Context *context, ham_status_t status,
                Transaction *local_txn)
{
  LocalEnvironment *env = lenv();

  if (status) {
    if (local_txn) {
      context->changeset.clear();
      env->txn_manager()->abort(local_txn);
    }
    return (status);
  }

  if (local_txn) {
    context->changeset.clear();
    env->txn_manager()->commit(local_txn);
  }
  else if (env->get_flags() & HAM_ENABLE_RECOVERY
      && !(env->get_flags() & HAM_ENABLE_TRANSACTIONS)) {
    context->changeset.flush(env->next_lsn());
  }
  return (0);
}

LocalTransaction *
LocalDatabase::begin_temp_txn()
{
  LocalTransaction *txn;
  ham_status_t st = lenv()->txn_begin((Transaction **)&txn, 0,
                        HAM_TXN_TEMPORARY | HAM_DONT_LOCK);
  if (st)
    throw Exception(st);
  return (txn);
}

} // namespace hamsterdb

max_stars_repo_name: mathieurodic/hamsterdb

max_issues_repo_name: mathieurodic/hamsterdb

max_forks_repo_name: mathieurodic/hamsterdb


text:
/*
 * Copyright 2009-2021 The VOTCA Development Team (http://www.votca.org)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Standard includes
#include <iostream>

// Third party includes
#include <boost/program_options.hpp>

// VOTCA includes
#include <votca/tools/property.h>
#include <votca/tools/tokenizer.h>

// Local VOTCA includes
#include "votca/csg/version.h"

using namespace std;
using namespace votca::csg;
using namespace votca::tools;

void help_text() {
  votca::csg::HelpTextHeader("csg_property");
  cout << "Helper program called by inverse scripts to parse xml file.\n\n";
}

int main(int argc, char **argv) {
  string filter, file, path, print;
  bool short_output = false;
  bool with_path = false;

  // lets read in some program options
  namespace po = boost::program_options;

  // Declare the supported options.
  po::options_description desc("Allowed options");
  desc.add_options()("help", "produce this help message")(
      "path", po::value<string>(&path)->default_value(""),
      "path to part of the xml file to print")(
      "filter", po::value<string>(&filter)->default_value(""),
      "list option values that match given criteria")(
      "print", po::value<string>(&print)->default_value("."),
      "specifies which children or root to print")(
      "file", po::value<string>(&file), "xml file to parse")(
      "short", "short version of output")("with-path",
                                          "include path of node in output");

  // now read in the command line
  po::variables_map vm;
  try {
    po::store(po::parse_command_line(argc, argv, desc), vm);
    po::notify(vm);
  } catch (po::error &err) {
    cout << "error parsing command line: " << err.what() << endl;
    return -1;
  }
  // does the user want help?
  if (vm.count("help")) {
    help_text();
    cout << desc << endl;
    return 0;
  }
  // file specified
  if (!vm.count("file")) {
    cout << "please specify file\n";
    cout << desc << endl;
    return -1;
  }
  if (vm.count("short")) {
    short_output = true;
  }
  if (vm.count("with-path")) {
    with_path = true;
  }

  try {
    Property p;
    p.LoadFromXML(file);

    for (Property *prop : p.Select(path)) {
      if (filter != "") {
        Tokenizer tokenizer(filter, "=");
        Tokenizer::iterator tok;
        tok = tokenizer.begin();
        if (tok == tokenizer.end()) {
          throw std::invalid_argument("error, specified invalid filter");
        }

        string field = *tok;
        ++tok;
        if (tok == tokenizer.end()) {
          throw std::invalid_argument("error, specified invalid filter");
        }

        string value = *tok;
        if (!wildcmp(value, prop->get(field).value())) {
          continue;
        }
      }

      std::vector<Property *> printvalues;

      // Select returns empty on "." thus we have to handle that case explicitly
      if (print == ".") {
        printvalues.push_back(&(prop->get(print)));
      } else {
        printvalues = prop->Select(print);
      }

      for (Property *p2 : printvalues) {

        if (!short_output && with_path) {
          cout << p2->path() << ".";
        }
        if (!short_output) {
          cout << p2->name() << " = ";
        }
        cout << p2->value();
        cout << endl;
      }
    }
  } catch (std::exception &error) {
    cerr << "Warning from parsing xml file '" << file << "':\n"
         << error.what() << endl;
    return -1;
  }
  return 0;
}

ext: cc

max_stars_repo_name: BuildJet/csg

max_issues_repo_name: BuildJet/csg

max_forks_repo_name: BuildJet/csg


text:
#ifndef MODEL_TRACKER_HPP
#define MODEL_TRACKER_HPP

#include <set>
#include <boost/lexical_cast.hpp>

#include "../exrpc/exrpc.hpp"
#include "frovedis/ml/glm/linear_model.hpp"
#include "frovedis/ml/recommendation/matrix_factorization_model.hpp"
#include "frovedis/ml/clustering/kmeans.hpp"
#include "frovedis/ml/tree/tree_model.hpp"
#include "frovedis/ml/fm/model.hpp"
#include "frovedis/ml/nb/nb_model.hpp"

namespace frovedis {

enum { NONE = 0xDEAD };

enum MODEL_KIND {
  GLM = 0,
  LRM,
  SVM,
  LNRM,
  MFM,
  KMEANS,
  DTM,
  NBM,
  FMM,
  FPM,
  FPR,
  SPARSE_CONV_INFO
};

enum MAT_KIND {
  RMJR = 1,
  CMJR = 2,
  BCLC = 3,
  SCRS = 4,
  SCCS = 5,
  SELL = 6,
  SHYBRID = 7,
  RMJR_L = 101,
  CMJR_L = 102,
  BCLC_L = 103,
  SCRS_L = 104,
  SCCS_L = 105,
  SELL_L = 106,
  SHYBRID_L = 107
};

enum DTYPE {
  INT = 1,
  LONG = 2,
  FLOAT = 3,
  DOUBLE = 4,
  STRING = 5,
  BOOL = 6
};

enum OPTYPE {
  EQ = 1,
  NE = 2,
  GT = 3,
  GE = 4,
  LT = 5,
  LE = 6,
  AND = 11,
  OR = 12
};

// [MODEL_ID] => [MODEL_KIND, MODEL_PTR]
extern std::map<int,std::pair<MODEL_KIND,exrpc_ptr_t>> model_table;
extern std::set<int> deleted_model_tracker;
extern std::set<int> under_training_model_tracker;

void register_model(int mid, MODEL_KIND m, exrpc_ptr_t mptr);
void register_for_train(int mid);
void unregister_from_train(int mid);
bool is_deleted(int mid);
bool is_under_training(int mid);

void finalize_model_table();
void finalize_trackers();
void cleanup_frovedis_server();

// retuns the model head for the requested registered model id
template <class M>
M* get_model_ptr(int mid) {
  if(model_table.find(mid) == model_table.end()) { // if not registered
    if(!is_under_training(mid)) { // if not under training
      std::string message = "request for either non-registered or deleted model: [";
      message += boost::lexical_cast<std::string>(mid) + " ]\n";
      REPORT_ERROR(USER_ERROR, message);
    }
    else while(is_under_training(mid)); // waits until training is completed
  }
  auto p = frovedis::model_table[mid];
  return reinterpret_cast<M*>(p.second);
}

}

#endif

ext: hpp

max_stars_repo_name: wmeddie/frovedis

max_issues_repo_name: wmeddie/frovedis

max_forks_repo_name: wmeddie/frovedis


text:
/**
   @author Shizuko Hattori
   @author Shin'ichiro Nakaoka
*/

#include "RTCItem.h"
#include "OpenRTMUtil.h"
#include <cnoid/BodyItem>
#include <cnoid/Link>
#include <cnoid/BasicSensors>
#include <cnoid/ItemManager>
#include <cnoid/Archive>
#include <cnoid/FileUtil>
#include <cnoid/ExecutablePath>
#include <cnoid/MessageView>
#include <cnoid/Sleep>
#include <cnoid/ProjectManager>
#include <rtm/RTObject.h>
#include <rtm/CorbaNaming.h>
#include <boost/format.hpp>
#include <boost/regex.hpp>
#include "gettext.h"

using namespace std;
using namespace cnoid;
using boost::format;
namespace filesystem = boost::filesystem;

namespace {
const bool TRACE_FUNCTIONS = false;
}

void RTCItem::initialize(ExtensionManager* ext)
{
    static bool initialized = false;
    if(!initialized){
        ext->itemManager().registerClass<RTCItem>(N_("RTCItem"));
        ext->itemManager().addCreationPanel<RTCItem>();
        initialized = true;
    }
}

RTCItem::RTCItem()
    : os(MessageView::instance()->cout()),
      periodicType(N_PERIODIC_TYPE, CNOID_GETTEXT_DOMAIN_NAME),
      baseDirectoryType(N_BASE_DIRECTORY_TYPES, CNOID_GETTEXT_DOMAIN_NAME)
{
    rtcomp = 0;
    moduleName.clear();
    mv = MessageView::instance();

    periodicRate = 1000;
    periodicType.setSymbol(PERIODIC_EXECUTION_CONTEXT,  N_("PeriodicExecutionContext"));
    periodicType.setSymbol(SYNCH_EXT_TRIGGER,  N_("SynchExtTriggerEC"));
    periodicType.setSymbol(EXT_TRIG_EXECUTION_CONTEXT,  N_("ExtTrigExecutionContext"));
    periodicType.setSymbol(CHOREONOID_EXECUTION_CONTEXT,  N_("ChoreonoidExecutionContext"));
    periodicType.select(PERIODIC_EXECUTION_CONTEXT);
    oldPeriodicType = periodicType.which();

    properties.clear();
    properties.insert(make_pair(string("exec_cxt.periodic.type"), periodicType.selectedSymbol()));
    stringstream ss;
    ss << periodicRate;
    properties.insert(make_pair(string("exec_cxt.periodic.rate"), ss.str()));

    baseDirectoryType.setSymbol(RTC_DIRECTORY, N_("RTC directory"));
    baseDirectoryType.setSymbol(PROJECT_DIRECTORY, N_("Project directory"));
    baseDirectoryType.select(RTC_DIRECTORY);
    oldBaseDirectoryType = baseDirectoryType.which();

    rtcDirectory = filesystem::path(executableTopDirectory()) / CNOID_PLUGIN_SUBDIR / "rtc";
}

RTCItem::RTCItem(const RTCItem& org)
    : Item(org),
      os(MessageView::instance()->cout()),
      periodicType(org.periodicType),
      baseDirectoryType(org.baseDirectoryType),
      rtcDirectory(org.rtcDirectory)
{
    rtcomp = org.rtcomp;
    moduleName = org.moduleName;
    mv = MessageView::instance();
    periodicRate = org.periodicRate;
    oldPeriodicType = org.oldPeriodicType;
    properties = org.properties;
    oldBaseDirectoryType = org.oldBaseDirectoryType;
}

RTCItem::~RTCItem()
{

}

void RTCItem::onPositionChanged()
{
    if(!rtcomp){
        if(convertAbsolutePath())
            rtcomp = new RTComponent(modulePath, properties);
    }
}

void RTCItem::onDisconnectedFromRoot()
{
    if(rtcomp){
        rtcomp->deleteRTC();
        delete rtcomp;
        rtcomp = 0;
    }
}

Item* RTCItem::doDuplicate() const
{
    return new RTCItem(*this);
}

void RTCItem::setModuleName(const std::string& name)
{
    if(moduleName!=name){
        moduleName = name;
        if(rtcomp){
            delete rtcomp;
        }
        if (convertAbsolutePath())
            rtcomp = new RTComponent(modulePath, properties);
    }
}

void RTCItem::setPeriodicType(int type)
{
    if(oldPeriodicType != type){
        oldPeriodicType = type;
        properties["exec_cxt.periodic.type"] = periodicType.symbol(type);
        if(rtcomp){
            delete rtcomp;
        }
        if (convertAbsolutePath())
            rtcomp = new RTComponent(modulePath, properties);
    }
}

void RTCItem::setPeriodicRate(int rate)
{
    if(periodicRate!=rate){
        periodicRate = rate;
        stringstream ss;
        ss << periodicRate;
        properties["exec_cxt.periodic.rate"] = ss.str();
        if(rtcomp){
            delete rtcomp;
        }
        if (convertAbsolutePath())
            rtcomp = new RTComponent(modulePath, properties);
    }
}

void RTCItem::setBaseDirectoryType(int base)
{
    baseDirectoryType.select(base);
    if (oldBaseDirectoryType != base){
        oldBaseDirectoryType = base;
        if (rtcomp){
            delete rtcomp;
        }
        if (convertAbsolutePath())
            rtcomp = new RTComponent(modulePath, properties);
    }
}

void RTCItem::doPutProperties(PutPropertyFunction& putProperty)
{
    Item::doPutProperties(putProperty);

    FilePathProperty moduleProperty(
        moduleName,
        { str(format(_("RT-Component module (*%1%)")) % DLL_SUFFIX) });

    if(baseDirectoryType.is(RTC_DIRECTORY)){
        moduleProperty.setBaseDirectory(rtcDirectory.string());
    } else if(baseDirectoryType.is(PROJECT_DIRECTORY)){
        moduleProperty.setBaseDirectory(ProjectManager::instance()->currentProjectDirectory());
    }

    putProperty(_("RTC module"), moduleProperty,
                [&](const string& name){ setModuleName(name); return true; });
    putProperty(_("Base directory"), baseDirectoryType,
                [&](int which){ setBaseDirectoryType(which); return true; });

    putProperty(_("Execution context"), periodicType,
                [&](int which){ return periodicType.select(which); });
    setPeriodicType(periodicType.selectedIndex());
    putProperty(_("Periodic rate"), periodicRate,
                [&](int rate){ setPeriodicRate(rate); return true; });
}

bool RTCItem::store(Archive& archive)
{
    if(!Item::store(archive)){
        return false;
    }
    archive.writeRelocatablePath("module", moduleName);
    archive.write("baseDirectory", baseDirectoryType.selectedSymbol(), DOUBLE_QUOTED);
    archive.write("periodicType", periodicType.selectedSymbol());
    archive.write("periodicRate", periodicRate);
    return true;
}

bool RTCItem::restore(const Archive& archive)
{
    if(!Item::restore(archive)){
        return false;
    }
    string value;
    if(archive.read("module", value) || archive.read("moduleName", value)){
        filesystem::path path(archive.expandPathVariables(value));
        moduleName = path.make_preferred().string();
    }
    if(archive.read("baseDirectory", value) || archive.read("RelativePathBase", value)){
        baseDirectoryType.select(value);
        oldBaseDirectoryType = baseDirectoryType.selectedIndex();
    }

    if(archive.read("periodicType", value)){
        periodicType.select(value);
        oldPeriodicType = periodicType.selectedIndex();
        properties["exec_cxt.periodic.type"] = value;
    }
    if(archive.read("periodicRate", periodicRate)){
        stringstream ss;
        ss << periodicRate;
        properties["exec_cxt.periodic.rate"] = ss.str();
    }

    return true;
}

bool RTCItem::convertAbsolutePath()
{
    modulePath = moduleName;
    if(!modulePath.is_absolute()){
        if(baseDirectoryType.is(RTC_DIRECTORY)){
            modulePath = rtcDirectory / modulePath;
        } else if(baseDirectoryType.is(PROJECT_DIRECTORY)){
            string projectDir = ProjectManager::instance()->currentProjectDirectory();
            if(projectDir.empty()){
                mv->putln(_("Please save the project."));
                return false;
            } else {
                modulePath = filesystem::path(projectDir) / modulePath;
            }
        }
    }
    return true;
}

RTComponent::RTComponent(const filesystem::path& modulePath, PropertyMap& prop)
{
    rtc_ = 0;
    rtcRef = 0;
    if (modulePath.empty()){
        return;
    }
    init(modulePath, prop);
}

RTComponent::~RTComponent()
{
    deleteRTC();
}

void RTComponent::init(const filesystem::path& modulePath_, PropertyMap& prop)
{
    mv = MessageView::instance();
    modulePath = modulePath_;
    createRTC(prop);
}

bool  RTComponent::createRTC(PropertyMap& prop)
{
    string moduleNameLeaf = modulePath.leaf().string();
    size_t i = moduleNameLeaf.rfind('.');
    if(i != string::npos){
        componentName = moduleNameLeaf.substr(0, i);
    } else {
        componentName = moduleNameLeaf;
    }

    string actualFilename;

    if(filesystem::exists(modulePath)){
        actualFilename = getNativePathString(modulePath);
        if(modulePath.extension() == DLL_SUFFIX){
            string initFunc(componentName + "Init");
            setupModules(actualFilename, initFunc, componentName, prop);
        } else {
            createProcess(actualFilename, prop);
        }
    } else {
        filesystem::path exePath(modulePath.string() + EXEC_SUFFIX);
        if(filesystem::exists(exePath)){
            actualFilename = getNativePathString(exePath);
            createProcess(actualFilename, prop);
        } else {
            filesystem::path dllPath(modulePath.string() + DLL_SUFFIX);
            if(filesystem::exists(dllPath)){
                actualFilename = getNativePathString(dllPath);
                string initFunc(componentName + "Init");
                setupModules(actualFilename, initFunc, componentName, prop);
            } else {
                mv->putln(fmt(_("A file of RTC \"%1%\" does not exist.")) % componentName);
            }
        }
    }

    bool created = isValid();

    if(created){
        mv->putln(fmt(_("RTC \"%1%\" has been created from \"%2%\".")) % componentName % actualFilename);
    } else {
        mv->putln(fmt(_("RTC \"%1%\" cannot be created.")) % componentName);
    }

    return created;
}

void RTComponent::setupModules(string& fileName, string& initFuncName, string& componentName, PropertyMap& prop)
{
    RTC::Manager& rtcManager = RTC::Manager::instance();

    rtcManager.load(fileName.c_str(), initFuncName.c_str());

    string option("?");
    for(PropertyMap::iterator it = prop.begin(); it != prop.end(); ){
        option += it->first + "=" + it->second;
        if(++it != prop.end()){
            option += "&";
        }
    }
    rtc_ = createManagedRTC((componentName + option).c_str());

    if(!rtc_){
        mv->putln(fmt(_("RTC \"%1%\" cannot be created by the RTC manager.\n"
                        " RTC module file: \"%2%\"\n"
                        " Init function: %3%\n"
                        " option: %4%"))
                  % componentName % fileName % initFuncName % option);
    }
}

void RTComponent::onReadyReadServerProcessOutput()
{
    mv->put(QString(rtcProcess.readAll()));
}

bool RTComponent::isValid() const
{
    return (rtc_ || rtcProcess.state() != QProcess::NotRunning);
}

void RTComponent::createProcess(string& command, PropertyMap& prop)
{
    QStringList argv;
    argv.push_back(QString("-o"));
    argv.push_back(QString("naming.formats: %n.rtc"));
    argv.push_back(QString("-o"));
    argv.push_back(QString("logger.enable: NO"));
    for(PropertyMap::iterator it = prop.begin(); it != prop.end(); it++){
        argv.push_back(QString("-o"));
        argv.push_back(QString(string(it->first+":"+it->second).c_str()));
    }
    if(rtcProcess.state() != QProcess::NotRunning){
        rtcProcess.kill();
        rtcProcess.waitForFinished(100);
    }
#ifdef _WIN32
    rtcProcess.start(QString("\"") + command.c_str() + "\"", argv );
#else
    rtcProcess.start(command.c_str(), argv);
#endif
    if(!rtcProcess.waitForStarted()){
        mv->putln(fmt(_("RT Component process \"%1%\" cannot be executed.")) % command);
    } else {
        mv->putln(fmt(_("RT Component process \"%1%\" has been executed.")) % command );
        rtcProcess.sigReadyReadStandardOutput().connect(
            std::bind(&RTComponent::onReadyReadServerProcessOutput, this));
    }
}

void RTComponent::deleteRTC()
{
    if(TRACE_FUNCTIONS){
        cout << "BodyRTComponent::deleteRTC()" << endl;
    }

    if(rtc_){
        string rtcName(rtc_->getInstanceName());
        mv->putln(fmt(_("delete %1%")) % rtcName);
        if(!cnoid::deleteRTC(rtc_)){
            mv->putln(fmt(_("%1% cannot be deleted.")) % rtcName);
        }
        rtc_ = 0;

    } else if(rtcProcess.state() != QProcess::NotRunning){
        mv->putln(fmt(_("delete %1%")) % componentName);
        rtcProcess.kill();
        rtcProcess.waitForFinished(100);
    }

    if(TRACE_FUNCTIONS){
        cout << "End of BodyRTCItem::deleteModule()" << endl;
    }
}

ext: cpp


text:
/*
 * Copyright (c) 2020-2021 Adrian Georg Herrmann
 *
 * This is the code for the HEMS Launcher. The HEMS Launcher is the executable that starts the
 * entire HEMS, launching all modules and initializing their message queues. The lifecycle of the
 * HEMS is identical with the launcher's lifecycle, therefore when the launcher terminates, the
 * entire HEMS does. The launcher also collects errors, outputs and log messages from all modules
 * and writes them to files and standard I/O streams.
 */

#include <chrono>
#include <fstream>
#include <iostream>
#include <mutex>
#include <thread>
#include <csignal>
#include <cstring>
#include <cstdlib>
#include <mqueue.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>

#include <boost/thread/shared_mutex.hpp>

#include "extras/semaphore.hpp"

#include "hems/modules/launcher/launcher.h"
#include "hems/modules/launcher/local_logger.h"
#include "hems/common/exit.h"
#include "hems/common/logger.h"
#include "hems/common/messenger.h"
#include "hems/common/modules.h"
#include "hems/common/types.h"
#include "hems/messages/storage.h"

namespace hems { namespace modules { namespace launcher {

    hems_launcher* hems_launcher::this_instance = nullptr;

    hems_launcher::hems_launcher(
        bool debug, bool test_mode, std::string storage_path, std::string db_path,
        std::string collection_path, std::string ui_path, std::string ui_server_root,
        std::string inference_path, std::string automation_path, std::string training_path
    ) : hems_launcher(
            debug, test_mode, storage_path, db_path, collection_path, ui_path, ui_server_root,
            inference_path, automation_path, training_path, types::settings_undefined
        ) {};

    hems_launcher::hems_launcher(
        bool debug, bool test_mode,
        std::string storage_path, std::string db_path, std::string collection_path,
        std::string ui_path, std::string ui_server_root, std::string inference_path,
        std::string automation_path, std::string training_path, types::settings_t settings
    ) : debug(debug) {

        logger::this_logger->log(
            "Starting " + modules::to_string_extended(module_type) + ".",
            logger::level::LOG
        );

        /* Delete message queues to remove junk messages from previous runs. */
        delete_msg_queues();

        /* Create message queues. */
        if (create_msg_queues()) {
            logger::this_logger->log("Error opening message queues, aborting.", logger::level::ERR);
            throw EXIT_FAILURE;
        } else {
            logger::this_logger->log("Message queues created successfully.", logger::level::DBG);
        }

        /* Open messenger object. */
        messenger::this_messenger = new messenger(module_type, test_mode);

        /* Begin listening for messages. */
        if (!messenger::this_messenger->listen(handler_map)) {
            logger::this_logger->log("Cannot listen for messages, aborting.", logger::level::ERR);
            throw EXIT_FAILURE;
        } else {
            logger::this_logger->log("Listening for messages.", logger::level::LOG);
        }

        /*  If debugging mode is on, we skip launching the modules and watching if they are still
            running, and let the user launch the modules themselves instead. */
        if (!debug) {
            /* BEGIN Launch modules. */

            auto start_module = [this](modules::type module, const char* file, char* const* argv) {
                if (!std::ifstream(file).good()) {
                    logger::this_logger->log(
                        "Cannot find binary '" + std::string(file) + "' for "
                        + modules::to_string_extended(module) + ", aborting.",
                        logger::level::ERR
                    );
                    throw EXIT_FAILURE;
                }
                switch (pid_t pid = fork()) {
                    case -1:
                        logger::this_logger->log(
                            "Couldn't start " + modules::to_string_extended(module) + ", aborting: " +
                                strerror(errno),
                            logger::level::ERR
                        );
                        throw EXIT_FAILURE;
                        break;
                    case 0:
                        execv(file, argv);
                        /*  `exec` only returns if something failed, so if the next line is ever
                            reached, something went wrong. We use `std::exit()` directly here
                            because we don't want a fork of the launcher module to call its destructor
                            as well. */
                        logger::this_logger->log(
                            "Couldn't start " + modules::to_string_extended(module) + ", aborting: " +
                                strerror(errno),
                            logger::level::ERR
                        );
                        std::exit(EXIT_FAILURE);
                        break;
                    default:
                        pids.mutex.lock();
                        pids.map[module] = pid;
                        pids.mutex.unlock();
                        break;
                }
            };

            const char* argv1[] = { storage_path.c_str(), "--db", db_path.c_str(), (char*) nullptr };
            start_module(modules::type::STORAGE, storage_path.c_str(), const_cast<char* const*>(argv1));
            watch_storage = new std::thread(&hems_launcher::watch_module, this, modules::type::STORAGE);

            const char* argv2[] = { collection_path.c_str(), (char*) nullptr };
            start_module(modules::type::COLLECTION, collection_path.c_str(), const_cast<char* const*>(argv2));
            watch_collection = new std::thread(&hems_launcher::watch_module, this, modules::type::COLLECTION);

            const char* argv3[] = { ui_path.c_str(), "--root", ui_server_root.c_str(), (char*) nullptr };
            start_module(modules::type::UI, ui_path.c_str(), const_cast<char* const*>(argv3));
            watch_ui = new std::thread(&hems_launcher::watch_module, this, modules::type::UI);

            const char* argv4[] = { inference_path.c_str(), (char*) nullptr };
            start_module(modules::type::INFERENCE, inference_path.c_str(), const_cast<char* const*>(argv4));
            watch_inference = new std::thread(&hems_launcher::watch_module, this, modules::type::INFERENCE);

            const char* argv5[] = { automation_path.c_str(), (char*) nullptr };
            start_module(modules::type::AUTOMATION, automation_path.c_str(), const_cast<char* const*>(argv5));
            watch_automation = new std::thread(&hems_launcher::watch_module, this, modules::type::AUTOMATION);

            const char* argv6[] = { training_path.c_str(), (char*) nullptr };
            start_module(modules::type::TRAINING, training_path.c_str(), const_cast<char* const*>(argv6));
            watch_training = new std::thread(&hems_launcher::watch_module, this, modules::type::TRAINING);

            /* END Launch modules. */
        }

        if (!test_mode) {
            /*  This must run in a separate thread because while it is initiated during the Launcher
                Module's construction, the constructor must finish before the settings
                initialization does. */
            init_settings_thread = new std::thread(&hems_launcher::init_settings, this, settings);
        }

        /* Message handlers must not be called before the module's constructor has finished. */
        logger::this_logger->log("Begin handling incoming messages.", logger::level::LOG);
        messenger::this_messenger->start_handlers();
    }

    hems_launcher::~hems_launcher() {
        logger::this_logger->log(
            "Shutting down " + modules::to_string_extended(module_type) + ".",
            logger::level::LOG
        );

        if (init_settings_thread != nullptr) {
            init_settings_thread->join();
            delete init_settings_thread;
        }

        /* BEGIN Shut down modules. */

        /*  If debugging mode is on, the modules were not launched automatically, so they cannot be
            shut down either. */
        if (!debug) {
            pids.mutex.lock_shared();
            for (const auto& item : pids.map) {
                modules::type module = item.first;
                pid_t pid = item.second;

                logger::this_logger->log(
                    "Signaling " + modules::to_string_extended(module) + " to shut down.",
                    logger::level::LOG
                );
                kill(pid, SIGTERM);
            }
            pids.mutex.unlock_shared();

            logger::this_logger->log(
                "Waiting for modules to shut down gracefully ...",
                logger::level::LOG
            );

            /*  No mutex is needed here because `watch_count.num` is atomic, so it cannot be changed
                by a module watcher while it's being read by the while loop. Were its value to
                change right after entering the while loop such that the loop condition were no
                longer true, the lock would succeed so no harm would have been done. */
            while (watch_count.num > 0) {
                if (!watch_count.sem.wait_for(std::chrono::seconds(5))) {
                    logger::this_logger->log(
                        "There was a timeout waiting for modules to shut down gracefully, terminating.",
                        logger::level::ERR
                    );

                    /* Forcefully kill all remaining processes. */
                    pids.mutex.lock_shared();
                    for (const auto& item : pids.map) {
                        kill(item.second, SIGKILL);
                    }
                    pids.mutex.unlock_shared();
                    break;
                }
            }

            if (watch_count.num <= 0) {
                logger::this_logger->log("All modules shut down.", logger::level::LOG);
            }

            logger::this_logger->log(
                "Successfully shut down " + modules::to_string_extended(module_type) +
                    ", stop listening for messages.",
                logger::level::LOG
            );

            watch_storage->join();
            watch_collection->join();
            watch_ui->join();
            watch_inference->join();
            watch_automation->join();
            watch_training->join();

            delete watch_storage;
            delete watch_collection;
            delete watch_ui;
            delete watch_inference;
            delete watch_automation;
            delete watch_training;
        }

        /* END Shut down modules. */

        /* Delete messenger object. */
        delete messenger::this_messenger;

        /* Delete message queues. */
        delete_msg_queues();
    }

    int hems_launcher::create_msg_queues() {
        auto create = [this] (const std::map<modules::type, std::string>& mqs) {
            for (const auto& item : mqs) {
                modules::type owner = item.first;
                std::string name = item.second;

                struct mq_attr attr = {
                    mq_flags    : 0,
                    mq_maxmsg   : 10,
                    mq_msgsize  : sizeof(messenger::msg_t),
                    mq_curmsgs  : 0
                };

                mqd_t id = mq_open(name.c_str(), O_RDWR | O_CLOEXEC | O_CREAT, 0666, &attr);
                if (id < 0) {
                    logger::this_logger->log(
                        "Could not create message queue for " + modules::to_string_extended(owner) + ": " +
                            strerror(errno),
                        logger::level::ERR
                    );
                    return 1;
                } else {
                    mq_close(id);
                }
            }
            return 0;
        };

        return (create(messenger::mq_names) | create(messenger::mq_res_names));
    }

    void hems_launcher::delete_msg_queues() {
        for (const auto& item : messenger::mq_names) {
            mq_unlink(item.second.c_str());
        }

        for (const auto& item : messenger::mq_res_names) {
            mq_unlink(item.second.c_str());
        }
    }

    void hems_launcher::watch_module(modules::type module) {
        int wait_status;
        int exit_status = EXIT_FAILURE;

        /*  There needs to be a second static mutex for the watcher count, which is used by module
            watchers to protect modifications of the variable. They can't use `watch_count.mutex`
            because they would lock up otherwise, as it is locked up by the main thread initially
            once at the end of the constructor. */
        static std::mutex wc_mutex;

        pids.mutex.lock_shared();
        pid_t pid = pids.map.at(module);
        pids.mutex.unlock_shared();

        wc_mutex.lock();
        ++watch_count.num;
        wc_mutex.unlock();

        pid_t pid_ = waitpid(pid, &wait_status, 0);
        if (pid_ == -1) {
            logger::this_logger->log(
                "Error watching process " + std::to_string(pid) + " of " +
                    modules::to_string_extended(module) + ", aborting: " + strerror(errno),
                logger::level::ERR
            );
            exit_status = EXIT_FAILURE;
        } else if (pid_ == pid) {
            exit_status = WIFEXITED(wait_status) ? WEXITSTATUS(wait_status) : EXIT_FAILURE;
            pids.mutex.lock();
            pids.map.erase(module);
            pids.mutex.unlock();
        }

        wc_mutex.lock();
        --watch_count.num;
        wc_mutex.unlock();

        watch_count.sem.notify();

        if (exit_status != EXIT_SUCCESS) {
            logger::this_logger->log(
                modules::to_string_extended(module) + " terminated with status " +
                    std::to_string(exit_status) + ", aborting.",
                logger::level::ERR
            );
        }
        hems::exit(exit_status);
    }

    void hems_launcher::init_settings(types::settings_t settings) {
        logger::this_logger->log(
            "Initializing settings: Waiting for Data Storage Module ...",
            logger::level::LOG
        );

        /*  Get the current settings from the Data Storage Module, if there are any. Broadcast them
            to all modules in that case. */
        std::string response;
        int res = messenger::this_messenger->send(
            2 * DEFAULT_SEND_TIMEOUT,
            messages::storage::msg_type::MSG_GET_SETTINGS,
            modules::STORAGE,
            messenger::this_messenger->serialize(""),
            &response
        );

        if (res == messages::storage::response_code::SUCCESS) {
            modules::current_settings = messenger::this_messenger->deserialize<types::settings_t>(response);
            logger::this_logger->log(
                "Prior settings found: " + to_string(modules::current_settings), logger::level::LOG
            );
        } else if (res == messages::storage::response_code::MSG_GET_NONE_AVAILABLE) {
            logger::this_logger->log(
                "No prior settings found.", logger::level::LOG
            );
        } else {
            logger::this_logger->log(
                "Error retrieving settings for initialization (" + std::to_string(res) + "). Terminating.",
                logger::level::ERR
            );
            exit(EXIT_FAILURE);
        }

        res = messenger::this_messenger->broadcast_settings_init(modules::current_settings);
        if (res != messenger::settings_code::SUCCESS) {
            logger::this_logger->log(
                "Error during settings initialization. Terminating.",
                logger::level::ERR
            );
            exit(EXIT_FAILURE);
        }

        logger::this_logger->log("Settings initialization finished.", logger::level::LOG);

        /* If new settings were given through command line arguments, broadcast them now. */
        if (!types::is_undefined(settings)) {
            logger::this_logger->log(
                "New settings were given through command line arguments, broadcasting them now: " +
                    types::to_string(settings),
                logger::level::LOG
            );

            res = messenger::this_messenger->broadcast_settings(settings);
            if (res != messenger::settings_code::SUCCESS) {
                logger::this_logger->log(
                    "The new settings were not accepted by all modules (code " + std::to_string(res) + "). " +
                        "Proceeding without applying new settings from command line arguments.",
                    logger::level::LOG
                );
            } else {
                logger::this_logger->log(
                    "New settings accepted by all modules: " + types::to_string(settings),
                    logger::level::LOG
                );
            }
        }
    }

}}}

max_stars_repo_name: adrianghc/HEMS

max_issues_repo_name: adrianghc/HEMS

max_forks_repo_name: adrianghc/HEMS


text:
//  Unit test for boost::lexical_cast.
//
//  See http://www.boost.org for most recent version, including documentation.
//
//  Copyright Alexander Nasonov, 2007.
//
//  Distributed under the Boost
//  Software License, Version 1.0. (See accompanying file
//  LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt).
//
//  Test that Source can be non-copyable.

#include <boost/config.hpp>

#if defined(__INTEL_COMPILER)
#pragma warning(disable: 193 383 488 981 1418 1419)
#elif defined(BOOST_MSVC)
#pragma warning(disable: 4097 4100 4121 4127 4146 4244 4245 4511 4512 4701 4800)
#endif

#include <boost/lexical_cast.hpp>
#include <boost/noncopyable.hpp>
#include <boost/test/unit_test.hpp>

using namespace boost;

void test_noncopyable();

unit_test::test_suite *init_unit_test_suite(int, char *[])
{
    unit_test::test_suite *suite =
        BOOST_TEST_SUITE("lexical_cast unit test");
    suite->add(BOOST_TEST_CASE(&test_noncopyable));

    return suite;
}

class Noncopyable : private boost::noncopyable
{
public:
    Noncopyable() {}
};

inline std::ostream &operator<<(std::ostream &out, const Noncopyable&)
{
    return out << "Noncopyable";
}

void test_noncopyable()
{
    Noncopyable x;
    BOOST_CHECK(boost::lexical_cast<std::string>(x) == "Noncopyable");
}