revert

Author: kostasrim

src/facade/dragonfly_connection.cc

@@ -272,7 +272,7 @@ void LogTraffic(uint32_t id, bool has_more, absl::Span<RespExpr> resp,
 
   // Write the data itself.
   array<iovec, 16> blobs;
-  uint64_t index = 0;
+  unsigned index = 0;
   if (next != stack_buf) {
     blobs[index++] = iovec{.iov_base = stack_buf, .iov_len = size_t(next - stack_buf)};
   }
@@ -538,7 +538,7 @@ void Connection::AsyncOperations::operator()(const PubMessage& pub_msg) {
     return;
   }
 
-  uint64_t i = 0;
+  unsigned i = 0;
   array<string_view, 4> arr;
   if (pub_msg.pattern.empty()) {
     arr[i++] = pub_msg.is_sharded ? "smessage" : "message";
@@ -609,11 +609,11 @@ void UpdateLibNameVerMap(const string& name, const string& ver, int delta) {
 }
 }  // namespace
 
-void Connection::Init(uint64_t io_threads) {
+void Connection::Init(unsigned io_threads) {
   CHECK(thread_queue_backpressure == nullptr);
   thread_queue_backpressure = new QueueBackpressure[io_threads];
 
-  for (uint64_t i = 0; i < io_threads; ++i) {
+  for (unsigned i = 0; i < io_threads; ++i) {
     auto& qbp = thread_queue_backpressure[i];
     qbp.publish_buffer_limit = GetFlag(FLAGS_publish_buffer_limit);
     qbp.pipeline_cache_limit = GetFlag(FLAGS_request_cache_limit);
@@ -861,7 +861,7 @@ void Connection::HandleRequests() {
   }
 }
 
-uint64_t Connection::GetSendWaitTimeSec() const {
+unsigned Connection::GetSendWaitTimeSec() const {
   if (reply_builder_ && reply_builder_->IsSendActive()) {
     return (util::fb2::ProactorBase::GetMonotonicTimeNs() - reply_builder_->GetLastSendTimeNs()) /
            1'000'000'000;
@@ -877,11 +877,11 @@ void Connection::RegisterBreakHook(BreakerCb breaker_cb) {
 pair<string, string> Connection::GetClientInfoBeforeAfterTid() const {
   if (!socket_) {
     LOG(DFATAL) << "unexpected null socket_ "
-                << " phase " << uint64_t(phase_) << ", is_http: " << unsigned(is_http_);
+                << " phase " << unsigned(phase_) << ", is_http: " << unsigned(is_http_);
     return {};
   }
 
-  CHECK_LT(uint64_t(phase_), NUM_PHASES);
+  CHECK_LT(unsigned(phase_), NUM_PHASES);
 
   string before;
   auto le = LocalBindStr();
@@ -947,7 +947,7 @@ pair<string, string> Connection::GetClientInfoBeforeAfterTid() const {
   return {std::move(before), std::move(after)};
 }
 
-string Connection::GetClientInfo(uint64_t thread_id) const {
+string Connection::GetClientInfo(unsigned thread_id) const {
   auto [before, after] = GetClientInfoBeforeAfterTid();
   absl::StrAppend(&before, " tid=", thread_id);
   absl::StrAppend(&before, after);
@@ -1213,7 +1213,7 @@ void Connection::DispatchSingle(bool has_more, absl::FunctionRef<void()> invoke_
   }
 }
 
-Connection::ParserStatus Connection::ParseRedis(uint64_t max_busy_cycles) {
+Connection::ParserStatus Connection::ParseRedis(unsigned max_busy_cycles) {
   uint32_t consumed = 0;
   RedisParser::Result result = RedisParser::OK;
 
@@ -1827,7 +1827,7 @@ bool Connection::Migrate(util::fb2::ProactorBase* dest) {
 Connection::WeakRef Connection::Borrow() {
   DCHECK(self_);
 
-  return {self_, uint64_t(socket_->proactor()->GetPoolIndex()), id_};
+  return {self_, unsigned(socket_->proactor()->GetPoolIndex()), id_};
 }
 
 void Connection::ShutdownThreadLocal() {
@@ -2095,7 +2095,7 @@ bool Connection::IsReplySizeOverLimit() const {
 }
 
 void Connection::UpdateFromFlags() {
-  uint64_t tid = fb2::ProactorBase::me()->GetPoolIndex();
+  unsigned tid = fb2::ProactorBase::me()->GetPoolIndex();
   thread_queue_backpressure[tid].pipeline_queue_max_len = GetFlag(FLAGS_pipeline_queue_limit);
   thread_queue_backpressure[tid].pipeline_buffer_limit = GetFlag(FLAGS_pipeline_buffer_limit);
   thread_queue_backpressure[tid].pipeline_cnd.notify_all();
@@ -2126,11 +2126,11 @@ void Connection::TrackRequestSize(bool enable) {
   }
 }
 
-void Connection::EnsureMemoryBudget(uint64_t tid) {
+void Connection::EnsureMemoryBudget(unsigned tid) {
   thread_queue_backpressure[tid].EnsureBelowLimit();
 }
 
-Connection::WeakRef::WeakRef(const std::shared_ptr<Connection>& ptr, uint64_t thread_id,
+Connection::WeakRef::WeakRef(const std::shared_ptr<Connection>& ptr, unsigned thread_id,
                              uint32_t client_id)
     : ptr_{ptr}, last_known_thread_id_{thread_id}, client_id_{client_id} {
 }

src/facade/dragonfly_connection.h

@@ -53,7 +53,7 @@ class SinkReplyBuilder;
 // a separate dispatch queue that is processed on a separate fiber.
 class Connection : public util::Connection {
  public:
-  static void Init(uint64_t io_threads);
+  static void Init(unsigned io_threads);
   static void Shutdown();
   static void ShutdownThreadLocal();
 
@@ -186,7 +186,7 @@ class Connection : public util::Connection {
   struct WeakRef {
    public:
     // Get residing thread of connection. Thread-safe.
-    uint64_t LastKnownThreadId() const {
+    unsigned LastKnownThreadId() const {
       return last_known_thread_id_;
     }
     // Get pointer to connection if still valid, nullptr if expired.
@@ -206,10 +206,10 @@ class Connection : public util::Connection {
    private:
     friend class Connection;
 
-    WeakRef(const std::shared_ptr<Connection>& ptr, uint64_t thread_id, uint32_t client_id);
+    WeakRef(const std::shared_ptr<Connection>& ptr, unsigned thread_id, uint32_t client_id);
 
     std::weak_ptr<Connection> ptr_;
-    uint64_t last_known_thread_id_;
+    unsigned last_known_thread_id_;
     uint32_t client_id_;
   };
 
@@ -247,7 +247,7 @@ class Connection : public util::Connection {
 
   bool IsCurrentlyDispatching() const;
 
-  std::string GetClientInfo(uint64_t thread_id) const;
+  std::string GetClientInfo(unsigned thread_id) const;
   std::string GetClientInfo() const;
 
   virtual std::string RemoteEndpointStr() const;  // virtual because overwritten in test_utils
@@ -313,14 +313,14 @@ class Connection : public util::Connection {
   static std::vector<std::string> GetMutableFlagNames();  // Triggers UpdateFromFlags
 
   static void TrackRequestSize(bool enable);
-  static void EnsureMemoryBudget(uint64_t tid);
+  static void EnsureMemoryBudget(unsigned tid);
   static void GetRequestSizeHistogramThreadLocal(std::string* hist);
 
-  uint64_t idle_time() const {
+  unsigned idle_time() const {
     return time(nullptr) - last_interaction_;
   }
 
-  uint64_t GetSendWaitTimeSec() const;
+  unsigned GetSendWaitTimeSec() const;
 
   Phase phase() const {
     return phase_;
@@ -370,7 +370,7 @@ class Connection : public util::Connection {
   // Create new pipeline request, re-use from pool when possible.
   PipelineMessagePtr FromArgs(const RespVec& args);
 
-  ParserStatus ParseRedis(uint64_t max_busy_cycles);
+  ParserStatus ParseRedis(unsigned max_busy_cycles);
   ParserStatus ParseMemcache();
 
   void OnBreakCb(int32_t mask);
@@ -456,7 +456,7 @@ class Connection : public util::Connection {
   std::string lib_name_;
   std::string lib_ver_;
 
-  uint64_t parser_error_ = 0;
+  unsigned parser_error_ = 0;
 
   BreakerCb breaker_cb_;