redis_cache.h

Go to the documentation of this file.

/*
 * Copyright 2016 Google Inc.
 *
 * Licensed 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 PAGESPEED_SYSTEM_REDIS_CACHE_H_
#define PAGESPEED_SYSTEM_REDIS_CACHE_H_

#include <stdarg.h>

#include <memory>
#include <initializer_list>
#include <map>
#include <vector>

#include "pagespeed/kernel/base/abstract_mutex.h"
#include "pagespeed/kernel/base/basictypes.h"
#include "pagespeed/kernel/base/message_handler.h"
#include "pagespeed/kernel/base/scoped_ptr.h"
#include "pagespeed/kernel/base/shared_string.h"
#include "pagespeed/kernel/base/statistics.h"
#include "pagespeed/kernel/base/string_util.h"
#include "pagespeed/kernel/base/string.h"
#include "pagespeed/kernel/base/thread_annotations.h"
#include "pagespeed/kernel/base/thread_system.h"
#include "pagespeed/kernel/base/timer.h"
#include "pagespeed/kernel/cache/cache_interface.h"
#include "pagespeed/kernel/thread/thread_synchronizer.h"
#include "pagespeed/system/external_server_spec.h"
#include "third_party/hiredis/src/hiredis.h"

namespace net_instaweb {

class RedisCache : public CacheInterface {
 public:
  RedisCache(StringPiece host, int port, ThreadSystem* thread_system,
             MessageHandler* message_handler, Timer* timer,
             int64 reconnection_delay_ms, int64 timeout_us,
             Statistics* stats, int database_index);
  ~RedisCache() override { ShutDown(); }

  static void InitStats(Statistics* stats);

  GoogleString ServerDescription() const;

  void StartUp(bool connect_now = true);

  GoogleString Name() const override { return FormatName(); }
  bool IsBlocking() const override { return true; }
  bool IsHealthy() const override;
  void ShutDown() override;

  static GoogleString FormatName() { return "RedisCache"; }

  void Get(const GoogleString& key, Callback* callback) override;
  void Put(const GoogleString& key, const SharedString& value) override;
  void Delete(const GoogleString& key) override;

  void GetStatus(GoogleString* status_string);

  static int HashSlot(StringPiece key);

  int64 Redirections() {
    return redirections_->Get();
  }

  int64 ClusterSlotsFetches() {
    return cluster_slots_fetches_->Get();
  }

 private:
  struct RedisReplyDeleter {
    void operator()(redisReply* ptr) {
      if (ptr != nullptr) {  
        freeReplyObject(ptr);
      }
    }
  };
  typedef std::unique_ptr<redisReply, RedisReplyDeleter> RedisReply;

  struct RedisContextDeleter {
    void operator()(redisContext* ptr) {
      if (ptr != nullptr) {  
        redisFree(ptr);
      }
    }
  };
  typedef std::unique_ptr<redisContext, RedisContextDeleter> RedisContext;

  class Connection {
   public:
    Connection(RedisCache* redis_cache, StringPiece host, int port,
        int database_index);

    void StartUp(bool connect_now = true)
        LOCKS_EXCLUDED(redis_mutex_, state_mutex_);
    bool IsHealthy() const LOCKS_EXCLUDED(redis_mutex_, state_mutex_);
    void ShutDown() LOCKS_EXCLUDED(redis_mutex_, state_mutex_);

    GoogleString ToString() {
      return StrCat(host_, ":", IntegerToString(port_));
    }

    AbstractMutex* GetOperationMutex() const LOCK_RETURNED(redis_mutex_) {
      return redis_mutex_.get();
    }

    RedisReply RedisCommand(const char* format, va_list args)
        EXCLUSIVE_LOCKS_REQUIRED(redis_mutex_) LOCKS_EXCLUDED(state_mutex_);
    RedisReply RedisCommand(const char* format, ...)
        EXCLUSIVE_LOCKS_REQUIRED(redis_mutex_) LOCKS_EXCLUDED(state_mutex_);

    bool ValidateRedisReply(const RedisReply& reply,
                            std::initializer_list<int> valid_types,
                            const char* command_executed)
        EXCLUSIVE_LOCKS_REQUIRED(redis_mutex_) LOCKS_EXCLUDED(state_mutex_);

   private:
    enum State {
      kShutDown,
      kDisconnected,
      kConnecting,
      kConnected
    };

    bool IsHealthyLockHeld() const EXCLUSIVE_LOCKS_REQUIRED(state_mutex_);
    void UpdateState() EXCLUSIVE_LOCKS_REQUIRED(redis_mutex_, state_mutex_);

    bool EnsureConnectionAndDatabaseSelection()
        EXCLUSIVE_LOCKS_REQUIRED(redis_mutex_)
        LOCKS_EXCLUDED(state_mutex_);
    bool EnsureConnection() EXCLUSIVE_LOCKS_REQUIRED(redis_mutex_)
        LOCKS_EXCLUDED(state_mutex_);
    bool EnsureDatabaseSelection() EXCLUSIVE_LOCKS_REQUIRED(state_mutex_)
        LOCKS_EXCLUDED(state_mutex_);

    RedisContext TryConnect() LOCKS_EXCLUDED(redis_mutex_, state_mutex_);

    void LogRedisContextError(redisContext* redis, const char* cause);

    const RedisCache* redis_cache_;
    const GoogleString host_;
    const int port_;
    const scoped_ptr<AbstractMutex> redis_mutex_;
    const scoped_ptr<AbstractMutex> state_mutex_;

    RedisContext redis_ GUARDED_BY(redis_mutex_);
    State state_ GUARDED_BY(state_mutex_);
    int64 next_reconnect_at_ms_ GUARDED_BY(state_mutex_);

    const int database_index_;

    
  };
  typedef std::map<GoogleString, std::unique_ptr<Connection>> ConnectionsMap;

  struct ClusterMapping {
    ClusterMapping(int start_slot_range,
                   int end_slot_range,
                   Connection* connection) :
        start_slot_range_(start_slot_range),
        end_slot_range_(end_slot_range),
        connection_(connection) {}
    int start_slot_range_;
    int end_slot_range_;
    Connection* connection_;
  };

  RedisReply RedisCommand(Connection* connection, const char* format,
                          std::initializer_list<int> valid_reply_types, ...);

  ThreadSynchronizer* GetThreadSynchronizerForTesting() const {
    return thread_synchronizer_.get();
  }

  ExternalServerSpec ParseRedirectionError(StringPiece error);

  Connection* GetOrCreateConnection(ExternalServerSpec spec,
      const int database_index);

  void FetchClusterSlotMapping(Connection* connection)
      LOCKS_EXCLUDED(cluster_map_lock_);

  Connection* LookupConnection(StringPiece key)
      LOCKS_EXCLUDED(cluster_map_lock_);

  const GoogleString main_host_;
  const int main_port_;
  ThreadSystem* thread_system_;
  MessageHandler* message_handler_;
  Timer* timer_;
  const int64 reconnection_delay_ms_;
  const int64 timeout_us_;
  const scoped_ptr<ThreadSynchronizer> thread_synchronizer_;
  const scoped_ptr<ThreadSystem::RWLock> connections_lock_;
  const scoped_ptr<ThreadSystem::RWLock> cluster_map_lock_;
  Variable* redirections_;
  Variable* cluster_slots_fetches_;

  ConnectionsMap connections_ GUARDED_BY(connections_lock_);
  std::vector<ClusterMapping> cluster_mappings_ GUARDED_BY(cluster_map_lock_);

  Connection* main_connection_;

  const int database_index_;

  friend class RedisCacheTest;
  
};

}  

#endif