662 lines
27 KiB
C
662 lines
27 KiB
C
/* tracking.c - Client side caching: keys tracking and invalidation
|
|
*
|
|
* Copyright (c) 2019, Salvatore Sanfilippo <antirez at gmail dot com>
|
|
* All rights reserved.
|
|
*
|
|
* Redistribution and use in source and binary forms, with or without
|
|
* modification, are permitted provided that the following conditions are met:
|
|
*
|
|
* * Redistributions of source code must retain the above copyright notice,
|
|
* this list of conditions and the following disclaimer.
|
|
* * Redistributions in binary form must reproduce the above copyright
|
|
* notice, this list of conditions and the following disclaimer in the
|
|
* documentation and/or other materials provided with the distribution.
|
|
* * Neither the name of Redis nor the names of its contributors may be used
|
|
* to endorse or promote products derived from this software without
|
|
* specific prior written permission.
|
|
*
|
|
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
|
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
|
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
|
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
|
|
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
|
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
|
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
|
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
|
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
|
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
|
* POSSIBILITY OF SUCH DAMAGE.
|
|
*/
|
|
|
|
#include "server.h"
|
|
|
|
/* The tracking table is constituted by a radix tree of keys, each pointing
|
|
* to a radix tree of client IDs, used to track the clients that may have
|
|
* certain keys in their local, client side, cache.
|
|
*
|
|
* When a client enables tracking with "CLIENT TRACKING on", each key served to
|
|
* the client is remembered in the table mapping the keys to the client IDs.
|
|
* Later, when a key is modified, all the clients that may have local copy
|
|
* of such key will receive an invalidation message.
|
|
*
|
|
* Clients will normally take frequently requested objects in memory, removing
|
|
* them when invalidation messages are received. */
|
|
rax *TrackingTable = NULL;
|
|
rax *PrefixTable = NULL;
|
|
uint64_t TrackingTableTotalItems = 0; /* Total number of IDs stored across
|
|
the whole tracking table. This gives
|
|
an hint about the total memory we
|
|
are using server side for CSC. */
|
|
robj *TrackingChannelName;
|
|
|
|
/* This is the structure that we have as value of the PrefixTable, and
|
|
* represents the list of keys modified, and the list of clients that need
|
|
* to be notified, for a given prefix. */
|
|
typedef struct bcastState {
|
|
rax *keys; /* Keys modified in the current event loop cycle. */
|
|
rax *clients; /* Clients subscribed to the notification events for this
|
|
prefix. */
|
|
} bcastState;
|
|
|
|
/* Remove the tracking state from the client 'c'. Note that there is not much
|
|
* to do for us here, if not to decrement the counter of the clients in
|
|
* tracking mode, because we just store the ID of the client in the tracking
|
|
* table, so we'll remove the ID reference in a lazy way. Otherwise when a
|
|
* client with many entries in the table is removed, it would cost a lot of
|
|
* time to do the cleanup. */
|
|
void disableTracking(client *c) {
|
|
/* If this client is in broadcasting mode, we need to unsubscribe it
|
|
* from all the prefixes it is registered to. */
|
|
if (c->flag.tracking_bcast) {
|
|
raxIterator ri;
|
|
raxStart(&ri, c->client_tracking_prefixes);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
while (raxNext(&ri)) {
|
|
void *result;
|
|
int found = raxFind(PrefixTable, ri.key, ri.key_len, &result);
|
|
serverAssert(found);
|
|
bcastState *bs = result;
|
|
raxRemove(bs->clients, (unsigned char *)&c, sizeof(c), NULL);
|
|
/* Was it the last client? Remove the prefix from the
|
|
* table. */
|
|
if (raxSize(bs->clients) == 0) {
|
|
raxFree(bs->clients);
|
|
raxFree(bs->keys);
|
|
zfree(bs);
|
|
raxRemove(PrefixTable, ri.key, ri.key_len, NULL);
|
|
}
|
|
}
|
|
raxStop(&ri);
|
|
raxFree(c->client_tracking_prefixes);
|
|
c->client_tracking_prefixes = NULL;
|
|
}
|
|
|
|
/* Clear flags and adjust the count. */
|
|
if (c->flag.tracking) {
|
|
server.tracking_clients--;
|
|
c->flag.tracking = 0;
|
|
c->flag.tracking_broken_redir = 0;
|
|
c->flag.tracking_bcast = 0;
|
|
c->flag.tracking_optin = 0;
|
|
c->flag.tracking_optout = 0;
|
|
c->flag.tracking_caching = 0;
|
|
c->flag.tracking_noloop = 0;
|
|
}
|
|
}
|
|
|
|
static int stringCheckPrefix(unsigned char *s1, size_t s1_len, unsigned char *s2, size_t s2_len) {
|
|
size_t min_length = s1_len < s2_len ? s1_len : s2_len;
|
|
return memcmp(s1, s2, min_length) == 0;
|
|
}
|
|
|
|
/* Check if any of the provided prefixes collide with one another or
|
|
* with an existing prefix for the client. A collision is defined as two
|
|
* prefixes that will emit an invalidation for the same key. If no prefix
|
|
* collision is found, 1 is return, otherwise 0 is returned and the client
|
|
* has an error emitted describing the error. */
|
|
int checkPrefixCollisionsOrReply(client *c, robj **prefixes, size_t numprefix) {
|
|
for (size_t i = 0; i < numprefix; i++) {
|
|
/* Check input list has no overlap with existing prefixes. */
|
|
if (c->client_tracking_prefixes) {
|
|
raxIterator ri;
|
|
raxStart(&ri, c->client_tracking_prefixes);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
while (raxNext(&ri)) {
|
|
if (stringCheckPrefix(ri.key, ri.key_len, prefixes[i]->ptr, sdslen(prefixes[i]->ptr))) {
|
|
sds collision = sdsnewlen(ri.key, ri.key_len);
|
|
addReplyErrorFormat(c,
|
|
"Prefix '%s' overlaps with an existing prefix '%s'. "
|
|
"Prefixes for a single client must not overlap.",
|
|
(unsigned char *)prefixes[i]->ptr, (unsigned char *)collision);
|
|
sdsfree(collision);
|
|
raxStop(&ri);
|
|
return 0;
|
|
}
|
|
}
|
|
raxStop(&ri);
|
|
}
|
|
/* Check input has no overlap with itself. */
|
|
for (size_t j = i + 1; j < numprefix; j++) {
|
|
if (stringCheckPrefix(prefixes[i]->ptr, sdslen(prefixes[i]->ptr), prefixes[j]->ptr,
|
|
sdslen(prefixes[j]->ptr))) {
|
|
addReplyErrorFormat(c,
|
|
"Prefix '%s' overlaps with another provided prefix '%s'. "
|
|
"Prefixes for a single client must not overlap.",
|
|
(unsigned char *)prefixes[i]->ptr, (unsigned char *)prefixes[j]->ptr);
|
|
return i;
|
|
}
|
|
}
|
|
}
|
|
return 1;
|
|
}
|
|
|
|
/* Set the client 'c' to track the prefix 'prefix'. If the client 'c' is
|
|
* already registered for the specified prefix, no operation is performed. */
|
|
void enableBcastTrackingForPrefix(client *c, char *prefix, size_t plen) {
|
|
void *result;
|
|
bcastState *bs;
|
|
/* If this is the first client subscribing to such prefix, create
|
|
* the prefix in the table. */
|
|
if (!raxFind(PrefixTable, (unsigned char *)prefix, plen, &result)) {
|
|
bs = zmalloc(sizeof(*bs));
|
|
bs->keys = raxNew();
|
|
bs->clients = raxNew();
|
|
raxInsert(PrefixTable, (unsigned char *)prefix, plen, bs, NULL);
|
|
} else {
|
|
bs = result;
|
|
}
|
|
if (raxTryInsert(bs->clients, (unsigned char *)&c, sizeof(c), NULL, NULL)) {
|
|
if (c->client_tracking_prefixes == NULL) c->client_tracking_prefixes = raxNew();
|
|
raxInsert(c->client_tracking_prefixes, (unsigned char *)prefix, plen, NULL, NULL);
|
|
}
|
|
}
|
|
|
|
/* Enable the tracking state for the client 'c', and as a side effect allocates
|
|
* the tracking table if needed. If the 'redirect_to' argument is non zero, the
|
|
* invalidation messages for this client will be sent to the client ID
|
|
* specified by the 'redirect_to' argument. Note that if such client will
|
|
* eventually get freed, we'll send a message to the original client to
|
|
* inform it of the condition. Multiple clients can redirect the invalidation
|
|
* messages to the same client ID. */
|
|
void enableTracking(client *c, uint64_t redirect_to, struct ClientFlags options, robj **prefix, size_t numprefix) {
|
|
if (!c->flag.tracking) server.tracking_clients++;
|
|
c->flag.tracking = 1;
|
|
c->flag.tracking_broken_redir = 0;
|
|
c->flag.tracking_bcast = 0;
|
|
c->flag.tracking_optin = 0;
|
|
c->flag.tracking_optout = 0;
|
|
c->flag.tracking_noloop = 0;
|
|
c->client_tracking_redirection = redirect_to;
|
|
|
|
/* This may be the first client we ever enable. Create the tracking
|
|
* table if it does not exist. */
|
|
if (TrackingTable == NULL) {
|
|
TrackingTable = raxNew();
|
|
PrefixTable = raxNew();
|
|
TrackingChannelName = createStringObject("__redis__:invalidate", 20);
|
|
}
|
|
|
|
/* For broadcasting, set the list of prefixes in the client. */
|
|
if (options.tracking_bcast) {
|
|
c->flag.tracking_bcast = 1;
|
|
if (numprefix == 0) enableBcastTrackingForPrefix(c, "", 0);
|
|
for (size_t j = 0; j < numprefix; j++) {
|
|
sds sdsprefix = prefix[j]->ptr;
|
|
enableBcastTrackingForPrefix(c, sdsprefix, sdslen(sdsprefix));
|
|
}
|
|
}
|
|
|
|
/* Set the remaining flags that don't need any special handling. */
|
|
c->flag.tracking_optin = options.tracking_optin;
|
|
c->flag.tracking_optout = options.tracking_optout;
|
|
c->flag.tracking_noloop = options.tracking_noloop;
|
|
}
|
|
|
|
/* This function is called after the execution of a readonly command in the
|
|
* case the client 'c' has keys tracking enabled and the tracking is not
|
|
* in BCAST mode. It will populate the tracking invalidation table according
|
|
* to the keys the user fetched, so that the server will know what are the clients
|
|
* that should receive an invalidation message with certain groups of keys
|
|
* are modified. */
|
|
void trackingRememberKeys(client *tracking, client *executing) {
|
|
/* Return if we are in optin/out mode and the right CACHING command
|
|
* was/wasn't given in order to modify the default behavior. */
|
|
uint64_t optin = tracking->flag.tracking_optin;
|
|
uint64_t optout = tracking->flag.tracking_optout;
|
|
uint64_t caching_given = tracking->flag.tracking_caching;
|
|
if ((optin && !caching_given) || (optout && caching_given)) return;
|
|
|
|
getKeysResult result;
|
|
initGetKeysResult(&result);
|
|
int numkeys = getKeysFromCommand(executing->cmd, executing->argv, executing->argc, &result);
|
|
if (!numkeys) {
|
|
getKeysFreeResult(&result);
|
|
return;
|
|
}
|
|
/* Shard channels are treated as special keys for client
|
|
* library to rely on `COMMAND` command to discover the node
|
|
* to connect to. These channels doesn't need to be tracked. */
|
|
if (executing->cmd->flags & CMD_PUBSUB) {
|
|
return;
|
|
}
|
|
|
|
keyReference *keys = result.keys;
|
|
|
|
for (int j = 0; j < numkeys; j++) {
|
|
int idx = keys[j].pos;
|
|
sds sdskey = executing->argv[idx]->ptr;
|
|
void *result;
|
|
rax *ids;
|
|
if (!raxFind(TrackingTable, (unsigned char *)sdskey, sdslen(sdskey), &result)) {
|
|
ids = raxNew();
|
|
int inserted = raxTryInsert(TrackingTable, (unsigned char *)sdskey, sdslen(sdskey), ids, NULL);
|
|
serverAssert(inserted == 1);
|
|
} else {
|
|
ids = result;
|
|
}
|
|
if (raxTryInsert(ids, (unsigned char *)&tracking->id, sizeof(tracking->id), NULL, NULL))
|
|
TrackingTableTotalItems++;
|
|
}
|
|
getKeysFreeResult(&result);
|
|
}
|
|
|
|
/* Given a key name, this function sends an invalidation message in the
|
|
* proper channel (depending on RESP version: PubSub or Push message) and
|
|
* to the proper client (in case of redirection), in the context of the
|
|
* client 'c' with tracking enabled.
|
|
*
|
|
* In case the 'proto' argument is non zero, the function will assume that
|
|
* 'keyname' points to a buffer of 'keylen' bytes already expressed in the
|
|
* form of RESP protocol. This is used for:
|
|
* - In BCAST mode, to send an array of invalidated keys to all
|
|
* applicable clients
|
|
* - Following a flush command, to send a single RESP NULL to indicate
|
|
* that all keys are now invalid. */
|
|
void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
|
|
struct ClientFlags old_flags = c->flag;
|
|
c->flag.pushing = 1;
|
|
|
|
int using_redirection = 0;
|
|
if (c->client_tracking_redirection) {
|
|
client *redir = lookupClientByID(c->client_tracking_redirection);
|
|
if (!redir) {
|
|
c->flag.tracking_broken_redir = 1;
|
|
/* We need to signal to the original connection that we
|
|
* are unable to send invalidation messages to the redirected
|
|
* connection, because the client no longer exist. */
|
|
if (c->resp > 2) {
|
|
addReplyPushLen(c, 2);
|
|
addReplyBulkCBuffer(c, "tracking-redir-broken", 21);
|
|
addReplyLongLong(c, c->client_tracking_redirection);
|
|
}
|
|
if (!old_flags.pushing) c->flag.pushing = 0;
|
|
return;
|
|
}
|
|
if (!old_flags.pushing) c->flag.pushing = 0;
|
|
c = redir;
|
|
using_redirection = 1;
|
|
old_flags = c->flag;
|
|
c->flag.pushing = 1;
|
|
}
|
|
|
|
/* Only send such info for clients in RESP version 3 or more. However
|
|
* if redirection is active, and the connection we redirect to is
|
|
* in Pub/Sub mode, we can support the feature with RESP 2 as well,
|
|
* by sending Pub/Sub messages in the __redis__:invalidate channel. */
|
|
if (c->resp > 2) {
|
|
addReplyPushLen(c, 2);
|
|
addReplyBulkCBuffer(c, "invalidate", 10);
|
|
} else if (using_redirection && c->flag.pubsub) {
|
|
/* We use a static object to speedup things, however we assume
|
|
* that addReplyPubsubMessage() will not take a reference. */
|
|
addReplyPubsubMessage(c, TrackingChannelName, NULL, shared.messagebulk);
|
|
} else {
|
|
/* If are here, the client is not using RESP3, nor is
|
|
* redirecting to another client. We can't send anything to
|
|
* it since RESP2 does not support push messages in the same
|
|
* connection. */
|
|
if (!old_flags.pushing) c->flag.pushing = 0;
|
|
return;
|
|
}
|
|
|
|
/* Send the "value" part, which is the array of keys. */
|
|
if (proto) {
|
|
addReplyProto(c, keyname, keylen);
|
|
} else {
|
|
addReplyArrayLen(c, 1);
|
|
addReplyBulkCBuffer(c, keyname, keylen);
|
|
}
|
|
updateClientMemUsageAndBucket(c);
|
|
if (!old_flags.pushing) c->flag.pushing = 0;
|
|
}
|
|
|
|
/* This function is called when a key is modified in the server and in the case
|
|
* we have at least one client with the BCAST mode enabled.
|
|
* Its goal is to set the key in the right broadcast state if the key
|
|
* matches one or more prefixes in the prefix table. Later when we
|
|
* return to the event loop, we'll send invalidation messages to the
|
|
* clients subscribed to each prefix. */
|
|
void trackingRememberKeyToBroadcast(client *c, char *keyname, size_t keylen) {
|
|
raxIterator ri;
|
|
raxStart(&ri, PrefixTable);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
while (raxNext(&ri)) {
|
|
if (ri.key_len > keylen) continue;
|
|
if (ri.key_len != 0 && memcmp(ri.key, keyname, ri.key_len) != 0) continue;
|
|
bcastState *bs = ri.data;
|
|
/* We insert the client pointer as associated value in the radix
|
|
* tree. This way we know who was the client that did the last
|
|
* change to the key, and can avoid sending the notification in the
|
|
* case the client is in NOLOOP mode. */
|
|
raxInsert(bs->keys, (unsigned char *)keyname, keylen, c, NULL);
|
|
}
|
|
raxStop(&ri);
|
|
}
|
|
|
|
/* This function is called from signalModifiedKey() or other places in the server
|
|
* when a key changes value. In the context of keys tracking, our task here is
|
|
* to send a notification to every client that may have keys about such caching
|
|
* slot.
|
|
*
|
|
* Note that 'c' may be NULL in case the operation was performed outside the
|
|
* context of a client modifying the database (for instance when we delete a
|
|
* key because of expire).
|
|
*
|
|
* The last argument 'bcast' tells the function if it should also schedule
|
|
* the key for broadcasting to clients in BCAST mode. This is the case when
|
|
* the function is called from the server core once a key is modified, however
|
|
* we also call the function in order to evict keys in the key table in case
|
|
* of memory pressure: in that case the key didn't really change, so we want
|
|
* just to notify the clients that are in the table for this key, that would
|
|
* otherwise miss the fact we are no longer tracking the key for them. */
|
|
void trackingInvalidateKey(client *c, robj *keyobj, int bcast) {
|
|
if (TrackingTable == NULL) return;
|
|
|
|
unsigned char *key = (unsigned char *)keyobj->ptr;
|
|
size_t keylen = sdslen(keyobj->ptr);
|
|
|
|
if (bcast && raxSize(PrefixTable) > 0) trackingRememberKeyToBroadcast(c, (char *)key, keylen);
|
|
|
|
void *result;
|
|
if (!raxFind(TrackingTable, key, keylen, &result)) return;
|
|
rax *ids = result;
|
|
|
|
raxIterator ri;
|
|
raxStart(&ri, ids);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
while (raxNext(&ri)) {
|
|
uint64_t id;
|
|
memcpy(&id, ri.key, sizeof(id));
|
|
client *target = lookupClientByID(id);
|
|
/* Note that if the client is in BCAST mode, we don't want to
|
|
* send invalidation messages that were pending in the case
|
|
* previously the client was not in BCAST mode. This can happen if
|
|
* TRACKING is enabled normally, and then the client switches to
|
|
* BCAST mode. */
|
|
if (target == NULL || !(target->flag.tracking) || target->flag.tracking_bcast) {
|
|
continue;
|
|
}
|
|
|
|
/* If the client enabled the NOLOOP mode, don't send notifications
|
|
* about keys changed by the client itself. */
|
|
if (target->flag.tracking_noloop && target == server.current_client) {
|
|
continue;
|
|
}
|
|
|
|
/* If target is current client and it's executing a command, we need schedule key invalidation.
|
|
* As the invalidation messages may be interleaved with command
|
|
* response and should after command response. */
|
|
if (target == server.current_client && (server.current_client->flag.executing_command)) {
|
|
incrRefCount(keyobj);
|
|
listAddNodeTail(server.tracking_pending_keys, keyobj);
|
|
} else {
|
|
sendTrackingMessage(target, (char *)keyobj->ptr, sdslen(keyobj->ptr), 0);
|
|
}
|
|
}
|
|
raxStop(&ri);
|
|
|
|
/* Free the tracking table: we'll create the radix tree and populate it
|
|
* again if more keys will be modified in this caching slot. */
|
|
TrackingTableTotalItems -= raxSize(ids);
|
|
raxFree(ids);
|
|
raxRemove(TrackingTable, (unsigned char *)key, keylen, NULL);
|
|
}
|
|
|
|
void trackingHandlePendingKeyInvalidations(void) {
|
|
if (!listLength(server.tracking_pending_keys)) return;
|
|
|
|
/* Flush pending invalidation messages only when we are not in nested call.
|
|
* So the messages are not interleaved with transaction response. */
|
|
if (server.execution_nesting) return;
|
|
|
|
listNode *ln;
|
|
listIter li;
|
|
|
|
listRewind(server.tracking_pending_keys, &li);
|
|
while ((ln = listNext(&li)) != NULL) {
|
|
robj *key = listNodeValue(ln);
|
|
/* current_client maybe freed, so we need to send invalidation
|
|
* message only when current_client is still alive */
|
|
if (server.current_client != NULL) {
|
|
if (key != NULL) {
|
|
sendTrackingMessage(server.current_client, (char *)key->ptr, sdslen(key->ptr), 0);
|
|
} else {
|
|
sendTrackingMessage(server.current_client, shared.null[server.current_client->resp]->ptr,
|
|
sdslen(shared.null[server.current_client->resp]->ptr), 1);
|
|
}
|
|
}
|
|
if (key != NULL) decrRefCount(key);
|
|
}
|
|
listEmpty(server.tracking_pending_keys);
|
|
}
|
|
|
|
/* This function is called when one or all of the databases are
|
|
* flushed. Caching keys are not specific for each DB but are global:
|
|
* currently what we do is send a special notification to clients with
|
|
* tracking enabled, sending a RESP NULL, which means, "all the keys",
|
|
* in order to avoid flooding clients with many invalidation messages
|
|
* for all the keys they may hold.
|
|
*/
|
|
void freeTrackingRadixTreeCallback(void *rt) {
|
|
raxFree(rt);
|
|
}
|
|
|
|
void freeTrackingRadixTree(rax *rt) {
|
|
raxFreeWithCallback(rt, freeTrackingRadixTreeCallback);
|
|
}
|
|
|
|
/* A RESP NULL is sent to indicate that all keys are invalid */
|
|
void trackingInvalidateKeysOnFlush(int async) {
|
|
if (server.tracking_clients) {
|
|
listNode *ln;
|
|
listIter li;
|
|
listRewind(server.clients, &li);
|
|
while ((ln = listNext(&li)) != NULL) {
|
|
client *c = listNodeValue(ln);
|
|
if (c->flag.tracking) {
|
|
if (c == server.current_client) {
|
|
/* We use a special NULL to indicate that we should send null */
|
|
listAddNodeTail(server.tracking_pending_keys, NULL);
|
|
} else {
|
|
sendTrackingMessage(c, shared.null[c->resp]->ptr, sdslen(shared.null[c->resp]->ptr), 1);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/* In case of FLUSHALL, reclaim all the memory used by tracking. */
|
|
if (TrackingTable) {
|
|
if (async) {
|
|
freeTrackingRadixTreeAsync(TrackingTable);
|
|
} else {
|
|
freeTrackingRadixTree(TrackingTable);
|
|
}
|
|
TrackingTable = raxNew();
|
|
TrackingTableTotalItems = 0;
|
|
}
|
|
}
|
|
|
|
/* Tracking forces the server to remember information about which client may have
|
|
* certain keys. In workloads where there are a lot of reads, but keys are
|
|
* hardly modified, the amount of information we have to remember server side
|
|
* could be a lot, with the number of keys being totally not bound.
|
|
*
|
|
* So the server allows the user to configure a maximum number of keys for the
|
|
* invalidation table. This function makes sure that we don't go over the
|
|
* specified fill rate: if we are over, we can just evict information about
|
|
* a random key, and send invalidation messages to clients like if the key was
|
|
* modified. */
|
|
void trackingLimitUsedSlots(void) {
|
|
static unsigned int timeout_counter = 0;
|
|
if (TrackingTable == NULL) return;
|
|
if (server.tracking_table_max_keys == 0) return; /* No limits set. */
|
|
size_t max_keys = server.tracking_table_max_keys;
|
|
if (raxSize(TrackingTable) <= max_keys) {
|
|
timeout_counter = 0;
|
|
return; /* Limit not reached. */
|
|
}
|
|
|
|
/* We have to invalidate a few keys to reach the limit again. The effort
|
|
* we do here is proportional to the number of times we entered this
|
|
* function and found that we are still over the limit. */
|
|
int effort = 100 * (timeout_counter + 1);
|
|
|
|
/* We just remove one key after another by using a random walk. */
|
|
raxIterator ri;
|
|
raxStart(&ri, TrackingTable);
|
|
while (effort > 0) {
|
|
effort--;
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
raxRandomWalk(&ri, 0);
|
|
if (raxEOF(&ri)) break;
|
|
robj *keyobj = createStringObject((char *)ri.key, ri.key_len);
|
|
trackingInvalidateKey(NULL, keyobj, 0);
|
|
decrRefCount(keyobj);
|
|
if (raxSize(TrackingTable) <= max_keys) {
|
|
timeout_counter = 0;
|
|
raxStop(&ri);
|
|
return; /* Return ASAP: we are again under the limit. */
|
|
}
|
|
}
|
|
|
|
/* If we reach this point, we were not able to go under the configured
|
|
* limit using the maximum effort we had for this run. */
|
|
raxStop(&ri);
|
|
timeout_counter++;
|
|
}
|
|
|
|
/* Generate RESP for an array containing all the key names
|
|
* in the 'keys' radix tree. If the client is not NULL, the list will not
|
|
* include keys that were modified the last time by this client, in order
|
|
* to implement the NOLOOP option.
|
|
*
|
|
* If the resulting array would be empty, NULL is returned instead. */
|
|
sds trackingBuildBroadcastReply(client *c, rax *keys) {
|
|
raxIterator ri;
|
|
uint64_t count;
|
|
|
|
if (c == NULL) {
|
|
count = raxSize(keys);
|
|
} else {
|
|
count = 0;
|
|
raxStart(&ri, keys);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
while (raxNext(&ri)) {
|
|
if (ri.data != c) count++;
|
|
}
|
|
raxStop(&ri);
|
|
|
|
if (count == 0) return NULL;
|
|
}
|
|
|
|
/* Create the array reply with the list of keys once, then send
|
|
* it to all the clients subscribed to this prefix. */
|
|
char buf[32];
|
|
size_t len = ll2string(buf, sizeof(buf), count);
|
|
sds proto = sdsempty();
|
|
proto = sdsMakeRoomFor(proto, count * 15);
|
|
proto = sdscatlen(proto, "*", 1);
|
|
proto = sdscatlen(proto, buf, len);
|
|
proto = sdscatlen(proto, "\r\n", 2);
|
|
raxStart(&ri, keys);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
while (raxNext(&ri)) {
|
|
if (c && ri.data == c) continue;
|
|
len = ll2string(buf, sizeof(buf), ri.key_len);
|
|
proto = sdscatlen(proto, "$", 1);
|
|
proto = sdscatlen(proto, buf, len);
|
|
proto = sdscatlen(proto, "\r\n", 2);
|
|
proto = sdscatlen(proto, ri.key, ri.key_len);
|
|
proto = sdscatlen(proto, "\r\n", 2);
|
|
}
|
|
raxStop(&ri);
|
|
return proto;
|
|
}
|
|
|
|
/* This function will run the prefixes of clients in BCAST mode and
|
|
* keys that were modified about each prefix, and will send the
|
|
* notifications to each client in each prefix. */
|
|
void trackingBroadcastInvalidationMessages(void) {
|
|
raxIterator ri, ri2;
|
|
|
|
/* Return ASAP if there is nothing to do here. */
|
|
if (TrackingTable == NULL || !server.tracking_clients) return;
|
|
|
|
raxStart(&ri, PrefixTable);
|
|
raxSeek(&ri, "^", NULL, 0);
|
|
|
|
/* For each prefix... */
|
|
while (raxNext(&ri)) {
|
|
bcastState *bs = ri.data;
|
|
|
|
if (raxSize(bs->keys)) {
|
|
/* Generate the common protocol for all the clients that are
|
|
* not using the NOLOOP option. */
|
|
sds proto = trackingBuildBroadcastReply(NULL, bs->keys);
|
|
|
|
/* Send this array of keys to every client in the list. */
|
|
raxStart(&ri2, bs->clients);
|
|
raxSeek(&ri2, "^", NULL, 0);
|
|
while (raxNext(&ri2)) {
|
|
client *c;
|
|
memcpy(&c, ri2.key, sizeof(c));
|
|
if (c->flag.tracking_noloop) {
|
|
/* This client may have certain keys excluded. */
|
|
sds adhoc = trackingBuildBroadcastReply(c, bs->keys);
|
|
if (adhoc) {
|
|
sendTrackingMessage(c, adhoc, sdslen(adhoc), 1);
|
|
sdsfree(adhoc);
|
|
}
|
|
} else {
|
|
sendTrackingMessage(c, proto, sdslen(proto), 1);
|
|
}
|
|
}
|
|
raxStop(&ri2);
|
|
|
|
/* Clean up: we can remove everything from this state, because we
|
|
* want to only track the new keys that will be accumulated starting
|
|
* from now. */
|
|
sdsfree(proto);
|
|
}
|
|
raxFree(bs->keys);
|
|
bs->keys = raxNew();
|
|
}
|
|
raxStop(&ri);
|
|
}
|
|
|
|
/* This is just used in order to access the amount of used slots in the
|
|
* tracking table. */
|
|
uint64_t trackingGetTotalItems(void) {
|
|
return TrackingTableTotalItems;
|
|
}
|
|
|
|
uint64_t trackingGetTotalKeys(void) {
|
|
if (TrackingTable == NULL) return 0;
|
|
return raxSize(TrackingTable);
|
|
}
|
|
|
|
uint64_t trackingGetTotalPrefixes(void) {
|
|
if (PrefixTable == NULL) return 0;
|
|
return raxSize(PrefixTable);
|
|
}
|