Cache: Enable LRU eviction

The maximum count and maximum size of the cache can now be
specified and a storage can declare what capabilities it has.
If a storage modile cannot enforce the maximum count or maximum
size limits, the storage is decorated with an LRU storage that
can.

Documentation/Filters/Cache.md

@@ -91,6 +91,34 @@ If nothing is specified, the default _ttl_ value is 10.
 ttl=60
 ```
 
+#### `max_count`
+
+The maximum number of items the cache may contain. If the limit has been
+reached and a new item should be stored, then an older item will be evicted.
+
+Note that if `cached_data` is `thread_specific` then this limit will be
+applied to each cache _separately_.
+```
+max_size=1000
+```
+The default value is 0, which means no limit.
+
+#### `max_size`
+
+The maximum size - expressed in kibibytes - the cache may occupy. If the limit
+has been reached and a new item should be stored, then some older item(s) will
+be evicted to make space.
+
+Note that the value of `max_size` must be at least as large as the value of
+`max_resultset_size`.
+
+Note that if `cached_data` is `thread_specific` then this limit will be
+applied to each cache _separately_.
+```
+max_count=10000
+```
+The default value is 0, which means no limit.
+
 #### `rules`
 
 Specifies the path of the file where the caching rules are stored. A relative
@@ -113,7 +141,12 @@ allowed values are:
      on the other hand that the very same data may be fetched and stored
      multiple times.
 
-Default is `shared`.
+```
+cached_data=thread_specific
+```
+
+Default is `shared`. See `max_count` and `max_size` what implication changing
+this setting to `thread_specific` has.
 
 #### `debug`
 

server/modules/filter/cache/cache_storage_api.h

@@ -65,6 +65,11 @@ typedef enum cache_storage_capabilities
     CACHE_STORAGE_CAP_MAX_SIZE  = 0x10, /*< Storage capable of capping size of cache.*/
 } cache_storage_capabilities_t;
 
+static inline bool cache_storage_has_cap(uint32_t capabilities, uint32_t mask)
+{
+    return (capabilities & mask) == mask;
+}
+
 typedef struct cache_storage_api
 {
     /**
@@ -105,7 +110,7 @@ typedef struct cache_storage_api
                                      const char *name,
                                      uint32_t ttl,
                                      uint32_t max_count,
-                                     uint32_t max_size,
+                                     uint64_t max_size,
                                      int argc, char* argv[]);
 
     /**

server/modules/filter/cache/cachefilter.cc

@@ -36,6 +36,8 @@ static const CACHE_CONFIG DEFAULT_CONFIG =
     NULL,
     0,
     CACHE_DEFAULT_TTL,
+    CACHE_DEFAULT_MAX_COUNT,
+    CACHE_DEFAULT_MAX_SIZE,
     CACHE_DEFAULT_DEBUG,
     CACHE_DEFAULT_THREAD_MODEL,
 };
@@ -333,24 +335,42 @@ static bool process_params(char **pzOptions, FILTER_PARAMETER **ppParams, CACHE_
 
         if (strcmp(pParam->name, "max_resultset_rows") == 0)
         {
-            int v = atoi(pParam->value);
+            char* end;
+            int32_t value = strtol(pParam->value, &end, 0);
 
-            if (v > 0)
+            if ((*end == 0) && (value >= 0))
             {
-                config.max_resultset_rows = v;
+                if (value != 0)
+                {
+                    config.max_resultset_rows = value;
+                }
+                else
+                {
+                    config.max_resultset_rows = CACHE_DEFAULT_MAX_RESULTSET_ROWS;
+                }
             }
             else
             {
-                config.max_resultset_rows = CACHE_DEFAULT_MAX_RESULTSET_ROWS;
+                MXS_ERROR("The value of the configuration entry '%s' must "
+                          "be an integer larger than 0.", pParam->name);
+                error = true;
             }
         }
         else if (strcmp(pParam->name, "max_resultset_size") == 0)
         {
-            int v = atoi(pParam->value);
+            char* end;
+            int64_t value = strtoll(pParam->value, &end, 0);
 
-            if (v > 0)
+            if ((*end == 0) && (value >= 0))
             {
-                config.max_resultset_size = v * 1024;
+                if (value != 0)
+                {
+                    config.max_resultset_size = value * 1024;
+                }
+                else
+                {
+                    config.max_resultset_size = CACHE_DEFAULT_MAX_RESULTSET_SIZE;
+                }
             }
             else
             {
@@ -452,6 +472,52 @@ static bool process_params(char **pzOptions, FILTER_PARAMETER **ppParams, CACHE_
                 error = true;
             }
         }
+        else if (strcmp(pParam->name, "max_count") == 0)
+        {
+            char* end;
+            int32_t value = strtoul(pParam->value, &end, 0);
+
+            if ((*end == 0) && (value >= 0))
+            {
+                if (value != 0)
+                {
+                    config.max_count = value;
+                }
+                else
+                {
+                    config.max_count = CACHE_DEFAULT_MAX_COUNT;
+                }
+            }
+            else
+            {
+                MXS_ERROR("The value of the configuration entry '%s' must "
+                          "be an integer larger than or equal to 0.", pParam->name);
+                error = true;
+            }
+        }
+        else if (strcmp(pParam->name, "max_size") == 0)
+        {
+            char* end;
+            int64_t value = strtoull(pParam->value, &end, 0);
+
+            if ((*end == 0) && (value >= 0))
+            {
+                if (value != 0)
+                {
+                    config.max_size = value * 1024;
+                }
+                else
+                {
+                    config.max_size = CACHE_DEFAULT_MAX_SIZE;
+                }
+            }
+            else
+            {
+                MXS_ERROR("The value of the configuration entry '%s' must "
+                          "be an integer larger than or equal to 0.", pParam->name);
+                error = true;
+            }
+        }
         else if (strcmp(pParam->name, "debug") == 0)
         {
             int v = atoi(pParam->value);
@@ -492,6 +558,17 @@ static bool process_params(char **pzOptions, FILTER_PARAMETER **ppParams, CACHE_
         }
     }
 
+    if (!error)
+    {
+        if (config.max_size < config.max_resultset_size)
+        {
+            MXS_ERROR("The value of 'max_size' must be at least as larged as that "
+                      "of 'max_resultset_size'.");
+
+            error = true;
+        }
+    }
+
     if (error)
     {
         cache_config_finish(config);

server/modules/filter/cache/cachefilter.h

@@ -39,13 +39,17 @@ class StorageFactory;
 #define CACHE_DEBUG_MAX          (CACHE_DEBUG_RULES | CACHE_DEBUG_USAGE | CACHE_DEBUG_DECISIONS)
 
 // Count
-#define CACHE_DEFAULT_MAX_RESULTSET_ROWS UINT_MAX
+#define CACHE_DEFAULT_MAX_RESULTSET_ROWS UINT32_MAX
 // Bytes
 #define CACHE_DEFAULT_MAX_RESULTSET_SIZE 64 * 1024
 // Seconds
 #define CACHE_DEFAULT_TTL                10
 // Integer value
 #define CACHE_DEFAULT_DEBUG              0
+// Positive integer
+#define CACHE_DEFAULT_MAX_COUNT          UINT32_MAX
+// Positive integer
+#define CACHE_DEFAULT_MAX_SIZE           UINT64_MAX
 // Thread model
 #define CACHE_DEFAULT_THREAD_MODEL       CACHE_THREAD_MODEL_MT
 
@@ -59,6 +63,8 @@ typedef struct cache_config
     char** storage_argv;               /**< Cooked options for storage module. */
     int storage_argc;                  /**< Number of cooked options. */
     uint32_t ttl;                      /**< Time to live. */
+    uint32_t max_count;                /**< Maximum number of entries in the cache.*/
+    uint64_t max_size;                 /**< Maximum size of the cache.*/
     uint32_t debug;                    /**< Debug settings. */
     cache_thread_model_t thread_model; /**< Thread model. */
 } CACHE_CONFIG;

server/modules/filter/cache/cachemt.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "cachemt.h"
 #include "storage.h"
 #include "storagefactory.h"
@@ -23,6 +24,8 @@ CacheMT::CacheMT(const std::string& name,
     : CacheSimple(name, pConfig, pRules, pFactory, pStorage)
 {
     spinlock_init(&m_lockPending);
+
+    MXS_NOTICE("Created multi threaded cache.");
 }
 
 CacheMT::~CacheMT()
@@ -89,10 +92,15 @@ CacheMT* CacheMT::Create(const std::string&  name,
     CacheMT* pCache = NULL;
 
     uint32_t ttl = pConfig->ttl;
+    uint32_t maxCount = pConfig->max_count;
+    uint32_t maxSize = pConfig->max_size;
+
     int argc = pConfig->storage_argc;
     char** argv = pConfig->storage_argv;
 
-    Storage* pStorage = pFactory->createStorage(CACHE_THREAD_MODEL_MT, name.c_str(), ttl, argc, argv);
+    Storage* pStorage = pFactory->createStorage(CACHE_THREAD_MODEL_MT, name.c_str(),
+                                                ttl, maxCount, maxSize,
+                                                argc, argv);
 
     if (pStorage)
     {

server/modules/filter/cache/cachept.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "cachept.h"
 #include <maxscale/atomic.h>
 #include <maxscale/platform.h>
@@ -52,6 +53,7 @@ CachePT::CachePT(const std::string&  name,
     : Cache(name, pConfig, pRules, pFactory)
     , m_caches(caches)
 {
+    MXS_NOTICE("Created cache per thread.");
 }
 
 CachePT::~CachePT()

server/modules/filter/cache/cachesimple.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "cachesimple.h"
 #include "storage.h"
 #include "storagefactory.h"

server/modules/filter/cache/cachest.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "cachest.h"
 #include "storage.h"
 #include "storagefactory.h"
@@ -22,6 +23,7 @@ CacheST::CacheST(const std::string& name,
                  Storage* pStorage)
     : CacheSimple(name, pConfig, pRules, pFactory, pStorage)
 {
+    MXS_NOTICE("Created single threaded cache.");
 }
 
 CacheST::~CacheST()
@@ -82,10 +84,15 @@ CacheST* CacheST::Create(const std::string&  name,
     CacheST* pCache = NULL;
 
     uint32_t ttl = pConfig->ttl;
+    uint32_t maxCount = pConfig->max_count;
+    uint32_t maxSize = pConfig->max_size;
+
     int argc = pConfig->storage_argc;
     char** argv = pConfig->storage_argv;
 
-    Storage* pStorage = pFactory->createStorage(CACHE_THREAD_MODEL_ST, name.c_str(), ttl, argc, argv);
+    Storage* pStorage = pFactory->createStorage(CACHE_THREAD_MODEL_ST, name.c_str(),
+                                                ttl, maxCount, maxSize,
+                                                argc, argv);
 
     if (pStorage)
     {

server/modules/filter/cache/lrustorage.cc

@@ -11,9 +11,9 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "lrustorage.h"
 
-
 LRUStorage::LRUStorage(Storage* pstorage, size_t max_count, size_t max_size)
     : pstorage_(pstorage)
     , max_count_(max_count)

server/modules/filter/cache/lrustoragemt.cc

@@ -11,12 +11,15 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "lrustoragemt.h"
 
 LRUStorageMT::LRUStorageMT(Storage* pstorage, size_t max_count, size_t max_size)
     : LRUStorage(pstorage, max_count, max_size)
 {
     spinlock_init(&lock_);
+
+    MXS_NOTICE("Created multi threaded LRU storage.");
 }
 
 LRUStorageMT::~LRUStorageMT()

server/modules/filter/cache/lrustoragemt.h

@@ -21,7 +21,7 @@ class LRUStorageMT : public LRUStorage
 public:
     ~LRUStorageMT();
 
-    LRUStorageMT* create(Storage* pstorage, size_t max_count, size_t max_size);
+    static LRUStorageMT* create(Storage* pstorage, size_t max_count, size_t max_size);
 
     cache_result_t get_value(const CACHE_KEY& key,
                              uint32_t flags,

server/modules/filter/cache/lrustoragest.cc

@@ -11,11 +11,13 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "cache"
 #include "lrustoragest.h"
 
 LRUStorageST::LRUStorageST(Storage* pstorage, size_t max_count, size_t max_size)
     : LRUStorage(pstorage, max_count, max_size)
 {
+    MXS_NOTICE("Created single threaded LRU storage.");
 }
 
 LRUStorageST::~LRUStorageST()

server/modules/filter/cache/lrustoragest.h

@@ -20,7 +20,7 @@ class LRUStorageST : public LRUStorage
 public:
     ~LRUStorageST();
 
-    LRUStorageST* create(Storage* pstorage, size_t max_count, size_t max_size);
+    static LRUStorageST* create(Storage* pstorage, size_t max_count, size_t max_size);
 
     cache_result_t get_value(const CACHE_KEY& key,
                              uint32_t flags,

server/modules/filter/cache/storage/storage_rocksdb/rocksdbinternals.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "storage_rocksdb"
 #include "rocksdbinternals.h"
 #include <rocksdb/env.h>
 #include <util/coding.h>

server/modules/filter/cache/storage/storage_rocksdb/rocksdbstorage.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "storage_rocksdb"
 #include "rocksdbstorage.h"
 #include <openssl/sha.h>
 #include <sys/stat.h>

server/modules/filter/cache/storage/storage_rocksdb/storage_rocksdb.cc

@@ -11,7 +11,9 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "storage_rocksdb"
 #include "storage_rocksdb.h"
+#include <inttypes.h>
 #include "../../cache_storage_api.h"
 #include "rocksdbstorage.h"
 
@@ -29,7 +31,7 @@ CACHE_STORAGE* createInstance(cache_thread_model_t, // Ignored, RocksDB always M
                               const char* zName,
                               uint32_t ttl,
                               uint32_t maxCount,
-                              uint32_t maxSize,
+                              uint64_t maxSize,
                               int argc, char* argv[])
 {
     ss_dassert(zName);
@@ -38,13 +40,13 @@ CACHE_STORAGE* createInstance(cache_thread_model_t, // Ignored, RocksDB always M
 
     if (maxCount != 0)
     {
-        MXS_WARNING("A maximum item count of %u specifed, although 'storage_rocksdb' "
+        MXS_WARNING("A maximum item count of %" PRIu32 " specifed, although 'storage_rocksdb' "
                     "does not enforce such a limit.", maxCount);
     }
 
     if (maxSize != 0)
     {
-        MXS_WARNING("A maximum size of %u specified, although 'storage_rocksdb' "
+        MXS_WARNING("A maximum size of %" PRIu64 " specified, although 'storage_rocksdb' "
                     "does not enforce such a limit.", maxSize);
     }
 

server/modules/filter/cache/storagefactory.cc

@@ -20,6 +20,8 @@
 #include <maxscale/gwdirs.h>
 #include <maxscale/log_manager.h>
 #include "cachefilter.h"
+#include "lrustoragest.h"
+#include "lrustoragemt.h"
 #include "storagereal.h"
 
 
@@ -142,24 +144,63 @@ StorageFactory* StorageFactory::Open(const char* zName)
 Storage* StorageFactory::createStorage(cache_thread_model_t model,
                                        const char* zName,
                                        uint32_t ttl,
+                                       uint32_t maxCount,
+                                       uint64_t maxSize,
                                        int argc, char* argv[])
 {
     ss_dassert(m_handle);
     ss_dassert(m_pApi);
 
     Storage* pStorage = 0;
-    // TODO: Handle max_count and max_size.
-    uint32_t max_count = 0;
-    uint32_t max_size = 0;
 
-    CACHE_STORAGE* pRawStorage = m_pApi->createInstance(model, zName, ttl, max_count, max_size,
+    uint32_t mc = cache_storage_has_cap(m_capabilities, CACHE_STORAGE_CAP_MAX_COUNT) ? maxCount : 0;
-                                                        argc, argv);
+    uint64_t ms = cache_storage_has_cap(m_capabilities, CACHE_STORAGE_CAP_MAX_SIZE) ? maxSize : 0;
+
+    CACHE_STORAGE* pRawStorage = m_pApi->createInstance(model, zName, ttl, mc, ms, argc, argv);
 
     if (pRawStorage)
     {
-        CPP_GUARD(pStorage = new StorageReal(m_pApi, pRawStorage));
+        StorageReal* pStorageReal = NULL;
 
-        if (!pStorage)
+        CPP_GUARD(pStorageReal = new StorageReal(m_pApi, pRawStorage));
+
+        if (pStorageReal)
+        {
+            uint32_t mask = CACHE_STORAGE_CAP_MAX_COUNT | CACHE_STORAGE_CAP_MAX_SIZE;
+
+            if (!cache_storage_has_cap(m_capabilities, mask))
+            {
+                // Ok, so the cache cannot handle eviction. Let's decorate the
+                // real storage with a storage than can.
+
+                LRUStorage *pLruStorage = NULL;
+
+                if (model == CACHE_THREAD_MODEL_ST)
+                {
+                    pLruStorage = LRUStorageST::create(pStorageReal, maxCount, maxSize);
+                }
+                else
+                {
+                    ss_dassert(model == CACHE_THREAD_MODEL_MT);
+
+                    pLruStorage = LRUStorageMT::create(pStorageReal, maxCount, maxSize);
+                }
+
+                if (pLruStorage)
+                {
+                    pStorage = pLruStorage;
+                }
+                else
+                {
+                    delete pStorageReal;
+                }
+            }
+            else
+            {
+                pStorage = pStorageReal;
+            }
+        }
+        else
         {
             m_pApi->freeInstance(pRawStorage);
         }

server/modules/filter/cache/storagefactory.h

@@ -29,6 +29,8 @@ public:
     Storage* createStorage(cache_thread_model_t model,
                            const char* zName,
                            uint32_t ttl,
+                           uint32_t max_count,
+                           uint64_t max_size,
                            int argc, char* argv[]);
 
 private: