feat: add and implement the incrby commandfeat (#44)

* feat: add and implement the incrby command

* fix: deleted print log

* feat: resolve the issues found in pr

* fix: initialize pointer

* fix: params nums

* fix: fix commond params num

* feat: fix review bug

src/base_cmd.h

@@ -24,6 +24,7 @@ namespace pikiwidb {
 // string cmd
 const std::string kCmdNameSet = "set";
 const std::string kCmdNameGet = "get";
+const std::string kCmdNameIncrby = "incrby";
 
 // multi
 const std::string kCmdNameMulti = "multi";

src/cmd_kv.cc

@@ -215,4 +215,39 @@ void BitCountCmd::DoCmd(PClient* client) {
   client->AppendInteger(static_cast<int64_t>(count));
 }
 
+IncrbyCmd::IncrbyCmd(const std::string& name, int16_t arity)
+    : BaseCmd(name, arity, CmdFlagsWrite, AclCategoryWrite | AclCategoryString) {}
+
+bool IncrbyCmd::DoInitial(PClient* client) {
+  int64_t by_ = 0;
+  if (!(pstd::String2int(client->argv_[2].data(), client->argv_[2].size(), &by_))) {
+    client->SetRes(CmdRes::kInvalidInt);
+    return false;
+  }
+  client->SetKey(client->argv_[1]);
+  return true;
+}
+
+void IncrbyCmd::DoCmd(PClient* client) {
+  int64_t new_value_ = 0;
+  int64_t by_ = 0;
+  pstd::String2int(client->argv_[2].data(), client->argv_[2].size(), &by_);
+  PError err = PSTORE.Incrby(client->Key(), by_, &new_value_);
+  switch (err) {
+    case PError_type:
+      client->SetRes(CmdRes::kInvalidInt);
+      break;
+    case PError_notExist:                 // key not exist, set a new value
+      PSTORE.ClearExpire(client->Key());  // clear key's old ttl
+      PSTORE.SetValue(client->Key(), PObject::CreateString(by_));
+      client->AppendInteger(by_);
+      break;
+    case PError_ok:
+      client->AppendInteger(new_value_);
+      break;
+    default:
+      client->SetRes(CmdRes::kErrOther, "incrby cmd error");
+      break;
+  }
+}
 }  // namespace pikiwidb

src/cmd_kv.h

@@ -88,4 +88,15 @@ class BitCountCmd : public BaseCmd {
   void DoCmd(PClient *client) override;
 };
 
+class IncrbyCmd : public BaseCmd {
+ public:
+  IncrbyCmd(const std::string &name, int16_t arity);
+
+ protected:
+  bool DoInitial(PClient *client) override;
+
+ private:
+  void DoCmd(PClient *client) override;
+};
+
 }  // namespace pikiwidb

src/cmd_table_manager.cc

@@ -42,6 +42,9 @@ void CmdTableManager::InitCmdTable() {
   cmds_->insert(std::make_pair(kCmdNameMset, std::move(msetPtr)));
   std::unique_ptr<BaseCmd> bitcountPtr = std::make_unique<BitCountCmd>(kCmdNameBitCount, -2);
   cmds_->insert(std::make_pair(kCmdNameBitCount, std::move(bitcountPtr)));
+
+  std::unique_ptr<BaseCmd> incrbyPtr = std::make_unique<IncrbyCmd>(kCmdNameIncrby, 3);
+  cmds_->insert(std::make_pair(kCmdNameIncrby, std::move(incrbyPtr)));
 }
 
 std::pair<BaseCmd*, CmdRes::CmdRet> CmdTableManager::GetCommand(const std::string& cmdName, PClient* client) {

src/store.cc

@@ -556,6 +556,39 @@ PObject* PStore::SetValue(const PString& key, PObject&& value) {
   return const_cast<PObject*>(&obj);
 }
 
+PError PStore::Incrby(const PString& key, int64_t value, int64_t* ret) {
+  PObject* old_value = nullptr;
+  auto db = &dbs_[dbno_];
+
+  // shared when reading
+  std::unique_lock<std::shared_mutex> lock(mutex_);
+  PError err = getValueByType(key, old_value, PType_string);
+  if (err != PError_ok) {
+    return err;
+  }
+  char* end = nullptr;
+  auto str = pikiwidb::GetDecodedString(old_value);
+  int64_t ival = strtoll(str->c_str(), &end, 10);
+  if (*end != 0) {
+    // value is not a integer
+    return PError_type;
+  }
+
+  PObject new_value;
+  *ret = ival + value;
+  new_value = PObject::CreateString((long)(*ret));
+  new_value.lru = PObject::lruclock;
+  auto [realObj, status] = db->insert_or_assign(key, std::move(new_value));
+  const PObject& obj = realObj->second;
+
+  // put this key to sync list
+  if (!waitSyncKeys_.empty()) {
+    waitSyncKeys_[dbno_].insert_or_assign(key, &obj);
+  }
+
+  return PError_ok;
+}
+
 void PStore::SetExpire(const PString& key, uint64_t when) const { expiredDBs_[dbno_].SetExpire(key, when); }
 
 int64_t PStore::TTL(const PString& key, uint64_t now) { return expiredDBs_[dbno_].TTL(key, now); }

src/store.h

@@ -19,6 +19,8 @@
 #include <folly/concurrency/ConcurrentHashMap.h>
 #include <map>
 #include <memory>
+#include <mutex>
+#include <shared_mutex>
 #include <vector>
 
 namespace pikiwidb {
@@ -117,6 +119,8 @@ class PStore {
   PError GetValueByTypeNoTouch(const PString& key, PObject*& value, PType type = PType_invalid);
 
   PObject* SetValue(const PString& key, PObject&& value);
+  // incr
+  PError Incrby(const PString& key, int64_t value, int64_t* ret);
 
   // for expire key
   enum ExpireResult : std::int8_t {
@@ -157,6 +161,8 @@ class PStore {
 
  private:
   PStore() : dbno_(0) {}
+  // mutex
+  mutable std::shared_mutex mutex_;
 
   PError getValueByType(const PString& key, PObject*& value, PType type = PType_invalid, bool touch = true);