Blog article sources

article.cpp

/* compile with
 *    g++ -std=c++11 article.cpp -I ../include/ -Wall -pedantic -L ../src/.libs/ -lupscaledb
 *
 * run with
 *    LD_LIBRARY_PATH=../src/.libs ./a.out
 */

#include <iostream>
#include <chrono>
#include <functional>
#include <stdlib.h> // for ::exit

#include <ups/upscaledb.h>

// A dummy for our time series values
struct TsValue {
  char data[64];
};

// Our database will have the name "1"
enum DatabaseIds : uint16_t {
  kDbId = 1
};

static void
handle_error(const char *function_name, ups_status_t status)
{
  std::cerr << function_name << " failed with status " << status
            << " (" << ups_strerror(status) << ")" << std::endl;
  ::exit(-1);
}

static uint64_t
nanoseconds()
{
  return (std::chrono::high_resolution_clock::now().time_since_epoch() / 
            std::chrono::nanoseconds(1));
}

static void
benchmark(const char *description, ups_db_t *db,
            const std::function<void (ups_db_t *)> &foo)
{
  auto t1 = std::chrono::high_resolution_clock::now();
  foo(db);
  auto t2 = std::chrono::high_resolution_clock::now();
  std::cout << description << ": "
            << std::chrono::duration <double, std::milli>(t2 - t1).count()
            << " ms" << std::endl;
}

static void
add_time_series_event(ups_db_t *db)
{
  // Store timestamps in nanonsecond resolution
  uint64_t now = nanoseconds();

  // Our value is just a placeholder for our example. A real application
  // would obviously use real data here.
  TsValue value = {0};

  ups_key_t key = ups_make_key(&now, sizeof(now));
  ups_record_t rec = ups_make_record(&value, sizeof(value));

  // Now insert the key/value pair
  ups_status_t st = ups_db_insert(db, 0, &key, &rec, 0);
  if (st != UPS_SUCCESS)
    handle_error("ups_db_insert", st);
}

static void
analyze_time_series(ups_db_t *db)
{
  // Analyzing time series data usually means to read and process the
  // data from a certain time window. Our window will be the last 0.1 seconds
  // that were stored. Create a cursor and locate it on a
  // key at "now - 0.1 seconds".
  uint64_t start_time = nanoseconds() - (1000000000 / 10);

  ups_key_t key = ups_make_key(&start_time, sizeof(start_time));
  ups_record_t rec = {0};

  // Create a new database cursor
  ups_cursor_t *cursor;
  ups_status_t st = ups_cursor_create(&cursor, db, 0, 0);
  if (st != UPS_SUCCESS)
    handle_error("ups_cursor_create", st);

  // Locate a key/value pair with a timestamp about 0.1 sec ago
  st = ups_cursor_find(cursor, &key, &rec, UPS_FIND_GEQ_MATCH);
  if (st != UPS_SUCCESS)
    handle_error("ups_cursor_find", st);

  int count = 0;
  do {
    // Process the key/value pair; we just count them
    count++;

    // And move to the next key, till we reach "the end" of the database
    st = ups_cursor_move(cursor, &key, &rec, UPS_CURSOR_NEXT);
    if (st != UPS_SUCCESS && st != UPS_KEY_NOT_FOUND)
      handle_error("ups_cursor_move", st);
  } while (st == 0);

  // Clean up
  ups_cursor_close(cursor);

  std::cout << "In the last 0.1 seconds, " << count << " events were inserted"
            << std::endl;
}

int
main()
{
  ups_status_t st;
  ups_env_t *env;          // upscaledb environment object
  ups_db_t *db;            // upscaledb database object

  // First create a new Environment (filename is "timeseries.db")
  st = ups_env_create(&env, "timeseries.db", 0, 0, 0);
  if (st != UPS_SUCCESS)
    handle_error("ups_env_create", st);

  // parameters for the new database: 64bit numeric keys, fixed length records
  ups_parameter_t db_params[] = {
    {UPS_PARAM_KEY_TYPE, UPS_TYPE_UINT64},
    {UPS_PARAM_RECORD_SIZE, sizeof(TsValue)},
    {0, }
  };

  // Then create a new Database in this Environment
  st = ups_env_create_db(env, &db, kDbId, 0, &db_params[0]);
  if (st != UPS_SUCCESS)
    handle_error("ups_env_create_db", st);

  // We will perform our work in here
  // ...
  // First we will simulate incoming time series data. This could be events
  // from a sensor, from a network of machines or visitor data from a
  // web server.
  benchmark("Inserting 1 mio events", db, [] (ups_db_t *db) {
          for (uint64_t i = 0; i < 1000000; i++)
            add_time_series_event(db);
        });

  // Now let's analyze our time series data. We could calculate highs and
  // lows, but our code will simply print the values from the last 0.1 seconds.
  benchmark("Analyzing 0.1 sec of data", db, [] (ups_db_t *db) {
            analyze_time_series(db);
        });

  // Close the Environment before the program terminates. The flag
  // UPS_AUTO_CLEANUP will automatically close all databases and related
  // objects for us.
  st = ups_env_close(env, UPS_AUTO_CLEANUP);
  if (st)
    handle_error("ups_env_close", st);
  
  return (0);
}