text:
// Copyright 2014 Cloudera Inc.
//
// 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 "scheduling/admission-controller.h"

#include <boost/algorithm/string.hpp>
#include <boost/bind.hpp>
#include <boost/foreach.hpp>
#include <boost/mem_fn.hpp>
#include <gutil/strings/substitute.h>

#include "common/logging.h"
#include "statestore/simple-scheduler.h"
#include "runtime/exec-env.h"
#include "runtime/mem-tracker.h"
#include "util/debug-util.h"
#include "util/time.h"
#include "util/runtime-profile.h"

using namespace std;
using namespace boost;
using namespace strings;

DEFINE_int64(queue_wait_timeout_ms, 60 * 1000, "Maximum amount of time (in "
    "milliseconds) that a request will wait to be admitted before timing out.");

namespace impala {

const string AdmissionController::IMPALA_REQUEST_QUEUE_TOPIC("impala-request-queue");

// Delimiter used for topic keys of the form "<pool_name><delimiter><backend_id>".
// "!" is used because the backend id contains a colon, but it should not contain "!".
// When parsing the topic key we need to be careful to find the last instance in
// case the pool name contains it as well.
const char TOPIC_KEY_DELIMITER = '!';

// Define metric key format strings for metrics in PoolMetrics
// '$0' is replaced with the pool name by strings::Substitute
const string LOCAL_ADMITTED_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-admitted";
const string LOCAL_QUEUED_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-queued";
const string LOCAL_DEQUEUED_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-dequeued";
const string LOCAL_REJECTED_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-rejected";
const string LOCAL_TIMED_OUT_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-timed-out";
const string LOCAL_COMPLETED_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-completed";
const string LOCAL_TIME_IN_QUEUE_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-time-in-queue-ms";
const string CLUSTER_NUM_RUNNING_METRIC_KEY_FORMAT =
  "admission-controller.$0.cluster-num-running";
const string CLUSTER_IN_QUEUE_METRIC_KEY_FORMAT =
  "admission-controller.$0.cluster-in-queue";
const string CLUSTER_MEM_USAGE_METRIC_KEY_FORMAT =
  "admission-controller.$0.cluster-mem-usage";
const string CLUSTER_MEM_ESTIMATE_METRIC_KEY_FORMAT =
  "admission-controller.$0.cluster-mem-estimate";
const string LOCAL_NUM_RUNNING_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-num-running";
const string LOCAL_IN_QUEUE_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-in-queue";
const string LOCAL_MEM_USAGE_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-mem-usage";
const string LOCAL_MEM_ESTIMATE_METRIC_KEY_FORMAT =
  "admission-controller.$0.local-mem-estimate";

// Profile query events
const string QUERY_EVENT_SUBMIT_FOR_ADMISSION = "Submit for admission";
const string QUERY_EVENT_COMPLETED_ADMISSION = "Completed admission";

// Profile info string for admission result
const string PROFILE_INFO_KEY_ADMISSION_RESULT = "Admission result";
const string PROFILE_INFO_VAL_ADMIT_IMMEDIATELY = "Admitted immediately";
const string PROFILE_INFO_VAL_ADMIT_QUEUED = "Admitted (queued)";
const string PROFILE_INFO_VAL_REJECTED = "Rejected";
const string PROFILE_INFO_VAL_TIME_OUT = "Timed out (queued)";

// Error status string formats
// $0 = query_id, $1 = pool, $2 = rejection reason (see REASON_XXX below)
const string STATUS_REJECTED = "Rejected query id=$0 from pool $1 : $2";
const string REASON_DISABLED_MEM_LIMIT = "disabled by mem limit set to 0";
const string REASON_DISABLED_REQUESTS_LIMIT = "disabled by requests limit set to 0";
const string REASON_QUEUE_FULL = "queue full, limit=$0, num_queued=$1";
const string REASON_REQ_OVER_MEM_LIMIT =
    "request memory estimate $0, greater than pool limit $1";
// $0 = query_id, $1 = timeout in milliseconds
const string STATUS_TIME_OUT = "Admission for query id=$0 exceeded timeout $1ms";

// Parses the pool name and backend_id from the topic key if it is valid.
// Returns true if the topic key is valid and pool_name and backend_id are set.
static inline bool ParsePoolTopicKey(const string& topic_key, string* pool_name,
    string* backend_id) {
  size_t pos = topic_key.find_last_of(TOPIC_KEY_DELIMITER);
  if (pos == string::npos || pos >= topic_key.size() - 1) {
    VLOG_QUERY << "Invalid topic key for pool: " << topic_key;
    return false;
  }
  *pool_name = topic_key.substr(0, pos);
  *backend_id = topic_key.substr(pos + 1);
  return true;
}

// Returns the topic key for the pool at this backend, i.e. a string of the
// form: "<pool_name><delimiter><backend_id>".
static inline string MakePoolTopicKey(const string& pool_name,
    const string& backend_id) {
  // Ensure the backend_id does not contain the delimiter to ensure that the topic key
  // can be parsed properly by finding the last instance of the delimiter.
  DCHECK_EQ(backend_id.find(TOPIC_KEY_DELIMITER), string::npos);
  return Substitute("$0$1$2", pool_name, TOPIC_KEY_DELIMITER, backend_id);
}

// Returns a debug string for the given local and total pool stats. Either
// 'total_stats' or 'local_stats' may be NULL to skip writing those stats.
static string DebugPoolStats(const string& pool_name,
    const TPoolStats* total_stats,
    const TPoolStats* local_stats) {
  stringstream ss;
  ss << "pool=" << pool_name;
  if (total_stats != NULL) {
    ss << " Total(";
    ss << "num_running=" << total_stats->num_running << ", ";
    ss << "num_queued=" << total_stats->num_queued << ", ";
    ss << "mem_usage=" <<
        PrettyPrinter::Print(total_stats->mem_usage, TCounterType::BYTES) << ", ";
    ss << "mem_estimate=" <<
        PrettyPrinter::Print(total_stats->mem_estimate, TCounterType::BYTES);
    ss << ")";
  }
  if (local_stats != NULL) {
    ss << " Local(";
    ss << "num_running=" << local_stats->num_running << ", ";
    ss << "num_queued=" << local_stats->num_queued << ", ";
    ss << "mem_usage=" <<
        PrettyPrinter::Print(local_stats->mem_usage, TCounterType::BYTES) << ", ";
    ss << "mem_estimate=" <<
        PrettyPrinter::Print(local_stats->mem_estimate, TCounterType::BYTES);
    ss << ")";
  }
  return ss.str();
}

AdmissionController::AdmissionController(RequestPoolService* request_pool_service,
    Metrics* metrics, const string& backend_id)
    : request_pool_service_(request_pool_service),
      metrics_(metrics),
      backend_id_(backend_id),
      thrift_serializer_(false),
      done_(false) {
  dequeue_thread_.reset(new Thread("scheduling", "admission-thread",
        &AdmissionController::DequeueLoop, this));
}

AdmissionController::~AdmissionController() {
  // The AdmissionController should live for the lifetime of the impalad, but
  // for unit tests we need to ensure that no thread is waiting on the
  // condition variable. This notifies the dequeue thread to stop and waits
  // for it to finish.
  {
    // Lock to ensure the dequeue thread will see the update to done_
    lock_guard<mutex> l(admission_ctrl_lock_);
    done_ = true;
    dequeue_cv_.notify_one();
  }
  dequeue_thread_->Join();
}

Status AdmissionController::Init(StatestoreSubscriber* subscriber) {
  StatestoreSubscriber::UpdateCallback cb =
    bind<void>(mem_fn(&AdmissionController::UpdatePoolStats), this, _1, _2);
  Status status = subscriber->AddTopic(IMPALA_REQUEST_QUEUE_TOPIC, true, cb);
  if (!status.ok()) {
    status.AddErrorMsg("AdmissionController failed to register request queue topic");
  }
  return status;
}

bool AdmissionController::CanAdmitRequest(const string& pool_name,
    const int64_t max_requests, const int64_t mem_limit, const QuerySchedule& schedule,
    bool admit_from_queue) {
  const TPoolStats& total_stats = cluster_pool_stats_[pool_name];
  DCHECK_GE(total_stats.mem_usage, 0);
  DCHECK_GE(total_stats.mem_estimate, 0);
  const int64_t cluster_estimated_memory = schedule.GetClusterMemoryEstimate() +
      max(total_stats.mem_usage, total_stats.mem_estimate);
  DCHECK_GE(cluster_estimated_memory, 0);

  // Can't admit if:
  //  (a) Already over the maximum number of requests
  //  (b) Request will go over the mem limit
  //  (c) This is not admitting from the queue and there are already queued requests
  if ((max_requests >= 0 && total_stats.num_running >= max_requests) ||
      (mem_limit >= 0 && cluster_estimated_memory >= mem_limit) ||
      (!admit_from_queue && total_stats.num_queued > 0)) {
    return false;
  }
  return true;
}

bool AdmissionController::RejectRequest(const string& pool_name,
    const int64_t max_requests, const int64_t mem_limit, const int64_t max_queued,
    const QuerySchedule& schedule, string* error_msg) {
  TPoolStats* total_stats = &cluster_pool_stats_[pool_name];
  const int64_t expected_mem_usage = schedule.GetClusterMemoryEstimate();
  string reject_reason;
  if (max_requests == 0) {
    reject_reason = REASON_DISABLED_REQUESTS_LIMIT;
  } else if (mem_limit == 0) {
    reject_reason = REASON_DISABLED_MEM_LIMIT;
  } else if (mem_limit > 0 && expected_mem_usage >= mem_limit) {
    reject_reason = Substitute(REASON_REQ_OVER_MEM_LIMIT,
        PrettyPrinter::Print(expected_mem_usage, TCounterType::BYTES),
        PrettyPrinter::Print(mem_limit, TCounterType::BYTES));
  } else if (total_stats->num_queued >= max_queued) {
    reject_reason = Substitute(REASON_QUEUE_FULL, max_queued, total_stats->num_queued);
  } else {
    return false; // Not rejected
  }
  *error_msg = Substitute(STATUS_REJECTED, PrintId(schedule.query_id()), pool_name,
      reject_reason);
  return true;
}

Status AdmissionController::AdmitQuery(QuerySchedule* schedule) {
  const string& pool_name = schedule->request_pool();
  TPoolConfigResult pool_config;
  RETURN_IF_ERROR(request_pool_service_->GetPoolConfig(pool_name, &pool_config));
  const int64_t max_requests = pool_config.max_requests;
  const int64_t max_queued = pool_config.max_queued;
  const int64_t mem_limit = pool_config.mem_limit;

  // Note the queue_node will not exist in the queue when this method returns.
  QueueNode queue_node(*schedule);

  schedule->query_events()->MarkEvent(QUERY_EVENT_SUBMIT_FOR_ADMISSION);
  ScopedEvent completedEvent(schedule->query_events(), QUERY_EVENT_COMPLETED_ADMISSION);
  {
    lock_guard<mutex> lock(admission_ctrl_lock_);
    RequestQueue* queue = &request_queue_map_[pool_name];
    pool_config_cache_[pool_name] = pool_config;
    PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
    TPoolStats* total_stats = &cluster_pool_stats_[pool_name];
    TPoolStats* local_stats = &local_pool_stats_[pool_name];
    const int64_t cluster_mem_estimate = schedule->GetClusterMemoryEstimate();
    VLOG_QUERY << "Schedule for id=" << schedule->query_id()
               << " in pool_name=" << pool_name << " PoolConfig(max_requests="
               << max_requests << " max_queued=" << max_queued
               << " mem_limit=" << PrettyPrinter::Print(mem_limit, TCounterType::BYTES)
               << ") query cluster_mem_estimate="
               << PrettyPrinter::Print(cluster_mem_estimate, TCounterType::BYTES);
    VLOG_QUERY << "Stats: " << DebugPoolStats(pool_name, total_stats, local_stats);

    if (CanAdmitRequest(pool_name, max_requests, mem_limit, *schedule, false)) {
      // Execute immediately
      pools_for_updates_.insert(pool_name);
      // The local and total stats get incremented together when we queue so if
      // there were any locally queued queries we should not admit immediately.
      DCHECK_EQ(local_stats->num_queued, 0);
      schedule->set_is_admitted(true);
      schedule->summary_profile()->AddInfoString(PROFILE_INFO_KEY_ADMISSION_RESULT,
          PROFILE_INFO_VAL_ADMIT_IMMEDIATELY);
      ++total_stats->num_running;
      ++local_stats->num_running;
      int64_t mem_estimate = schedule->GetClusterMemoryEstimate();
      local_stats->mem_estimate += mem_estimate;
      total_stats->mem_estimate += mem_estimate;
      if (pool_metrics != NULL) {
        pool_metrics->local_admitted->Increment(1L);
        pool_metrics->local_mem_estimate->Increment(mem_estimate);
        pool_metrics->cluster_mem_estimate->Increment(mem_estimate);
      }
      VLOG_QUERY << "Admitted query id=" << schedule->query_id();
      VLOG_RPC << "Final: " << DebugPoolStats(pool_name, total_stats, local_stats);
      return Status::OK;
    }

    string error_msg; // detailed error message returned if rejected
    if (RejectRequest(pool_name, max_requests, mem_limit, max_queued, *schedule,
        &error_msg)) {
      schedule->set_is_admitted(false);
      schedule->summary_profile()->AddInfoString(PROFILE_INFO_KEY_ADMISSION_RESULT,
          PROFILE_INFO_VAL_REJECTED);
      if (pool_metrics != NULL) pool_metrics->local_rejected->Increment(1L);
      VLOG_QUERY << error_msg;
      return Status(error_msg);
    }

    // We cannot immediately admit but do not need to reject, so queue the request
    VLOG_QUERY << "Queuing, query id=" << schedule->query_id();
    DCHECK_LT(total_stats->num_queued, max_queued);
    DCHECK(max_requests > 0 || mem_limit > 0);
    pools_for_updates_.insert(pool_name);
    ++local_stats->num_queued;
    ++total_stats->num_queued;
    queue->Enqueue(&queue_node);
    if (pool_metrics != NULL) pool_metrics->local_queued->Increment(1L);
  }

  int64_t wait_start_ms = ms_since_epoch();
  int64_t queue_wait_timeout_ms = max(0L, FLAGS_queue_wait_timeout_ms);
  // We just call Get() to block until the result is set or it times out. Note that we
  // don't hold the admission_ctrl_lock_ while we wait on this promise so we need to
  // check the state after acquiring the lock in order to avoid any races because it is
  // Set() by the dequeuing thread while holding admission_ctrl_lock_.
  // TODO: handle cancellation
  bool timed_out;
  queue_node.is_admitted.Get(queue_wait_timeout_ms, &timed_out);
  int64_t wait_time_ms = ms_since_epoch() - wait_start_ms;

  // Take the lock in order to check the result of is_admitted as there could be a race
  // with the timeout. If the Get() timed out, then we need to dequeue the request.
  // Otherwise, the request was admitted and we update the number of running queries
  // stats.
  {
    lock_guard<mutex> lock(admission_ctrl_lock_);
    RequestQueue* queue = &request_queue_map_[pool_name];
    PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
    pools_for_updates_.insert(pool_name);
    if (pool_metrics != NULL) {
      pool_metrics->local_time_in_queue_ms->Increment(wait_time_ms);
    }
    // Now that we have the lock, check again if the query was actually admitted (i.e.
    // if the promise still hasn't been set), in which case we just admit the query.
    timed_out = !queue_node.is_admitted.IsSet();
    TPoolStats* total_stats = &cluster_pool_stats_[pool_name];
    TPoolStats* local_stats = &local_pool_stats_[pool_name];
    if (timed_out) {
      queue->Remove(&queue_node);
      queue_node.is_admitted.Set(false);
      schedule->set_is_admitted(false);
      schedule->summary_profile()->AddInfoString(PROFILE_INFO_KEY_ADMISSION_RESULT,
          PROFILE_INFO_VAL_TIME_OUT);
      --local_stats->num_queued;
      --total_stats->num_queued;
      if (pool_metrics != NULL) pool_metrics->local_timed_out->Increment(1L);
      return Status(Substitute(STATUS_TIME_OUT, PrintId(schedule->query_id()),
            queue_wait_timeout_ms));
    }
    // The dequeue thread updates the stats (to avoid a race condition) so we do
    // not change them here.
    DCHECK(queue_node.is_admitted.Get());
    DCHECK(!queue->Contains(&queue_node));
    schedule->set_is_admitted(true);
    schedule->summary_profile()->AddInfoString(PROFILE_INFO_KEY_ADMISSION_RESULT,
        PROFILE_INFO_VAL_ADMIT_QUEUED);
    if (pool_metrics != NULL) pool_metrics->local_admitted->Increment(1L);
    VLOG_QUERY << "Admitted queued query id=" << schedule->query_id();
    VLOG_RPC << "Final: " << DebugPoolStats(pool_name, total_stats, local_stats);
    return Status::OK;
  }
}

Status AdmissionController::ReleaseQuery(QuerySchedule* schedule) {
  if (!schedule->is_admitted()) return Status::OK; // No-op if query was not admitted
  const string& pool_name = schedule->request_pool();
  {
    lock_guard<mutex> lock(admission_ctrl_lock_);
    TPoolStats* total_stats = &cluster_pool_stats_[pool_name];
    TPoolStats* local_stats = &local_pool_stats_[pool_name];
    DCHECK_GT(total_stats->num_running, 0);
    DCHECK_GT(local_stats->num_running, 0);
    --total_stats->num_running;
    --local_stats->num_running;

    int64_t mem_estimate = schedule->GetClusterMemoryEstimate();
    local_stats->mem_estimate -= mem_estimate;
    total_stats->mem_estimate -= mem_estimate;
    PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
    if (pool_metrics != NULL) {
      pool_metrics->local_completed->Increment(1L);
      pool_metrics->local_mem_estimate->Increment(-1 * mem_estimate);
      pool_metrics->cluster_mem_estimate->Increment(-1 * mem_estimate);
    }
    pools_for_updates_.insert(pool_name);
    VLOG_RPC << "Released query id=" << schedule->query_id() << " "
             << DebugPoolStats(pool_name, total_stats, local_stats);
  }
  dequeue_cv_.notify_one();
  return Status::OK;
}

// Statestore subscriber callback for IMPALA_REQUEST_QUEUE_TOPIC. First, add any local
// pool stats updates. Then, per_backend_pool_stats_map_ is updated with the updated
// stats from any topic deltas that are received and we recompute the cluster-wide
// aggregate stats.
void AdmissionController::UpdatePoolStats(
    const StatestoreSubscriber::TopicDeltaMap& incoming_topic_deltas,
    vector<TTopicDelta>* subscriber_topic_updates) {
  {
    lock_guard<mutex> lock(admission_ctrl_lock_);
    BOOST_FOREACH(PoolStatsMap::value_type& entry, local_pool_stats_) {
      UpdateLocalMemUsage(entry.first);
    }
    AddPoolUpdates(subscriber_topic_updates);

    StatestoreSubscriber::TopicDeltaMap::const_iterator topic =
        incoming_topic_deltas.find(IMPALA_REQUEST_QUEUE_TOPIC);
    if (topic != incoming_topic_deltas.end()) {
      const TTopicDelta& delta = topic->second;
      // Delta and non-delta updates are handled the same way, except for a full update
      // we first clear the per_backend_pool_stats_map_. We then update the global map
      // and then re-compute the pool stats for any pools that changed.
      if (!delta.is_delta) {
        VLOG_ROW << "Full impala-request-queue stats update";
        per_backend_pool_stats_map_.clear();
      }
      HandleTopicUpdates(delta.topic_entries);
      HandleTopicDeletions(delta.topic_deletions);
    }
    BOOST_FOREACH(PoolStatsMap::value_type& entry, local_pool_stats_) {
      UpdateClusterAggregates(entry.first);
    }
  }
  dequeue_cv_.notify_one(); // Dequeue and admit queries on the dequeue thread
}

void AdmissionController::HandleTopicUpdates(const vector<TTopicItem>& topic_updates) {
  BOOST_FOREACH(const TTopicItem& item, topic_updates) {
    string pool_name;
    string topic_backend_id;
    if (!ParsePoolTopicKey(item.key, &pool_name, &topic_backend_id)) continue;
    // The topic entry from this subscriber is handled specially; the stats coming
    // from the statestore are likely already outdated.
    if (topic_backend_id == backend_id_) continue;
    local_pool_stats_[pool_name]; // Create an entry in the local map if it doesn't exist
    TPoolStats pool_update;
    uint32_t len = item.value.size();
    Status status = DeserializeThriftMsg(reinterpret_cast<const uint8_t*>(
          item.value.data()), &len, false, &pool_update);
    if (!status.ok()) {
      VLOG_QUERY << "Error deserializing pool update with key: " << item.key;
      continue;
    }
    PoolStatsMap& pool_map = per_backend_pool_stats_map_[pool_name];

    // Debug logging
    if (pool_map.find(topic_backend_id) != pool_map.end()) {
      VLOG_ROW << "Stats update for key=" << item.key << " previous: "
               << DebugPoolStats(pool_name, NULL, &pool_map[topic_backend_id]);
    }
    VLOG_ROW << "Stats update for key=" << item.key << " updated: "
             << DebugPoolStats(pool_name, NULL, &pool_update);

    pool_map[topic_backend_id] = pool_update;
    DCHECK(per_backend_pool_stats_map_[pool_name][topic_backend_id].num_running ==
        pool_update.num_running);
    DCHECK(per_backend_pool_stats_map_[pool_name][topic_backend_id].num_queued ==
        pool_update.num_queued);
  }
}

void AdmissionController::HandleTopicDeletions(const vector<string>& topic_deletions) {
  BOOST_FOREACH(const string& topic_key, topic_deletions) {
    string pool_name;
    string topic_backend_id;
    if (!ParsePoolTopicKey(topic_key, &pool_name, &topic_backend_id)) continue;
    PoolStatsMap& pool_map = per_backend_pool_stats_map_[pool_name];
    VLOG_ROW << "Deleting stats for key=" << topic_key << " "
             << DebugPoolStats(pool_name, NULL, &pool_map[topic_backend_id]);
    pool_map.erase(topic_backend_id);
    DCHECK(per_backend_pool_stats_map_[pool_name].find(topic_backend_id) ==
           per_backend_pool_stats_map_[pool_name].end());
  }
}

void AdmissionController::UpdateClusterAggregates(const string& pool_name) {
  const TPoolStats& local_stats = local_pool_stats_[pool_name];
  const PoolStatsMap& pool_map = per_backend_pool_stats_map_[pool_name];
  TPoolStats total_stats;
  BOOST_FOREACH(const PoolStatsMap::value_type& entry, pool_map) {
    // Skip an update from this subscriber as the information may be outdated.
    // The current local_stats will be added below.
    if (entry.first == backend_id_) continue;
    DCHECK_GE(entry.second.num_running, 0);
    DCHECK_GE(entry.second.num_queued, 0);
    DCHECK_GE(entry.second.mem_usage, 0);
    DCHECK_GE(entry.second.mem_estimate, 0);
    total_stats.num_running += entry.second.num_running;
    total_stats.num_queued += entry.second.num_queued;
    total_stats.mem_usage += entry.second.mem_usage;
    total_stats.mem_estimate += entry.second.mem_estimate;
  }
  total_stats.num_running += local_stats.num_running;
  total_stats.num_queued += local_stats.num_queued;
  total_stats.mem_usage += local_stats.mem_usage;
  total_stats.mem_estimate += local_stats.mem_estimate;

  DCHECK_GE(total_stats.num_running, 0);
  DCHECK_GE(total_stats.num_queued, 0);
  DCHECK_GE(total_stats.mem_usage, 0);
  DCHECK_GE(total_stats.mem_estimate, 0);
  DCHECK_GE(total_stats.num_running, local_stats.num_running);
  DCHECK_GE(total_stats.num_queued, local_stats.num_queued);

  cluster_pool_stats_[pool_name] = total_stats;
  PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
  if (pool_metrics != NULL) {
    pool_metrics->cluster_num_running->Update(total_stats.num_running);
    pool_metrics->cluster_in_queue->Update(total_stats.num_queued);
    pool_metrics->cluster_mem_usage->Update(total_stats.mem_usage);
    pool_metrics->cluster_mem_estimate->Update(total_stats.mem_estimate);
  }

  if (cluster_pool_stats_[pool_name] != total_stats) {
    VLOG_ROW << "Recomputed stats, previous: "
             << DebugPoolStats(pool_name, &cluster_pool_stats_[pool_name], NULL);
    VLOG_ROW << "Recomputed stats, updated: "
             << DebugPoolStats(pool_name, &total_stats, NULL);
  }
}

void AdmissionController::UpdateLocalMemUsage(const string& pool_name) {
  TPoolStats* stats = &local_pool_stats_[pool_name];
  MemTracker* tracker = MemTracker::GetRequestPoolMemTracker(pool_name, NULL);
  const int64_t current_usage = tracker == NULL ? 0L : tracker->consumption();
  if (current_usage != stats->mem_usage) {
    stats->mem_usage = current_usage;
    pools_for_updates_.insert(pool_name);
    PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
    if (pool_metrics != NULL) {
      pool_metrics->local_mem_usage->Update(current_usage);
    }
  }
}

void AdmissionController::AddPoolUpdates(vector<TTopicDelta>* topic_updates) {
  if (pools_for_updates_.empty()) return;
  topic_updates->push_back(TTopicDelta());
  TTopicDelta& topic_delta = topic_updates->back();
  topic_delta.topic_name = IMPALA_REQUEST_QUEUE_TOPIC;
  BOOST_FOREACH(const string& pool_name, pools_for_updates_) {
    DCHECK(local_pool_stats_.find(pool_name) != local_pool_stats_.end());
    TPoolStats& pool_stats = local_pool_stats_[pool_name];
    VLOG_ROW << "Sending topic update " << DebugPoolStats(pool_name, NULL, &pool_stats);
    topic_delta.topic_entries.push_back(TTopicItem());
    TTopicItem& topic_item = topic_delta.topic_entries.back();
    topic_item.key = MakePoolTopicKey(pool_name, backend_id_);
    Status status = thrift_serializer_.Serialize(&pool_stats, &topic_item.value);
    if (!status.ok()) {
      LOG(WARNING) << "Failed to serialize query pool stats: " << status.GetErrorMsg();
      topic_updates->pop_back();
    }
    PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
    if (pool_metrics != NULL) {
      pool_metrics->local_num_running->Update(pool_stats.num_running);
      pool_metrics->local_in_queue->Update(pool_stats.num_queued);
      pool_metrics->local_mem_usage->Update(pool_stats.mem_usage);
    }
  }
  pools_for_updates_.clear();
}

void AdmissionController::DequeueLoop() {
  while (true) {
    unique_lock<mutex> lock(admission_ctrl_lock_);
    if (done_) break;
    dequeue_cv_.wait(lock);
    BOOST_FOREACH(PoolStatsMap::value_type& entry, local_pool_stats_) {
      const string& pool_name = entry.first;
      TPoolStats* local_stats = &entry.second;

      PoolConfigMap::iterator it = pool_config_cache_.find(pool_name);
      if (it == pool_config_cache_.end()) continue; // No local requests in this pool
      const TPoolConfigResult& pool_config = it->second;

      const int64_t max_requests = pool_config.max_requests;
      const int64_t mem_limit = pool_config.mem_limit;

      // We should never have queued any requests in pools where either limit is 0 as no
      // requests should ever be admitted or when both limits are less than 0, i.e.
      // unlimited requests can be admitted and should never be queued.
      if (max_requests == 0 || mem_limit == 0 || (max_requests < 0 && mem_limit < 0)) {
        DCHECK_EQ(local_stats->num_queued, 0);
      }

      if (local_stats->num_queued == 0) continue; // Nothing to dequeue
      DCHECK(max_requests > 0 || mem_limit > 0);
      TPoolStats* total_stats = &cluster_pool_stats_[pool_name];

      DCHECK_GT(local_stats->num_queued, 0);
      DCHECK_GE(total_stats->num_queued, local_stats->num_queued);

      // Determine the maximum number of requests that can possibly be dequeued based
      // on the max_requests limit and the current queue size. We will attempt to
      // dequeue up to this number of requests until reaching the per-pool memory limit.
      int64_t max_to_dequeue = 0;
      if (max_requests > 0) {
        const int64_t total_available = max_requests - total_stats->num_running;
        if (total_available <= 0) continue;
        double queue_size_ratio = static_cast<double>(local_stats->num_queued) /
            static_cast<double>(total_stats->num_queued);
        // The maximum number of requests that can possibly be dequeued is the total
        // number of available requests scaled by the ratio of the size of the local
        // queue to the size of the total queue. We attempt to dequeue at least one
        // request and at most the size of the local queue.
        // TODO: Use a simple heuristic rather than a lower bound of 1 to avoid admitting
        // too many requests globally when only a single request can be admitted.
        max_to_dequeue = min(local_stats->num_queued,
            max(1L, static_cast<int64_t>(queue_size_ratio * total_available)));
      } else {
        max_to_dequeue = local_stats->num_queued; // No limit on num running requests
      }

      RequestQueue& queue = request_queue_map_[pool_name];
      VLOG_RPC << "Dequeue thread will try to admit " << max_to_dequeue << " requests"
               << ", pool=" << pool_name << ", num_queued=" << local_stats->num_queued;

      PoolMetrics* pool_metrics = GetPoolMetrics(pool_name);
      while (max_to_dequeue > 0 && !queue.empty()) {
        QueueNode* queue_node = queue.head();
        DCHECK(queue_node != NULL);
        DCHECK(!queue_node->is_admitted.IsSet());
        const QuerySchedule& schedule = queue_node->schedule;
        if (!CanAdmitRequest(pool_name, max_requests, mem_limit, schedule, true)) {
          VLOG_RPC << "Could not dequeue query id=" << queue_node->schedule.query_id();
          break;
        }
        queue.Dequeue();
        --local_stats->num_queued;
        --total_stats->num_queued;
        ++local_stats->num_running;
        ++total_stats->num_running;
        int64_t mem_estimate = schedule.GetClusterMemoryEstimate();
        local_stats->mem_estimate += mem_estimate;
        total_stats->mem_estimate += mem_estimate;
        if (pool_metrics != NULL) {
          pool_metrics->local_dequeued->Increment(1L);
          pool_metrics->local_mem_estimate->Increment(mem_estimate);
          pool_metrics->cluster_mem_estimate->Increment(mem_estimate);
        }
        VLOG_ROW << "Dequeuing query id=" << queue_node->schedule.query_id();
        queue_node->is_admitted.Set(true);
        --max_to_dequeue;
      }
      pools_for_updates_.insert(pool_name);
    }
  }
}

AdmissionController::PoolMetrics*
AdmissionController::GetPoolMetrics(const string& pool_name) {
  if (metrics_ == NULL) return NULL;
  PoolMetricsMap::iterator it = pool_metrics_map_.find(pool_name);
  if (it != pool_metrics_map_.end()) return &it->second;

  PoolMetrics* pool_metrics = &pool_metrics_map_[pool_name];
  pool_metrics->local_admitted = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_ADMITTED_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_queued = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_QUEUED_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_dequeued = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_DEQUEUED_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_rejected = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_REJECTED_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_timed_out = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_TIMED_OUT_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_completed = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_COMPLETED_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_time_in_queue_ms = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_TIME_IN_QUEUE_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->cluster_num_running = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(CLUSTER_NUM_RUNNING_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->cluster_in_queue = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(CLUSTER_IN_QUEUE_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->cluster_mem_usage = metrics_->RegisterMetric(new Metrics::BytesMetric(
      Substitute(CLUSTER_MEM_USAGE_METRIC_KEY_FORMAT, pool_name), 0L));
  pool_metrics->cluster_mem_estimate = metrics_->RegisterMetric(new Metrics::BytesMetric(
      Substitute(CLUSTER_MEM_ESTIMATE_METRIC_KEY_FORMAT, pool_name), 0L));
  pool_metrics->local_num_running = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_NUM_RUNNING_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_in_queue = metrics_->CreateAndRegisterPrimitiveMetric(
      Substitute(LOCAL_IN_QUEUE_METRIC_KEY_FORMAT, pool_name), 0L);
  pool_metrics->local_mem_usage = metrics_->RegisterMetric(new Metrics::BytesMetric(
      Substitute(LOCAL_MEM_USAGE_METRIC_KEY_FORMAT, pool_name), 0L));
  pool_metrics->local_mem_estimate = metrics_->RegisterMetric(new Metrics::BytesMetric(
      Substitute(LOCAL_MEM_ESTIMATE_METRIC_KEY_FORMAT, pool_name), 0L));
  return pool_metrics;
}
}

max_stars_repo_name: Altiscale/Impala

max_issues_repo_name: Altiscale/Impala

max_forks_repo_name: Altiscale/Impala


text:
// ============================================================================
//
//  Project 2.1: 2D Feature Tracking (Udacity Sensor Fusion Nanodegree)
//
//  Authors:     Andreas Albrecht using code base/skeleton provided by Udacity
//
//  Source:      https://github.com/udacity/SFND_2D_Feature_Tracking
//
// ============================================================================

/* INCLUDES FOR THIS PROJECT */
#include <iostream>
#include <fstream>
#include <sstream>
#include <iomanip>
#include <vector>
#include <cmath>
#include <limits>
#include <opencv2/core.hpp>
#include <opencv2/highgui/highgui.hpp>
#include <opencv2/imgproc/imgproc.hpp>
#include <opencv2/features2d.hpp>
#include <opencv2/xfeatures2d.hpp>
#include <opencv2/xfeatures2d/nonfree.hpp>

#include <boost/circular_buffer.hpp>
#include <boost/circular_buffer/base.hpp>

#include "dataStructures.h"
#include "matching2D.hpp"

using namespace std;

/* GLOBAL VARIABLES */
string dataPath = "../"; // data location

/* 2D FEATURE TRACKING FUNCTION */
int fun_2DFeatureTracking(
    vector<boost::circular_buffer<EvalResults>> & evalResultBuffers,
    string detectorType = "FAST",
    bool bFocusOnVehicle = true,
    bool bLimitKpts = false,
    int maxKeypoints = 50,
    string descExtractorType = "BRIEF",
    string matcherType = "MAT_BF",
    string descriptorType = "DES_BINARY",
    string selectorType = "SEL_KNN",
    bool bVis = true,
    bool bVisDebug = false,
    bool bExportResultsToCSV = true)
{

    /* INIT VARIABLES AND DATA STRUCTURES */

    // camera data frames
    string imgBasePath = dataPath + "images/";
    string imgPrefix = "KITTI/2011_09_26/image_00/data/000000";  // left camera, color
    string imgFileType = ".png";
    int imgStartIndex = 0; // first file index to load (assumes Lidar and camera names have identical naming convention)
    int imgEndIndex = 9;   // last file index to load
    int imgFillWidth = 4;  // no. of digits which make up the file index (e.g. img-0001.png)

    // using boost circular data buffer class template
    // https://www.boost.org/doc/libs/1_65_1/doc/html/circular_buffer.html
    int dataBufferSize = 2; // no. of images which are held in memory (ring buffer) at the same time
    //vector<DataFrame> dataBuffer; // list of data frames which are held in memory at the same time (constantly growing! => replace with ringbuffer)
    boost::circular_buffer<DataFrame> dataBuffer(dataBufferSize); // create circular buffer for DataFrame structures

    // print data buffer capacity
    cout << "Circular data buffer capacity in use = "
        << dataBuffer.size()
        << " of max. "
        << dataBuffer.capacity()
        << " data frames."
        << endl;

    // visualization (debugging)
    //bool bVis = true;        // visualize results
    //bool bVisDebug = false;  // visualize intermediate results (for debugging or to look into details)

    // circular data buffer to hold evaluation results
    //bool bExportResultsToCSV = true;
    int resultBufferSize = imgEndIndex - imgStartIndex + 1;
    boost::circular_buffer<EvalResults> resultBuffer(resultBufferSize);

    /* MAIN LOOP OVER ALL IMAGES */

    for (size_t imgIndex = 0; imgIndex <= imgEndIndex - imgStartIndex; imgIndex++)
    {

        /* LOAD IMAGE INTO BUFFER */

        // assemble filenames for current index
        ostringstream imgNumber;
        imgNumber << setfill('0') << setw(imgFillWidth) << imgStartIndex + imgIndex;
        string imgFullFilename = imgBasePath + imgPrefix + imgNumber.str() + imgFileType;

        // load image from file and convert to grayscale
        cv::Mat img, imgGray;
        img = cv::imread(imgFullFilename);
        cv::cvtColor(img, imgGray, cv::COLOR_BGR2GRAY);

        //// STUDENT ASSIGNMENT
        //// TASK MP.1 -> replace the following code with ring buffer of size dataBufferSize

        // create data frame and push it back to the tail of the (circular) data frame buffer, or ringbuffer, resp.
        DataFrame frame;
        dataBuffer.push_back(frame);  // push back empty data frame to the tail of the data frame ringbuffer

        // store filename without path and grayscale image in the last element of the data frame ringbuffer
        (dataBuffer.end() - 1)->imgFilename = imgNumber.str() + imgFileType;
        (dataBuffer.end() - 1)->cameraImg = imgGray;

        // clear temporary variables that are no longer needed (avoid memory leaks)
        img.release();
        imgGray.release();

        // create structure to hold the evaluation results and push it back to tail of the (circular) result buffer, or ringbuffer, resp.
        EvalResults results;
        resultBuffer.push_back(results);  // push back empty results structure to the tail fo the result ringbuffer

        // store filename without path in the last element of the result ringbuffer
        (resultBuffer.end() - 1)->imgFilename = imgNumber.str() + imgFileType;

        //// EOF STUDENT ASSIGNMENT
        cout << "#1 : LOAD IMAGE INTO BUFFER done" << endl;

        // Print data buffer capacity in use
        cout << "Circular data buffer capacity in use = "
            << dataBuffer.size()
            << " of max. "
            << dataBuffer.capacity()
            << " data frames."
            << endl;

        /* DETECT 2D KEYPOINTS IN CURRENT IMAGE */

        //// STUDENT ASSIGNMENT
        //// TASK MP.2 -> add the following keypoint detectors in file matching2D.cpp and enable string-based selection based on detectorType
        //// -> HARRIS, FAST, BRISK, ORB, AKAZE, SIFT

        // Select keypoint detector type
        //string detectorType = "SIFT";  // options: "SHITOMASI", "HARRIS", "FAST", "BRISK", "ORB", "KAZE", "AKAZE", "SIFT", "SURF"

        // store selected detector type in the last element of result ringbuffer
        (resultBuffer.end() - 1)->detectorType = detectorType;
        cout << "Seletect keypoint detector tpye = " << detectorType << endl;

        // create empty feature list for current image
        vector<cv::KeyPoint> keypoints;

        // Initialize processing time for keypoint detection
        double t_detKeypoints = 0.0;

        // detect keypoints
        if (detectorType.compare("SHITOMASI") == 0)
        {
            try
            {
                // Detect keypoints using Shi-Tomasi detector
                t_detKeypoints = detKeypointsShiTomasi(keypoints, (dataBuffer.end() - 1)->cameraImg, bVisDebug);
            }
            catch(const exception& e)
            {
                // show exeption and return 1
                cerr << e.what() << endl;
                return 1;
            }
        }
        else if (detectorType.compare("HARRIS") == 0)
        {
            try
            {
                // detect keypoints using Harris detector
                t_detKeypoints = detKeypointsHarris(keypoints, (dataBuffer.end() - 1)->cameraImg, bVisDebug);
            }
            catch(const exception& e)
            {
                // show exeption and return 1
                cerr << e.what() << endl;
                return 1;
            }
        }
        else
        {
            try
            {
                // detect keypoints using other user-specified detector types
                t_detKeypoints = detKeypointsModern(keypoints, (dataBuffer.end() - 1)->cameraImg, detectorType, bVisDebug);
            }
            catch(const char *msg)
            {
                // show error message and return 1
                cout << msg << endl;
                return 1;
            }
            catch(const exception& e)
            {
                // show exeption and return 1
                cerr << e.what() << endl;
                return 1;
            }
        }

        // store the number of detected keypoints and the processing time for keypoint detection in the last element of the result ringbuffer
        (resultBuffer.end() - 1)->numKeypoints = keypoints.size();
        (resultBuffer.end() - 1)->t_detKeypoints = t_detKeypoints;

        //// EOF STUDENT ASSIGNMENT

        //// STUDENT ASSIGNMENT
        //// TASK MP.3 -> only keep keypoints on the preceding vehicle

        // only keep keypoints on the preceding vehicle
        // remark: This is a temporary solution that will be replace with bounding boxes provided by an object detection CNN!
        //bool bFocusOnVehicle = true;  // true: only consider keypoints within the bounding box; false: consider all keypoints
        // store bFocusOnVehicle flag in the last result ringbuffer element
        (resultBuffer.end() - 1)->bFocusOnVehicle = bFocusOnVehicle;
        cout << "Focus on keypoints on vehicle = " << bFocusOnVehicle << endl;
        cv::Rect vehicleRect(535, 180, 180, 150);
        if (bFocusOnVehicle)
        {
            // create a copy the original keypoint vector
            vector<cv::KeyPoint> keypoints_cpy;
            copy(keypoints.begin(), keypoints.end(), back_inserter(keypoints_cpy));

            // clear the contents of the original keypoint vector leaving the container with size 0
            keypoints.clear();

            // loop over all keypoints found in the current image
            for (auto keyPoint : keypoints_cpy)
            {
                // check wheter a keypoint is within the bounding box of the preceeding vehicle
                if (vehicleRect.contains(keyPoint.pt))
                {
                    // keep only those keypoints within the bounding box of the preceeding vehicle
                    keypoints.push_back(keyPoint);
                }
            }

            // viszalize keypoints before and after bounding box filtering (for debugging)
            if (bVisDebug)
            {
                // plot original keypoints (copy) before filtering
                cv::Mat visImgGray_all_Kpts = (dataBuffer.end() - 1)->cameraImg.clone();
                cv::drawKeypoints(
                    (dataBuffer.end() - 1)->cameraImg, keypoints_cpy, visImgGray_all_Kpts,
                    cv::Scalar::all(-1), cv::DrawMatchesFlags::DRAW_RICH_KEYPOINTS);
                cv::rectangle(visImgGray_all_Kpts, vehicleRect, cv::Scalar(0, 255, 0), 2, 8);
                // plot keypoints after filtering
                cv::Mat visImgGray_filtered_Kpts = (dataBuffer.end() - 1)->cameraImg.clone();
                cv::drawKeypoints(
                    (dataBuffer.end() - 1)->cameraImg, keypoints, visImgGray_filtered_Kpts,
                    cv::Scalar::all(-1), cv::DrawMatchesFlags::DRAW_RICH_KEYPOINTS);
                cv::rectangle(visImgGray_filtered_Kpts, vehicleRect, cv::Scalar(0, 255, 0), 2, 8);
                // vertically concatenate both plots
                cv::Mat visImgGray_Kpts;
                cv::vconcat(visImgGray_all_Kpts, visImgGray_filtered_Kpts, visImgGray_Kpts);
                // show concatenated plots
                string windowName_Kpts = "Keypoints before and after bounding box filtering";
                cv::namedWindow(windowName_Kpts, 5);
                cv::imshow(windowName_Kpts, visImgGray_Kpts);

                // wait for user key press
                cout << "Press key to continue to next image" << endl;
                cv::waitKey(0); // wait for key to be pressed

                // clear temporary images and variables
                visImgGray_all_Kpts.release();
                visImgGray_filtered_Kpts.release();
                visImgGray_Kpts.release();
                windowName_Kpts.clear();
            }

            // number of keypoints befor and after filtering
            cout << "Total number of keypoints found in the overall image:    n_total  = " << keypoints_cpy.size() << endl;
            cout << "Number of keypoints in the target vehicle bounding box:  n_target = " << keypoints.size() << endl;

            // clear the copy of the original keypoint vector
            keypoints_cpy.clear();
        }

        // store the number of keypoints within the region of interest (target bounding box) in the last result ringbuffer element
        (resultBuffer.end() - 1)->numKeypointsInROI = keypoints.size();  // equal to total number of keypoints if bFocusOnVehicle == false

        //// EOF STUDENT ASSIGNMENT

        // optional : limit number of keypoints (helpful only for debugging and learning => Do not use in real application!)
        //bool bLimitKpts = false;
        (resultBuffer.end() - 1)->bLimitKpts = bLimitKpts;  // store bLimitKpts flag in the last result ringbuffer element
        cout << "Limit number of keypoints = " << bLimitKpts << endl;
        if (bLimitKpts)
        {
            // copy keypoints for debugging and visualization purpose
            vector<cv::KeyPoint> keypoints_cpy;
            copy(keypoints.begin(), keypoints.end(), back_inserter(keypoints_cpy));

            // int maxKeypoints = 50;  // only for testing => Do not limit keypoints in real application!

            if (detectorType.compare("SHITOMASI") == 0)
            { // there is no response info, so keep the first maxKepyoints as they are sorted in descending quality order
                keypoints.erase(keypoints.begin() + maxKeypoints, keypoints.end());
            }
            else
            {
                // sort keypoints according to the strength of the detector response (keypoints are not always sorted automatically!)
                sort(keypoints.begin(), keypoints.end(), compareKeypointResponse);

                // keep the first maxKeypoints from the list sorted by descending order of the detector response
                keypoints.erase(keypoints.begin() + maxKeypoints, keypoints.end());
            }
            cv::KeyPointsFilter::retainBest(keypoints, maxKeypoints);
            cout << "NOTE: Keypoints have been limited (n_max = " << maxKeypoints << ")!" << endl;
            cout << "The first n_max = " << keypoints.size() << " keypoints are kept. " << endl;

            // vViszalize keypoints before and after limiting their number (for debugging)
            if (bVisDebug)
            {
                // plot original keypoints (copy) before filtering
                cv::Mat visImgGray_all_Kpts = (dataBuffer.end() - 1)->cameraImg.clone();
                cv::drawKeypoints(
                    (dataBuffer.end() - 1)->cameraImg, keypoints_cpy, visImgGray_all_Kpts,
                    cv::Scalar::all(-1), cv::DrawMatchesFlags::DRAW_RICH_KEYPOINTS);
                // plot keypoints after filtering
                cv::Mat visImgGray_filtered_Kpts = (dataBuffer.end() - 1)->cameraImg.clone();
                cv::drawKeypoints(
                    (dataBuffer.end() - 1)->cameraImg, keypoints, visImgGray_filtered_Kpts,
                    cv::Scalar::all(-1), cv::DrawMatchesFlags::DRAW_RICH_KEYPOINTS);
                // vertically concatenate both plots
                cv::Mat visImgGray_Kpts;
                cv::vconcat(visImgGray_all_Kpts, visImgGray_filtered_Kpts, visImgGray_Kpts);
                // show concatenated plots
                string windowName_Kpts = "Keypoints before and after bounding box filtering";
                cv::namedWindow(windowName_Kpts, 5);
                cv::imshow(windowName_Kpts, visImgGray_Kpts);

                // wait for user key press
                cout << "Press key to continue to next image" << endl;
                cv::waitKey(0); // wait for key to be pressed

                // clear temporary images and variables
                visImgGray_all_Kpts.release();
                visImgGray_filtered_Kpts.release();
                visImgGray_Kpts.release();
                windowName_Kpts.clear();
            }

            // clear the copy of the original keypoint vector
            keypoints_cpy.clear();
        }

        // store the number of limited keypoints within the region of interest (target bounding box) in the last result ringbuffer element
        (resultBuffer.end() - 1)->numKeypointsInROILimited = keypoints.size();  // equal to number of keypoints in ROI if bLimitKpts == false

        // push keypoints and descriptor for current frame to end of data buffer
        (dataBuffer.end() - 1)->keypoints = keypoints;

        // clear temporary variables
        keypoints.clear();

        cout << "#2 : DETECT KEYPOINTS done" << endl;

        /* EVALUATE THE MEAN STRENGTH AND NEIGHBORHOOD SIZE (MEAN AND VARIANCE) OF THE REMAINING KEYPOINTS

           Note:
           - Keypoints.response = strength of the keypoint detectors's response
           - keypoints.size = keypoint diameter
           - keypoints.size() = length of keypoints vector

        */

        // calculate the mean detector response and the mean keypoint diameter over all remaining keypoints
        double meanDetectorResponse = 0.0;
        double meanKeypointDiam = 0.0;
        for (auto kPt = (dataBuffer.end() - 1)->keypoints.begin(); kPt < (dataBuffer.end() - 1)->keypoints.end(); kPt++)
        {
            meanDetectorResponse += kPt->response;
            meanKeypointDiam += kPt->size;
        }
        meanDetectorResponse /= (dataBuffer.end() - 1)->keypoints.size();
        meanKeypointDiam /= (dataBuffer.end() - 1)->keypoints.size();

        // calculate the keypoint diameter variance over all remaining keypoints
        double varianceKeypointDiam = 0.0;
        for (auto kPt = (dataBuffer.end() - 1)->keypoints.begin(); kPt < (dataBuffer.end() - 1)->keypoints.end(); kPt++)
        {
            varianceKeypointDiam += (kPt->size - meanKeypointDiam) * (kPt->size - meanKeypointDiam);
        }
        varianceKeypointDiam /= (dataBuffer.end() - 1)->keypoints.size();

        // output for debugging
        if (true) {
            cout << "meanDetectorResponse = " << meanDetectorResponse << endl;
            cout << "meanKeypointDiam = " << meanKeypointDiam << endl;
            cout << "varianceKeypointDiam = " << varianceKeypointDiam << endl;
        }

        // store the mean strength of the keypoint detector and the statistical neighborhood size in the last result ringbuffer element
        (resultBuffer.end() - 1)->meanDetectorResponse = meanDetectorResponse;
        (resultBuffer.end() - 1)->meanKeypointDiam = meanKeypointDiam;
        (resultBuffer.end() - 1)->varianceKeypointDiam = varianceKeypointDiam;

        /* EXTRACT KEYPOINT DESCRIPTORS */

        //// STUDENT ASSIGNMENT
        //// TASK MP.4 -> add the following descriptors in file matching2D.cpp and enable string-based selection based on descriptorType
        //// -> BRIEF, ORB, FREAK, AKAZE, SIFT

        // select keypoint descriptor extractor type
        //string descExtractorType = "BRIEF";  // options: "BRISK", "BRIEF", "ORB", "FREAK", "KAZE", "AKAZE", "SIFT", "SURF"

        // store selected keypoint descriptor type in the last result ringbuffer element
        (resultBuffer.end() - 1)->descExtractorType = descExtractorType;
        cout << "Seletect descriptor extractor tpye = " << descExtractorType << endl;

        // initialize descriptor matrix
        cv::Mat descriptors;

        // initialize processing time for keypoint descriptor extraction
        double t_descKeypoints = 0.0;

        // check if descriptor extractor and keypoint detector are

        if ( (detectorType == "SIFT") && (descExtractorType == "ORB") )
        { // skip the combination "SIFT" (keypoint detector) and "ORB" (descriptor extractor) due to memory allocation issues
            // inform user about current memory allocation problem using "SIFT" keypoint detector and "ORB"
            cout << "The keypoint detector type "
                << detectorType
                << " in combination with the descriptor extractor type "
                << descExtractorType
                << " causes memory allocation issues ..."
                << endl;
            cout << "... so this combination will be skipped" << endl;
        }
        else if (
            ((descExtractorType != "KAZE") && (descExtractorType != "AKAZE")) ||
            (((descExtractorType == "KAZE") || (descExtractorType == "AKAZE")) && ((detectorType == "KAZE") || (detectorType == "AKAZE")))
            )
        {  // "KAZE" and "AKAZE" descriptor extractors are only compatible with "KAZE" or "AKAZE" keypoints

            // extract keypoint descriptors
            try
            {
                // extract descriptors using user-specified descriptor extractor types
                t_descKeypoints = descKeypoints(
                    (dataBuffer.end() - 1)->keypoints, (dataBuffer.end() - 1)->cameraImg, descriptors, descExtractorType);
            }
            catch(const char *msg)
            {
                // show error message and return 1
                cout << msg << endl;
                return 1;
            }
            catch(const exception& e)
            {
                // show exeption and return 1
                cerr << e.what() << endl;
                return 1;
            }

        }
        else
        {
            // inform user about incompatibility of descriptor extractor and keypoint detector type
            cout << "No descriptor extraction possible ... as descriptor extractor type "
                << descExtractorType
                << " is not compatible with keypoint detector type "
                << detectorType
                << endl;
        }

        // store the processing time for keypoint descriptor extration in the last result ringbuffer element
        (resultBuffer.end() - 1)->t_descKeypoints = t_descKeypoints;

        // store the cumulated processing time for keypoint detection and descriptor extraction in the last result ringbuffer element
        (resultBuffer.end() - 1)->t_sum_det_desc = t_detKeypoints + t_descKeypoints;

        //// EOF STUDENT ASSIGNMENT

        // push descriptors for current frame to end of data buffer
        (dataBuffer.end() - 1)->descriptors = descriptors;

        // clear temporary variables
        descriptors.release();

        cout << "#3 : EXTRACT DESCRIPTORS done" << endl;

        /* MATCH KEYPOINT DESCRIPTORS */

        // Initialize processing time for keypoint descriptor matching
        double t_matchDescriptors = 0.0;

        // Match keypoint descriptors
        if (dataBuffer.size() > 1)  // wait until at least two images have been processed
        {
            // check if descriptor extractor and keypoint detector are compatible
            if ( (detectorType == "SIFT") && (descExtractorType == "ORB") )
            { // skip the combination "SIFT" (keypoint detector) and "ORB" (descriptor extractor) due to memory allocation issues
                // inform user about current memory allocation problem using "SIFT" keypoint detector and "ORB"
                cout << "The keypoint detector type "
                    << detectorType
                    << " in combination with the descriptor extractor type "
                    << descExtractorType
                    << " causes memory allocation issues ..."
                    << endl;
                cout << "... so this combination will be skipped" << endl;

                // store configuration in result buffer
                (resultBuffer.end() - 1)->matcherType = matcherType;     // store selected matcher type in result buffer
                (resultBuffer.end() - 1)->descriptorType = descriptorType;  // store selected descriptor type in result buffer
                (resultBuffer.end() - 1)->selectorType = selectorType;  // store selected selector type in result buffer
                (resultBuffer.end() - 1)->numDescMatches = -1;  // store number of descriptor matches in result buffer
            }
            else if (
                ((descExtractorType != "KAZE") && (descExtractorType != "AKAZE")) ||
                (((descExtractorType == "KAZE") || (descExtractorType == "AKAZE")) && ((detectorType == "KAZE") || (detectorType == "AKAZE")))
                )
            { // "KAZE" and "AKAZE" descriptor extractors are only compatible with "KAZE" or "AKAZE" keypoints

                // create vector of keypoint descriptor matches
                vector<cv::DMatch> matches;
                //string matcherType = "MAT_BF";            // MAT_BF, MAT_FLANN
                //string descriptorType = "DES_BINARY";     // DES_BINARY, DES_HOG
                //string selectorType = "SEL_KNN";          // SEL_NN, SEL_KNN

                // store selected matcher type in the last result ringbuffer element
                (resultBuffer.end() - 1)->matcherType = matcherType;
                cout << "Seletect descriptor matcher tpye = " << matcherType << endl;

                // store selected descriptor type in the last result ringbuffer element
                (resultBuffer.end() - 1)->descriptorType = descriptorType;
                cout << "Seletect descriptor tpye = " << descriptorType << endl;

                // store selected selector type in the last result ringbuffer element
                (resultBuffer.end() - 1)->selectorType = selectorType;
                cout << "Seletect selector tpye = " << selectorType << endl;

                //// STUDENT ASSIGNMENT
                //// TASK MP.5 -> add FLANN matching in file matching2D.cpp
                //// TASK MP.6 -> add KNN match selection and perform descriptor distance ratio filtering with t=0.8 in file matching2D.cpp

                // print out which images are used
                cout << "Matching keypoint descriptors between the last and the second last image stored in the ringbuffer:" << endl;
                cout << "Filename of last image in ringbuffer     = " << (dataBuffer.end() - 1)->imgFilename << endl;
                cout << "Filename of 2nd last image in ringbuffer = " << (dataBuffer.end() - 2)->imgFilename << endl;

                try
                {
                    // match keypoint descriptors between the last and the second last image stored in the ringbuffer
                    t_matchDescriptors = matchDescriptors(
                        (dataBuffer.end() - 2)->keypoints, (dataBuffer.end() - 1)->keypoints,
                        (dataBuffer.end() - 2)->descriptors, (dataBuffer.end() - 1)->descriptors,
                        matches, descriptorType, matcherType, selectorType);
                }
                catch(const exception& e)
                {
                    // show exeption and return 1
                    cerr << e.what() << endl;
                    return 1;
                }

                //// EOF STUDENT ASSIGNMENT

                // store matches in current data frame
                (dataBuffer.end() - 1)->kptMatches = matches;

                // store number of matches in the last result ringbuffer element
                (resultBuffer.end() - 1)->numDescMatches = matches.size();

                // clear temporary variables
                matches.clear();

                cout << "#4 : MATCH KEYPOINT DESCRIPTORS done" << endl;

                // visualize matches between the current and the previous image
                // bVis = true;
                if (bVis)
                {
                    // plot keypoint matches between the current and the previous image
                    cv::Mat matchImg = ((dataBuffer.end() - 1)->cameraImg).clone();
                    cv::drawMatches((dataBuffer.end() - 2)->cameraImg, (dataBuffer.end() - 2)->keypoints,
                                    (dataBuffer.end() - 1)->cameraImg, (dataBuffer.end() - 1)->keypoints,
                                    (dataBuffer.end() - 1)->kptMatches, matchImg,
                                    cv::Scalar::all(-1), cv::Scalar::all(-1),
                                    vector<char>(), cv::DrawMatchesFlags::DRAW_RICH_KEYPOINTS);
                    string windowName = "Matching keypoints between two camera images";
                    cv::namedWindow(windowName, 7);
                    cv::imshow(windowName, matchImg);

                    // wait for user key press
                    cout << "Press key to continue to next image" << endl;
                    cv::waitKey(0);  // wait for key to be pressed

                    // clear temporary images and variables
                    matchImg.release();
                    windowName.clear();
                }
                // bVis = false;

            }
            else
            {
                // inform user about incompatibility of "KAZE" descriptor extractor with keypoint detectors other than "KAZE" or "AKAZE"
                cout << "Descriptor extractor type "
                    << descExtractorType
                    << " is not compatible with keypoint detector type "
                    << detectorType
                    << endl;

                // store configuration in result buffer
                (resultBuffer.end() - 1)->matcherType = matcherType;     // store selected matcher type in result buffer
                (resultBuffer.end() - 1)->descriptorType = descriptorType;  // store selected descriptor type in result buffer
                (resultBuffer.end() - 1)->selectorType = selectorType;  // store selected selector type in result buffer
                (resultBuffer.end() - 1)->numDescMatches = -1;  // store number of descriptor matches in result buffer
            }
        }
        else
        {
            // store empty strings for the first image in result buffer
            (resultBuffer.end() - 1)->matcherType = "";     // store selected matcher type in result buffer
            (resultBuffer.end() - 1)->descriptorType = "";  // store selected descriptor type in result buffer
            (resultBuffer.end() - 1)->selectorType = "";    // store selected selector type in result buffer
            (resultBuffer.end() - 1)->numDescMatches = 0;   // store number of descriptor matches in result buffer
        }

        // store processing time for keypoint descriptor matching in the last result ringbuffer element
        (resultBuffer.end() - 1)->t_matchDescriptors = t_matchDescriptors; // is always zero for the first image

        // store the cumulated processing time for keypoint detection, descriptor extraction and descriptor matching
        // in the last result ringbuffer element
        (resultBuffer.end() - 1)->t_sum_det_desc_match = t_detKeypoints + t_descKeypoints + t_matchDescriptors;

    } // eof loop over all images

    // push current result buffer into the vector of result buffers
    evalResultBuffers.push_back(resultBuffer);

    /*
    // debugging
    cout << "Test access of last element in evalResultBuffers:" << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->detectorType << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->descExtractorType << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->matcherType << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->descriptorType << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->selectorType << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->numKeypoints << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->numKeypointsInROI << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->numKeypointsInROILimited << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->numDescMatches << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->t_detKeypoints << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->t_descKeypoints << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->t_sum_det_desc << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->t_matchDescriptors << endl;
    cout << ((*(evalResultBuffers.end() - 1)).end() - 1)->t_sum_det_desc_match << endl;

    cout << "Current size of evalResultsBuffers = " << evalResultBuffers.size() << endl;

    cout << "Press any key to continue ...";
    cin.get();
    */

    if (bExportResultsToCSV)
    {
        // define filepath and filename for result file
        string resultFilepath = dataPath + "results/";
        string resultFilename = "2D_feature_tracking_using_";
        string resultFiletype = ".csv";
        string resultFullFilename = resultFilepath
                                + resultFilename
                                + resultBuffer.begin()->detectorType
                                + "_and_"
                                + resultBuffer.begin()->descExtractorType
                                + resultFiletype;
        cout << "Export evaluation results to " << resultFullFilename << endl;

        // try to export the results to csv file
        try
        {
            // export results to csv file
            exportResultsToCSV(resultFullFilename, resultBuffer);
        }
        catch(const exception& e)
        {
            // show exeption and return 1
            cerr << e.what() << endl;
            return 1;
        }

        cout << "#5 : EXPORT RESULTS TO CSV FILE done" << endl;
    }

    // return 0 if program terminates without errors
    return 0;
}

/* MAIN PROGRAM */
int main(int argc, const char *argv[])
{
    /* SWITCH BETWEEN SINGLE RUN (USING MANUAL CONFIGURATION) AND BATCH RUN (EVALUATING DIFFERENT DETECTOR EXTRACTOR COMBINATIONS) */

    // Select batch run or single run mode
    bool BatchMode = false; // options: true => batch run; false => single run

    if (!BatchMode)
    {
        /* --- SINGLE RUN MODE --- */

        // print selected run mode
        cout << "Evaluation of 2D feature tracking in single run mode:" << endl;

        //// TASK MP.1 -> replace the following code with ring buffer of size dataBufferSize

        // ringbuffer implementation based on boost circular buffer, s. fun_2DFeatureTracking()

        // check installed boost version
        cout << "Using Boost version "
            << BOOST_VERSION / 100000     << "."  // major version
            << BOOST_VERSION / 100 % 1000 << "."  // minor version
            << BOOST_VERSION % 100                // patch level
            << endl;

        /* MANUAL CONFIGURATION FOR 2D FEATURE TRACKING STUDENT ASSIGNMENT */

        //// TASK MP.2 -> add the following keypoint detectors in file matching2D.cpp and enable string-based selection based on detectorType
        //// -> HARRIS, FAST, BRISK, ORB, AKAZE, SIFT

        // Select keypoint detector type
        string detectorType = "FAST";  // options: "SHITOMASI", "HARRIS", "FAST", "BRISK", "ORB", "KAZE", "AKAZE", "SIFT", "SURF"

        //// TASK MP.3 -> only keep keypoints on the preceding vehicle
        bool bFocusOnVehicle = true;  // true: only consider keypoints within the bounding box; false: consider all keypoints

        // optional : limit number of keypoints (helpful only for debugging and learning => Do not use in real application!)
        bool bLimitKpts = false;
        int maxKeypoints = 50;  // only for testing => Do not limit keypoints in real application!

        //// TASK MP.4 -> add the following descriptors in file matching2D.cpp and enable string-based selection based on descriptorType
        //// -> BRIEF, ORB, FREAK, AKAZE, SIFT

        // select keypoint descriptor extractor type ("KAZE" and "AKAZE" only work with "KAZE" or "AKAZE" keypoints)
        string descExtractorType = "BRIEF";  // options: "BRISK", "BRIEF", "ORB", "FREAK", "KAZE", "AKAZE", "SIFT", "SURF"

        //// TASK MP.5 -> add FLANN matching in file matching2D.cpp
        //// TASK MP.6 -> add KNN match selection and perform descriptor distance ratio filtering with t = 0.8 in file matching2D.cpp

        // select descriptor matcher tpye
        string matcherType = "MAT_BF";  // options: MAT_BF, MAT_FLANN

        // select descriptor type (use "DES_HOG" for "KAZE", "SIFT" and "SURF", otherwise use "DES_BINARY")
        string descriptorType = "DES_BINARY";  // options: DES_BINARY, DES_HOG

        // select selector type
        string selectorType = "SEL_KNN";  // SEL_NN, SEL_KNN

        // result visualization
        bool bVis = true;        // visualization of keypoint matching results
        bool bVisDebug = false;  // visualize intermediate results (for debugging or to look into details)

        // export evaluation results to csv file
        bool bExportResultsToCSV = true;

        // initialize (empty) vector of evaluation result buffers
        vector<boost::circular_buffer<EvalResults>> evalResultBuffers;

        // try to run 2D feature tracking in single run mode
        try
        {
            // evaluate 2D feature tracking performance
            if (fun_2DFeatureTracking(
                evalResultBuffers,
                detectorType,
                bFocusOnVehicle,
                bLimitKpts,
                maxKeypoints,
                descExtractorType,
                matcherType,
                descriptorType,
                selectorType,
                bVis,
                bVisDebug,
                bExportResultsToCSV) == 0)
            {
                // return 0 if program terminates without errors
                return 0;
            }
            else
            {
                // return 1 if program terminates with errors
                return 1;
            }
        }
        catch(const exception& e)
        {
            // show exeption and return 1
            cerr << e.what() << endl;
            return 1;
        }

    }
    else
    {
        /* --- BATCH RUN MODE --- */

        // print selected run mode
        cout << "Evaluation of 2D feature tracking in batch run mode:" << endl;

        // vector of keypoint detector types to evaluate
        vector<string> vec_detectorTypes = {"SHITOMASI", "HARRIS", "FAST", "BRISK", "ORB", "KAZE", "AKAZE", "SIFT", "SURF"};

        // only keep keypoints on the preceding vehicle
        bool bFocusOnVehicle = true;

        // optional : limit number of keypoints (helpful only for debugging and learning => Do not use in real application!)
        bool bLimitKpts = false;
        int maxKeypoints = 100;

        //// TASK MP.4 -> add the following descriptors in file matching2D.cpp and enable string-based selection based on descriptorType
        //// -> BRIEF, ORB, FREAK, AKAZE, SIFT

        // vector of keypoint descriptor extractor types to evaluate
        vector<string> vec_descExtractorTypes = {"BRISK", "BRIEF", "ORB", "FREAK", "KAZE", "AKAZE", "SIFT", "SURF"};

        //// TASK MP.5 -> add FLANN matching in file matching2D.cpp
        //// TASK MP.6 -> add KNN match selection and perform descriptor distance ratio filtering with t = 0.8 in file matching2D.cpp

        // set descriptor matcher tpye
        string matcherType = "MAT_BF";  // options: MAT_BF, MAT_FLANN

        // set binary descriptor type => will be automatically adapted to the descriptor extractor type in the loop over all combinations
        string descriptorType = "DES_BINARY";  // options: DES_BINARY, DES_HOG

        // set selector type
        string selectorType = "SEL_KNN";  // SEL_NN, SEL_KNN

        // result visualization
        bool bVis = false;       // visualization of keypoint matching results
        bool bVisDebug = false;  // visualize intermediate results (for debugging or to look into details)

        // export evaluation results to csv file
        bool bExportResultsToCSV = true;

        // vector of evaluation result buffers
        //int evalResultBuffersSize = 1;
        //vector<boost::circular_buffer<EvalResults>> evalResultBuffers(evalResultBuffersSize);
        vector<boost::circular_buffer<EvalResults>> evalResultBuffers;

        // iterator
        int itr = 0;

        for (vector<string>::const_iterator ptrDetType = vec_detectorTypes.begin(); ptrDetType != vec_detectorTypes.end(); ptrDetType++)
        {

            for (vector<string>::const_iterator ptrDescExtType = vec_descExtractorTypes.begin(); ptrDescExtType != vec_descExtractorTypes.end(); ptrDescExtType++)
            {

                // print current iterator
                cout << "\nIteration no. " << itr++ << endl;

                if ((*ptrDescExtType) == "KAZE" || (*ptrDescExtType) == "SIFT" || (*ptrDescExtType) == "SURF")
                {
                    // use gradient based descriptor type for SIFT and SURF
                    descriptorType = "DES_HOG";
                }
                else
                {
                    // use binary descriptor type for all other descriptor extractor types
                    descriptorType = "DES_BINARY";
                }

                // print current configuration
                cout << "\n" << "Next configuration for 2D feature tracking:" << endl;
                cout << "Feature detector type     = " << (*ptrDetType) << endl;
                cout << "Descriptor extractor type = " << (*ptrDescExtType) << endl;
                cout << "Matcher type              = " << matcherType << endl;
                cout << "Descriptor type           = " << descriptorType << endl;
                cout << "Selector type             = " << selectorType << "\n" << endl;

                // KADZE and AKAZE feature extractors only work with KAZE or AKAZE keypoints
                // => skip other configurations

                // try to run 2D feature tracking in batch run mode
                try
                {
                    // evaluate 2D feature tracking performance in batch mode
                    if (fun_2DFeatureTracking(
                        evalResultBuffers,
                        (*ptrDetType),
                        bFocusOnVehicle,
                        bLimitKpts,
                        maxKeypoints,
                        (*ptrDescExtType),
                        matcherType,
                        descriptorType,
                        selectorType,
                        bVis,
                        bVisDebug,
                        bExportResultsToCSV) == 0)
                    {
                        continue;
                    }
                    else
                    {
                        // return 1 if program terminates with errors
                        return 1;
                    }
                }
                catch(const exception& e)
                {
                    // show exeption and return 1
                    cerr << e.what() << endl;
                    return 1;
                }

                // wait for user key press
                string tmp;
                cout << "Press any key to continue: ";
                cin >> tmp;
                cout << "endl";

            }

        }

        // export overall results in an overview on all keypoint detector - descriptor extractor combinations to a csv file
        if (bExportResultsToCSV)
        {
            // define filepath and filename for result file
            string resultFilepath = dataPath + "results/";
            string resultFilename = "2D_feature_tracking_overall_results";
            string resultFiletype = ".csv";
            string resultFullFilename = resultFilepath
                                    + resultFilename
                                    + resultFiletype;
            cout << "Export overall evaluation results to " << resultFullFilename << endl;

            // try to export the results to csv file
            try
            {
                // export overall results to csv file
                exportOverallResultsToCSV(resultFullFilename, evalResultBuffers);
            }
            catch(const exception& e)
                {
                // show exeption and return 1
                cerr << e.what() << endl;
                return 1;
            }

            cout << "#6 : EXPORT OVERALL RESULTS TO CSV FILE done" << endl;

        }

        // return 0 if program terminates without errors
        return 0;

    }

}

ext: cpp


text:
/*
* Software License Agreement (BSD License)
*
*  Copyright (c) 2014-2015, Timm Linder, Social Robotics Lab, University of Freiburg
*  All rights reserved.
*
*  Redistribution and use in source and binary forms, with or without
*  modification, are permitted provided that the following conditions are met:
*
*  * Redistributions of source code must retain the above copyright notice, this
*    list of conditions and the following disclaimer.
*  * Redistributions in binary form must reproduce the above copyright notice,
*    this list of conditions and the following disclaimer in the documentation
*    and/or other materials provided with the distribution.
*  * Neither the name of the copyright holder nor the names of its contributors
*    may be used to endorse or promote products derived from this software
*    without specific prior written permission.
*
*  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
*  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
*  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
*  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
*  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
*  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
*  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
*  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
*  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
*  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

#include <ros/ros.h>
#include <pcl_ros/point_cloud.h>
#include <sensor_msgs/PointCloud2.h>
#include <string>
#include <cstdio>

#include "../features.h"
#include "../volume.h"

using namespace std;

#include <boost/random.hpp>
#include <boost/random/normal_distribution.hpp>
#include <boost/random/uniform_real.hpp>
#include <boost/foreach.hpp>
#define foreach BOOST_FOREACH

///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void sampleSphere(PointCloud::Ptr& cloud, const Eigen::Vector3d& origin, double radius, int pointCount)
{
    boost::mt19937 rng;
    boost::normal_distribution<> nd(0.0, 1.0);
    boost::variate_generator<boost::mt19937&, boost::normal_distribution<> > var_nor(rng, nd);

    for(int i = 0; i < pointCount; i++) {
        Eigen::Vector3d v(var_nor(), var_nor(), var_nor());
        Eigen::Vector3d pointOnSphere = radius / v.norm() * v;
        cloud->points.push_back( PointType(pointOnSphere(0), pointOnSphere(1), pointOnSphere(2)) );
    }
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void samplePlane(PointCloud::Ptr& cloud, const Eigen::Vector3d& normalVector, double distanceToOrigin, int pointCount, double scale1 = 1.0, double scale2 = 1.0)
{
    boost::mt19937 rng;
    boost::variate_generator<boost::mt19937&, boost::uniform_real<> > uni1(rng, boost::uniform_real<>(-0.2 * scale1, +0.2 * scale1));
    boost::variate_generator<boost::mt19937&, boost::uniform_real<> > uni2(rng, boost::uniform_real<>(-0.2 * scale2, +0.2 * scale2));

    for(int i = 0; i < pointCount; i++) {
        double x0 = uni1();
        double x1 = uni2();
        double x2 = (-distanceToOrigin - normalVector(0)*x0 - normalVector(1)*x1) / normalVector(2);

        Eigen::Vector3d pointOnPlane(x0, x1, x2);
        cloud->points.push_back( PointType(pointOnPlane(0), pointOnPlane(1), pointOnPlane(2)) );
    }
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
int main(int argc, char **argv)
{
    ros::init(argc, argv, "test_features");
    ros::NodeHandle nodeHandle("");
    ros::NodeHandle privateHandle("~");

    // Create feature calculator
    FeatureCalculator featureCalculator;
    const std::vector<string>& featureNames = featureCalculator.getFeatureNames();

    pcl::PointXYZ minCoords(-0.3, -0.3, 0), maxCoords(0.3, 0.3, 1.8);
    Volume parentVolume(minCoords, maxCoords);

    // Create point cloud publisher
    ros::Publisher pointCloudPublisher = nodeHandle.advertise<sensor_msgs::PointCloud2>("cloud", 1, true);

    for(int testCase = 0; testCase < 7 && ros::ok(); testCase++) {
        PointCloud::Ptr testCloud(new PointCloud);
        testCloud->header.stamp = ros::Time::now().toNSec() / 1000;
        testCloud->header.frame_id = "extracted_cloud_frame";

        switch(testCase)
        {
            case 0:
            {
                ROS_INFO("=== SMALL SPHERE ===");
                Eigen::Vector3d origin(1, 2, 3);
                sampleSphere(testCloud, origin, 0.2, 100);
                break;
            }
            case 1:
            {
                ROS_INFO("=== LARGE SPHERE ===");
                Eigen::Vector3d origin(0, 0, 0);
                sampleSphere(testCloud, origin, 0.4, 500);
                break;
            }
            case 2:
            {
                ROS_INFO("=== EXTRA-LARGE SPHERE ===");
                Eigen::Vector3d origin(1, 1, 0);
                sampleSphere(testCloud, origin, 0.8, 500);
                break;
            }

            case 3:
            {
                ROS_INFO("=== FIRST PLANE ===");
                Eigen::Vector3d normalVector(0.9, 0, 0.3);
                samplePlane(testCloud, normalVector.normalized(), 0.1, 100);
                break;
            }
            case 4:
            {
                ROS_INFO("=== SECOND PLANE ===");
                Eigen::Vector3d normalVector(0.2, 0, 0.5);
                samplePlane(testCloud, normalVector.normalized(), -1.0, 500);
                break;
            }
            case 5:
            {
                ROS_INFO("=== THIRD PLANE ===");
                Eigen::Vector3d normalVector(0, 0, 1);
                samplePlane(testCloud, normalVector.normalized(), 0.0, 200);
                break;
            }
            case 6:
            {
                ROS_INFO("=== FOURTH PLANE ===");
                Eigen::Vector3d normalVector(0, 0, 1);
                samplePlane(testCloud, normalVector.normalized(), 0.0, 200, 0.1, 10.0);
                break;
            }
        }

        std::vector<int> indices;
        for(int i = 0; i < testCloud->points.size(); i++) indices.push_back(i);

        std::vector<double> featureVector;
        featureCalculator.calculateFeatures(parentVolume, *testCloud, indices, featureCalculator.maskAllFeaturesActive(), featureVector);

        for(int f = 0; f < featureVector.size(); f++) {
            ROS_INFO("Feature %s:  %g", featureNames[f].c_str(), featureVector[f]);
        }

        pointCloudPublisher.publish(testCloud);
        ros::spinOnce();

        ROS_INFO("Press ENTER to show next test case...");
        getchar();
    }

    return 0;
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////


text:
#include "clientversion.h"
#include "primitives/transaction.h"
#include "random.h"
#include "sync.h"
#include "utilstrencodings.h"
#include "utilmoneystr.h"
#include "test/test_bitcoin.h"

#include <stdint.h>
#include <vector>

#include "chainparams.h"
#include "consensus/consensus.h"
#include "consensus/validation.h"
#include "key.h"
#include "validation.h"
#include "miner.h"
#include "pubkey.h"
#include "random.h"
#include "txdb.h"
#include "txmempool.h"
#include "ui_interface.h"
#include "rpc/server.h"
#include "rpc/register.h"
#include "zerocoin.h"
#include "sigma.h"

#include "test/fixtures.h"
#include "test/testutil.h"

#include "wallet/db.h"
#include "wallet/wallet.h"

#include <boost/filesystem.hpp>
#include <boost/test/unit_test.hpp>
#include <boost/thread.hpp>

BOOST_FIXTURE_TEST_SUITE(zerocoin_tests2_v3, ZerocoinTestingSetup200)

BOOST_AUTO_TEST_CASE(zerocoin_mintspend2_v3)
{
    sigma::CSigmaState *sigmaState = sigma::CSigmaState::GetState();

    //200 blocks already mined, create another 350. See Params::nSigmaPaddingBlock

    CreateAndProcessEmptyBlocks(350, scriptPubKey);

    std::vector<string> denominations = {"0.1", "0.5", "1"};
    for(string denomination : denominations) {
        string stringError;
        //Make sure that transactions get to mempool
        pwalletMain->SetBroadcastTransactions(true);

        //Block 201 create 5 mints
        //Verify Mint is successful
        for(int i = 0; i < 5; i++)
        {
            BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinMintModel(stringError, denomination.c_str(), SIGMA), stringError + " - Create Mint failed");
        }

        //Put 5 in the same block
        BOOST_CHECK_MESSAGE(mempool.size() == 5, "Mints were not added to mempool");

        int previousHeight = chainActive.Height();
        CBlock b = CreateAndProcessBlock(scriptPubKey);
        BOOST_CHECK_MESSAGE(previousHeight + 1 == chainActive.Height(), "Block not added to chain");
        BOOST_CHECK_MESSAGE(mempool.size() == 0, "Mempool not cleared");

        //Block 111, put 6 mints
        for(int i = 0; i < 6; i++)
            BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinMintModel(stringError, denomination.c_str(), SIGMA), stringError + " - Create Mint failed");

        //Put 6 in the same block
        BOOST_CHECK_MESSAGE(mempool.size() == 6, "Mints were not added to mempool");

        previousHeight = chainActive.Height();
        b = CreateAndProcessBlock(scriptPubKey);
        BOOST_CHECK_MESSAGE(previousHeight + 1 == chainActive.Height(), "Block not added to chain");
        BOOST_CHECK_MESSAGE(mempool.size() == 0, "Mempool not cleared");

        for (int i = 0; i < 5; i++)
        {
            CBlock b = CreateAndProcessBlock(scriptPubKey);
        }

        //Block 117, put 10 mints and one spend
        for(int i = 0; i < 10; i++)
            BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinMintModel(stringError, denomination.c_str(), SIGMA), stringError + " - Create Mint failed");
        BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinSpendModel(stringError, "", denomination.c_str()), "Spend failed");

        //Put 11 in the same block
        BOOST_CHECK_MESSAGE(mempool.size() == 11, "Mints were not added to mempool");

        previousHeight = chainActive.Height();
        b = CreateAndProcessBlock(scriptPubKey);
        BOOST_CHECK_MESSAGE(previousHeight + 1 == chainActive.Height(), "Block not added to chain");
        BOOST_CHECK_MESSAGE(mempool.size() == 0, "Mempool not cleared");

        //20 spends in 20 blocks
        for(int i = 0; i < 20; i++) {

            BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinSpendModel(stringError, "", denomination.c_str()), "Spend failed");
            BOOST_CHECK_MESSAGE(mempool.size() == 1, "Spends were not added to mempool");
            previousHeight = chainActive.Height();
            b = CreateAndProcessBlock(scriptPubKey);
            BOOST_CHECK_MESSAGE(previousHeight + 1 == chainActive.Height(), "Block not added to chain");
            BOOST_CHECK_MESSAGE(mempool.size() == 0, "Mempool not cleared");
        }

        //Put 19 mints
        for(int i = 0; i < 19; i++)
            BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinMintModel(stringError, denomination.c_str(), SIGMA), stringError + " - Create Mint failed");

        //Put 19 in the same block
        BOOST_CHECK_MESSAGE(mempool.size() == 19, "Mints were not added to mempool");

        previousHeight = chainActive.Height();
        b = CreateAndProcessBlock(scriptPubKey);
        BOOST_CHECK_MESSAGE(previousHeight + 1 == chainActive.Height(), "Block not added to chain");
        BOOST_CHECK_MESSAGE(mempool.size() == 0, "Mempool not cleared");

        for (int i = 0; i < 5; i++)
        {
            CBlock b = CreateAndProcessBlock(scriptPubKey);
        }

        //19 spends in 19 blocks
        for(int i = 0; i < 19; i++) {
            BOOST_CHECK_MESSAGE(pwalletMain->CreateZerocoinSpendModel(stringError, "", denomination.c_str()), "Spend failed");
            BOOST_CHECK_MESSAGE(mempool.size() == 1, "Spends were not added to mempool");
            previousHeight = chainActive.Height();
            b = CreateAndProcessBlock(scriptPubKey);
            BOOST_CHECK_MESSAGE(previousHeight + 1 == chainActive.Height(), "Block not added to chain");
            BOOST_CHECK_MESSAGE(mempool.size() == 0, "Mempool not cleared");
        }
    }

    mempool.clear();
    sigmaState->Reset();
}

BOOST_AUTO_TEST_SUITE_END()

max_stars_repo_name: MintPond/coin-zcoin

max_issues_repo_name: MintPond/coin-zcoin

max_forks_repo_name: MintPond/coin-zcoin


text:
/*
 * Auth.cpp
 *
 *  Created on: Apr 24, 2016
 *      Author: James Hong
 */

#include <controller/access/DynamicAuth.h>

#define BOOST_NETWORK_ENABLE_HTTPS

#include <boost/network/protocol/http/client.hpp>
#include <boost/network/protocol/http/request.hpp>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <cstdint>
#include <exception>
#include <iostream>
#include <sstream>

#include "controller/ControllerClient.h"
#include "Debug.h"
#include "Device.h"
#include "device/socket/tcp/TCPClient.h"

DynamicAuth::DynamicAuth(rule_t ruleId_) {
	ruleId = ruleId_;
}

void DynamicAuth::reset() {
	state = UNATTEMPTED;
}

static bool isPrivateAddress(uint32_t ip) {
	uint8_t b1, b2;
	b1 = (uint8_t) (ip >> 24);
	b2 = (uint8_t) ((ip >> 16) & 0x0ff);
	if (b1 == 10) {
		return true;
	}
	if ((b1 == 172) && (b2 >= 16) && (b2 <= 31)) {
		return true;
	}
	if ((b1 == 192) && (b2 == 168)) {
		return true;
	}
	return false;
}

NetworkAuth::NetworkAuth(rule_t r, std::string ip_, bool isPrivate_) :
		DynamicAuth(r) {
	isPrivate = isPrivate_;
	ip = ip_;
}

void NetworkAuth::evaluate(std::shared_ptr<ControllerClient> cc, std::shared_ptr<Device> from,
		std::shared_ptr<Device> to) {
	auto tcpTo = std::dynamic_pointer_cast<TCPClient>(to);
	if (tcpTo) {
		struct sockaddr_in addr = tcpTo->getSockaddr();
		if (isPrivate) {
			state = isPrivateAddress(ntohl(addr.sin_addr.s_addr)) ? SATISFIED : DENIED;
		} else {
			struct in_addr toAddr;
			inet_aton(ip.c_str(), &toAddr);
			state = (addr.sin_addr.s_addr == toAddr.s_addr) ? SATISFIED : DENIED;
		}
	} else {
		state = SATISFIED;
	}
	if (debug_controller) {
		std::stringstream ss;
		ss << "evaluated network auth : " << ((state == SATISFIED) ? "satisfied" : "denied");
		pdebug(ss.str());
	}
}

PasscodeAuth::PasscodeAuth(rule_t r) :
		DynamicAuth(r) {

}

void PasscodeAuth::evaluate(std::shared_ptr<ControllerClient> cc, std::shared_ptr<Device> from,
		std::shared_ptr<Device> to) {
	if (state == SATISFIED) {
		return;
	}

	/*
	 * Must be an endpoint, not a device created by Beetle.
	 */
	switch (to->getType()) {
	case Device::LE_PERIPHERAL:
	case Device::IPC_APPLICATION:
	case Device::TCP_CLIENT: {
		std::stringstream resource;
		resource << "state/passcode/isLive/" << std::fixed << ruleId << "/" << cc->getName() << "/" << std::fixed
				<< to->getId();
		std::string url = cc->getApiUrl(resource.str());

		using namespace boost::network;
		http::client::request request(url);
		request << header("User-Agent", "linux");

		try {
			auto response = cc->getClient()->get(request);
			if (response.status() == 200) {
				std::stringstream ss;
				ss << body(response);
				expire = static_cast<time_t>(std::stod(ss.str()));
				state = SATISFIED;
			} else {
				state = UNATTEMPTED;
			}
		} catch (std::exception &e) {
			pexcept(e);
			state = DENIED;
		}
		break;
	}
	default:
		state = SATISFIED;
	}

	if (debug_controller) {
		std::stringstream ss;
		ss << "evaluated passcode auth : " << ((state == SATISFIED) ? "satisfied" : "not satisfied");
		pdebug(ss.str());
	}
}

AdminAuth::AdminAuth(rule_t r) :
		DynamicAuth(r) {

}

void AdminAuth::evaluate(std::shared_ptr<ControllerClient> cc, std::shared_ptr<Device> from,
		std::shared_ptr<Device> to) {
	if (state == SATISFIED) {
		return;
	}

	if (state == UNATTEMPTED) {
		switch (to->getType()) {
		case Device::LE_PERIPHERAL:
		case Device::IPC_APPLICATION:
		case Device::TCP_CLIENT: {
			std::stringstream resource;
			resource << "state/admin/request/" << std::fixed << ruleId << "/" << cc->getName() << "/" << std::fixed
					<< to->getId();
			std::string url = cc->getApiUrl(resource.str());

			using namespace boost::network;
			http::client::request request(url);
			request << header("User-Agent", "linux");

			try {
				auto response = cc->getClient()->post(request);
				if (response.status() == 202) {
					std::stringstream ss;
					ss << body(response);
					expire = static_cast<time_t>(std::stod(ss.str()));
					state = SATISFIED;
				} else {
					state = DENIED;
				}
			} catch (std::exception &e) {
				pexcept(e);
				state = DENIED;
			}
			break;
		}
		default:
			state = SATISFIED;
		}
	}

	if (debug_controller) {
		std::stringstream ss;
		ss << "evaluated admin auth : " << ((state == SATISFIED) ? "satisfied" : "not satisfied");
		pdebug(ss.str());
	}
}

UserAuth::UserAuth(rule_t r) :
		DynamicAuth(r) {

}

void UserAuth::evaluate(std::shared_ptr<ControllerClient> cc, std::shared_ptr<Device> from,
		std::shared_ptr<Device> to) {
	if (state == SATISFIED) {
		return;
	}

	if (state == UNATTEMPTED) {
		switch (to->getType()) {
		case Device::LE_PERIPHERAL:
		case Device::IPC_APPLICATION:
		case Device::TCP_CLIENT: {
			std::stringstream resource;
			resource << "state/user/request/" << std::fixed << ruleId << "/" << cc->getName() << "/" << std::fixed
					<< to->getId();
			std::string url = cc->getApiUrl(resource.str());

			using namespace boost::network;
			http::client::request request(url);
			request << header("User-Agent", "linux");

			try {
				auto response = cc->getClient()->post(request);
				if (response.status() == 202) {
					std::stringstream ss;
					ss << body(response);
					expire = static_cast<time_t>(std::stod(ss.str()));
					state = SATISFIED;
				} else {
					state = DENIED;
				}
			} catch (std::exception &e) {
				pexcept(e);
				state = DENIED;
			}
			break;
		}
		default:
			state = SATISFIED;
		}
	}

	if (debug_controller) {
		std::stringstream ss;
		ss << "evaluated user auth : " << ((state == SATISFIED) ? "satisfied" : "not satisfied");
		pdebug(ss.str());
	}
}

max_stars_repo_name: helena-project/beetle

max_issues_repo_name: helena-project/beetle

max_forks_repo_name: helena-project/beetle


text:
#include "app_delegate.hpp"
#include <boost/program_options.hpp>
#include <sstream>
#include <iostream>
sgns::AppDelegate g_app_delegate;
int main (int argc, char * const * argv)
{
	std::cout << "--------------main()---------------" << std::endl;
    int result (0);
    g_app_delegate.init(argc, argv);
    // boost::program_options::variables_map vm;
    // auto data_path_it = vm.find ("data_path");
	// if (data_path_it == vm.end ())
	// {
	// 	std::string error_string;
	// 	if (!sgns::migrate_working_path (error_string))
	// 	{
	// 		std::cerr << error_string << std::endl;

	// 		return 1;
	// 	}
	// }
    // boost::filesystem::path data_path ((data_path_it != vm.end ()) ? data_path_it->second.as<std::string> () : sgns::working_path ());
    g_app_delegate.run(/*data_path*/);
    g_app_delegate.exit();
    return result;
}

max_stars_repo_name: GeniusVentures/SuperGenius

max_issues_repo_name: GeniusVentures/SuperGenius

max_forks_repo_name: GeniusVentures/SuperGenius