Commit 79866a63 authored by antirez's avatar antirez
Browse files

Streams: 12 commits squashed into the initial Streams implementation.

parent 045d65c3
...@@ -144,7 +144,7 @@ endif ...@@ -144,7 +144,7 @@ endif
REDIS_SERVER_NAME=redis-server REDIS_SERVER_NAME=redis-server
REDIS_SENTINEL_NAME=redis-sentinel REDIS_SENTINEL_NAME=redis-sentinel
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.c
REDIS_CLI_NAME=redis-cli REDIS_CLI_NAME=redis-cli
REDIS_CLI_OBJ=anet.o adlist.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o REDIS_CLI_OBJ=anet.o adlist.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o
REDIS_BENCHMARK_NAME=redis-benchmark REDIS_BENCHMARK_NAME=redis-benchmark
......
This diff is collapsed.
/* Listpack -- A lists of strings serialization format
*
* This file implements the specification you can find at:
*
* https://github.com/antirez/listpack
*
* Copyright (c) 2017, 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.
*/
#ifndef __LISTPACK_H
#define __LISTPACK_H
#include <stdint.h>
#define LP_INTBUF_SIZE 21 /* 20 digits of -2^63 + 1 null term = 21. */
/* lpInsert() where argument possible values: */
#define LP_BEFORE 0
#define LP_AFTER 1
#define LP_REPLACE 2
unsigned char *lpNew(void);
void lpFree(unsigned char *lp);
unsigned char *lpInsert(unsigned char *lp, unsigned char *ele, uint32_t size, unsigned char *p, int where, unsigned char **newp);
unsigned char *lpAppend(unsigned char *lp, unsigned char *ele, uint32_t size);
unsigned char *lpDelete(unsigned char *lp, unsigned char *p, unsigned char **newp);
uint32_t lpLength(unsigned char *lp);
unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf);
unsigned char *lpFirst(unsigned char *lp);
unsigned char *lpLast(unsigned char *lp);
unsigned char *lpNext(unsigned char *lp, unsigned char *p);
unsigned char *lpPrev(unsigned char *lp, unsigned char *p);
uint32_t lpBytes(unsigned char *lp);
unsigned char *lpSeek(unsigned char *lp, long index);
#endif
/* Listpack -- A lists of strings serialization format
* https://github.com/antirez/listpack
*
* Copyright (c) 2017, 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.
*/
/* Allocator selection.
*
* This file is used in order to change the Rax allocator at compile time.
* Just define the following defines to what you want to use. Also add
* the include of your alternate allocator if needed (not needed in order
* to use the default libc allocator). */
#ifndef LISTPACK_ALLOC_H
#define LISTPACK_ALLOC_H
#define lp_malloc malloc
#define lp_realloc realloc
#define lp_free free
#endif
...@@ -232,6 +232,13 @@ robj *createZsetZiplistObject(void) { ...@@ -232,6 +232,13 @@ robj *createZsetZiplistObject(void) {
return o; return o;
} }
robj *createStreamObject(void) {
stream *s = streamNew();
robj *o = createObject(OBJ_STREAM,s);
o->encoding = OBJ_ENCODING_STREAM;
return o;
}
robj *createModuleObject(moduleType *mt, void *value) { robj *createModuleObject(moduleType *mt, void *value) {
moduleValue *mv = zmalloc(sizeof(*mv)); moduleValue *mv = zmalloc(sizeof(*mv));
mv->type = mt; mv->type = mt;
......
...@@ -131,7 +131,7 @@ static inline void raxStackFree(raxStack *ts) { ...@@ -131,7 +131,7 @@ static inline void raxStackFree(raxStack *ts) {
} }
/* ---------------------------------------------------------------------------- /* ----------------------------------------------------------------------------
* Radis tree implementation * Radix tree implementation
* --------------------------------------------------------------------------*/ * --------------------------------------------------------------------------*/
/* Allocate a new non compressed node with the specified number of children. /* Allocate a new non compressed node with the specified number of children.
...@@ -873,7 +873,8 @@ raxNode *raxRemoveChild(raxNode *parent, raxNode *child) { ...@@ -873,7 +873,8 @@ raxNode *raxRemoveChild(raxNode *parent, raxNode *child) {
memmove(((char*)cp)-1,cp,(parent->size-taillen-1)*sizeof(raxNode**)); memmove(((char*)cp)-1,cp,(parent->size-taillen-1)*sizeof(raxNode**));
/* Move the remaining "tail" pointer at the right position as well. */ /* Move the remaining "tail" pointer at the right position as well. */
memmove(((char*)c)-1,c+1,taillen*sizeof(raxNode**)+parent->iskey*sizeof(void*)); size_t valuelen = (parent->iskey && !parent->isnull) ? sizeof(void*) : 0;
memmove(((char*)c)-1,c+1,taillen*sizeof(raxNode**)+valuelen);
/* 4. Update size. */ /* 4. Update size. */
parent->size--; parent->size--;
...@@ -1175,7 +1176,7 @@ void raxIteratorDelChars(raxIterator *it, size_t count) { ...@@ -1175,7 +1176,7 @@ void raxIteratorDelChars(raxIterator *it, size_t count) {
* The function returns 1 on success or 0 on out of memory. */ * The function returns 1 on success or 0 on out of memory. */
int raxIteratorNextStep(raxIterator *it, int noup) { int raxIteratorNextStep(raxIterator *it, int noup) {
if (it->flags & RAX_ITER_EOF) { if (it->flags & RAX_ITER_EOF) {
return 0; return 1;
} else if (it->flags & RAX_ITER_JUST_SEEKED) { } else if (it->flags & RAX_ITER_JUST_SEEKED) {
it->flags &= ~RAX_ITER_JUST_SEEKED; it->flags &= ~RAX_ITER_JUST_SEEKED;
return 1; return 1;
...@@ -1187,10 +1188,6 @@ int raxIteratorNextStep(raxIterator *it, int noup) { ...@@ -1187,10 +1188,6 @@ int raxIteratorNextStep(raxIterator *it, int noup) {
size_t orig_stack_items = it->stack.items; size_t orig_stack_items = it->stack.items;
raxNode *orig_node = it->node; raxNode *orig_node = it->node;
/* Clear the EOF flag: it will be set again if the EOF condition
* is still valid. */
it->flags &= ~RAX_ITER_EOF;
while(1) { while(1) {
int children = it->node->iscompr ? 1 : it->node->size; int children = it->node->iscompr ? 1 : it->node->size;
if (!noup && children) { if (!noup && children) {
...@@ -1291,7 +1288,7 @@ int raxSeekGreatest(raxIterator *it) { ...@@ -1291,7 +1288,7 @@ int raxSeekGreatest(raxIterator *it) {
* effect to the one of raxIteratorPrevSte(). */ * effect to the one of raxIteratorPrevSte(). */
int raxIteratorPrevStep(raxIterator *it, int noup) { int raxIteratorPrevStep(raxIterator *it, int noup) {
if (it->flags & RAX_ITER_EOF) { if (it->flags & RAX_ITER_EOF) {
return 0; return 1;
} else if (it->flags & RAX_ITER_JUST_SEEKED) { } else if (it->flags & RAX_ITER_JUST_SEEKED) {
it->flags &= ~RAX_ITER_JUST_SEEKED; it->flags &= ~RAX_ITER_JUST_SEEKED;
return 1; return 1;
...@@ -1412,6 +1409,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1412,6 +1409,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
it->node = it->rt->head; it->node = it->rt->head;
if (!raxSeekGreatest(it)) return 0; if (!raxSeekGreatest(it)) return 0;
assert(it->node->iskey); assert(it->node->iskey);
it->data = raxGetData(it->node);
return 1; return 1;
} }
...@@ -1430,6 +1428,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1430,6 +1428,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
/* We found our node, since the key matches and we have an /* We found our node, since the key matches and we have an
* "equal" condition. */ * "equal" condition. */
if (!raxIteratorAddChars(it,ele,len)) return 0; /* OOM. */ if (!raxIteratorAddChars(it,ele,len)) return 0; /* OOM. */
it->data = raxGetData(it->node);
} else if (lt || gt) { } else if (lt || gt) {
/* Exact key not found or eq flag not set. We have to set as current /* Exact key not found or eq flag not set. We have to set as current
* key the one represented by the node we stopped at, and perform * key the one represented by the node we stopped at, and perform
...@@ -1502,6 +1501,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1502,6 +1501,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
* the previous sub-tree. */ * the previous sub-tree. */
if (nodechar < keychar) { if (nodechar < keychar) {
if (!raxSeekGreatest(it)) return 0; if (!raxSeekGreatest(it)) return 0;
it->data = raxGetData(it->node);
} else { } else {
if (!raxIteratorAddChars(it,it->node->data,it->node->size)) if (!raxIteratorAddChars(it,it->node->data,it->node->size))
return 0; return 0;
...@@ -1647,6 +1647,14 @@ void raxStop(raxIterator *it) { ...@@ -1647,6 +1647,14 @@ void raxStop(raxIterator *it) {
raxStackFree(&it->stack); raxStackFree(&it->stack);
} }
/* Return if the iterator is in an EOF state. This happens when raxSeek()
* failed to seek an appropriate element, so that raxNext() or raxPrev()
* will return zero, or when an EOF condition was reached while iterating
* with raxNext() and raxPrev(). */
int raxEOF(raxIterator *it) {
return it->flags & RAX_ITER_EOF;
}
/* ----------------------------- Introspection ------------------------------ */ /* ----------------------------- Introspection ------------------------------ */
/* This function is mostly used for debugging and learning purposes. /* This function is mostly used for debugging and learning purposes.
......
...@@ -155,6 +155,7 @@ int raxPrev(raxIterator *it); ...@@ -155,6 +155,7 @@ int raxPrev(raxIterator *it);
int raxRandomWalk(raxIterator *it, size_t steps); int raxRandomWalk(raxIterator *it, size_t steps);
int raxCompare(raxIterator *iter, const char *op, unsigned char *key, size_t key_len); int raxCompare(raxIterator *iter, const char *op, unsigned char *key, size_t key_len);
void raxStop(raxIterator *it); void raxStop(raxIterator *it);
int raxEOF(raxIterator *it);
void raxShow(rax *rax); void raxShow(rax *rax);
#endif #endif
...@@ -89,10 +89,11 @@ ...@@ -89,10 +89,11 @@
#define RDB_TYPE_ZSET_ZIPLIST 12 #define RDB_TYPE_ZSET_ZIPLIST 12
#define RDB_TYPE_HASH_ZIPLIST 13 #define RDB_TYPE_HASH_ZIPLIST 13
#define RDB_TYPE_LIST_QUICKLIST 14 #define RDB_TYPE_LIST_QUICKLIST 14
#define RDB_TYPE_STREAM_LISTPACKS 15
/* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */ /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
/* Test if a type is an object type. */ /* Test if a type is an object type. */
#define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 14)) #define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 15))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_AUX 250 #define RDB_OPCODE_AUX 250
......
...@@ -302,6 +302,8 @@ struct redisCommand redisCommandTable[] = { ...@@ -302,6 +302,8 @@ struct redisCommand redisCommandTable[] = {
{"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0}, {"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0},
{"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0}, {"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0},
{"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0}, {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0},
{"xadd",xaddCommand,-4,"wmF",0,NULL,1,1,1,0,0},
{"xrange",xrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
{"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0}, {"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
{"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0}, {"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
{"latency",latencyCommand,-2,"aslt",0,NULL,0,0,0,0,0} {"latency",latencyCommand,-2,"aslt",0,NULL,0,0,0,0,0}
......
...@@ -59,6 +59,7 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -59,6 +59,7 @@ typedef long long mstime_t; /* millisecond time type. */
#include "anet.h" /* Networking the easy way */ #include "anet.h" /* Networking the easy way */
#include "ziplist.h" /* Compact list data structure */ #include "ziplist.h" /* Compact list data structure */
#include "intset.h" /* Compact integer set structure */ #include "intset.h" /* Compact integer set structure */
#include "stream.h" /* Stream data type header file. */
#include "version.h" /* Version macro */ #include "version.h" /* Version macro */
#include "util.h" /* Misc functions useful in many places */ #include "util.h" /* Misc functions useful in many places */
#include "latency.h" /* Latency monitor API */ #include "latency.h" /* Latency monitor API */
...@@ -451,6 +452,7 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -451,6 +452,7 @@ typedef long long mstime_t; /* millisecond time type. */
#define OBJ_SET 2 #define OBJ_SET 2
#define OBJ_ZSET 3 #define OBJ_ZSET 3
#define OBJ_HASH 4 #define OBJ_HASH 4
#define OBJ_STREAM 5
/* The "module" object type is a special one that signals that the object /* The "module" object type is a special one that signals that the object
* is one directly managed by a Redis module. In this case the value points * is one directly managed by a Redis module. In this case the value points
...@@ -575,6 +577,7 @@ typedef struct RedisModuleDigest { ...@@ -575,6 +577,7 @@ typedef struct RedisModuleDigest {
#define OBJ_ENCODING_SKIPLIST 7 /* Encoded as skiplist */ #define OBJ_ENCODING_SKIPLIST 7 /* Encoded as skiplist */
#define OBJ_ENCODING_EMBSTR 8 /* Embedded sds string encoding */ #define OBJ_ENCODING_EMBSTR 8 /* Embedded sds string encoding */
#define OBJ_ENCODING_QUICKLIST 9 /* Encoded as linked list of ziplists */ #define OBJ_ENCODING_QUICKLIST 9 /* Encoded as linked list of ziplists */
#define OBJ_ENCODING_STREAM 10 /* Encoded as a radix tree of listpacks */
#define LRU_BITS 24 #define LRU_BITS 24
#define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */ #define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */
...@@ -1414,6 +1417,9 @@ void handleClientsBlockedOnLists(void); ...@@ -1414,6 +1417,9 @@ void handleClientsBlockedOnLists(void);
void popGenericCommand(client *c, int where); void popGenericCommand(client *c, int where);
void signalListAsReady(redisDb *db, robj *key); void signalListAsReady(redisDb *db, robj *key);
/* Stream data type. */
stream *streamNew(void);
/* MULTI/EXEC/WATCH... */ /* MULTI/EXEC/WATCH... */
void unwatchAllKeys(client *c); void unwatchAllKeys(client *c);
void initClientMultiState(client *c); void initClientMultiState(client *c);
...@@ -1455,6 +1461,7 @@ robj *createIntsetObject(void); ...@@ -1455,6 +1461,7 @@ robj *createIntsetObject(void);
robj *createHashObject(void); robj *createHashObject(void);
robj *createZsetObject(void); robj *createZsetObject(void);
robj *createZsetZiplistObject(void); robj *createZsetZiplistObject(void);
robj *createStreamObject(void);
robj *createModuleObject(moduleType *mt, void *value); robj *createModuleObject(moduleType *mt, void *value);
int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg); int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg);
int checkType(client *c, robj *o, int type); int checkType(client *c, robj *o, int type);
...@@ -1992,6 +1999,8 @@ void pfdebugCommand(client *c); ...@@ -1992,6 +1999,8 @@ void pfdebugCommand(client *c);
void latencyCommand(client *c); void latencyCommand(client *c);
void moduleCommand(client *c); void moduleCommand(client *c);
void securityWarningCommand(client *c); void securityWarningCommand(client *c);
void xaddCommand(client *c);
void xrangeCommand(client *c);
#if defined(__GNUC__) #if defined(__GNUC__)
void *calloc(size_t count, size_t size) __attribute__ ((deprecated)); void *calloc(size_t count, size_t size) __attribute__ ((deprecated));
......
#ifndef STREAM_H
#define STREAM_H
#include "rax.h"
/* Stream item ID: a 128 bit number composed of a milliseconds time and
* a sequence counter. IDs generated in the same millisecond (or in a past
* millisecond if the clock jumped backward) will use the millisecond time
* of the latest generated ID and an incremented sequence. */
typedef struct streamID {
uint64_t ms; /* Unix time in milliseconds. */
uint64_t seq; /* Sequence number. */
} streamID;
typedef struct stream {
rax *rax; /* The radix tree holding the stream. */
uint64_t length; /* Number of elements inside this stream. */
streamID last_id; /* Zero if there are yet no items. */
} stream;
#endif
/*
* Copyright (c) 2017, 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.
*/
/* TODO:
* - After loading a stream, populate the last ID.
*/
#include "server.h"
#include "listpack.h"
#include "endianconv.h"
#include "stream.h"
#define STREAM_BYTES_PER_LISTPACK 4096
/* -----------------------------------------------------------------------
* Low level stream encoding: a radix tree of listpacks.
* ----------------------------------------------------------------------- */
/* Create a new stream data structure. */
stream *streamNew(void) {
stream *s = zmalloc(sizeof(*s));
s->rax = raxNew();
s->length = 0;
s->last_id.ms = 0;
s->last_id.seq = 0;
return s;
}
/* Generate the next stream item ID given the previous one. If the current
* milliseconds Unix time is greater than the previous one, just use this
* as time part and start with sequence part of zero. Otherwise we use the
* previous time (and never go backward) and increment the sequence. */
void streamNextID(streamID *last_id, streamID *new_id) {
uint64_t ms = mstime();
if (ms > last_id->ms) {
new_id->ms = ms;
new_id->seq = 0;
} else {
new_id->ms = last_id->ms;
new_id->seq = last_id->seq+1;
}
}
/* This is just a wrapper for lpAppend() to directly use a 64 bit integer
* instead of a string. */
unsigned char *lpAppendInteger(unsigned char *lp, int64_t value) {
char buf[LONG_STR_SIZE];
int slen = ll2string(buf,sizeof(buf),value);
return lpAppend(lp,(unsigned char*)buf,slen);
}
/* This is a wrapper function for lpGet() to directly get an integer value
* from the listpack (that may store numbers as a string), converting
* the string if needed. */
int64_t lpGetInteger(unsigned char *ele) {
int64_t v;
unsigned char *e = lpGet(ele,&v,NULL);
if (e == NULL) return v;
/* The following code path should never be used for how listpacks work:
* they should always be able to store an int64_t value in integer
* encoded form. However the implementation may change. */
int retval = string2ll((char*)e,v,&v);
serverAssert(retval != 0);
return v;
}
/* Convert the specified stream entry ID as a 128 bit big endian number, so
* that the IDs can be sorted lexicographically. */
void streamEncodeID(void *buf, streamID *id) {
uint64_t e[2];
e[0] = htonu64(id->ms);
e[1] = htonu64(id->seq);
memcpy(buf,e,sizeof(e));
}
/* This is the reverse of streamEncodeID(): the decoded ID will be stored
* in the 'id' structure passed by reference. The buffer 'buf' must point
* to a 128 bit big-endian encoded ID. */
void streamDecodeID(void *buf, streamID *id) {
uint64_t e[2];
memcpy(e,buf,sizeof(e));
id->ms = ntohu64(e[0]);
id->seq = ntohu64(e[1]);
}
/* Adds a new item into the stream 's' having the specified number of
* field-value pairs as specified in 'numfields' and stored into 'argv'.
* Returns the new entry ID populating the 'added_id' structure. */
void streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id) {
raxIterator ri;
raxStart(&ri,s->rax);
raxSeek(&ri,"$",NULL,0);
size_t lp_bytes = 0; /* Total bytes in the tail listpack. */
unsigned char *lp = NULL; /* Tail listpack pointer. */
/* Get a reference to the tail node listpack. */
if (raxNext(&ri)) {
lp = ri.data;
lp_bytes = lpBytes(lp);
}
raxStop(&ri);
/* Generate the new entry ID. */
streamID id;
streamNextID(&s->last_id,&id);
/* We have to add the key into the radix tree in lexicographic order,
* to do so we consider the ID as a single 128 bit number written in
* big endian, so that the most significant bytes are the first ones. */
uint64_t rax_key[2]; /* Key in the radix tree containing the listpack.*/
uint64_t entry_id[2]; /* Entry ID of the new item as 128 bit string. */
streamEncodeID(entry_id,&id);
/* Create a new listpack and radix tree node if needed. */
if (lp == NULL || lp_bytes > STREAM_BYTES_PER_LISTPACK) {
lp = lpNew();
rax_key[0] = entry_id[0];
rax_key[1] = entry_id[1];
raxInsert(s->rax,(unsigned char*)&rax_key,sizeof(rax_key),lp,NULL);
} else {
serverAssert(ri.key_len == sizeof(rax_key));
memcpy(rax_key,ri.key,sizeof(rax_key));
}
/* Populate the listpack with the new entry. */
lp = lpAppend(lp,(unsigned char*)entry_id,sizeof(entry_id));
lp = lpAppendInteger(lp,numfields);
for (int i = 0; i < numfields; i++) {
sds field = argv[i*2]->ptr, value = argv[i*2+1]->ptr;
lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
lp = lpAppend(lp,(unsigned char*)value,sdslen(value));
}
/* Insert back into the tree in order to update the listpack pointer. */
raxInsert(s->rax,(unsigned char*)&rax_key,sizeof(rax_key),lp,NULL);
s->length++;
s->last_id = id;
if (added_id) *added_id = id;
raxShow(s->rax);
}
/* Send the specified range to the client 'c'. The range the client will
* receive is between start and end inclusive, if 'count' is non zero, no more
* than 'count' elemnets are sent. The 'end' pointer can be NULL to mean that
* we want all the elements from 'start' till the end of the stream. */
size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count) {
void *arraylen_ptr = addDeferredMultiBulkLength(c);
size_t arraylen = 0;
/* Seek the radix tree node that contains our start item. */
uint64_t key[2];
uint64_t end_key[2];
streamEncodeID(key,start);
if (end) streamEncodeID(end_key,end);
raxIterator ri;
raxStart(&ri,s->rax);
/* Seek the correct node in the radix tree. */
if (start->ms || start->seq) {
raxSeek(&ri,"<=",(unsigned char*)key,sizeof(key));
if (raxEOF(&ri)) raxSeek(&ri,">",(unsigned char*)key,sizeof(key));
} else {
raxSeek(&ri,"^",NULL,0);
}
/* For every radix tree node, iterate the corresponding listpack,
* returning elmeents when they are within range. */
while (raxNext(&ri)) {
serverAssert(ri.key_len == sizeof(key));
unsigned char *lp = ri.data;
unsigned char *lp_ele = lpFirst(lp);
while(lp_ele) {
int64_t e_len;
unsigned char buf[LP_INTBUF_SIZE];
unsigned char *e = lpGet(lp_ele,&e_len,buf);
serverAssert(e_len == sizeof(streamID));
/* Seek next field: number of elements. */
lp_ele = lpNext(lp,lp_ele);
if (memcmp(e,key,sizeof(key)) >= 0) { /* If current >= start */
if (end && memcmp(e,end_key,sizeof(key)) > 0) {
break; /* We are already out of range. */
}
streamID thisid;
streamDecodeID(e,&thisid);
sds replyid = sdscatfmt(sdsempty(),"+%U.%U\r\n",
thisid.ms,thisid.seq);
/* Emit this stream entry in the client output. */
addReplyMultiBulkLen(c,2);
addReplySds(c,replyid);
int64_t numfields = lpGetInteger(lp_ele);
lp_ele = lpNext(lp,lp_ele);
addReplyMultiBulkLen(c,numfields*2);
for (int64_t i = 0; i < numfields; i++) {
/* Emit two items (key-value) per iteration. */
for (int k = 0; k < 2; k++) {
e = lpGet(lp_ele,&e_len,buf);
addReplyBulkCBuffer(c,e,e_len);
lp_ele = lpNext(lp,lp_ele);
}
}
arraylen++;
if (count && count == arraylen) break;
} else {
/* If we do not emit, we have to discard. */
int64_t numfields = lpGetInteger(lp_ele);
lp_ele = lpNext(lp,lp_ele);
for (int64_t i = 0; i < numfields*2; i++)
lp_ele = lpNext(lp,lp_ele);
}
}
if (count && count == arraylen) break;
}
raxStop(&ri);
setDeferredMultiBulkLength(c,arraylen_ptr,arraylen);
return arraylen;
}
/* -----------------------------------------------------------------------
* Stream commands implementation
* ----------------------------------------------------------------------- */
/* Look the stream at 'key' and return the corresponding stream object.
* The function creates a key setting it to an empty stream if needed. */
robj *streamTypeLookupWriteOrCreate(client *c, robj *key) {
robj *o = lookupKeyWrite(c->db,key);
if (o == NULL) {
o = createStreamObject();
dbAdd(c->db,key,o);
} else {
if (o->type != OBJ_STREAM) {
addReply(c,shared.wrongtypeerr);
return NULL;
}
}
return o;
}
/* Helper function to convert a string to an unsigned long long value.
* The function attempts to use the faster string2ll() function inside
* Redis: if it fails, strtoull() is used instead. The function returns
* 1 if the conversion happened successfully or 0 if the number is
* invalid or out of range. */
int string2ull(const char *s, unsigned long long *value) {
long long ll;
if (string2ll(s,strlen(s),&ll)) {
if (ll < 0) return 0; /* Negative values are out of range. */
*value = ll;
return 1;
}
errno = 0;
*value = strtoull(s,NULL,10);
if (errno == EINVAL || errno == ERANGE) return 0; /* strtoull() failed. */
return 1; /* Conversion done! */
}
/* Parse a stream ID in the format given by clients to Redis, that is
* <ms>.<seq>, and converts it into a streamID structure. If
* the specified ID is invalid C_ERR is returned and an error is reported
* to the client, otherwise C_OK is returned. The ID may be in incomplete
* form, just stating the milliseconds time part of the stream. In such a case
* the missing part is set according to the value of 'missing_seq' parameter.
* The IDs "-" and "+" specify respectively the minimum and maximum IDs
* that can be represented. */
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
char buf[128];
if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid;
memcpy(buf,o->ptr,sdslen(o->ptr)+1);
/* Handle the "-" and "+" special cases. */
if (buf[0] == '-' && buf[1] == '\0') {
id->ms = 0;
id->seq = 0;
return C_OK;
} else if (buf[0] == '+' && buf[1] == '\0') {
id->ms = UINT64_MAX;
id->seq = UINT64_MAX;
return C_OK;
}
/* Parse <ms>.<seq> form. */
char *dot = strchr(buf,'.');
if (dot) *dot = '\0';
uint64_t ms, seq;
if (string2ull(buf,&ms) == 0) goto invalid;
if (dot && string2ull(dot+1,&seq) == 0) goto invalid;
if (!dot) seq = missing_seq;
id->ms = ms;
id->seq = seq;
return C_OK;
invalid:
addReplyError(c,"Invalid stream ID specified as stream command argument");
return C_ERR;
}
/* XADD key [field value] [field value] ... */
void xaddCommand(client *c) {
if ((c->argc % 2) == 1) {
addReplyError(c,"wrong number of arguments for XADD");
return;
}
/* Lookup the stream at key. */
robj *o;
stream *s;
if ((o = streamTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
s = o->ptr;
/* Append using the low level function and return the ID. */
streamID id;
streamAppendItem(s,c->argv+2,(c->argc-2)/2,&id);
sds reply = sdscatfmt(sdsempty(),"+%U.%U\r\n",id.ms,id.seq);
addReplySds(c,reply);
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_HASH,"xadd",c->argv[1],c->db->id);
server.dirty++;
}
/* XRANGE key start end [COUNT <n>] */
void xrangeCommand(client *c) {
robj *o;
stream *s;
streamID startid, endid;
long long count = 0;
if (streamParseIDOrReply(c,c->argv[2],&startid,0) == C_ERR) return;
if (streamParseIDOrReply(c,c->argv[3],&endid,UINT64_MAX) == C_ERR) return;
/* Parse the COUNT option if any. */
if (c->argc > 4) {
if (strcasecmp(c->argv[4]->ptr,"COUNT") == 0) {
if (getLongLongFromObjectOrReply(c,c->argv[5],&count,NULL) != C_OK)
return;
} else {
addReply(c,shared.syntaxerr);
return;
}
}
/* Return the specified range to the user. */
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
|| checkType(c,o,OBJ_STREAM)) return;
s = o->ptr;
streamReplyWithRange(c,s,&startid,&endid,count);
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment