ceph/src/rocksdb/utilities/lua/rocks_lua_compaction_filter.cc

   1 //  Copyright (c) 2016, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5
   6 #if defined(LUA) && !defined(ROCKSDB_LITE)
   7 #include "rocksdb/utilities/lua/rocks_lua_compaction_filter.h"
   8
   9 extern "C" {
  10 #include <luaconf.h>
  11 }
  12
  13 #include "rocksdb/compaction_filter.h"
  14
  15 namespace rocksdb {
  16 namespace lua {
  17
  18 const std::string kFilterFunctionName = "Filter";
  19 const std::string kNameFunctionName = "Name";
  20
  21 void RocksLuaCompactionFilter::LogLuaError(const char* format, ...) const {
  22   if (options_.error_log.get() != nullptr &&
  23       error_count_ < options_.error_limit_per_filter) {
  24     error_count_++;
  25
  26     va_list ap;
  27     va_start(ap, format);
  28     options_.error_log->Logv(InfoLogLevel::ERROR_LEVEL, format, ap);
  29     va_end(ap);
  30   }
  31 }
  32
  33 bool RocksLuaCompactionFilter::Filter(int level, const Slice& key,
  34                                       const Slice& existing_value,
  35                                       std::string* new_value,
  36                                       bool* value_changed) const {
  37   auto* lua_state = lua_state_wrapper_.GetLuaState();
  38   // push the right function into the lua stack
  39   lua_getglobal(lua_state, kFilterFunctionName.c_str());
  40
  41   int error_no = 0;
  42   int num_input_values;
  43   int num_return_values;
  44   if (options_.ignore_value == false) {
  45     // push input arguments into the lua stack
  46     lua_pushnumber(lua_state, level);
  47     lua_pushlstring(lua_state, key.data(), key.size());
  48     lua_pushlstring(lua_state, existing_value.data(), existing_value.size());
  49     num_input_values = 3;
  50     num_return_values = 3;
  51   } else {
  52     // If ignore_value is set to true, then we only put two arguments
  53     // and expect one return value
  54     lua_pushnumber(lua_state, level);
  55     lua_pushlstring(lua_state, key.data(), key.size());
  56     num_input_values = 2;
  57     num_return_values = 1;
  58   }
  59
  60   // perform the lua call
  61   if ((error_no =
  62            lua_pcall(lua_state, num_input_values, num_return_values, 0)) != 0) {
  63     LogLuaError("[Lua] Error(%d) in Filter function --- %s", error_no,
  64                 lua_tostring(lua_state, -1));
  65     // pops out the lua error from stack
  66     lua_pop(lua_state, 1);
  67     return false;
  68   }
  69
  70   // As lua_pcall went successfully, it can be guaranteed that the top
  71   // three elements in the Lua stack are the three returned values.
  72
  73   bool has_error = false;
  74   const int kIndexIsFiltered = -num_return_values;
  75   const int kIndexValueChanged = -num_return_values + 1;
  76   const int kIndexNewValue = -num_return_values + 2;
  77
  78   // check the types of three return values
  79   // is_filtered
  80   if (!lua_isboolean(lua_state, kIndexIsFiltered)) {
  81     LogLuaError(
  82         "[Lua] Error in Filter function -- "
  83         "1st return value (is_filtered) is not a boolean "
  84         "while a boolean is expected.");
  85     has_error = true;
  86   }
  87
  88   if (options_.ignore_value == false) {
  89     // value_changed
  90     if (!lua_isboolean(lua_state, kIndexValueChanged)) {
  91       LogLuaError(
  92           "[Lua] Error in Filter function -- "
  93           "2nd return value (value_changed) is not a boolean "
  94           "while a boolean is expected.");
  95       has_error = true;
  96     }
  97     // new_value
  98     if (!lua_isstring(lua_state, kIndexNewValue)) {
  99       LogLuaError(
 100           "[Lua] Error in Filter function -- "
 101           "3rd return value (new_value) is not a string "
 102           "while a string is expected.");
 103       has_error = true;
 104     }
 105   }
 106
 107   if (has_error) {
 108     lua_pop(lua_state, num_return_values);
 109     return false;
 110   }
 111
 112   // Fetch the return values
 113   bool is_filtered = false;
 114   if (!has_error) {
 115     is_filtered = lua_toboolean(lua_state, kIndexIsFiltered);
 116     if (options_.ignore_value == false) {
 117       *value_changed = lua_toboolean(lua_state, kIndexValueChanged);
 118       if (*value_changed) {
 119         const char* new_value_buf = lua_tostring(lua_state, kIndexNewValue);
 120         const size_t new_value_size = lua_strlen(lua_state, kIndexNewValue);
 121         // Note that any string that lua_tostring returns always has a zero at
 122         // its end, bu/t it can have other zeros inside it
 123         assert(new_value_buf[new_value_size] == '\0');
 124         assert(strlen(new_value_buf) <= new_value_size);
 125         new_value->assign(new_value_buf, new_value_size);
 126       }
 127     } else {
 128       *value_changed = false;
 129     }
 130   }
 131   // pops the three return values.
 132   lua_pop(lua_state, num_return_values);
 133   return is_filtered;
 134 }
 135
 136 const char* RocksLuaCompactionFilter::Name() const {
 137   if (name_ != "") {
 138     return name_.c_str();
 139   }
 140   auto* lua_state = lua_state_wrapper_.GetLuaState();
 141   // push the right function into the lua stack
 142   lua_getglobal(lua_state, kNameFunctionName.c_str());
 143
 144   // perform the call (0 arguments, 1 result)
 145   int error_no;
 146   if ((error_no = lua_pcall(lua_state, 0, 1, 0)) != 0) {
 147     LogLuaError("[Lua] Error(%d) in Name function --- %s", error_no,
 148                 lua_tostring(lua_state, -1));
 149     // pops out the lua error from stack
 150     lua_pop(lua_state, 1);
 151     return name_.c_str();
 152   }
 153
 154   // check the return value
 155   if (!lua_isstring(lua_state, -1)) {
 156     LogLuaError(
 157         "[Lua] Error in Name function -- "
 158         "return value is not a string while string is expected");
 159   } else {
 160     const char* name_buf = lua_tostring(lua_state, -1);
 161     const size_t name_size __attribute__((__unused__)) = lua_strlen(lua_state, -1);
 162     assert(name_buf[name_size] == '\0');
 163     assert(strlen(name_buf) <= name_size);
 164     name_ = name_buf;
 165   }
 166   lua_pop(lua_state, 1);
 167   return name_.c_str();
 168 }
 169
 170 /* Not yet supported
 171 bool RocksLuaCompactionFilter::FilterMergeOperand(
 172     int level, const Slice& key, const Slice& operand) const {
 173   auto* lua_state = lua_state_wrapper_.GetLuaState();
 174   // push the right function into the lua stack
 175   lua_getglobal(lua_state, "FilterMergeOperand");
 176
 177   // push input arguments into the lua stack
 178   lua_pushnumber(lua_state, level);
 179   lua_pushlstring(lua_state, key.data(), key.size());
 180   lua_pushlstring(lua_state, operand.data(), operand.size());
 181
 182   // perform the call (3 arguments, 1 result)
 183   int error_no;
 184   if ((error_no = lua_pcall(lua_state, 3, 1, 0)) != 0) {
 185     LogLuaError("[Lua] Error(%d) in FilterMergeOperand function --- %s",
 186         error_no, lua_tostring(lua_state, -1));
 187     // pops out the lua error from stack
 188     lua_pop(lua_state, 1);
 189     return false;
 190   }
 191
 192   bool is_filtered = false;
 193   // check the return value
 194   if (!lua_isboolean(lua_state, -1)) {
 195     LogLuaError("[Lua] Error in FilterMergeOperand function -- "
 196                 "return value is not a boolean while boolean is expected");
 197   } else {
 198     is_filtered = lua_toboolean(lua_state, -1);
 199   }
 200
 201   lua_pop(lua_state, 1);
 202
 203   return is_filtered;
 204 }
 205 */
 206
 207 bool RocksLuaCompactionFilter::IgnoreSnapshots() const {
 208   return options_.ignore_snapshots;
 209 }
 210
 211 RocksLuaCompactionFilterFactory::RocksLuaCompactionFilterFactory(
 212     const RocksLuaCompactionFilterOptions opt)
 213     : opt_(opt) {
 214   auto filter = CreateCompactionFilter(CompactionFilter::Context());
 215   name_ = std::string("RocksLuaCompactionFilterFactory::") +
 216           std::string(filter->Name());
 217 }
 218
 219 std::unique_ptr<CompactionFilter>
 220 RocksLuaCompactionFilterFactory::CreateCompactionFilter(
 221     const CompactionFilter::Context& /*context*/) {
 222   std::lock_guard<std::mutex> lock(opt_mutex_);
 223   return std::unique_ptr<CompactionFilter>(new RocksLuaCompactionFilter(opt_));
 224 }
 225
 226 std::string RocksLuaCompactionFilterFactory::GetScript() {
 227   std::lock_guard<std::mutex> lock(opt_mutex_);
 228   return opt_.lua_script;
 229 }
 230
 231 void RocksLuaCompactionFilterFactory::SetScript(const std::string& new_script) {
 232   std::lock_guard<std::mutex> lock(opt_mutex_);
 233   opt_.lua_script = new_script;
 234 }
 235
 236 const char* RocksLuaCompactionFilterFactory::Name() const {
 237   return name_.c_str();
 238 }
 239
 240 }  // namespace lua
 241 }  // namespace rocksdb
 242 #endif  // defined(LUA) && !defined(ROCKSDB_LITE)