srvdsapr_modified.cpp

#include <future>
#include <thread>
#include <mutex>
#include <iostream>
#include <list>
#include <exception>
#include <sstream>
#include <cassandra.h>


using namespace std;
using namespace std::chrono;

namespace {

const auto hosts = string("127.0.0.1");
const auto port = int(9042);
const auto num_concurrent = size_t(20);
const auto req_timeout = milliseconds(2000);
const auto update_timeout = milliseconds(1000);

mutex mtx_cerr;

string get_future_error_text(CassFuture* future) {
    const char* message;
    size_t message_length;
    cass_future_error_message(future, &message, &message_length);
    return string(message, message_length);
}

enum class EC {
    SUCCESS,
    TIMEOUT,
    ERROR
};

EC execute_test_query(CassSession* session) {
    EC res = EC::SUCCESS;
    std::string error;

    CassStatement* statement = nullptr;
    CassFuture* result_future = nullptr;

    try {
        statement = cass_statement_new("SELECT keyspace_name, columnfamily_name FROM system.schema_columnfamilies;", 0);
        if (!statement) {
            throw runtime_error("Unable to create a statement (cass_statement_new)");
        }

        result_future = cass_session_execute(session, statement);
        if (!result_future) {
            throw runtime_error("Unable to execute, no future is returned (cass_session_execute)");
        }

        if (cass_future_wait_timed(result_future, req_timeout.count() * 1000) == cass_false) {
            res = EC::TIMEOUT;
        }
        else {
            if (cass_future_error_code(result_future) != CASS_OK) {
                error = "Unable to run query: " +  get_future_error_text(result_future);
                // Added this. You need to throw out of here, otherwise
                // cass_future_get_result() will return NULL.
                throw runtime_error(error);
            }

            const CassResult* result = cass_future_get_result(result_future);

            CassIterator* rows = cass_iterator_from_result(result);

            while(cass_iterator_next(rows)) {
                const CassRow* row = cass_iterator_get_row(rows);
                const CassValue* value = cass_row_get_column_by_name(row, "keyspace_name");
                const char* keyspace;
                size_t keyspace_length;
                cass_value_get_string(value, &keyspace, &keyspace_length);
            }

            cass_result_free(result);
            cass_iterator_free(rows);
        }
    }
    catch (std::exception& ex) {
        error = ex.what();
        res = EC::ERROR;
    }
    catch (...) {
        error = "unknown";
        res = EC::ERROR;
    }

    if (statement) {
        cass_statement_free(statement);
    }

    if (result_future) {
        cass_future_free(result_future);
    }

    if (!error.empty()) {
        lock_guard<decltype(mtx_cerr)> lock(mtx_cerr);
        std::cerr << error << std::endl;
    }

    return res;
}

}

int main(int argc, char* argv[]) {

    int ret = 0;

    CassCluster* cluster = nullptr;
    CassSession* session = nullptr;
    CassFuture* connect_future = nullptr;

    try {
        cout << "Creating cassandra cluster..." << endl;
        CassCluster* cluster = cass_cluster_new();
        if (cluster == nullptr) {
            throw runtime_error("Can't create the Cassandra cluster (cass_cluster_new)");
        }

        cout << "Set contact points..." << endl;
        cass_cluster_set_contact_points(cluster, hosts.c_str());

        //cout << "Set request timeout " <<  req_timeout << endl;
        //cass_cluster_set_request_timeout(cluster, req_timeout.count());

        cout << "Creating cassandra session..." << endl;
        auto session = cass_session_new();
        if (session == nullptr) {
            throw runtime_error("Can't create the Cassandra session (cass_session_new)");
        }

        cout << "Connecting cassandra session..." << endl;
        connect_future = cass_session_connect(session, cluster);
        if (!connect_future) {
            throw runtime_error("Can't create connect (cass_session_connect)");
        }

        auto er = cass_future_error_code(connect_future);
        if (er != CASS_OK) {
            throw runtime_error("Session 'connect' error: " + get_future_error_text(connect_future));
        }

        cass_future_free(connect_future);


        struct task_info {
            std::future<EC> f;
            system_clock::time_point started;
        };

        std::list<task_info> tasks;

        cout << "Running " << num_concurrent << " working async tasks requesting Cassandra..." << endl;
        while(true) {

            while(tasks.size() < num_concurrent) {
                task_info ti = { std::async(execute_test_query, session), system_clock::now() };
                tasks.emplace_back(move(ti));
            }

            this_thread::sleep_for(update_timeout);

            size_t finished_ok = 0;
            size_t finished_timeout = 0;
            size_t finished_error = 0;
            size_t continued_ok = 0;
            size_t continued_toolong = 0;

            for (auto it_ti = tasks.begin(); it_ti != tasks.end(); ) {

                if (it_ti->f.valid()) {
                    const auto ended = it_ti->f.get();
                    if (ended == EC::SUCCESS) {
                        ++finished_ok;
                    }
                    else if (ended == EC::TIMEOUT) {
                        ++finished_timeout;
                    }
                    else if (ended == EC::ERROR) {
                        ++finished_error;
                    }

                    it_ti = tasks.erase(it_ti);
                }
                else {
                    const auto now = system_clock::now();
                    if (duration_cast<milliseconds>(now - it_ti->started) > milliseconds(req_timeout) + milliseconds(1000)) {
                        ++continued_ok;
                    }
                    else {
                        ++continued_toolong;
                    }
                    ++it_ti;
                }
            }

            //const auto now = system_clock::now();

            cout << "Sucessfully completed: " << finished_ok << "\n";
            cout << "Stopped with timeout:  " << finished_timeout << "\n";
            cout << "Stopped with error:    " << finished_error << "\n";
            cout << "Working:               " << continued_ok << "\n";
            cout << "Stuck:                 " << continued_toolong << endl;
        }

    }
    catch(std::exception& e) {
        lock_guard<decltype(mtx_cerr)> lock(mtx_cerr);
        cerr << "Exception: " << e.what() << "\n";
        ret = -1;
    }
    catch(...) {
        lock_guard<decltype(mtx_cerr)> lock(mtx_cerr);
        cerr << "Unknown exception" << "\n";
        ret = -1;
    }

    cout << "Deleting the session..." << endl;
    if (session) {
        auto close_future = cass_session_close(session);
        cass_future_wait(close_future);
        cass_future_free(close_future);
        cass_session_free(session);
    }

    if (cluster) {
        cass_cluster_free(cluster);
    }

    return ret;
}