// ------------------------------------------------------------------- // // eleveldb: Erlang Wrapper for LevelDB (http://code.google.com/p/leveldb/) // // Copyright (c) 2011-2013 Basho Technologies, Inc. All Rights Reserved. // // This file is provided to you 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. // // ------------------------------------------------------------------- #ifndef __EROCKSDB_DETAIL_HPP #include "detail.hpp" #endif #ifndef INCL_WORKITEMS_H #include "workitems.h" #endif #ifndef INCL_UTIL_H #include "util.h" #endif #ifndef ATOMS_H #include "atoms.h" #endif #include "rocksdb/cache.h" #include "rocksdb/filter_policy.h" namespace erocksdb { /** * WorkTask functions */ WorkTask::WorkTask(ErlNifEnv *caller_env, ERL_NIF_TERM& caller_ref) : terms_set(false), resubmit_work(false) { if (NULL!=caller_env) { local_env_ = enif_alloc_env(); caller_ref_term = enif_make_copy(local_env_, caller_ref); caller_pid_term = enif_make_pid(local_env_, enif_self(caller_env, &local_pid)); terms_set=true; } // if else { local_env_=NULL; terms_set=false; } // else return; } // WorkTask::WorkTask WorkTask::WorkTask(ErlNifEnv *caller_env, ERL_NIF_TERM& caller_ref, DbObject * DbPtr) : m_DbPtr(DbPtr), terms_set(false), resubmit_work(false) { if (NULL!=caller_env) { local_env_ = enif_alloc_env(); caller_ref_term = enif_make_copy(local_env_, caller_ref); caller_pid_term = enif_make_pid(local_env_, enif_self(caller_env, &local_pid)); terms_set=true; } // if else { local_env_=NULL; terms_set=false; } // else return; } // WorkTask::WorkTask WorkTask::WorkTask(ErlNifEnv *caller_env, ERL_NIF_TERM& caller_ref, DbObject * DbPtr, ColumnFamilyObject * CfPtr) : m_DbPtr(DbPtr), m_CfPtr(CfPtr), terms_set(false), resubmit_work(false) { if (NULL!=caller_env) { local_env_ = enif_alloc_env(); caller_ref_term = enif_make_copy(local_env_, caller_ref); caller_pid_term = enif_make_pid(local_env_, enif_self(caller_env, &local_pid)); terms_set=true; } // if else { local_env_=NULL; terms_set=false; } // else return; } // WorkTask::WorkTask WorkTask::~WorkTask() { ErlNifEnv * env_ptr; // this is likely overkill in the present code, but seemed // important at one time and leaving for safety env_ptr=local_env_; if (compare_and_swap(&local_env_, env_ptr, (ErlNifEnv *)NULL) && NULL!=env_ptr) { enif_free_env(env_ptr); } // if return; } // WorkTask::~WorkTask void WorkTask::prepare_recycle() { // does not work by default resubmit_work=false; } // WorkTask::prepare_recycle void WorkTask::recycle() { // does not work by default } // WorkTask::recycle /** * OpenTask functions */ OpenTask::OpenTask( ErlNifEnv* caller_env, ERL_NIF_TERM& _caller_ref, const std::string& db_name_, rocksdb::Options *Options_) : WorkTask(caller_env, _caller_ref), db_name(db_name_), options(Options_) { } // OpenTask::OpenTask work_result OpenTask::operator()() { DbObject * db_ptr; rocksdb::DB *db(0); rocksdb::Status status = rocksdb::DB::Open(*options, db_name, &db); if(!status.ok()) return error_tuple(local_env(), ATOM_ERROR_DB_OPEN, status); db_ptr=DbObject::CreateDbObject(db, options); // create a resource reference to send erlang ERL_NIF_TERM result = enif_make_resource(local_env(), db_ptr); // clear the automatic reference from enif_alloc_resource in CreateDbObject enif_release_resource(db_ptr); return work_result(local_env(), ATOM_OK, result); } // OpenTask::operator() /** * OpenCfTask functions */ OpenCfTask::OpenCfTask( ErlNifEnv* caller_env, ERL_NIF_TERM& _caller_ref, const std::string& db_name_, rocksdb::Options *options_, std::vector column_families_, unsigned int num_cols_) : WorkTask(caller_env, _caller_ref), db_name(db_name_), options(options_), column_families(column_families_), num_cols(num_cols_) { } // OpenTask::OpenTask work_result OpenCfTask::operator()() { DbObject * db_ptr; rocksdb::DB *db(0); std::vector handles; rocksdb::Status status = rocksdb::DB::Open(*options, db_name, column_families, &handles, &db); if(!status.ok()) return error_tuple(local_env(), ATOM_ERROR_DB_OPEN, status); // create db respirce db_ptr = DbObject::CreateDbObject(db, options); ERL_NIF_TERM result = enif_make_resource(local_env(), db_ptr); // creare cf list ERL_NIF_TERM cf_list = enif_make_list(local_env(), 0); try { for (unsigned int i = 0; i < num_cols; ++i) { ColumnFamilyObject * handle_ptr; handle_ptr = ColumnFamilyObject::CreateColumnFamilyObject(db_ptr, handles[i]); ERL_NIF_TERM cf = enif_make_resource(local_env(), handle_ptr); enif_release_resource(handle_ptr); handle_ptr = NULL; cf_list = enif_make_list_cell(local_env(), cf, cf_list); } } catch (const std::exception& e) { // pass through } // clear the automatic reference from enif_alloc_resource in CreateDbObject enif_release_resource(db_ptr); ERL_NIF_TERM cf_list_out; enif_make_reverse_list(local_env(), cf_list, &cf_list_out); return work_result(local_env(), ATOM_OK, result, cf_list_out); } // OpenCfTask::operator() /** * MoveTask functions */ work_result MoveTask::operator()() { rocksdb::Iterator* itr = m_ItrWrap->get(); if(NULL == itr) return work_result(local_env(), ATOM_ERROR, ATOM_ITERATOR_CLOSED); switch(action) { case FIRST: itr->SeekToFirst(); break; case LAST: itr->SeekToLast(); break; case PREFETCH: case NEXT: if(itr->Valid()) itr->Next(); break; case PREV: if(itr->Valid()) itr->Prev(); break; case SEEK: { rocksdb::Slice key_slice(seek_target); itr->Seek(key_slice); break; } // case default: // JFW: note: *not* { ERROR, badarg() } here-- we want the exception: // JDB: note: We can't send an exception as a message. It crashes Erlang. // Changing to be {error, badarg}. return work_result(local_env(), ATOM_ERROR, ATOM_BADARG); break; } // switch // who got back first, us or the erlang loop if (compare_and_swap(&m_ItrWrap->m_HandoffAtomic, 0, 1)) { // this is prefetch of next iteration. It returned faster than actual // request to retrieve it. Stop and wait for erlang to catch up. // (even if this result is an Invalid() ) } // if else { // setup next race for the response m_ItrWrap->m_HandoffAtomic=0; if(itr->Valid()) { if (PREFETCH==action) prepare_recycle(); // erlang is waiting, send message if(m_ItrWrap->m_KeysOnly) return work_result(local_env(), ATOM_OK, slice_to_binary(local_env(), itr->key())); return work_result(local_env(), ATOM_OK, slice_to_binary(local_env(), itr->key()), slice_to_binary(local_env(), itr->value())); } // if else { return work_result(local_env(), ATOM_ERROR, ATOM_INVALID_ITERATOR); } // else } // else return(work_result()); } ErlNifEnv * MoveTask::local_env() { if (NULL==local_env_) local_env_ = enif_alloc_env(); if (!terms_set) { caller_ref_term = enif_make_copy(local_env_, m_ItrWrap->itr_ref); caller_pid_term = enif_make_pid(local_env_, &local_pid); terms_set=true; } // if return(local_env_); } // MoveTask::local_env void MoveTask::prepare_recycle() { resubmit_work=true; } // MoveTask::prepare_recycle void MoveTask::recycle() { // test for race condition of simultaneous delete & recycle if (1