Compare commits

..
Author SHA1 Message Date
Paul ChenandPaul Chen 37fd5d25fe [BUG] Moves key to another DB, the source key was removed if the move failed due to the key exists in the destination db #497 (#498)
Co-authored-by: Paul Chen <mingchen@Mings-MacBook-Pro.local>
2022-10-20 18:06:58 -04:00
John Sully e8e3666dcb Workaround bug in snapshot sync - abort don't crash 2022-09-28 19:00:26 +00:00
John Sully 597b5669e3 Fix Issue #486 2022-09-14 14:48:28 +00:00
Ilya Shipitsin 772bf47ef7 src\redis-cli.c: fix potential null pointer dereference found by cppcheck
src\redis-cli.c:5488:35: warning: Either the condition
'!table' is redundant or there is possible null pointer dereference:
table. [nullPointerRedundantCheck]
2022-09-14 10:03:08 -04:00
root 7bf9c02954 avoid from instatiating EpochHolder multiple times to improve performance and cpu utilization 2022-09-14 10:02:02 -04:00
root 8661cfe7bd avoid from instatiating EpochHolder multiple times to improve performance and cpu utilization 2022-09-14 10:02:02 -04:00
Karthick Ariyaratnam (A) 2173975aaf Remove old assertion which is commented out. 2022-09-14 10:01:21 -04:00
Karthick Ariyaratnam (A) b989cd80c1 Fixes a crash where the server assertion failed when the key exists in DB during RDB load 2022-09-14 10:01:21 -04:00
John Sully 79b6c65185 Prevent crash on load in long running KeyDB instances 2022-09-01 15:26:26 +00:00
John Sully e114f41d5c Fix race removing key cache 2022-09-01 15:16:27 +00:00
John Sully a2e6e35d5e Fix crash running INFO command while a disk based backlog is set 2022-08-01 04:53:56 +00:00
John Sully 9e5d1c8ec9 Fix issue #480 2022-08-01 03:14:09 +00:00
John Sully 1948f21d0d Do not allow commands to run in background when in eval, Issue #452 2022-07-06 04:26:13 +00:00
John Sully 3cba0e0ba4 Fix issue #454 (BSD build break) 2022-07-06 04:04:47 +00:00
guybe7 3a4f46104c lpGetInteger returns int64_t, avoid overflow (#10068)
Fix #9410

Crucial for the ms and sequence deltas, but I changed all
calls, just in case (e.g. "flags")

Before this commit:
`ms_delta` and `seq_delta` could have overflown, causing `currid` to be wrong,
which in turn would cause `streamTrim` to trim the entire rax node (see new test)
2022-07-06 03:44:24 +00:00
benschermel aa032d3066 Merge pull request #438 from Snapchat/update_pkg
Packaging Updates
2022-06-06 22:38:28 -04:00
benschermel 7c9c448441 Merge pull request #443 from Snapchat/update_gem_ref
update gem reference
2022-06-06 22:35:11 -04:00
benschermel accd3871be update gem reference 2022-06-06 14:55:24 -04:00
benschermel 6ad5cabe15 update deb pkg references and changelog 2022-05-25 01:10:56 -04:00
benschermel a9ec2fc738 update dockerfiles 2022-05-25 01:08:24 -04:00
72 changed files with 767 additions and 1653 deletions

No files matched your search

+18 -23
View File
@@ -8,52 +8,40 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
with:
submodules: recursive
- name: make
run: |
sudo apt-get update
sudo apt-get -y remove libzstd || true
sudo apt-get -y install uuid-dev libcurl4-openssl-dev libbz2-dev zlib1g-dev libsnappy-dev liblz4-dev libzstd-dev libgflags-dev
make BUILD_TLS=yes -j2 KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror'
sudo apt-get -y install uuid-dev libcurl4-openssl-dev
make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' BUILD_TLS=yes -j2
- name: gen-cert
run: ./utils/gen-test-certs.sh
- name: test-tls
run: |
sudo apt-get -y install tcl tcl-tls
./runtest --clients 1 --verbose --tls --config server-threads 3
./runtest --clients 2 --verbose --tls
- name: cluster-test
run: |
./runtest-cluster --tls --config server-threads 3
./runtest-cluster --tls
- name: sentinel test
run: |
./runtest-sentinel
- name: module tests
run: |
./runtest-moduleapi
- name: rotation test
run: |
./runtest-rotation
build-ubuntu-old:
runs-on: ubuntu-20.04
runs-on: ubuntu-18.04
steps:
- uses: actions/checkout@v2
with:
submodules: recursive
- uses: actions/checkout@v1
- name: make -j2
run: |
sudo apt-get update
sudo apt-get -y remove libzstd || true
sudo apt-get -y install uuid-dev libcurl4-openssl-dev libbz2-dev zlib1g-dev libsnappy-dev liblz4-dev libzstd-dev libgflags-dev
sudo apt-get -y install uuid-dev libcurl4-openssl-dev
make -j2
build-macos-latest:
runs-on: macos-latest
steps:
- uses: actions/checkout@v2
with:
submodules: recursive
- name: make
run: make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' -j2
@@ -61,11 +49,18 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
with:
submodules: recursive
- name: make
run: |
sudo apt-get update
sudo apt-get -y remove libzstd || true
sudo apt-get -y install uuid-dev libcurl4-openssl-dev libbz2-dev zlib1g-dev libsnappy-dev liblz4-dev libzstd-dev libgflags-dev
sudo apt-get -y install uuid-dev libcurl4-openssl-dev
make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' MALLOC=libc -j2
build-ubuntu-32bit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: make
run: |
sudo dpkg --add-architecture i386
sudo apt-get update
sudo apt-get -y install gcc-multilib g++-multilib libc6-dev-i386 lib32z1 uuid-dev:i386 libcurl4-openssl-dev:i386
make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' 32bit -j2
+164
View File
@@ -0,0 +1,164 @@
.standard-pipeline:
rules:
- if: '$COVERAGE'
when: never
- if: '$ENDURANCE'
when: never
- if: '$CI_PIPELINE_SOURCE == "push"'
build:
extends: .standard-pipeline
tags:
- docker
stage: build
script:
- git submodule update --init
- make distclean
- make -j
artifacts:
paths:
- src/
runtest:
extends: .standard-pipeline
dependencies:
- build
tags:
- docker
stage: test
script:
- ./runtest --config server-threads 3
runtest-cluster:
extends: .standard-pipeline
dependencies:
- build
tags:
- docker
stage: test
script:
- ./runtest-cluster
runtest-moduleapi:
extends: .standard-pipeline
dependencies:
- build
tags:
- docker
stage: test
script:
- ./runtest-moduleapi
runtest-sentinel:
extends: .standard-pipeline
dependencies:
- build
tags:
- docker
stage: test
script:
- ./runtest-sentinel
node-redis-test:
extends: .standard-pipeline
dependencies:
- build
rules:
- when: never
tags:
- docker
- ipv6
stage: test
script:
- cp -pf src/keydb-server /usr/local/bin
- cp -pf src/keydb-cli /usr/local/bin
- git clone https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.eqalpha.com/keydb-dev/node-redis.git
- cd node-redis
- npm install
- npm run test
jedis-test:
extends: .standard-pipeline
dependencies:
- build
tags:
- docker
- ipv4
stage: test
script:
- cp -pf src/keydb-server /usr/local/bin
- cp -pf src/keydb-cli /usr/local/bin
- git clone https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.eqalpha.com/keydb-dev/jedis.git
- cd jedis
- make test
redis-rs-test:
extends: .standard-pipeline
dependencies:
- build
tags:
- docker
stage: test
script:
- cp -pf src/keydb-server /usr/local/bin
- cp -pf src/keydb-cli /usr/local/bin
- git clone https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.eqalpha.com/keydb-dev/redis-rs.git
- cd redis-rs
- make test
endurance-test:
rules:
- if: '$ENDURANCE'
tags:
- docker
stage: test
script:
- git submodule update --init
- make distclean
- make -j
- ./runtest --config server-threads 3 --loop --stop
coverage-test:
rules:
- if: '$COVERAGE'
tags:
- docker
stage: test
script:
- git submodule update --init
- make distclean
- make gcov -j
- make install
- ./runtest || true
- pkill keydb-server || true
- pkill stunnel || true
- ./runtest-cluster || true
- pkill keydb-server || true
- pkill stunnel || true
- ./runtest-sentinel || true
- pkill keydb-server || true
- pkill stunnel || true
- ./runtest-moduleapi || true
- pkill keydb-server || true
- pkill stunnel || true
- git clone https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.eqalpha.com/keydb-dev/redis-rs.git
- cd redis-rs
- make test || true
- pkill keydb-server || true
- pkill stunnel || true
- cd ..
- git clone https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.eqalpha.com/keydb-dev/jedis.git
- cd jedis
- make test || true
- pkill keydb-server || true
- pkill stunnel || true
- cd ..
- git clone https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.eqalpha.com/keydb-dev/node-redis.git
- cd node-redis
- npm install
- npm run test || true
- pkill keydb-server || true
- pkill stunnel || true
- cd ..
- geninfo -o KeyDB.info --no-external .
- genhtml --legend -o lcov-html KeyDB.info
+4 -16
View File
@@ -4,7 +4,7 @@
##### KeyDB is now a part of Snap Inc! Check out the announcement [here](https://docs.keydb.dev/news/2022/05/12/keydb-joins-snap)
##### [Release v6.3.0](https://github.com/EQ-Alpha/KeyDB/releases/tag/v6.3.0) is here with major improvements as we consolidate our Open Source and Enterprise offerings into a single BSD-3 licensed project. See our [roadmap](https://docs.keydb.dev/docs/coming-soon) for details.
##### [Release v6.3.0](https://github.com/EQ-Alpha/KeyDB/releases/tag/v6.3.0) is here with major improvements as we consolodate our Open Source and Enterprise offerings into a single BSD-3 licensed project. See our [roadmap](https://docs.keydb.dev/docs/coming-soon) for details.
##### Want to extend KeyDB with Javascript? Try [ModJS](https://github.com/JohnSully/ModJS)
@@ -37,7 +37,7 @@ Because of this difference of opinion features which are right for KeyDB may not
Project Support
-------------------
The KeyDB team maintains this project as part of Snap Inc. KeyDB is used by Snap as part of its caching infrastructure and is fully open sourced. There is no separate commercial product and no paid support options available. We really value collaborating with the open source community and welcome PRs, bug reports, and open discussion. For community support or to get involved further with the project check out our community support options [here](https://docs.keydb.dev/docs/support) (slack, forum, meetup, github issues). Our team monitors these channels regularly.
The KeyDB team maintains this project as part of Snap Inc. KeyDB is used by Snap as part of its caching infrastructure and is fully open sourced. There is no separate commercial product and no paid support options available. We really value collaborating with the open source community and welcome PRs, bug reports, and open discussion. For community support or to get involved further with the project check out our community support options [here](https://docs.keydb.dev/docs/support) (slack, forum, meetup, github issues). Our team monitors these channlels regularly.
Additional Resources
@@ -104,12 +104,6 @@ Avoid forwarding RREPLAY messages to other masters? WARNING: This setting is dan
If you would like KeyDB to dump and load directly to AWS S3 this option specifies the bucket. Using this option with the traditional RDB options will result in KeyDB backing up twice to both locations. If both are specified KeyDB will first attempt to load from the local dump file and if that fails load from S3. This requires the AWS CLI tools to be installed and configured which are used under the hood to transfer the data.
```
storage-provider flash /path/to/flash
```
If you would like to use KeyDB FLASH storage, specify the storage medium followed by the directory path on your local SSD volume. Note that this feature is still considered experimental and should be used with discretion. See [FLASH Documentation](https://docs.keydb.dev/docs/flash) for more details on configuration and setting up your FLASH volume.
Building KeyDB
--------------
@@ -141,9 +135,9 @@ To append a suffix to KeyDB program names, use:
***Note that the following dependencies may be needed:
% sudo apt-get install autoconf autotools-dev libnuma-dev libtool
KeyDB by default is built with TLS enabled. To build without TLS support, use:
To buik=ld with TLS support, use:
% make BUILD_TLS=no
% make BUILD_TLS=yes
Running the tests with TLS enabled (you will need `tcl-tls`
installed):
@@ -151,12 +145,6 @@ installed):
% ./utils/gen-test-certs.sh
% ./runtest --tls
To build with KeyDB FLASH support, use:
% make ENABLE_FLASH=yes
***Note that the KeyDB FLASH feature is considered experimental (beta) and should used with discretion
Fixing build problems with dependencies or cached build options
---------
+48
View File
@@ -0,0 +1,48 @@
# Doc: https://wiki.sc-corp.net/pages/viewpage.action?pageId=121500284
version: 1
machamp:
keydb-build:
# Optional - build counter is linked to the build def
tag_template: 0.0.%build.counter%
# Optional - value in seconds before a build is terminated, default is 3600 seconds
timeout: 3600
# Optional - update ghe or not, default to true
update_ghe: true
code_coverage: false
# Required
steps:
make-build:
type: cmd
# https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
command: ./machamp_scripts/build.sh
tls-test:
type: cmd
parent: make-build
# https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
command: ./runtest --clients $(nproc) --verbose --tls
cluster-test:
type: cmd
parent: make-build
# https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
command: ./runtest-cluster --tls
sentinel-test:
type: cmd
parent: make-build
# https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
command: ./runtest-sentinel
module-test:
type: cmd
parent: make-build
# https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
command: ./runtest-moduleapi
rotation-test:
type: cmd
parent: make-build
# https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
command: ./runtest-rotation
+16
View File
@@ -0,0 +1,16 @@
# Doc: https://wiki.sc-corp.net/display/TOOL/ci.yaml+User+Guide
version: 1
on:
pull_request:
- workflows:
# All builds that use machamp should use the defined `backend_workflow`
- workflow_type: backend_workflow
# references a build defined in build.yaml
build_name: keydb-build
arch_types: ["amd64", "arm64"]
push:
- branches: [master]
workflows:
- workflow_type: backend_workflow
build_name: keydb-build
arch_types: ["amd64", "arm64"]
+3 -3
View File
@@ -101,11 +101,11 @@ jemalloc: .make-prerequisites
.PHONY: jemalloc
rocksdb: .make-prerequisites
@printf '%b %b\n' $(MAKECOLOR)MAKE$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR)
# @printf '%b %b\n' $(MAKECOLOR)MAKE$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR)
ifeq ($(uname_M),x86_64)
cd rocksdb && CFLAGS=-Wno-error PORTABLE=1 USE_SSE=1 FORCE_SSE42=1 $(MAKE) static_lib
# cd rocksdb && CFLAGS=-Wno-error PORTABLE=1 USE_SSE=1 FORCE_SSE42=1 $(MAKE) static_lib
else
cd rocksdb && PORTABLE=1 $(MAKE) static_lib
# cd rocksdb && PORTABLE=1 $(MAKE) static_lib
endif
.PHONY: rocksdb
+2 -18
View File
@@ -184,9 +184,7 @@ tcp-keepalive 300
#
# tls-client-key-file-pass secret
# Configure a DH parameters file to enable Diffie-Hellman (DH) key exchange,
# required by older versions of OpenSSL (<3.0). Newer versions do not require
# this configuration and recommend against it.
# Configure a DH parameters file to enable Diffie-Hellman (DH) key exchange:
#
# tls-dh-params-file keydb.dh
@@ -2073,18 +2071,4 @@ replica-weighting-factor 2
#
# By default this is enabled
#
active-client-balancing yes
# Enable FLASH support (Experimental Feature)
# storage-provider flash /path/to/flash/db
# Blob support is a way to store very large objects (>200MB) on disk
# The files are automatically cleaned up when KeyDB exits and are only
# for temporary use. This helps reduce memory pressure for very large
# data items at the cost of some performance.
#
# By default this config is disable. When enabled the disk associated
# with KeyDB's working directory will be used. If there is insufficient
# disk space or any other I/O error KeyDB will instead use memory.
#
# blob-support false
active-client-balancing yes
-14
View File
@@ -2063,17 +2063,3 @@ server-threads 2
#
# By default KeyDB sets this to 2.
replica-weighting-factor 2
# Enable FLASH support (Experimental Feature)
# storage-provider flash /path/to/flash/db
# Blob support is a way to store very large objects (>200MB) on disk
# The files are automatically cleaned up when KeyDB exits and are only
# for temporary use. This helps reduce memory pressure for very large
# data items at the cost of some performance.
#
# By default this config is disable. When enabled the disk associated
# with KeyDB's working directory will be used. If there is insufficient
# disk space or any other I/O error KeyDB will instead use memory.
#
# blob-support false
-5
View File
@@ -19,11 +19,6 @@ elif [ "$distributor" == "Ubuntu" ]; then
fi
codename=$(lsb_release --codename --short)
date=$(date +%a," "%d" "%b" "%Y" "%T)
# overwrite debian bookworm version until updated
if [ $codename == "bookworm" ]; then
distname=+deb12u1
fi
pkg_name=keydb-$majorv:$version$distname
# create build tree
-1
View File
@@ -4,7 +4,6 @@ include /usr/share/dpkg/buildflags.mk
export BUILD_TLS=yes
export USE_SYSTEMD=yes
export ENABLE_FLASH=yes
export CFLAGS CPPFLAGS LDFLAGS
export DEB_BUILD_MAINT_OPTIONS = hardening=+all
export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)
-1
View File
@@ -10,7 +10,6 @@ include /usr/share/dpkg/buildflags.mk
export BUILD_TLS=yes
export USE_SYSTEMD=yes
export ENABLE_FLASH=yes
export CFLAGS CPPFLAGS LDFLAGS
export DEB_BUILD_MAINT_OPTIONS = hardening=+all
export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)
+1 -1
View File
@@ -1,4 +1,4 @@
FROM ubuntu:20.04
FROM ubuntu:18.04
SHELL ["/bin/bash","-c"]
RUN groupadd -r keydb && useradd -r -g keydb keydb
# use gosu for easy step-down from root: https://github.com/tianon/gosu/releases
-4
View File
@@ -21,10 +21,6 @@ RUN set -eux; \
bash \
perl \
git \
bzip2-dev \
zstd-dev \
lz4-dev \
snappy-dev \
; \
cd /tmp && git clone --branch $BRANCH https://github.com/Snapchat/KeyDB.git --recursive; \
cd /tmp/KeyDB; \
@@ -2063,17 +2063,3 @@ server-threads 2
#
# By default KeyDB sets this to 2.
replica-weighting-factor 2
# Enable FLASH support (Experimental Feature)
# storage-provider flash /path/to/flash/db
# Blob support is a way to store very large objects (>200MB) on disk
# The files are automatically cleaned up when KeyDB exits and are only
# for temporary use. This helps reduce memory pressure for very large
# data items at the cost of some performance.
#
# By default this config is disable. When enabled the disk associated
# with KeyDB's working directory will be used. If there is insufficient
# disk space or any other I/O error KeyDB will instead use memory.
#
# blob-support false
-1
View File
@@ -37,6 +37,5 @@ $TCLSH tests/test_helper.tcl \
--single unit/moduleapi/hash \
--single unit/moduleapi/zset \
--single unit/moduleapi/stream \
--single unit/moduleapi/load \
--config server-threads 3 \
"${@}"
+3 -13
View File
@@ -20,7 +20,7 @@ release_hdr := $(shell sh -c './mkreleasehdr.sh')
uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not')
uname_M := $(shell sh -c 'uname -m 2>/dev/null || echo not')
OPTIMIZATION?=-O2 -flto
DEPENDENCY_TARGETS=hiredis linenoise lua hdr_histogram
DEPENDENCY_TARGETS=hiredis linenoise lua hdr_histogram rocksdb
NODEPS:=clean distclean
# Default settings
@@ -62,7 +62,6 @@ endif
endif
USEASM?=true
ENABLE_FLASH?=no
ifneq ($(strip $(SANITIZE)),)
CFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE -fno-omit-frame-pointer
@@ -72,15 +71,6 @@ ifneq ($(strip $(SANITIZE)),)
USEASM=false
endif
ifeq ($(ENABLE_FLASH),yes)
FINAL_LIBS+= -lz -lcrypto -lbz2 -lzstd -llz4 -lsnappy
CXXFLAGS+= -I../deps/rocksdb/include/ -DENABLE_ROCKSDB
STORAGE_OBJ+= storage/rocksdb.o storage/rocksdbfactory.o
FINAL_LIBS+= ../deps/rocksdb/librocksdb.a
DEPENDENCY_TARGETS+= rocksdb
endif
ifeq ($(CHECKED),true)
CXXFLAGS+= -DCHECKED_BUILD
endif
@@ -142,7 +132,7 @@ DEBUG=-g -ggdb
FINAL_CFLAGS=$(STD) $(WARN) $(OPT) $(DEBUG) $(CFLAGS) $(KEYDB_CFLAGS) $(REDIS_CFLAGS)
FINAL_CXXFLAGS=$(CXX_STD) $(WARN) $(OPT) $(DEBUG) $(CXXFLAGS) $(KEYDB_CFLAGS) $(REDIS_CFLAGS)
FINAL_LDFLAGS=$(LDFLAGS) $(KEYDB_LDFLAGS) $(DEBUG)
FINAL_LIBS+=-lm -lz -lcrypto -lbz2 -lzstd -llz4 -lsnappy
FINAL_LIBS+=-lm -lz -lcrypto
ifneq ($(uname_S),Darwin)
FINAL_LIBS+=-latomic
@@ -376,7 +366,7 @@ endif
REDIS_SERVER_NAME=keydb-server$(PROG_SUFFIX)
REDIS_SENTINEL_NAME=keydb-sentinel$(PROG_SUFFIX)
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 t_nhash.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 crcspeed.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.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o connection.o tls.o sha256.o motd_server.o timeout.o setcpuaffinity.o AsyncWorkQueue.o snapshot.o storage/teststorageprovider.o keydbutils.o StorageCache.o monotonic.o cli_common.o mt19937-64.o $(ASM_OBJ) $(STORAGE_OBJ)
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 t_nhash.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 crcspeed.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.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o connection.o tls.o sha256.o motd_server.o timeout.o setcpuaffinity.o AsyncWorkQueue.o snapshot.o storage/teststorageprovider.o keydbutils.o StorageCache.o monotonic.o cli_common.o mt19937-64.o $(ASM_OBJ)
KEYDB_SERVER_OBJ=SnapshotPayloadParseState.o
REDIS_CLI_NAME=keydb-cli$(PROG_SUFFIX)
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o redis-cli-cpphelper.o zmalloc.o release.o anet.o ae.o crcspeed.o crc64.o siphash.o crc16.o storage-lite.o fastlock.o motd_client.o monotonic.o cli_common.o mt19937-64.o $(ASM_OBJ)
-2
View File
@@ -43,8 +43,6 @@ void StorageCache::clear(void(callback)(void*))
void StorageCache::clearAsync()
{
std::unique_lock<fastlock> ul(m_lock);
if (count() == 0)
return;
if (m_pdict != nullptr) {
dict *dSav = m_pdict;
m_pdict = dictCreate(&dbStorageCacheType, nullptr);
+5 -10
View File
@@ -716,6 +716,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
}
}
if (eventLoop->flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
}
if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP) {
std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE)) {
@@ -726,11 +731,6 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
eventLoop->beforesleep(eventLoop);
}
if (eventLoop->flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
}
/* Call the multiplexing API, will return only on timeout or when
* some event fires. */
numevents = aeApiPoll(eventLoop, tvp);
@@ -859,11 +859,6 @@ void aeReleaseForkLock()
g_forkLock.downgradeWrite();
}
void aeForkLockInChild()
{
g_forkLock.setNotify(false);
}
int aeThreadOwnsLock()
{
return fOwnLockOverride || g_lock.fOwnLock();
-1
View File
@@ -171,7 +171,6 @@ int aeTryAcquireLock(int fWeak);
void aeThreadOffline();
void aeReleaseLock();
void aeReleaseForkLock();
void aeForkLockInChild();
int aeThreadOwnsLock();
void aeSetThreadOwnsLockOverride(int fOverride);
int aeLockContested(int threshold);
+1 -1
View File
@@ -441,7 +441,7 @@ static int _anetTcpServer(char *err, int port, const char *bindaddr, int af, int
char _port[6]; /* strlen("65535") */
struct addrinfo hints, *servinfo, *p;
snprintf(_port,sizeof(_port),"%d",port);
snprintf(_port,6,"%d",port);
memset(&hints,0,sizeof(hints));
hints.ai_family = af;
hints.ai_socktype = SOCK_STREAM;
+11 -11
View File
@@ -285,7 +285,7 @@ int startAppendOnly(void) {
strerror(errno));
return C_ERR;
}
if (hasActiveChildProcessOrBGSave() && g_pserver->child_type != CHILD_TYPE_AOF) {
if (hasActiveChildProcess() && g_pserver->child_type != CHILD_TYPE_AOF) {
g_pserver->aof_rewrite_scheduled = 1;
serverLog(LL_WARNING,"AOF was enabled but there is already another background operation. An AOF background was scheduled to start when possible.");
} else {
@@ -438,7 +438,7 @@ void flushAppendOnlyFile(int force) {
* useful for graphing / monitoring purposes. */
if (sync_in_progress) {
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
} else if (hasActiveChildProcessOrBGSave()) {
} else if (hasActiveChildProcess()) {
latencyAddSampleIfNeeded("aof-write-active-child",latency);
} else {
latencyAddSampleIfNeeded("aof-write-alone",latency);
@@ -535,7 +535,7 @@ void flushAppendOnlyFile(int force) {
try_fsync:
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
* children doing I/O in the background. */
if (g_pserver->aof_no_fsync_on_rewrite && hasActiveChildProcessOrBGSave())
if (g_pserver->aof_no_fsync_on_rewrite && hasActiveChildProcess())
return;
/* Perform the fsync if needed. */
@@ -752,7 +752,7 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
* accumulate the differences between the child DB and the current one
* in a buffer, so that when the child process will do its work we
* can append the differences to the new append only file. */
if (hasActiveChildProcess() && g_pserver->child_type == CHILD_TYPE_AOF)
if (g_pserver->child_type == CHILD_TYPE_AOF)
aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
sdsfree(buf);
@@ -1658,7 +1658,7 @@ int rewriteAppendOnlyFile(char *filename) {
{ // BEGIN GOTO SCOPED VARIABLES
/* Note that we have to use a different temp name here compared to the
* one used by rewriteAppendOnlyFileBackground() function. */
snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-%d.aof", (int) getpid());
snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
fp = fopen(tmpfile,"w");
if (!fp) {
serverLog(LL_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
@@ -1887,7 +1887,7 @@ void aofClosePipes(void) {
int rewriteAppendOnlyFileBackground(void) {
pid_t childpid;
if (hasActiveChildProcessOrBGSave()) return C_ERR;
if (hasActiveChildProcess()) return C_ERR;
if (aofCreatePipes() != C_OK) return C_ERR;
if ((childpid = redisFork(CHILD_TYPE_AOF)) == 0) {
char tmpfile[256];
@@ -1895,7 +1895,7 @@ int rewriteAppendOnlyFileBackground(void) {
/* Child */
redisSetProcTitle("keydb-aof-rewrite");
redisSetCpuAffinity(g_pserver->aof_rewrite_cpulist);
snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof", (int) getpid());
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
sendChildCowInfo(CHILD_INFO_TYPE_AOF_COW_SIZE, "AOF rewrite");
exitFromChild(0);
@@ -1930,7 +1930,7 @@ int rewriteAppendOnlyFileBackground(void) {
void bgrewriteaofCommand(client *c) {
if (g_pserver->child_type == CHILD_TYPE_AOF) {
addReplyError(c,"Background append only file rewriting already in progress");
} else if (hasActiveChildProcessOrBGSave()) {
} else if (hasActiveChildProcess()) {
g_pserver->aof_rewrite_scheduled = 1;
addReplyStatus(c,"Background append only file rewriting scheduled");
} else if (rewriteAppendOnlyFileBackground() == C_OK) {
@@ -1944,10 +1944,10 @@ void bgrewriteaofCommand(client *c) {
void aofRemoveTempFile(pid_t childpid) {
char tmpfile[256];
snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof", (int) childpid);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
bg_unlink(tmpfile);
snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-%d.aof", (int) childpid);
snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) childpid);
bg_unlink(tmpfile);
}
@@ -1985,7 +1985,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
/* Flush the differences accumulated by the parent to the
* rewritten AOF. */
latencyStartMonitor(latency);
snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof",
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
(int)g_pserver->child_pid);
newfd = open(tmpfile,O_WRONLY|O_APPEND);
if (newfd == -1) {
+1 -1
View File
@@ -322,7 +322,7 @@ handle_wrap:
return 1;
}
int checkSignedBitfieldOverflow(int64_t value, int64_t incr, int bits, int owtype, int64_t *limit) {
int checkSignedBitfieldOverflow(int64_t value, int64_t incr, uint64_t bits, int owtype, int64_t *limit) {
int64_t max = (bits == 64) ? INT64_MAX : (((int64_t)1<<(bits-1))-1);
int64_t min = (-max)-1;
+3 -5
View File
@@ -771,7 +771,7 @@ unsigned long getClusterConnectionsCount(void) {
/* We decrement the number of nodes by one, since there is the
* "myself" node too in the list. Each node uses two file descriptors,
* one incoming and one outgoing, thus the multiplication by 2. */
return g_pserver->cluster_enabled && g_pserver->cluster != nullptr ?
return g_pserver->cluster_enabled ?
((dictSize(g_pserver->cluster->nodes)-1)*2) : 0;
}
@@ -1525,9 +1525,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
* it's greater than our view but is not in the future
* (with 500 milliseconds tolerance) from the POV of our
* clock. */
mstime_t mstime;
__atomic_load(&g_pserver->mstime, &mstime, __ATOMIC_RELAXED);
if (pongtime <= (mstime+500) &&
if (pongtime <= (g_pserver->mstime+500) &&
pongtime > node->pong_received)
{
node->pong_received = pongtime;
@@ -5107,7 +5105,7 @@ void createDumpPayload(rio *payload, robj_roptr o, robj *key) {
serverAssert(rdbSaveObject(payload,o,key));
char szT[32];
uint64_t mvcc = mvccFromObj(o);
snprintf(szT, sizeof(szT), "%" PRIu64, mvcc);
snprintf(szT, 32, "%" PRIu64, mvcc);
serverAssert(rdbSaveAuxFieldStrStr(payload,"mvcc-tstamp", szT) != -1);
/* Write the footer, this is how it looks like:
+8 -18
View File
@@ -40,7 +40,6 @@
#endif
const char *KEYDB_SET_VERSION = KEYDB_REAL_VERSION;
size_t g_semiOrderedSetTargetBucketSize = 0; // Its a header only class so nowhere else for this to go
/*-----------------------------------------------------------------------------
* Config file name-value maps.
@@ -356,17 +355,16 @@ bool initializeStorageProvider(const char **err)
return true;
if (!strcasecmp(g_sdsProvider, "flash") && g_sdsArgs != nullptr)
{
#ifdef ENABLE_ROCKSDB
// Temporary: Disable FLASH
serverLog(LL_WARNING, "FLASH Is Not yet Supported");
_Exit(EXIT_FAILURE);
// Create The Storage Factory (if necessary)
serverLog(LL_NOTICE, "Initializing FLASH storage provider (this may take a long time)");
adjustOpenFilesLimit();
g_pserver->m_pstorageFactory = CreateRocksDBStorageFactory(g_sdsArgs, cserver.dbnum, cserver.storage_conf, cserver.storage_conf ? strlen(cserver.storage_conf) : 0);
#else
serverLog(LL_WARNING, "To use the flash storage provider please compile KeyDB with ENABLE_FLASH=yes");
serverLog(LL_WARNING, "Exiting due to the use of an unsupported storage provider");
exit(EXIT_FAILURE);
#endif
}
}
else if (!strcasecmp(g_sdsProvider, "test") && g_sdsArgs == nullptr)
{
g_pserver->m_pstorageFactory = new (MALLOC_LOCAL) TestStorageFactory();
@@ -768,12 +766,6 @@ void loadServerConfigFromString(char *config) {
g_sdsProvider = sdsdup(argv[1]);
if (argc > 2)
g_sdsArgs = sdsdup(argv[2]);
} else if (!strcasecmp(argv[0],"is-flash-enabled") && argc == 1) {
#ifdef ENABLE_ROCKSDB
exit(EXIT_SUCCESS);
#else
exit(EXIT_FAILURE);
#endif
} else {
err = "Bad directive or wrong number of arguments"; goto loaderr;
}
@@ -2588,7 +2580,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
adjustOpenFilesLimit();
if (g_pserver->maxclients != val) {
static char msg[128];
snprintf(msg, sizeof(msg), "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
sprintf(msg, "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
*err = msg;
if (g_pserver->maxclients > prev) {
g_pserver->maxclients = prev;
@@ -2627,7 +2619,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
if (res != AE_OK){
static char msg[128];
snprintf(msg, sizeof(msg),"Failed to post the request to change setsize for Thread %d", iel);
sprintf(msg, "Failed to post the request to change setsize for Thread %d", iel);
*err = msg;
return 0;
}
@@ -2910,8 +2902,6 @@ standardConfig configs[] = {
createBoolConfig("multi-master-no-forward", NULL, MODIFIABLE_CONFIG, cserver.multimaster_no_forward, 0, validateMultiMasterNoForward, NULL),
createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
createBoolConfig("force-backlog-disk-reserve", NULL, MODIFIABLE_CONFIG, cserver.force_backlog_disk, 0, NULL, NULL),
createBoolConfig("soft-shutdown", NULL, MODIFIABLE_CONFIG, g_pserver->config_soft_shutdown, 0, NULL, NULL),
createSizeTConfig("semi-ordered-set-bucket-size", NULL, MODIFIABLE_CONFIG, 0, 1024, g_semiOrderedSetTargetBucketSize, 0, INTEGER_CONFIG, NULL, NULL),
#ifdef USE_OPENSSL
createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */
+4 -17
View File
@@ -93,7 +93,7 @@ static void lookupKeyUpdateObj(robj *val, int flags)
/* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */
if (!hasActiveChildProcessOrBGSave() && !(flags & LOOKUP_NOTOUCH))
if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH))
{
if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
@@ -1059,7 +1059,7 @@ void keysCommand(client *c) {
sds pattern = szFromObj(c->argv[1]);
const redisDbPersistentDataSnapshot *snapshot = nullptr;
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)))
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
snapshot = c->db->createSnapshot(c->mvccCheckpoint, true /* fOptional */);
if (snapshot != nullptr)
{
@@ -1462,11 +1462,6 @@ void shutdownCommand(client *c) {
flags |= SHUTDOWN_NOSAVE;
} else if (!strcasecmp(szFromObj(c->argv[1]),"save")) {
flags |= SHUTDOWN_SAVE;
} else if (!strcasecmp(szFromObj(c->argv[1]), "soft")) {
g_pserver->soft_shutdown = true;
serverLog(LL_WARNING, "Soft Shutdown Initiated");
addReply(c, shared.ok);
return;
} else {
addReplyErrorObject(c,shared.syntaxerr);
return;
@@ -2607,14 +2602,6 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
}
void moduleLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
if (g_pserver->cluster_enabled) {
clusterStorageLoadCallback(rgchKey, cchKey, data);
}
robj *keyobj = createEmbeddedStringObject(rgchKey, cchKey);
moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", keyobj, *(int *)data);
}
void redisDb::initialize(int id)
{
redisDbPersistentData::initialize();
@@ -2633,8 +2620,8 @@ void redisDb::storageProviderInitialize()
{
if (g_pserver->m_pstorageFactory != nullptr)
{
IStorageFactory::key_load_iterator itr = moduleLoadCallback;
this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
IStorageFactory::key_load_iterator itr = (g_pserver->cluster_enabled) ? clusterStorageLoadCallback : nullptr;
this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, nullptr));
}
}
+1 -1
View File
@@ -1577,7 +1577,7 @@ char *stringFromLongLong(long long value) {
int len;
char *s;
len = snprintf(buf,sizeof(buf),"%lld",value);
len = sprintf(buf,"%lld",value);
s = zmalloc(len+1);
memcpy(s, buf, len);
s[len] = '\0';
+3 -3
View File
@@ -112,15 +112,15 @@ int endianconvTest(int argc, char *argv[], int accurate) {
UNUSED(argv);
UNUSED(accurate);
snprintf(buf,sizeof(buf),"ciaoroma");
sprintf(buf,"ciaoroma");
memrev16(buf);
printf("%s\n", buf);
snprintf(buf,sizeof(buf),"ciaoroma");
sprintf(buf,"ciaoroma");
memrev32(buf);
printf("%s\n", buf);
snprintf(buf,sizeof(buf),"ciaoroma");
sprintf(buf,"ciaoroma");
memrev64(buf);
printf("%s\n", buf);
+1 -1
View File
@@ -1071,7 +1071,7 @@ struct commandHelp {
1,
"2.2.0" },
{ "SHUTDOWN",
"[NOSAVE|SAVE|SOFT]",
"[NOSAVE|SAVE]",
"Synchronously save the dataset to disk and then shut down the server",
9,
"1.0.0" },
+1 -1
View File
@@ -904,7 +904,7 @@ int main(int argc, const char **argv) {
while (self_threads < config.max_threads) {
for (int i = 0; i < config.numclients; i++) {
snprintf(command, sizeof(command), "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
sprintf(command, "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
createClient(command, strlen(command), NULL,self_threads);
}
+2 -3
View File
@@ -51,9 +51,8 @@ int TimerCommand_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
for (int j = 0; j < 10; j++) {
int delay = rand() % 5000;
int bufsize = 256;
char *buf = RedisModule_Alloc(bufsize);
snprintf(buf,bufsize,"After %d", delay);
char *buf = RedisModule_Alloc(256);
snprintf(buf,256,"After %d", delay);
RedisModuleTimerID tid = RedisModule_CreateTimer(ctx,delay,timerHandler,buf);
REDISMODULE_NOT_USED(tid);
}
+3 -17
View File
@@ -145,7 +145,7 @@ client *createClient(connection *conn, int iel) {
client_id = g_pserver->next_client_id.fetch_add(1);
c->iel = iel;
c->id = client_id;
snprintf(c->lock.szName, sizeof(c->lock.szName), "client %" PRIu64, client_id);
sprintf(c->lock.szName, "client %" PRIu64, client_id);
c->resp = 2;
c->conn = conn;
c->name = NULL;
@@ -777,11 +777,11 @@ void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
* we return NULL in addReplyDeferredLen() */
if (node == NULL) return;
char lenstr[128];
size_t lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);
size_t lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
setDeferredReply(c, node, lenstr, lenstr_len);
} else {
char lenstr[128];
int lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);
int lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
size_t idxSplice = (size_t)node;
serverAssert(idxSplice <= c->replyAsync->used);
@@ -1256,20 +1256,6 @@ static void acceptCommonHandler(connection *conn, int flags, char *ip, int iel)
return;
}
/* Prevent new connections if we're in a soft shutdown situation */
if (g_pserver->soft_shutdown) {
const char *err = "-SHUTDOWN";
/* That's a best effort error message, don't check write errors.
* Note that for TLS connections, no handshake was done yet so nothing
* is written and the connection will just drop. */
if (connWrite(conn,err,strlen(err)) == -1) {
/* Nothing to do, Just to avoid the warning... */
}
g_pserver->stat_rejected_conn++;
connClose(conn);
return;
}
/* Limit the number of connections we take at the same time.
*
* Admission control will happen before a client is created and connAccept()
+2 -1
View File
@@ -1142,7 +1142,8 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mem_total+=mem;
std::unique_lock<fastlock> ul(g_expireLock);
mem = db->setexpire()->estimated_bytes_used();
mem = db->setexpire()->bytes_used();
mh->db[mh->num_dbs].overhead_ht_expires = mem;
mem_total+=mem;
+21 -21
View File
@@ -1163,7 +1163,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj_roptr key, robj_roptr val, const expireEn
char szT[32];
if (g_pserver->fActiveReplica) {
snprintf(szT, sizeof(szT), "%" PRIu64, mvccFromObj(val));
snprintf(szT, 32, "%" PRIu64, mvccFromObj(val));
if (rdbSaveAuxFieldStrStr(rdb,"mvcc-tstamp", szT) == -1) return -1;
}
@@ -1190,7 +1190,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj_roptr key, robj_roptr val, const expireEn
{
if (itr.subkey() == nullptr)
continue; // already saved
snprintf(szT, sizeof(szT), "%lld", itr.when());
snprintf(szT, 32, "%lld", itr.when());
rdbSaveAuxFieldStrStr(rdb,"keydb-subexpire-key",itr.subkey());
rdbSaveAuxFieldStrStr(rdb,"keydb-subexpire-when",szT);
}
@@ -1490,7 +1490,7 @@ int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdb
rio rdb;
int error = 0;
snprintf(tmpfile,sizeof(tmpfile),"temp-%d-%d.rdb", getpid(), g_pserver->rdbThreadVars.tmpfileNum);
snprintf(tmpfile,256,"temp-%d-%d.rdb", getpid(), g_pserver->rdbThreadVars.tmpfileNum);
fp = fopen(tmpfile,"w");
if (!fp) {
char *cwdp = getcwd(cwd,MAXPATHLEN);
@@ -1679,15 +1679,17 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
pthread_t child;
long long start;
if (hasActiveChildProcessOrBGSave()) return C_ERR;
if (hasActiveChildProcess()) return C_ERR;
g_pserver->dirty_before_bgsave = g_pserver->dirty;
g_pserver->lastbgsave_try = time(NULL);
openChildInfoPipe();
start = ustime();
latencyStartMonitor(g_pserver->rdb_save_latency);
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
if (launchRdbSaveThread(child, rsi) != C_OK) {
closeChildInfoPipe();
g_pserver->lastbgsave_status = C_ERR;
@@ -1695,9 +1697,6 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
strerror(errno));
return C_ERR;
}
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
serverLog(LL_NOTICE,"Background saving started");
g_pserver->rdb_save_time_start = time(NULL);
@@ -3116,12 +3115,13 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
std::unique_ptr<rdbInsertJob> spjob;
// If we're tracking changes we need to reset this
std::vector<bool> fTracking(cserver.dbnum);
// We don't want to track here because processChangesAsync is outside the normal scope handling
for (int idb = 0; idb < cserver.dbnum; ++idb) {
if ((fTracking[idb] = g_pserver->db[idb]->FTrackingChanges()))
bool fTracking = g_pserver->db[0]->FTrackingChanges();
if (fTracking) {
// We don't want to track here because processChangesAsync is outside the normal scope handling
for (int idb = 0; idb < cserver.dbnum; ++idb) {
if (g_pserver->db[idb]->processChanges(false))
g_pserver->db[idb]->commitChanges();
}
}
rdb->update_cksum = rdbLoadProgressCallback;
@@ -3489,10 +3489,11 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
}
wqueue.endWork();
// Reset track changes
for (int idb = 0; idb < cserver.dbnum; ++idb) {
if (fTracking[idb])
if (fTracking) {
// Reset track changes
for (int idb = 0; idb < cserver.dbnum; ++idb) {
g_pserver->db[idb]->trackChanges(false);
}
}
if (empty_keys_skipped) {
serverLog(LL_WARNING,
@@ -3510,10 +3511,11 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
* the RDB file from a socket during initial SYNC (diskless replica mode),
* we'll report the error to the caller, so that we can retry. */
eoferr:
if (fTracking) {
// Reset track changes
for (int idb = 0; idb < cserver.dbnum; ++idb) {
if (fTracking[idb])
for (int idb = 0; idb < cserver.dbnum; ++idb) {
g_pserver->db[idb]->trackChanges(false);
}
}
wqueue.endWork();
@@ -3602,8 +3604,6 @@ static void backgroundSaveDoneHandlerDisk(int exitcode, bool fCancelled) {
g_pserver->dirty = g_pserver->dirty - g_pserver->dirty_before_bgsave;
g_pserver->lastsave = time(NULL);
g_pserver->lastbgsave_status = C_OK;
latencyEndMonitor(g_pserver->rdb_save_latency);
latencyAddSampleIfNeeded("rdb-save",g_pserver->rdb_save_latency);
} else if (!fCancelled && exitcode != 0) {
serverLog(LL_WARNING, "Background saving error");
g_pserver->lastbgsave_status = C_ERR;
@@ -3694,7 +3694,7 @@ void killRDBChild(bool fSynchronous) {
serverAssert(GlobalLocksAcquired());
if (cserver.fForkBgSave) {
kill(g_pserver->child_pid,SIGUSR1);
kill(g_pserver->rdb_child_pid,SIGUSR1);
} else {
g_pserver->rdbThreadVars.fRdbThreadCancel = true;
if (g_pserver->rdb_child_type == RDB_CHILD_TYPE_SOCKET) {
@@ -3785,7 +3785,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
int pipefds[2];
rdbSaveSocketThreadArgs *args = nullptr;
if (hasActiveChildProcessOrBGSave()) return C_ERR;
if (hasActiveChildProcess()) return C_ERR;
/* Even if the previous fork child exited, don't start a new one until we
* drained the pipe. */
+2 -9
View File
@@ -8,7 +8,6 @@ class readWriteLock {
int m_readCount = 0;
int m_writeCount = 0;
bool m_writeWaiting = false;
bool m_notify = true;
public:
readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
@@ -66,8 +65,7 @@ public:
void releaseRead() {
std::unique_lock<fastlock> rm(m_readLock);
m_readCount--;
if (m_notify)
m_cv.notify_all();
m_cv.notify_all();
}
void releaseWrite(bool exclusive = true) {
@@ -76,8 +74,7 @@ public:
if (exclusive)
m_writeLock.unlock();
m_writeCount--;
if (m_notify)
m_cv.notify_all();
m_cv.notify_all();
}
void downgradeWrite(bool exclusive = true) {
@@ -85,10 +82,6 @@ public:
acquireRead();
}
void setNotify(bool notify) {
m_notify = notify;
}
bool hasReader() {
return m_readCount > 0;
}
+14 -14
View File
@@ -3293,8 +3293,8 @@ static redisReply *clusterManagerMigrateKeysInReply(clusterManagerNode *source,
argv_len = zcalloc(argc * sizeof(size_t), MALLOC_LOCAL);
char portstr[255];
char timeoutstr[255];
snprintf(portstr, sizeof(portstr), "%d", target->port);
snprintf(timeoutstr, sizeof(timeoutstr), "%d", timeout);
snprintf(portstr, 10, "%d", target->port);
snprintf(timeoutstr, 10, "%d", timeout);
argv[0] = "MIGRATE";
argv_len[0] = 7;
argv[1] = target->ip;
@@ -6825,7 +6825,7 @@ static long getLongInfoField(char *info, char *field) {
/* Convert number of bytes into a human readable string of the form:
* 100B, 2G, 100M, 4K, and so forth. */
void bytesToHuman(char *s, long long n, size_t bufsize) {
void bytesToHuman(char *s, long long n) {
double d;
if (n < 0) {
@@ -6835,17 +6835,17 @@ void bytesToHuman(char *s, long long n, size_t bufsize) {
}
if (n < 1024) {
/* Bytes */
snprintf(s,bufsize,"%lldB",n);
sprintf(s,"%lldB",n);
return;
} else if (n < (1024*1024)) {
d = (double)n/(1024);
snprintf(s,bufsize,"%.2fK",d);
sprintf(s,"%.2fK",d);
} else if (n < (1024LL*1024*1024)) {
d = (double)n/(1024*1024);
snprintf(s,bufsize,"%.2fM",d);
sprintf(s,"%.2fM",d);
} else if (n < (1024LL*1024*1024*1024)) {
d = (double)n/(1024LL*1024*1024);
snprintf(s,bufsize,"%.2fG",d);
sprintf(s,"%.2fG",d);
}
}
@@ -6875,38 +6875,38 @@ static void statMode(void) {
for (j = 0; j < 20; j++) {
long k;
snprintf(buf,sizeof(buf),"db%d:keys",j);
sprintf(buf,"db%d:keys",j);
k = getLongInfoField(reply->str,buf);
if (k == LONG_MIN) continue;
aux += k;
}
snprintf(buf,sizeof(buf),"%ld",aux);
sprintf(buf,"%ld",aux);
printf("%-11s",buf);
/* Used memory */
aux = getLongInfoField(reply->str,"used_memory");
bytesToHuman(buf,aux,sizeof(buf));
bytesToHuman(buf,aux);
printf("%-8s",buf);
/* Clients */
aux = getLongInfoField(reply->str,"connected_clients");
snprintf(buf,sizeof(buf),"%ld",aux);
sprintf(buf,"%ld",aux);
printf(" %-8s",buf);
/* Blocked (BLPOPPING) Clients */
aux = getLongInfoField(reply->str,"blocked_clients");
snprintf(buf,sizeof(buf),"%ld",aux);
sprintf(buf,"%ld",aux);
printf("%-8s",buf);
/* Requests */
aux = getLongInfoField(reply->str,"total_commands_processed");
snprintf(buf,sizeof(buf),"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
sprintf(buf,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
printf("%-19s",buf);
requests = aux;
/* Connections */
aux = getLongInfoField(reply->str,"total_connections_received");
snprintf(buf,sizeof(buf),"%ld",aux);
sprintf(buf,"%ld",aux);
printf(" %-12s",buf);
/* Children */
+3
View File
@@ -97,6 +97,9 @@ extern "C" {
#define CC_FORCE (1<<0) /* Re-connect if already connected. */
#define CC_QUIET (1<<1) /* Don't log connecting errors. */
struct clusterManagerLink;
typedef struct clusterManagerLink clusterManagerLink;
/* Dict Helpers */
uint64_t dictSdsHash(const void *key);
-1
View File
@@ -326,7 +326,6 @@ static const RedisModuleEvent
#define REDISMODULE_SUBEVENT_LOADING_ENDED 3
#define REDISMODULE_SUBEVENT_LOADING_FAILED 4
#define _REDISMODULE_SUBEVENT_LOADING_NEXT 5
#define REDISMODULE_SUBEVENT_LOADING_FLASH_START 6
#define REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED 0
#define REDISMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED 1
+9 -22
View File
@@ -1142,7 +1142,7 @@ public:
}
};
int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
int rdbSaveSnapshotForReplication(struct rdbSaveInfo *rsi) {
// TODO: This needs to be on a background thread
int retval = C_OK;
serverAssert(GlobalLocksAcquired());
@@ -1538,7 +1538,7 @@ void syncCommand(client *c) {
} else {
/* We don't have a BGSAVE in progress, let's start one. Diskless
* or disk-based mode is determined by replica's capacity. */
if (!hasActiveChildProcessOrBGSave()) {
if (!hasActiveChildProcess()) {
startBgsaveForReplication(c->slave_capa);
} else {
serverLog(LL_NOTICE,
@@ -2989,11 +2989,10 @@ error:
}
void readSyncBulkPayload(connection *conn) {
serverAssert(GlobalLocksAcquired());
rdbSaveInfo rsi;
redisMaster *mi = (redisMaster*)connGetPrivateData(conn);
static int usemark = 0;
if (mi == nullptr || conn != mi->repl_transfer_s) {
if (mi == nullptr) {
// We're about to be free'd so bail out
return;
}
@@ -3006,9 +3005,6 @@ void readSyncBulkPayload(connection *conn) {
return;
}
if (conn != mi->repl_transfer_s)
return;
/* Final setup of the connected slave <- master link */
replicationCreateMasterClient(mi,mi->repl_transfer_s,rsi.repl_stream_db);
if (mi->isRocksdbSnapshotRepl) {
@@ -3736,7 +3732,7 @@ retry_connect:
while(maxtries--) {
auto dt = std::chrono::system_clock::now().time_since_epoch();
auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
snprintf(tmpfile,sizeof(tmpfile),
snprintf(tmpfile,256,
"temp-%d.%ld.rdb",(int)dtMillisecond.count(),(long int)getpid());
dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
if (dfd != -1) break;
@@ -3817,14 +3813,12 @@ int connectWithMaster(redisMaster *mi) {
* Never call this function directly, use cancelReplicationHandshake() instead.
*/
void undoConnectWithMaster(redisMaster *mi) {
serverAssert(GlobalLocksAcquired());
auto conn = mi->repl_transfer_s;
connSetPrivateData(conn, nullptr);
mi->repl_transfer_s = NULL;
int result = aePostFunction(g_pserver->rgthreadvar[mi->ielReplTransfer].el, [conn]{
aePostFunction(g_pserver->rgthreadvar[mi->ielReplTransfer].el, [conn]{
connClose(conn);
}, false);
serverAssert(result == AE_OK);
});
mi->repl_transfer_s = NULL;
}
/* Abort the async download of the bulk dataset while SYNC-ing with master.
@@ -3979,10 +3973,6 @@ void freeMasterInfo(redisMaster *mi)
{
sdsfree(mi->masterauth);
zfree(mi->masteruser);
if (g_pserver->rdb_filename != nullptr && g_pserver->rdb_filename == mi->repl_transfer_tmpfile) {
unlink(g_pserver->rdb_filename);
g_pserver->rdb_filename = nullptr;
}
if (mi->repl_transfer_tmpfile)
zfree(mi->repl_transfer_tmpfile);
delete mi->staleKeyMap;
@@ -4984,7 +4974,7 @@ void replicationStartPendingFork(void) {
* In case of diskless replication, we make sure to wait the specified
* number of seconds (according to configuration) so that other slaves
* have the time to arrive before we start streaming. */
if (!hasActiveChildProcessOrBGSave()) {
if (!hasActiveChildProcess()) {
time_t idle, max_idle = 0;
int slaves_waiting = 0;
int mincapa = -1;
@@ -5652,10 +5642,7 @@ void flushReplBacklogToClients()
replica->repl_end_off = g_pserver->master_repl_offset;
/* Only if the there isn't already a pending write do we prepare the client to write */
if (replica->repl_curr_off == g_pserver->master_repl_offset) {
serverLog(LL_DEBUG, "Pending write when it's on repl_offset=%lld", g_pserver->master_repl_offset);
continue;
}
serverAssert(replica->repl_curr_off != g_pserver->master_repl_offset);
prepareClientToWrite(replica);
}
if (fAsyncWrite)
+9 -11
View File
@@ -27,7 +27,6 @@ namespace keydbutils
template<>
size_t hash(const sdsview &);
}
extern size_t g_semiOrderedSetTargetBucketSize;
template<typename T, typename T_KEY = T, bool MEMMOVE_SAFE = false>
class semiorderedset
@@ -42,14 +41,11 @@ class semiorderedset
size_t idxRehash = (1ULL << bits_min);
int cfPauseRehash = 0;
inline size_t targetElementsPerBucket()
constexpr size_t targetElementsPerBucket()
{
// Aim for roughly 4 cache lines per bucket (determined by imperical testing)
// lower values are faster but use more memory
if (g_semiOrderedSetTargetBucketSize == 0)
return std::max((64/sizeof(T))*8, (size_t)2);
else
return g_semiOrderedSetTargetBucketSize;
return std::max((64/sizeof(T))*8, (size_t)2);
}
public:
@@ -244,12 +240,14 @@ public:
bool empty() const noexcept { return celem == 0; }
size_t size() const noexcept { return celem; }
size_t estimated_bytes_used() const
size_t bytes_used() const
{
// This estimate does't include all the overhead of the internal vectors
size_t cb = sizeof(this)
+ (m_data.capacity() * sizeof(m_data[0]))
+ sizeof(T) * size();
size_t cb = sizeof(this) + (m_data.capacity()-m_data.size())*sizeof(T);
for (auto &vec : m_data)
{
if (vec != nullptr)
cb += vec->bytes_used();
}
return cb;
}
+63 -98
View File
@@ -779,7 +779,7 @@ struct redisCommand redisCommandTable[] = {
0,NULL,0,0,0,0,0,0},
{"shutdown",shutdownCommand,-1,
"admin no-script ok-loading ok-stale noprop",
"admin no-script ok-loading ok-stale",
0,NULL,0,0,0,0,0,0},
{"lastsave",lastsaveCommand,1,
@@ -1706,7 +1706,7 @@ int redisDbPersistentData::incrementallyRehash() {
* for dict.c to resize the hash tables accordingly to the fact we have an
* active fork child running. */
void updateDictResizePolicy(void) {
if (!hasActiveChildProcess())
if (!hasActiveChildProcess() || (g_pserver->FRdbSaveInProgress() && !cserver.fForkBgSave))
dictEnableResize();
else
dictDisableResize();
@@ -1725,11 +1725,7 @@ const char *strChildType(int type) {
/* Return true if there are active children processes doing RDB saving,
* AOF rewriting, or some side process spawned by a loaded module. */
int hasActiveChildProcess() {
return g_pserver->child_pid != -1;
}
int hasActiveChildProcessOrBGSave() {
return g_pserver->FRdbSaveInProgress() || hasActiveChildProcess();
return g_pserver->FRdbSaveInProgress() || g_pserver->child_pid != -1;
}
void resetChildState() {
@@ -2078,7 +2074,7 @@ void databasesCron(bool fMainThread) {
/* Perform hash tables rehashing if needed, but only if there are no
* other processes saving the DB on disk. Otherwise rehashing is bad
* as will cause a lot of copy-on-write of memory pages. */
if (!hasActiveChildProcess()) {
if (!hasActiveChildProcess() || g_pserver->FRdbSaveInProgress()) {
/* We use global counters so if we stop the computation at a given
* DB we'll be able to start from the successive in the next
* cron loop iteration. */
@@ -2472,14 +2468,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
/* Start a scheduled AOF rewrite if this was requested by the user while
* a BGSAVE was in progress. */
if (!hasActiveChildProcessOrBGSave() &&
if (!hasActiveChildProcess() &&
g_pserver->aof_rewrite_scheduled)
{
rewriteAppendOnlyFileBackground();
}
/* Check if a background saving or AOF rewrite in progress terminated. */
if (hasActiveChildProcessOrBGSave() || ldbPendingChildren())
if (hasActiveChildProcess() || ldbPendingChildren())
{
run_with_period(1000) receiveChildInfo();
checkChildrenDone();
@@ -2521,7 +2517,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
/* Trigger an AOF rewrite if needed. */
if (g_pserver->aof_state == AOF_ON &&
!hasActiveChildProcessOrBGSave() &&
!hasActiveChildProcess() &&
g_pserver->aof_rewrite_perc &&
g_pserver->aof_current_size > g_pserver->aof_rewrite_min_size)
{
@@ -2605,7 +2601,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
* Note: this code must be after the replicationCron() call above so
* make sure when refactoring this file to keep this order. This is useful
* because we want to give priority to RDB savings for replication. */
if (!hasActiveChildProcessOrBGSave() &&
if (!hasActiveChildProcess() &&
g_pserver->rdb_bgsave_scheduled &&
(g_pserver->unixtime-g_pserver->lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
g_pserver->lastbgsave_status == C_OK))
@@ -2652,26 +2648,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
}
}
if (g_pserver->soft_shutdown) {
/* Loop through our clients list and see if there are any active clients */
listIter li;
listNode *ln;
listRewind(g_pserver->clients, &li);
bool fActiveClient = false;
while ((ln = listNext(&li)) && !fActiveClient) {
client *c = (client*)listNodeValue(ln);
if (c->flags & CLIENT_IGNORE_SOFT_SHUTDOWN)
continue;
fActiveClient = true;
}
if (!fActiveClient) {
if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) {
serverLog(LL_WARNING, "All active clients have disconnected while a soft shutdown is pending. Shutting down now.");
throw ShutdownException();
}
}
}
g_pserver->cronloops++;
return 1000/g_pserver->hz;
}
@@ -2799,8 +2775,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
AeLocker locker;
int iel = ielFromEventLoop(eventLoop);
tlsProcessPendingData();
locker.arm();
/* end any snapshots created by fast async commands */
@@ -2819,6 +2793,9 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
runAndPropogateToReplicas(processClients);
/* Handle precise timeouts of blocked clients. */
handleBlockedClientsTimeout();
/* Just call a subset of vital functions in case we are re-entering
* the event loop from processEventsWhileBlocked(). Note that in this
* case we keep track of the number of events we are processing, since
@@ -2828,6 +2805,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
uint64_t processed = 0;
int aof_state = g_pserver->aof_state;
locker.disarm();
processed += tlsProcessPendingData();
processed += handleClientsWithPendingWrites(iel, aof_state);
locker.arm();
processed += freeClientsInAsyncFreeQueue(iel);
@@ -2838,6 +2816,13 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* Handle precise timeouts of blocked clients. */
handleBlockedClientsTimeout();
/* Handle TLS pending data. (must be done before flushAppendOnlyFile) */
if (tlsHasPendingData()) {
locker.disarm();
tlsProcessPendingData();
locker.arm();
}
/* If tls still has pending unread data don't sleep at all. */
aeSetDontWait(eventLoop, tlsHasPendingData());
@@ -3016,11 +3001,8 @@ void afterSleep(struct aeEventLoop *eventLoop) {
serverAssert(serverTL->gcEpoch.isReset());
serverTL->gcEpoch = g_pserver->garbageCollector.startEpoch();
aeAcquireLock();
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb]->trackChanges(false);
aeReleaseLock();
serverTL->disable_async_commands = false;
}
@@ -4068,6 +4050,12 @@ void initServer(void) {
}
}
/* We have to initialize storage providers after the cluster has been initialized */
for (int idb = 0; idb < cserver.dbnum; ++idb)
{
g_pserver->db[idb]->storageProviderInitialize();
}
saveMasterStatusToStorage(false); // eliminate the repl-offset field
/* Initialize ACL default password if it exists */
@@ -4080,15 +4068,6 @@ void initServer(void) {
* Thread Local Storage initialization collides with dlopen call.
* see: https://sourceware.org/bugzilla/show_bug.cgi?id=19329 */
void InitServerLast() {
/* We have to initialize storage providers after the cluster has been initialized */
moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_FLASH_START, NULL);
for (int idb = 0; idb < cserver.dbnum; ++idb)
{
g_pserver->db[idb]->storageProviderInitialize();
}
moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_ENDED, NULL);
bioInit();
set_jemalloc_bg_thread(cserver.jemalloc_bg_thread);
g_pserver->initial_memory_usage = zmalloc_used_memory();
@@ -5320,11 +5299,6 @@ void pingCommand(client *c) {
return;
}
if (g_pserver->soft_shutdown && !(c->flags & CLIENT_IGNORE_SOFT_SHUTDOWN)) {
addReplyError(c, "-SHUTDOWN PENDING");
return;
}
if (c->flags & CLIENT_PUBSUB && c->resp == 2) {
addReply(c,shared.mbulkhdr[2]);
addReplyBulkCBuffer(c,"pong",4);
@@ -5471,30 +5445,30 @@ NULL
/* Convert an amount of bytes into a human readable string in the form
* of 100B, 2G, 100M, 4K, and so forth. */
void bytesToHuman(char *s, unsigned long long n, size_t bufsize) {
void bytesToHuman(char *s, unsigned long long n) {
double d;
if (n < 1024) {
/* Bytes */
snprintf(s,bufsize,"%lluB",n);
sprintf(s,"%lluB",n);
} else if (n < (1024*1024)) {
d = (double)n/(1024);
snprintf(s,bufsize,"%.2fK",d);
sprintf(s,"%.2fK",d);
} else if (n < (1024LL*1024*1024)) {
d = (double)n/(1024*1024);
snprintf(s,bufsize,"%.2fM",d);
sprintf(s,"%.2fM",d);
} else if (n < (1024LL*1024*1024*1024)) {
d = (double)n/(1024LL*1024*1024);
snprintf(s,bufsize,"%.2fG",d);
sprintf(s,"%.2fG",d);
} else if (n < (1024LL*1024*1024*1024*1024)) {
d = (double)n/(1024LL*1024*1024*1024);
snprintf(s,bufsize,"%.2fT",d);
sprintf(s,"%.2fT",d);
} else if (n < (1024LL*1024*1024*1024*1024*1024)) {
d = (double)n/(1024LL*1024*1024*1024*1024);
snprintf(s,bufsize,"%.2fP",d);
sprintf(s,"%.2fP",d);
} else {
/* Let's hope we never need this */
snprintf(s,bufsize,"%lluB",n);
sprintf(s,"%lluB",n);
}
}
@@ -5670,13 +5644,13 @@ sds genRedisInfoString(const char *section) {
if (zmalloc_used > g_pserver->stat_peak_memory)
g_pserver->stat_peak_memory = zmalloc_used;
bytesToHuman(hmem,zmalloc_used,sizeof(hmem));
bytesToHuman(peak_hmem,g_pserver->stat_peak_memory,sizeof(peak_hmem));
bytesToHuman(total_system_hmem,total_system_mem,sizeof(total_system_hmem));
bytesToHuman(used_memory_lua_hmem,memory_lua,sizeof(used_memory_lua_hmem));
bytesToHuman(used_memory_scripts_hmem,mh->lua_caches,sizeof(used_memory_scripts_hmem));
bytesToHuman(used_memory_rss_hmem,g_pserver->cron_malloc_stats.process_rss,sizeof(used_memory_rss_hmem));
bytesToHuman(maxmemory_hmem,g_pserver->maxmemory,sizeof(maxmemory_hmem));
bytesToHuman(hmem,zmalloc_used);
bytesToHuman(peak_hmem,g_pserver->stat_peak_memory);
bytesToHuman(total_system_hmem,total_system_mem);
bytesToHuman(used_memory_lua_hmem,memory_lua);
bytesToHuman(used_memory_scripts_hmem,mh->lua_caches);
bytesToHuman(used_memory_rss_hmem,g_pserver->cron_malloc_stats.process_rss);
bytesToHuman(maxmemory_hmem,g_pserver->maxmemory);
if (sections++) info = sdscat(info,"\r\n");
info = sdscatprintf(info,
@@ -6563,8 +6537,7 @@ void usage(void) {
void redisAsciiArt(void) {
#include "asciilogo.h"
size_t bufsize = 1024*16;
char *buf = (char*)zmalloc(bufsize, MALLOC_LOCAL);
char *buf = (char*)zmalloc(1024*16, MALLOC_LOCAL);
const char *mode;
if (g_pserver->cluster_enabled) mode = "cluster";
@@ -6586,7 +6559,7 @@ void redisAsciiArt(void) {
);
} else {
sds motd = fetchMOTD(true, cserver.enable_motd);
snprintf(buf,bufsize,ascii_logo,
snprintf(buf,1024*16,ascii_logo,
KEYDB_REAL_VERSION,
redisGitSHA1(),
strtol(redisGitDirty(),NULL,10) > 0,
@@ -6720,7 +6693,7 @@ static void sigShutdownHandler(int sig) {
* If we receive the signal the second time, we interpret this as
* the user really wanting to quit ASAP without waiting to persist
* on disk. */
if ((g_pserver->shutdown_asap || g_pserver->soft_shutdown) && sig == SIGINT) {
if (g_pserver->shutdown_asap && sig == SIGINT) {
serverLogFromHandler(LL_WARNING, "You insist... exiting now.");
rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum, 1);
g_pserver->garbageCollector.shutdown();
@@ -6731,10 +6704,7 @@ static void sigShutdownHandler(int sig) {
}
serverLogFromHandler(LL_WARNING, msg);
if (g_pserver->config_soft_shutdown)
g_pserver->soft_shutdown = true;
else
g_pserver->shutdown_asap = 1;
g_pserver->shutdown_asap = 1;
}
void setupSignalHandlers(void) {
@@ -6889,7 +6859,6 @@ int redisFork(int purpose) {
latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
if ((childpid = fork()) == 0) {
/* Child */
aeForkLockInChild();
aeReleaseForkLock();
g_pserver->in_fork_child = purpose;
setOOMScoreAdj(CONFIG_OOM_BGCHILD);
@@ -7572,29 +7541,6 @@ int main(int argc, char **argv) {
validateConfiguration();
if (!g_pserver->sentinel_mode) {
#ifdef __linux__
linuxMemoryWarnings();
#if defined (__arm64__)
int ret;
if ((ret = linuxMadvFreeForkBugCheck())) {
if (ret == 1)
serverLog(LL_WARNING,"WARNING Your kernel has a bug that could lead to data corruption during background save. "
"Please upgrade to the latest stable kernel.");
else
serverLog(LL_WARNING, "Failed to test the kernel for a bug that could lead to data corruption during background save. "
"Your system could be affected, please report this error.");
if (!checkIgnoreWarning("ARM64-COW-BUG")) {
serverLog(LL_WARNING,"KeyDB will now exit to prevent data corruption. "
"Note that it is possible to suppress this warning by setting the following config: ignore-warnings ARM64-COW-BUG");
exit(1);
}
}
#endif /* __arm64__ */
#endif /* __linux__ */
}
const char *err;
if (!initializeStorageProvider(&err))
{
@@ -7620,6 +7566,25 @@ int main(int argc, char **argv) {
if (!g_pserver->sentinel_mode) {
/* Things not needed when running in Sentinel mode. */
serverLog(LL_WARNING,"Server initialized");
#ifdef __linux__
linuxMemoryWarnings();
#if defined (__arm64__)
int ret;
if ((ret = linuxMadvFreeForkBugCheck())) {
if (ret == 1)
serverLog(LL_WARNING,"WARNING Your kernel has a bug that could lead to data corruption during background save. "
"Please upgrade to the latest stable kernel.");
else
serverLog(LL_WARNING, "Failed to test the kernel for a bug that could lead to data corruption during background save. "
"Your system could be affected, please report this error.");
if (!checkIgnoreWarning("ARM64-COW-BUG")) {
serverLog(LL_WARNING,"KeyDB will now exit to prevent data corruption. "
"Note that it is possible to suppress this warning by setting the following config: ignore-warnings ARM64-COW-BUG");
exit(1);
}
}
#endif /* __arm64__ */
#endif /* __linux__ */
moduleInitModulesSystemLast();
moduleLoadFromQueue();
ACLLoadUsersAtStartup();
+1 -13
View File
@@ -508,7 +508,6 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
#define CLIENT_PREVENT_AOF_PROP (1<<19) /* Don't propagate to AOF. */
#define CLIENT_PREVENT_REPL_PROP (1<<20) /* Don't propagate to slaves. */
#define CLIENT_PREVENT_PROP (CLIENT_PREVENT_AOF_PROP|CLIENT_PREVENT_REPL_PROP)
#define CLIENT_IGNORE_SOFT_SHUTDOWN (CLIENT_MASTER | CLIENT_SLAVE | CLIENT_BLOCKED | CLIENT_MONITOR)
#define CLIENT_PENDING_WRITE (1<<21) /* Client has output to send but a write
handler is yet not installed. */
#define CLIENT_REPLY_OFF (1<<22) /* Don't send replies to client. */
@@ -1909,12 +1908,6 @@ struct MasterSaveInfo {
masterhost = sdsstring(sdsdup(mi.masterhost));
masterport = mi.masterport;
}
MasterSaveInfo(const MasterSaveInfo &other) {
masterhost = other.masterhost;
masterport = other.masterport;
memcpy(master_replid, other.master_replid, sizeof(master_replid));
master_initial_offset = other.master_initial_offset;
}
MasterSaveInfo &operator=(const MasterSaveInfo &other) {
masterhost = other.masterhost;
@@ -2470,8 +2463,7 @@ struct redisServer {
time_t lastbgsave_try; /* Unix time of last attempted bgsave */
time_t rdb_save_time_last; /* Time used by last RDB save run. */
time_t rdb_save_time_start; /* Current RDB save start time. */
mstime_t rdb_save_latency; /* Used to track end to end latency of rdb save*/
pid_t rdb_child_pid = -1; /* Used only during fork bgsave */
pid_t rdb_child_pid = -1; /* Used only during fork bgsave */
int rdb_bgsave_scheduled; /* BGSAVE when possible if true. */
int rdb_child_type; /* Type of save by active child. */
int lastbgsave_status; /* C_OK or C_ERR */
@@ -2729,9 +2721,6 @@ struct redisServer {
long long repl_batch_offStart = -1;
long long repl_batch_idxStart = -1;
int config_soft_shutdown = false;
bool soft_shutdown = false;
/* Lock Contention Ring Buffer */
static const size_t s_lockContentionSamples = 64;
uint16_t rglockSamples[s_lockContentionSamples];
@@ -3255,7 +3244,6 @@ void receiveChildInfo(void);
void executeWithoutGlobalLock(std::function<void()> func);
int redisFork(int type);
int hasActiveChildProcess();
int hasActiveChildProcessOrBGSave();
void resetChildState();
int isMutuallyExclusiveChildType(int type);
+4 -6
View File
@@ -194,8 +194,8 @@ void sortCommand(client *c) {
list *operations;
unsigned int outputlen = 0;
int desc = 0, alpha = 0;
long limit_start = 0, limit_count = -1, start, end, vectorlen;
int j, dontsort = 0;
long limit_start = 0, limit_count = -1, start, end;
int j, dontsort = 0, vectorlen;
int getop = 0; /* GET operation counter */
int int_conversion_error = 0;
int syntax_error = 0;
@@ -321,10 +321,8 @@ void sortCommand(client *c) {
default: vectorlen = 0; serverPanic("Bad SORT type"); /* Avoid GCC warning */
}
/* Perform LIMIT start,count sanity checking.
* And avoid integer overflow by limiting inputs to object sizes. */
start = std::min(std::max(limit_start, (long)0), vectorlen);
limit_count = std::min(std::max(limit_count, (long)-1), vectorlen);
/* Perform LIMIT start,count sanity checking. */
start = (limit_start < 0) ? 0 : limit_start;
end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
if (start >= vectorlen) {
start = vectorlen-1;
-236
View File
@@ -1,236 +0,0 @@
#include "rocksdb.h"
#include <string>
#include <sstream>
#include <mutex>
#include <unistd.h>
#include "rocksdbfactor_internal.h"
static const char keyprefix[] = INTERNAL_KEY_PREFIX;
rocksdb::Options DefaultRocksDBOptions();
extern "C" pid_t gettid();
bool FInternalKey(const char *key, size_t cch)
{
if (cch >= sizeof(INTERNAL_KEY_PREFIX))
{
if (memcmp(key, keyprefix, sizeof(INTERNAL_KEY_PREFIX)-1) == 0)
return true;
}
return false;
}
RocksDBStorageProvider::RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count)
: m_pfactory(pfactory), m_spdb(spdb), m_psnapshot(psnapshot), m_spcolfamily(spcolfam), m_count(count)
{
m_readOptionsTemplate = rocksdb::ReadOptions();
m_readOptionsTemplate.verify_checksums = false;
m_readOptionsTemplate.snapshot = m_psnapshot;
}
void RocksDBStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite)
{
rocksdb::Status status;
std::unique_lock<fastlock> l(m_lock);
if (m_spbatch != nullptr)
status = m_spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(key, cchKey), rocksdb::Slice((const char*)data, cb));
else
status = m_spdb->Put(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey), rocksdb::Slice((const char*)data, cb));
if (!status.ok())
throw status.ToString();
if (!fOverwrite)
++m_count;
}
void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
{
if (celem >= 16384) {
rocksdb::Options options = DefaultRocksDBOptions();
rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
std::string file_path = m_pfactory->getTempFolder() + "/tmpIngest.";
file_path += std::to_string(gettid());
file_path += ".sst";
rocksdb::Status s = sst_file_writer.Open(file_path);
if (!s.ok())
goto LFallback;
// Insert rows into the SST file, note that inserted keys must be
// strictly increasing (based on options.comparator)
for (size_t ielem = 0; ielem < celem; ++ielem) {
s = sst_file_writer.Put(rocksdb::Slice(rgkeys[ielem], rgcbkeys[ielem]), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
if (!s.ok()) {
unlink(file_path.c_str());
goto LFallback;
}
}
s = sst_file_writer.Finish();
if (!s.ok()) {
unlink(file_path.c_str());
goto LFallback;
}
auto ingestOptions = rocksdb::IngestExternalFileOptions();
ingestOptions.move_files = true;
ingestOptions.write_global_seqno = false;
ingestOptions.failed_move_fall_back_to_copy = false;
// Ingest the external SST file into the DB
s = m_spdb->IngestExternalFile(m_spcolfamily.get(), {file_path}, ingestOptions);
if (!s.ok()) {
unlink(file_path.c_str());
goto LFallback;
}
} else {
LFallback:
auto spbatch = std::make_unique<rocksdb::WriteBatch>();
for (size_t ielem = 0; ielem < celem; ++ielem) {
spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(rgkeys[ielem], rgcbkeys[ielem]), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
}
m_spdb->Write(WriteOptions(), spbatch.get());
}
std::unique_lock<fastlock> l(m_lock);
m_count += celem;
}
bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
{
rocksdb::Status status;
std::unique_lock<fastlock> l(m_lock);
if (!FKeyExists(key, cchKey))
return false;
if (m_spbatch != nullptr)
{
status = m_spbatch->Delete(m_spcolfamily.get(), rocksdb::Slice(key, cchKey));
}
else
{
status = m_spdb->Delete(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey));
}
if (status.ok())
--m_count;
return status.ok();
}
void RocksDBStorageProvider::retrieve(const char *key, size_t cchKey, callbackSingle fn) const
{
rocksdb::PinnableSlice slice;
auto status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey), &slice);
if (status.ok())
fn(key, cchKey, slice.data(), slice.size());
}
size_t RocksDBStorageProvider::clear()
{
size_t celem = count();
auto status = m_spdb->DropColumnFamily(m_spcolfamily.get());
auto strName = m_spcolfamily->GetName();
rocksdb::ColumnFamilyHandle *handle = nullptr;
rocksdb::ColumnFamilyOptions cf_options(m_pfactory->RocksDbOptions());
m_spdb->CreateColumnFamily(cf_options, strName, &handle);
m_spcolfamily = std::shared_ptr<rocksdb::ColumnFamilyHandle>(handle);
if (!status.ok())
throw status.ToString();
m_count = 0;
return celem;
}
size_t RocksDBStorageProvider::count() const
{
std::unique_lock<fastlock> l(m_lock);
return m_count;
}
bool RocksDBStorageProvider::enumerate(callback fn) const
{
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
size_t count = 0;
for (it->SeekToFirst(); it->Valid(); it->Next()) {
if (FInternalKey(it->key().data(), it->key().size()))
continue;
++count;
bool fContinue = fn(it->key().data(), it->key().size(), it->value().data(), it->value().size());
if (!fContinue)
break;
}
if (!it->Valid() && count != m_count)
{
if (const_cast<RocksDBStorageProvider*>(this)->m_count != count)
printf("WARNING: rocksdb count mismatch");
const_cast<RocksDBStorageProvider*>(this)->m_count = count;
}
assert(it->status().ok()); // Check for any errors found during the scan
return !it->Valid();
}
const IStorage *RocksDBStorageProvider::clone() const
{
std::unique_lock<fastlock> l(m_lock);
const rocksdb::Snapshot *psnapshot = const_cast<RocksDBStorageProvider*>(this)->m_spdb->GetSnapshot();
return new RocksDBStorageProvider(m_pfactory, const_cast<RocksDBStorageProvider*>(this)->m_spdb, const_cast<RocksDBStorageProvider*>(this)->m_spcolfamily, psnapshot, m_count);
}
RocksDBStorageProvider::~RocksDBStorageProvider()
{
if (m_spbatch != nullptr)
endWriteBatch();
if (m_spdb != nullptr && m_psnapshot == nullptr)
{
insert(count_key, sizeof(count_key), &m_count, sizeof(m_count), false);
flush();
}
if (m_spdb != nullptr)
{
if (m_psnapshot != nullptr)
m_spdb->ReleaseSnapshot(m_psnapshot);
}
}
rocksdb::WriteOptions RocksDBStorageProvider::WriteOptions() const
{
auto opt = rocksdb::WriteOptions();
return opt;
}
void RocksDBStorageProvider::beginWriteBatch()
{
m_lock.lock();
m_spbatch = std::make_unique<rocksdb::WriteBatchWithIndex>();
}
void RocksDBStorageProvider::endWriteBatch()
{
m_spdb->Write(WriteOptions(), m_spbatch.get()->GetWriteBatch());
m_spbatch = nullptr;
m_lock.unlock();
}
void RocksDBStorageProvider::batch_lock()
{
m_lock.lock();
}
void RocksDBStorageProvider::batch_unlock()
{
m_lock.unlock();
}
void RocksDBStorageProvider::flush()
{
m_spdb->SyncWAL();
}
bool RocksDBStorageProvider::FKeyExists(const char *key, size_t cch) const
{
rocksdb::PinnableSlice slice;
if (m_spbatch)
return m_spbatch->GetFromBatchAndDB(m_spdb.get(), ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cch), &slice).ok();
return m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cch), &slice).ok();
}
-57
View File
@@ -1,57 +0,0 @@
#pragma once
#include <memory>
#include "../IStorage.h"
#include <rocksdb/db.h>
#include <rocksdb/utilities/write_batch_with_index.h>
#include "../fastlock.h"
#define INTERNAL_KEY_PREFIX "\x00\x04\x03\x00\x05\x02\x04"
static const char count_key[] = INTERNAL_KEY_PREFIX "__keydb__count\1";
static const char version_key[] = INTERNAL_KEY_PREFIX "__keydb__version\1";
static const char meta_key[] = INTERNAL_KEY_PREFIX "__keydb__metadata\1";
class RocksDBStorageFactory;
class RocksDBStorageProvider : public IStorage
{
RocksDBStorageFactory *m_pfactory;
std::shared_ptr<rocksdb::DB> m_spdb; // Note: This must be first so it is deleted last
std::unique_ptr<rocksdb::WriteBatchWithIndex> m_spbatch;
const rocksdb::Snapshot *m_psnapshot = nullptr;
std::shared_ptr<rocksdb::ColumnFamilyHandle> m_spcolfamily;
rocksdb::ReadOptions m_readOptionsTemplate;
size_t m_count = 0;
mutable fastlock m_lock {"RocksDBStorageProvider"};
public:
RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count);
~RocksDBStorageProvider();
virtual void insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite) override;
virtual bool erase(const char *key, size_t cchKey) override;
virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const override;
virtual size_t clear() override;
virtual bool enumerate(callback fn) const override;
virtual const IStorage *clone() const override;
virtual void beginWriteBatch() override;
virtual void endWriteBatch() override;
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) override;
virtual void batch_lock() override;
virtual void batch_unlock() override;
virtual void flush() override;
size_t count() const override;
protected:
bool FKeyExists(const char *key, size_t cchKey) const;
const rocksdb::ReadOptions &ReadOptions() const { return m_readOptionsTemplate; }
rocksdb::WriteOptions WriteOptions() const;
};
bool FInternalKey(const char *key, size_t cch);
-204
View File
@@ -1,204 +0,0 @@
#include "rocksdb.h"
#include "../version.h"
#include <rocksdb/filter_policy.h>
#include <rocksdb/table.h>
#include <rocksdb/utilities/options_util.h>
#include <rocksdb/sst_file_manager.h>
#include <rocksdb/utilities/convenience.h>
#include <rocksdb/slice_transform.h>
#include "rocksdbfactor_internal.h"
#include <sys/types.h>
#include <sys/stat.h>
rocksdb::Options DefaultRocksDBOptions() {
rocksdb::Options options;
options.max_background_compactions = 4;
options.max_background_flushes = 2;
options.bytes_per_sync = 1048576;
options.compaction_pri = rocksdb::kMinOverlappingRatio;
options.compression = rocksdb::kNoCompression;
options.enable_pipelined_write = true;
options.allow_mmap_reads = true;
options.avoid_unnecessary_blocking_io = true;
options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(0));
rocksdb::BlockBasedTableOptions table_options;
table_options.block_size = 16 * 1024;
table_options.cache_index_and_filter_blocks = true;
table_options.pin_l0_filter_and_index_blocks_in_cache = true;
table_options.data_block_index_type = rocksdb::BlockBasedTableOptions::kDataBlockBinaryAndHash;
table_options.checksum = rocksdb::kNoChecksum;
table_options.format_version = 4;
options.table_factory.reset(NewBlockBasedTableFactory(table_options));
return options;
}
IStorageFactory *CreateRocksDBStorageFactory(const char *path, int dbnum, const char *rgchConfig, size_t cchConfig)
{
return new RocksDBStorageFactory(path, dbnum, rgchConfig, cchConfig);
}
rocksdb::Options RocksDBStorageFactory::RocksDbOptions()
{
rocksdb::Options options = DefaultRocksDBOptions();
options.max_open_files = filedsRequired();
options.sst_file_manager = m_pfilemanager;
options.create_if_missing = true;
options.create_missing_column_families = true;
options.info_log_level = rocksdb::ERROR_LEVEL;
options.max_total_wal_size = 1 * 1024 * 1024 * 1024;
return options;
}
RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, const char *rgchConfig, size_t cchConfig)
: m_path(dbfile)
{
dbnum++; // create an extra db for metadata
// Get the count of column families in the actual database
std::vector<std::string> vecT;
auto status = rocksdb::DB::ListColumnFamilies(rocksdb::Options(), dbfile, &vecT);
// RocksDB requires we know the count of col families before opening, if the user only wants to see less
// we still have to make room for all column family handles regardless
if (status.ok() && (int)vecT.size() > dbnum)
dbnum = (int)vecT.size();
std::vector<rocksdb::ColumnFamilyDescriptor> veccoldesc;
veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(rocksdb::kDefaultColumnFamilyName, rocksdb::ColumnFamilyOptions())); // ignore default col family
m_pfilemanager = std::shared_ptr<rocksdb::SstFileManager>(rocksdb::NewSstFileManager(rocksdb::Env::Default()));
rocksdb::DB *db = nullptr;
auto options = RocksDbOptions();
for (int idb = 0; idb < dbnum; ++idb)
{
rocksdb::ColumnFamilyOptions cf_options(options);
cf_options.level_compaction_dynamic_level_bytes = true;
veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(std::to_string(idb), cf_options));
}
if (rgchConfig != nullptr)
{
std::string options_string(rgchConfig, cchConfig);
rocksdb::Status status;
if (!(status = rocksdb::GetDBOptionsFromString(options, options_string, &options)).ok())
{
fprintf(stderr, "Failed to parse FLASH options: %s\r\n", status.ToString().c_str());
exit(EXIT_FAILURE);
}
}
std::vector<rocksdb::ColumnFamilyHandle*> handles;
status = rocksdb::DB::Open(options, dbfile, veccoldesc, &handles, &db);
if (!status.ok())
throw status.ToString();
m_spdb = std::shared_ptr<rocksdb::DB>(db);
for (auto handle : handles)
{
std::string strVersion;
auto status = m_spdb->Get(rocksdb::ReadOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), &strVersion);
if (!status.ok())
{
setVersion(handle);
}
else
{
SymVer ver = parseVersion(strVersion.c_str());
auto cmp = compareVersion(&ver);
if (cmp == NewerVersion)
throw "Cannot load FLASH database created by newer version of KeyDB";
if (cmp == OlderVersion)
setVersion(handle);
}
m_vecspcols.emplace_back(handle);
}
}
RocksDBStorageFactory::~RocksDBStorageFactory()
{
m_spdb->SyncWAL();
}
void RocksDBStorageFactory::setVersion(rocksdb::ColumnFamilyHandle *handle)
{
auto status = m_spdb->Put(rocksdb::WriteOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), rocksdb::Slice(KEYDB_REAL_VERSION, strlen(KEYDB_REAL_VERSION)+1));
if (!status.ok())
throw status.ToString();
}
size_t RocksDBStorageFactory::filedsRequired() const {
return 256;
}
std::string RocksDBStorageFactory::getTempFolder()
{
auto path = m_path + "/keydb_tmp/";
if (!m_fCreatedTempFolder) {
if (!mkdir(path.c_str(), 0700))
m_fCreatedTempFolder = true;
}
return path;
}
IStorage *RocksDBStorageFactory::createMetadataDb()
{
IStorage *metadataDb = this->create(-1, nullptr, nullptr);
metadataDb->insert(meta_key, sizeof(meta_key), (void*)METADATA_DB_IDENTIFIER, strlen(METADATA_DB_IDENTIFIER), false);
return metadataDb;
}
IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *privdata)
{
++db; // skip default col family
std::shared_ptr<rocksdb::ColumnFamilyHandle> spcolfamily(m_vecspcols[db].release());
size_t count = 0;
bool fUnclean = false;
std::string value;
auto status = m_spdb->Get(rocksdb::ReadOptions(), spcolfamily.get(), rocksdb::Slice(count_key, sizeof(count_key)), &value);
if (status.ok() && value.size() == sizeof(size_t))
{
count = *reinterpret_cast<const size_t*>(value.data());
m_spdb->Delete(rocksdb::WriteOptions(), spcolfamily.get(), rocksdb::Slice(count_key, sizeof(count_key)));
}
else
{
fUnclean = true;
}
if (fUnclean || iter != nullptr)
{
count = 0;
auto opts = rocksdb::ReadOptions();
opts.tailing = true;
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(opts, spcolfamily.get()));
it->SeekToFirst();
bool fFirstRealKey = true;
for (;it->Valid(); it->Next()) {
if (FInternalKey(it->key().data(), it->key().size()))
continue;
if (fUnclean && it->Valid() && fFirstRealKey)
printf("\tDatabase %d was not shutdown cleanly, recomputing metrics\n", db);
fFirstRealKey = false;
if (iter != nullptr)
iter(it->key().data(), it->key().size(), privdata);
++count;
}
}
return new RocksDBStorageProvider(this, m_spdb, spcolfamily, nullptr, count);
}
const char *RocksDBStorageFactory::name() const
{
return "flash";
}
size_t RocksDBStorageFactory::totalDiskspaceUsed() const
{
return m_pfilemanager->GetTotalSize();
}
+2 -4
View File
@@ -26,12 +26,10 @@ TestStorageProvider::~TestStorageProvider()
{
}
void TestStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite)
void TestStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool /*fOverwrite*/)
{
auto strkey = std::string(key, cchKey);
bool fActuallyExists = m_map.find(strkey) != m_map.end();
serverAssert(fActuallyExists == fOverwrite);
m_map.insert(std::make_pair(strkey, std::string((char*)data, cb)));
m_map[strkey] = std::string((char*)data, cb);
}
+1 -6
View File
@@ -1225,13 +1225,8 @@ void hrandfieldCommand(client *c) {
if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withvalues"))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
} else if (c->argc == 4) {
} else if (c->argc == 4)
withvalues = 1;
if (l < LONG_MIN/2 || l > LONG_MAX/2) {
addReplyError(c,"value is out of range");
return;
}
}
hrandfieldWithCountCommand(c, l, withvalues);
return;
}
+1 -1
View File
@@ -263,7 +263,7 @@ sds writeJsonValue(sds output, const char *valIn, size_t cchIn) {
serverAssert(!FSimpleJsonEscapeCh(valIn[ich]));
if (FExtendedJsonEscapeCh(valIn[ich])) {
dst[ichDst++] = '\\'; dst[ichDst++] = 'u';
snprintf(dst + ichDst, cchIn+cchEscapeExtra-ichDst, "%4x", valIn[ich]);
sprintf(dst + ichDst, "%4x", valIn[ich]);
ichDst += 4;
} else {
dst[ichDst++] = valIn[ich];
+6 -11
View File
@@ -38,14 +38,8 @@ int getGenericCommand(client *c);
* String Commands
*----------------------------------------------------------------------------*/
static int checkStringLength(client *c, long long size, long long append) {
if (c->flags & CLIENT_MASTER)
return C_OK;
/* 'uint64_t' cast is there just to prevent undefined behavior on overflow */
long long total = (uint64_t)size + append;
/* Test configured max-bulk-len represending a limit of the biggest string object,
* and also test for overflow. */
if (total > g_pserver->proto_max_bulk_len || total < size || total < append) {
static int checkStringLength(client *c, long long size) {
if (!(c->flags & CLIENT_MASTER) && size > g_pserver->proto_max_bulk_len) {
addReplyError(c,"string exceeds maximum allowed size (proto-max-bulk-len)");
return C_ERR;
}
@@ -451,7 +445,7 @@ void setrangeCommand(client *c) {
}
/* Return when the resulting string exceeds allowed size */
if (checkStringLength(c,offset,sdslen(value)) != C_OK)
if (checkStringLength(c,offset+sdslen(value)) != C_OK)
return;
o = createObject(OBJ_STRING,sdsnewlen(NULL, offset+sdslen(value)));
@@ -471,7 +465,7 @@ void setrangeCommand(client *c) {
}
/* Return when the resulting string exceeds allowed size */
if (checkStringLength(c,offset,sdslen(value)) != C_OK)
if (checkStringLength(c,offset+sdslen(value)) != C_OK)
return;
/* Create a copy when the object is shared or encoded. */
@@ -691,7 +685,8 @@ void appendCommand(client *c) {
/* "append" is an argument, so always an sds */
append = c->argv[2];
if (checkStringLength(c,stringObjectLen(o),sdslen((sds)ptrFromObj(append))) != C_OK)
totlen = stringObjectLen(o)+sdslen((sds)ptrFromObj(append));
if (checkStringLength(c,totlen) != C_OK)
return;
/* Append the value */
+1 -6
View File
@@ -4214,13 +4214,8 @@ void zrandmemberCommand(client *c) {
if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withscores"))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
} else if (c->argc == 4) {
} else if (c->argc == 4)
withscores = 1;
if (l < LONG_MIN/2 || l > LONG_MAX/2) {
addReplyError(c,"value is out of range");
return;
}
}
zrandmemberWithCountCommand(c, l, withscores);
return;
}
+8 -57
View File
@@ -46,9 +46,6 @@
#include <sys/stat.h>
#include <arpa/inet.h>
#if OPENSSL_VERSION_NUMBER >= 0x30000000L
#include <openssl/decoder.h>
#endif
#define REDIS_TLS_PROTO_TLSv1 (1<<0)
#define REDIS_TLS_PROTO_TLSv1_1 (1<<1)
@@ -157,13 +154,14 @@ void tlsInit(void) {
*/
#if OPENSSL_VERSION_NUMBER < 0x10100000L
OPENSSL_config(NULL);
SSL_load_error_strings();
SSL_library_init();
#elif OPENSSL_VERSION_NUMBER < 0x10101000L
OPENSSL_init_crypto(OPENSSL_INIT_LOAD_CONFIG, NULL);
#else
OPENSSL_init_crypto(OPENSSL_INIT_LOAD_CONFIG|OPENSSL_INIT_ATFORK, NULL);
#endif
ERR_load_crypto_strings();
SSL_load_error_strings();
SSL_library_init();
#ifdef USE_CRYPTO_LOCKS
initCryptoLocks();
@@ -366,46 +364,20 @@ int tlsConfigure(redisTLSContextConfig *ctx_config) {
if (ctx_config->prefer_server_ciphers)
SSL_CTX_set_options(ctx, SSL_OP_CIPHER_SERVER_PREFERENCE);
#if ((OPENSSL_VERSION_NUMBER < 0x30000000L) && defined(SSL_CTX_set_ecdh_auto))
#if defined(SSL_CTX_set_ecdh_auto)
SSL_CTX_set_ecdh_auto(ctx, 1);
#endif
SSL_CTX_set_options(ctx, SSL_OP_SINGLE_DH_USE);
if (ctx_config->dh_params_file) {
FILE *dhfile = fopen(ctx_config->dh_params_file, "r");
DH *dh = NULL;
if (!dhfile) {
serverLog(LL_WARNING, "Failed to load %s: %s", ctx_config->dh_params_file, strerror(errno));
goto error;
}
#if (OPENSSL_VERSION_NUMBER >= 0x30000000L)
EVP_PKEY *pkey = NULL;
OSSL_DECODER_CTX *dctx = OSSL_DECODER_CTX_new_for_pkey(
&pkey, "PEM", NULL, "DH", OSSL_KEYMGMT_SELECT_DOMAIN_PARAMETERS, NULL, NULL);
if (!dctx) {
serverLog(LL_WARNING, "No decoder for DH params.");
fclose(dhfile);
goto error;
}
if (!OSSL_DECODER_from_fp(dctx, dhfile)) {
serverLog(LL_WARNING, "%s: failed to read DH params.", ctx_config->dh_params_file);
OSSL_DECODER_CTX_free(dctx);
fclose(dhfile);
goto error;
}
OSSL_DECODER_CTX_free(dctx);
fclose(dhfile);
if (SSL_CTX_set0_tmp_dh_pkey(ctx, pkey) <= 0) {
ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
serverLog(LL_WARNING, "Failed to load DH params file: %s: %s", ctx_config->dh_params_file, errbuf);
EVP_PKEY_free(pkey);
goto error;
}
/* Not freeing pkey, it is owned by OpenSSL now */
#else
DH *dh = PEM_read_DHparams(dhfile, NULL, NULL, NULL);
dh = PEM_read_DHparams(dhfile, NULL, NULL, NULL);
fclose(dhfile);
if (!dh) {
serverLog(LL_WARNING, "%s: failed to read DH params.", ctx_config->dh_params_file);
@@ -420,11 +392,6 @@ int tlsConfigure(redisTLSContextConfig *ctx_config) {
}
DH_free(dh);
#endif
} else {
#if (OPENSSL_VERSION_NUMBER >= 0x30000000L)
SSL_CTX_set_dh_auto(ctx, 1);
#endif
}
/* If a client-side certificate is configured, create an explicit client context */
@@ -536,26 +503,11 @@ bool tlsCheckAgainstAllowlist(const char * client){
#define ASN1_STRING_get0_data ASN1_STRING_data
#endif
class TCleanup {
std::function<void()> fn;
public:
TCleanup(std::function<void()> fn)
: fn(fn)
{}
~TCleanup() {
fn();
}
};
bool tlsValidateCertificateName(tls_connection* conn){
if (g_pserver->tls_allowlist.empty())
return true; // Empty list implies acceptance of all
X509 * cert = SSL_get_peer_certificate(conn->ssl);
TCleanup certClen([cert]{X509_free(cert);});
/* Check the common name (CN) of the certificate first */
X509_NAME_ENTRY * ne = X509_NAME_get_entry(X509_get_subject_name(cert), X509_NAME_get_index_by_NID(X509_get_subject_name(cert), NID_commonName, -1));
const char * commonName = reinterpret_cast<const char*>(ASN1_STRING_get0_data(X509_NAME_ENTRY_get_data(ne)));
@@ -616,9 +568,8 @@ bool tlsValidateCertificateName(tls_connection* conn){
/* If neither the CN nor the SANs match, update the SSL error and return false */
conn->c.last_errno = 0;
if (conn->ssl_error) zfree(conn->ssl_error);
size_t bufsize = 512;
conn->ssl_error = (char*)zmalloc(bufsize);
snprintf(conn->ssl_error, bufsize, "Client CN (%s) and SANs not found in allowlist.", commonName);
conn->ssl_error = (char*)zmalloc(512);
snprintf(conn->ssl_error, 512, "Client CN (%s) and SANs not found in allowlist.", commonName);
return false;
}
+11 -11
View File
@@ -1704,17 +1704,17 @@ static unsigned char *createIntList() {
unsigned char *zl = ziplistNew();
char buf[32];
snprintf(buf, sizeof(buf), "100");
sprintf(buf, "100");
zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
snprintf(buf, sizeof(buf), "128000");
sprintf(buf, "128000");
zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
snprintf(buf, sizeof(buf), "-100");
sprintf(buf, "-100");
zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
snprintf(buf, sizeof(buf), "4294967296");
sprintf(buf, "4294967296");
zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
snprintf(buf, sizeof(buf), "non integer");
sprintf(buf, "non integer");
zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
snprintf(buf, sizeof(buf), "much much longer non integer");
sprintf(buf, "much much longer non integer");
zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
return zl;
}
@@ -2228,7 +2228,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
char buf[32];
int i,len;
for (i = 0; i < 1000; i++) {
len = snprintf(buf,sizeof(buf),"%d",i);
len = sprintf(buf,"%d",i);
zl = ziplistPush(zl,(unsigned char*)buf,len,ZIPLIST_TAIL);
}
for (i = 0; i < 1000; i++) {
@@ -2375,13 +2375,13 @@ int ziplistTest(int argc, char **argv, int accurate) {
} else {
switch(rand() % 3) {
case 0:
buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()) >> 20);
buflen = sprintf(buf,"%lld",(0LL + rand()) >> 20);
break;
case 1:
buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()));
buflen = sprintf(buf,"%lld",(0LL + rand()));
break;
case 2:
buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()) << 20);
buflen = sprintf(buf,"%lld",(0LL + rand()) << 20);
break;
default:
assert(NULL);
@@ -2410,7 +2410,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
assert(ziplistGet(p,&sstr,&slen,&sval));
if (sstr == NULL) {
buflen = snprintf(buf,sizeof(buf),"%lld",sval);
buflen = sprintf(buf,"%lld",sval);
} else {
buflen = slen;
memcpy(buf,sstr,buflen);
+3 -3
View File
@@ -370,7 +370,7 @@ size_t zmalloc_get_rss(void) {
int fd, count;
char *p, *x;
snprintf(filename,sizeof(filename),"/proc/%ld/stat",(long) getpid());
snprintf(filename,256,"/proc/%ld/stat",(long) getpid());
if ((fd = open(filename,O_RDONLY)) == -1) return 0;
if (read(fd,buf,4096) <= 0) {
close(fd);
@@ -462,7 +462,7 @@ size_t zmalloc_get_rss(void) {
char filename[256];
int fd;
snprintf(filename,sizeof(filename),"/proc/%ld/psinfo",(long) getpid());
snprintf(filename,256,"/proc/%ld/psinfo",(long) getpid());
if ((fd = open(filename,O_RDONLY)) == -1) return 0;
if (ioctl(fd, PIOCPSINFO, &info) == -1) {
@@ -521,7 +521,7 @@ int jemalloc_purge() {
unsigned narenas = 0;
size_t sz = sizeof(unsigned);
if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
snprintf(tmp, sizeof(tmp), "arena.%d.purge", narenas);
sprintf(tmp, "arena.%d.purge", narenas);
if (!je_mallctl(tmp, NULL, 0, NULL, 0))
return 0;
}
+2 -6
View File
@@ -53,12 +53,12 @@ start_server {overrides {active-replica {yes} multi-master {yes} client-output-b
set elapsed [expr {[clock milliseconds]-$cycle_start_time}]
if {$elapsed > $duration*1000} break
if {rand() < .05} {
test "PSYNC2 #3899 regression (multi-master): kill first replica" {
test "PSYNC2 #3899 regression: kill first replica" {
$R(1) client kill type master
}
}
if {rand() < .05} {
test "PSYNC2 #3899 regression (multi-master): kill chained replica" {
test "PSYNC2 #3899 regression: kill chained replica" {
$R(2) client kill type master
}
}
@@ -99,8 +99,4 @@ start_server {overrides {active-replica {yes} multi-master {yes} client-output-b
fail [format "The three instances have different data sets:\n\tnode 1: %s\n\tnode 2: %s\n\tnode 3: %s\nRun diff -u against /tmp/repldump*.txt for more info" [$R(0) debug digest] [$R(1) debug digest] [$R(2) debug digest] ]
}
}
assert {[s -2 sync_partial_ok] > 0}
assert {[s -1 sync_partial_ok] > 0}
assert {[s 0 sync_partial_ok] > 0}
}}}
+20 -22
View File
@@ -6,31 +6,29 @@ proc prepare_value {size} {
return $_v
}
if {$::flash_enabled} {
start_server {tags {"replication-fast"} overrides {storage-provider {flash ./rocks.db.master} databases 256}} {
set slave [srv 0 client]
set slave_host [srv 0 host]
set slave_port [srv 0 port]
start_server {tags {} overrides {storage-provider {flash ./rocks.db.replica} databases 256}} {
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
start_server {tags {"replication-fast"} overrides {storage-provider {flash ./rocks.db.master} databases 256}} {
set slave [srv 0 client]
set slave_host [srv 0 host]
set slave_port [srv 0 port]
start_server {tags {} overrides {storage-provider {flash ./rocks.db.replica} databases 256}} {
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
test "fast replication with large value" {
set _v [prepare_value [expr 64*1024*1024]]
# $master set key $_v
test "fast replication with large value" {
set _v [prepare_value [expr 64*1024*1024]]
# $master set key $_v
$slave replicaof $master_host $master_port
wait_for_condition 50 300 {
[lindex [$slave role] 0] eq {slave} &&
[string match {*master_link_status:up*} [$slave info replication]]
} else {
fail "Can't turn the instance into a replica"
}
assert_equal [$slave debug digest] [$master debug digest]
$slave replicaof no one
$slave replicaof $master_host $master_port
wait_for_condition 50 300 {
[lindex [$slave role] 0] eq {slave} &&
[string match {*master_link_status:up*} [$slave info replication]]
} else {
fail "Can't turn the instance into a replica"
}
assert_equal [$slave debug digest] [$master debug digest]
$slave replicaof no one
}
}
}
+103 -105
View File
@@ -8,130 +8,128 @@
# If reconnect is > 0, the test actually try to break the connection and
# reconnect with the master, otherwise just the initial synchronization is
# checked for consistency.
if {$::flash_enabled} {
proc test_psync {descr duration backlog_size backlog_ttl delay cond mdl sdl reconnect} {
start_server [list tags {"repl"} overrides [list storage-provider {flash .rocks.db.m} repl-backlog-size 64m]] {
start_server [list tags {flash} overrides [list storage-provider {flash ./rocks.db} delete-on-evict no storage-flush-period 10]] {
proc test_psync {descr duration backlog_size backlog_ttl delay cond mdl sdl reconnect} {
start_server [list tags {"repl"} overrides [list storage-provider {flash .rocks.db.m} repl-backlog-size 64m]] {
start_server [list tags {flash} overrides [list storage-provider {flash ./rocks.db} delete-on-evict no storage-flush-period 10]] {
set master [srv -1 client]
set master_host [srv -1 host]
set master_port [srv -1 port]
set slave [srv 0 client]
set master [srv -1 client]
set master_host [srv -1 host]
set master_port [srv -1 port]
set slave [srv 0 client]
set load_handle0 [start_bg_complex_data $master_host $master_port 9 100000]
set load_handle1 [start_bg_complex_data $master_host $master_port 11 100000]
set load_handle2 [start_bg_complex_data $master_host $master_port 12 100000]
set load_handle0 [start_bg_complex_data $master_host $master_port 9 100000]
set load_handle1 [start_bg_complex_data $master_host $master_port 11 100000]
set load_handle2 [start_bg_complex_data $master_host $master_port 12 100000]
test {Slave should be able to synchronize with the master} {
$slave slaveof $master_host $master_port
wait_for_condition 50 100 {
[lindex [r role] 0] eq {slave} &&
[lindex [r role] 3] eq {connected}
} else {
fail "Replication not started."
}
test {Slave should be able to synchronize with the master} {
$slave slaveof $master_host $master_port
wait_for_condition 50 100 {
[lindex [r role] 0] eq {slave} &&
[lindex [r role] 3] eq {connected}
} else {
fail "Replication not started."
}
}
# Check that the background clients are actually writing.
test {Detect write load to master} {
wait_for_condition 50 1000 {
[$master dbsize] > 100
} else {
fail "Can't detect write load from background clients."
}
# Check that the background clients are actually writing.
test {Detect write load to master} {
wait_for_condition 50 1000 {
[$master dbsize] > 100
} else {
fail "Can't detect write load from background clients."
}
}
test "Test replication partial resync: $descr (diskless: $mdl, $sdl, reconnect: $reconnect)" {
# Now while the clients are writing data, break the maste-slave
# link multiple times.
if ($reconnect) {
for {set j 0} {$j < $duration*10} {incr j} {
after 100
# catch {puts "MASTER [$master dbsize] keys, REPLICA [$slave dbsize] keys"}
test "Test replication partial resync: $descr (diskless: $mdl, $sdl, reconnect: $reconnect)" {
# Now while the clients are writing data, break the maste-slave
# link multiple times.
if ($reconnect) {
for {set j 0} {$j < $duration*10} {incr j} {
after 100
# catch {puts "MASTER [$master dbsize] keys, REPLICA [$slave dbsize] keys"}
if {($j % 20) == 0} {
catch {
$slave debug restart
}
if {($j % 20) == 0} {
catch {
$slave debug restart
}
}
}
stop_bg_complex_data $load_handle0
stop_bg_complex_data $load_handle1
stop_bg_complex_data $load_handle2
# Wait for the slave to reach the "online"
# state from the POV of the master.
set retry 5000
while {$retry} {
set info [$master info]
if {[string match {*slave0:*state=online*} $info]} {
break
} else {
incr retry -1
after 100
}
}
if {$retry == 0} {
error "assertion:Slave not correctly synchronized"
}
# Wait that slave acknowledge it is online so
# we are sure that DBSIZE and DEBUG DIGEST will not
# fail because of timing issues. (-LOADING error)
wait_for_condition 5000 100 {
[lindex [$slave role] 3] eq {connected}
} else {
fail "Slave still not connected after some time"
}
set retry 20
while {$retry && ([$master debug digest] ne [$slave debug digest])}\
{
after 1000
incr retry -1
}
assert {[$master dbsize] > 0}
if {[$master debug digest] ne [$slave debug digest]} {
set csv1 [csvdump r]
set csv2 [csvdump {r -1}]
set fd [open /tmp/repldump1.txt w]
puts -nonewline $fd $csv1
close $fd
set fd [open /tmp/repldump2.txt w]
puts -nonewline $fd $csv2
close $fd
puts "Master - Replica inconsistency"
puts "Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal [r debug digest] [r -1 debug digest]
eval $cond
}
stop_bg_complex_data $load_handle0
stop_bg_complex_data $load_handle1
stop_bg_complex_data $load_handle2
# Wait for the slave to reach the "online"
# state from the POV of the master.
set retry 5000
while {$retry} {
set info [$master info]
if {[string match {*slave0:*state=online*} $info]} {
break
} else {
incr retry -1
after 100
}
}
if {$retry == 0} {
error "assertion:Slave not correctly synchronized"
}
# Wait that slave acknowledge it is online so
# we are sure that DBSIZE and DEBUG DIGEST will not
# fail because of timing issues. (-LOADING error)
wait_for_condition 5000 100 {
[lindex [$slave role] 3] eq {connected}
} else {
fail "Slave still not connected after some time"
}
set retry 20
while {$retry && ([$master debug digest] ne [$slave debug digest])}\
{
after 1000
incr retry -1
}
assert {[$master dbsize] > 0}
if {[$master debug digest] ne [$slave debug digest]} {
set csv1 [csvdump r]
set csv2 [csvdump {r -1}]
set fd [open /tmp/repldump1.txt w]
puts -nonewline $fd $csv1
close $fd
set fd [open /tmp/repldump2.txt w]
puts -nonewline $fd $csv2
close $fd
puts "Master - Replica inconsistency"
puts "Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal [r debug digest] [r -1 debug digest]
eval $cond
}
}
}
}
foreach mdl {no yes} {
foreach sdl {disabled swapdb} {
test_psync {no reconnection, just sync} 6 1000000 3600 0 {
} $mdl $sdl 0
foreach mdl {no yes} {
foreach sdl {disabled swapdb} {
test_psync {no reconnection, just sync} 6 1000000 3600 0 {
} $mdl $sdl 0
test_psync {ok psync} 6 100000000 3600 0 {
assert {[s -1 sync_partial_ok] > 0}
} $mdl $sdl 1
test_psync {ok psync} 6 100000000 3600 0 {
assert {[s -1 sync_partial_ok] > 0}
} $mdl $sdl 1
test_psync {no backlog} 6 100 3600 0.5 {
assert {[s -1 sync_partial_err] > 0}
} $mdl $sdl 1
test_psync {no backlog} 6 100 3600 0.5 {
assert {[s -1 sync_partial_err] > 0}
} $mdl $sdl 1
test_psync {ok after delay} 3 100000000 3600 3 {
assert {[s -1 sync_partial_ok] > 0}
} $mdl $sdl 1
test_psync {ok after delay} 3 100000000 3600 3 {
assert {[s -1 sync_partial_ok] > 0}
} $mdl $sdl 1
test_psync {backlog expired} 3 100000000 1 3 {
assert {[s -1 sync_partial_err] > 0}
} $mdl $sdl 1
}
test_psync {backlog expired} 3 100000000 1 3 {
assert {[s -1 sync_partial_err] > 0}
} $mdl $sdl 1
}
}
+1 -2
View File
@@ -39,8 +39,7 @@ TEST_MODULES = \
defragtest.so \
hash.so \
zset.so \
stream.so \
load.so
stream.so
.PHONY: all
-1
View File
@@ -212,7 +212,6 @@ void loadingCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void
case REDISMODULE_SUBEVENT_LOADING_RDB_START: keyname = "loading-rdb-start"; break;
case REDISMODULE_SUBEVENT_LOADING_AOF_START: keyname = "loading-aof-start"; break;
case REDISMODULE_SUBEVENT_LOADING_REPL_START: keyname = "loading-repl-start"; break;
case REDISMODULE_SUBEVENT_LOADING_FLASH_START: keyname = "loading-flash-start"; break;
case REDISMODULE_SUBEVENT_LOADING_ENDED: keyname = "loading-end"; break;
case REDISMODULE_SUBEVENT_LOADING_FAILED: keyname = "loading-failed"; break;
}
-94
View File
@@ -1,94 +0,0 @@
/* Server hooks API example
*
* -----------------------------------------------------------------------------
*
* 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.
*/
#define REDISMODULE_EXPERIMENTAL_API
#include "redismodule.h"
size_t count, finalCount;
/* Client state change callback. */
void loadCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
REDISMODULE_NOT_USED(ctx);
REDISMODULE_NOT_USED(e);
REDISMODULE_NOT_USED(data);
if (sub == REDISMODULE_SUBEVENT_LOADING_FLASH_START || sub == REDISMODULE_SUBEVENT_LOADING_RDB_START || sub == REDISMODULE_SUBEVENT_LOADING_AOF_START || sub == REDISMODULE_SUBEVENT_LOADING_REPL_START) {
count = 0;
finalCount = 0;
} else if (sub == REDISMODULE_SUBEVENT_LOADING_ENDED) {
finalCount = count;
}
}
int loadKeyCallback(RedisModuleCtx *ctx, int type, const char *event, RedisModuleString *key) {
REDISMODULE_NOT_USED(ctx);
REDISMODULE_NOT_USED(type);
REDISMODULE_NOT_USED(event);
const char *keyname = RedisModule_StringPtrLen(key, NULL);
RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_NOTICE, "Loaded key: %s", keyname);
count++;
return 0;
}
int LoadCount_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
if (argc != 1) return RedisModule_WrongArity(ctx);
RedisModule_ReplyWithLongLong(ctx, finalCount);
return REDISMODULE_OK;
}
/* This function must be present on each Redis module. It is used in order to
* register the commands into the Redis server. */
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx,"load",1,REDISMODULE_APIVER_1)
== REDISMODULE_ERR) return REDISMODULE_ERR;
RedisModule_SubscribeToServerEvent(ctx,
RedisModuleEvent_Loading, loadCallback);
RedisModule_SubscribeToKeyspaceEvents(ctx,
REDISMODULE_NOTIFY_LOADED, loadKeyCallback);
if (RedisModule_CreateCommand(ctx, "load.count",
LoadCount_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
+5 -36
View File
@@ -66,33 +66,6 @@ proc redis {{server 127.0.0.1} {port 6379} {defer 0} {tls 0} {tlsoptions {}} {re
interp alias {} ::redis::redisHandle$id {} ::redis::__dispatch__ $id
}
# On recent versions of tcl-tls/OpenSSL, reading from a dropped connection
# results with an error we need to catch and mimic the old behavior.
proc ::redis::redis_safe_read {fd len} {
if {$len == -1} {
set err [catch {set val [read $fd]} msg]
} else {
set err [catch {set val [read $fd $len]} msg]
}
if {!$err} {
return $val
}
if {[string match "*connection abort*" $msg]} {
return {}
}
error $msg
}
proc ::redis::redis_safe_gets {fd} {
if {[catch {set val [gets $fd]} msg]} {
if {[string match "*connection abort*" $msg]} {
return {}
}
error $msg
}
return $val
}
# This is a wrapper to the actual dispatching procedure that handles
# reconnection if needed.
proc ::redis::__dispatch__ {id method args} {
@@ -173,10 +146,6 @@ proc ::redis::__method__read {id fd} {
::redis::redis_read_reply $id $fd
}
proc ::redis::__method__rawread {id fd {len -1}} {
return [redis_safe_read $fd $len]
}
proc ::redis::__method__write {id fd buf} {
::redis::redis_write $fd $buf
}
@@ -223,8 +192,8 @@ proc ::redis::redis_writenl {fd buf} {
}
proc ::redis::redis_readnl {fd len} {
set buf [redis_safe_read $fd $len]
redis_safe_read $fd 2 ; # discard CR LF
set buf [read $fd $len]
read $fd 2 ; # discard CR LF
return $buf
}
@@ -270,11 +239,11 @@ proc ::redis::redis_read_map {id fd} {
}
proc ::redis::redis_read_line fd {
string trim [redis_safe_gets $fd]
string trim [gets $fd]
}
proc ::redis::redis_read_null fd {
redis_safe_gets $fd
gets $fd
return {}
}
@@ -291,7 +260,7 @@ proc ::redis::redis_read_reply {id fd} {
}
while {1} {
set type [redis_safe_read $fd 1]
set type [read $fd 1]
switch -exact -- $type {
_ {return [redis_read_null $fd]}
: -
-6
View File
@@ -40,12 +40,6 @@ proc assert_failed {expected_err detail} {
error "assertion:$expected_err $detail"
}
proc assert_not_equal {value expected {detail ""}} {
if {!($expected ne $value)} {
assert_failed "Expected '$value' not equal to '$expected'" $detail
}
}
proc assert_equal {value expected {detail ""}} {
if {$expected ne $value} {
assert_failed "Expected '$value' to be equal to '$expected'" $detail
-14
View File
@@ -47,7 +47,6 @@ set ::all_tests {
integration/replication-3
integration/replication-4
integration/replication-psync
integration/replication-psync-flash
integration/replication-active
integration/replication-multimaster
integration/replication-multimaster-connect
@@ -60,13 +59,11 @@ set ::all_tests {
integration/failover
integration/keydb-cli
integration/keydb-benchmark
integration/replication-fast
integration/replication-psync-multimaster
unit/pubsub
unit/slowlog
unit/scripting
unit/maxmemory
unit/flash
unit/introspection
unit/introspection-2
unit/limits
@@ -89,7 +86,6 @@ set ::all_tests {
integration/logging
integration/corrupt-dump
integration/corrupt-dump-fuzzer
unit/soft_shutdown
}
# Index to the next test to run in the ::all_tests list.
set ::next_test 0
@@ -739,16 +735,6 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
}
}
# Check if we compiled with flash
set status [catch {exec src/keydb-server --is-flash-enabled}]
if {$status == 0} {
puts "KeyDB was built with FLASH, including FLASH tests"
set ::flash_enabled 1
} else {
puts "KeyDB was not built with FLASH. Excluding FLASH tests"
set ::flash_enabled 0
}
set filtered_tests {}
# Set the filtered tests to be the short list (single_tests) if exists.
+160 -161
View File
@@ -1,171 +1,170 @@
if {$::flash_enabled} {
start_server [list tags {flash} overrides [list storage-provider {flash ./rocks.db} delete-on-evict no storage-flush-period 10]] {
start_server [list tags {flash} overrides [list storage-provider {flash ./rocks.db} delete-on-evict no storage-flush-period 10]] {
test { FLASH - GET works after eviction } {
r set testkey foo
r flushall cache
assert_equal {foo} [r get testkey]
}
test { FLASH - GET works after eviction } {
r set testkey foo
r flushall cache
assert_equal {foo} [r get testkey]
}
test { DEL of nonexistant key returns 0 } {
r flushall
assert_equal {0} [r del foobar]
assert_equal {0} [r dbsize] "Key count is accurate after non-existant delete"
}
test { DEL of flushed key works } {
r flushall
r set testkey foo
assert_equal {1} [r dbsize] "Only one key after first insert"
r flushall cache
assert_equal {foo} [r get testkey] "Value still there after flushing cache"
r del testkey
assert_equal {0} [r dbsize] "No keys after delete"
}
test { SET of existing but flushed key works } {
r flushall
r set testkey foo
assert_equal {1} [r dbsize] "Only one key after first insert"
r flushall cache
assert_equal {1} [r dbsize] "Only one key after flushall cache"
r set testkey bar
assert_equal {1} [r dbsize] "Only one key after overwrite"
assert_equal {bar} [r get testkey]
}
test { SET of existing but flushed key with EXPIRE works } {
test { DEL of nonexistant key returns 0 } {
r flushall
assert_equal {0} [r dbsize]
r set testkey foo ex 10000
assert_equal {0} [r del foobar]
assert_equal {0} [r dbsize] "Key count is accurate after non-existant delete"
}
test { DEL of flushed key works } {
r flushall
r set testkey foo
assert_equal {1} [r dbsize] "Only one key after first insert"
r flushall cache
assert_equal {foo} [r get testkey] "Value still there after flushing cache"
r del testkey
assert_equal {0} [r dbsize] "No keys after delete"
}
test { SET of existing but flushed key works } {
r flushall
r set testkey foo
assert_equal {1} [r dbsize] "Only one key after first insert"
r flushall cache
assert_equal {1} [r dbsize] "Only one key after flushall cache"
r set testkey bar ex 10000
r set testkey bar
assert_equal {1} [r dbsize] "Only one key after overwrite"
assert_equal {bar} [r get testkey]
assert [expr [r ttl testkey] > 0]
}
test { EXPIRE of existing but flushed key } {
r flushall
assert_equal {0} [r dbsize]
r set testkey foo
assert_equal {1} [r dbsize]
r flushall cache
r expire testkey 10000
assert_equal {1} [r dbsize]
assert_equal {foo} [r get testkey]
assert [expr [r ttl testkey] > 0]
}
test { CREATE and UPDATE in transaction, key count is accurate } {
r flushall
r multi
r set testkey 2
r incr testkey
r exec
assert_equal {1} [r dbsize]
assert_equal {3} [r get testkey]
}
test { EXPIRE key count is accurate } {
r flushall
r set testkey foo ex 1
r flushall cache
assert_equal {1} [r dbsize]
after 1500
assert_equal {0} [r dbsize]
}
test { SUBKEY EXPIRE persists after cache flush } {
r flushall
r sadd testkey foo bar baz
r expiremember testkey foo 10000
r flushall cache
assert [expr [r ttl testkey foo] > 0]
}
test { LIST pop works after flushing cache } {
r flushall
r lpush testkey foo
r flushall cache
assert_equal {foo} [r lpop testkey]
}
test { DIGEST string the same after flushing cache } {
r flushall
r set testkey foo
r set testkey1 foo ex 10000
set expectedDigest [r debug digest]
r flushall cache
assert_equal $expectedDigest [r debug digest]
}
test { DIGEST list the same after flushing cache } {
r flushall
r lpush testkey foo bar
set expectedDigest [r debug digest]
r flushall cache
assert_equal $expectedDigest [r debug digest]
}
test { DELETE of flushed set member persists after another flush } {
r flushall
r sadd set1 val1 val2 val3
assert_equal {3} [r scard set1]
r flushall cache
r srem set1 val1
r flushall cache
assert_equal {2} [r scard set1]
}
r flushall
# If a weak storage memory model is set, wait for any pending snapshot writes to finish
after 500
foreach policy {
allkeys-random allkeys-lru allkeys-lfu
} {
test "FLASH - is eviction working without data loss (successfully stored to flash)? (policy $policy)" {
# Get the current memory limit and calculate a new limit.
# Set limit to 100M.
set used [s used_memory]
set limit [expr {$used+60*1024*1024}]
r config set maxmemory $limit
r config set maxmemory-policy $policy
# Now add keys equivalent to 1024b until the limit is almost reached.
set numkeys 0
r set first val
while 1 {
r set $numkeys xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
incr numkeys
if {[s used_memory]+1024 >= $limit} {
break
}
}
# Add additional keys to force eviction
# should still be under the limit for maxmemory, however all keys set should still exist between flash and memory
# check same number of keys exist in addition to values of first and last keys
set err 0
set extra_keys [expr floor([expr ($limit * 0.4) / 1024])]
for {set j 0} {$j < $extra_keys} {incr j} {
catch {
r set p2$j xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
} err
assert {$err == {OK}}
}
if {[log_file_matches [srv 0 stdout] "*Failed to evict*"]} {
fail "Server did not evict cleanly (detected full flush)"
}
r set last val
set dbsize [r dbsize]
assert {[s used_memory] < ($limit*1.2)}
assert {$dbsize == $numkeys+$extra_keys+2}
assert {[r get first] == {val}}
assert {[r get last] == {val}}
r flushall
}
}
}
test { SET of existing but flushed key with EXPIRE works } {
r flushall
assert_equal {0} [r dbsize]
r set testkey foo ex 10000
assert_equal {1} [r dbsize] "Only one key after first insert"
r flushall cache
assert_equal {1} [r dbsize] "Only one key after flushall cache"
r set testkey bar ex 10000
assert_equal {1} [r dbsize] "Only one key after overwrite"
assert_equal {bar} [r get testkey]
assert [expr [r ttl testkey] > 0]
}
test { EXPIRE of existing but flushed key } {
r flushall
assert_equal {0} [r dbsize]
r set testkey foo
assert_equal {1} [r dbsize]
r flushall cache
r expire testkey 10000
assert_equal {1} [r dbsize]
assert_equal {foo} [r get testkey]
assert [expr [r ttl testkey] > 0]
}
test { CREATE and UPDATE in transaction, key count is accurate } {
r flushall
r multi
r set testkey 2
r incr testkey
r exec
assert_equal {1} [r dbsize]
assert_equal {3} [r get testkey]
}
test { EXPIRE key count is accurate } {
r flushall
r set testkey foo ex 1
r flushall cache
assert_equal {1} [r dbsize]
after 1500
assert_equal {0} [r dbsize]
}
test { SUBKEY EXPIRE persists after cache flush } {
r flushall
r sadd testkey foo bar baz
r expiremember testkey foo 10000
r flushall cache
assert [expr [r ttl testkey foo] > 0]
}
test { LIST pop works after flushing cache } {
r flushall
r lpush testkey foo
r flushall cache
assert_equal {foo} [r lpop testkey]
}
test { DIGEST string the same after flushing cache } {
r flushall
r set testkey foo
r set testkey1 foo ex 10000
set expectedDigest [r debug digest]
r flushall cache
assert_equal $expectedDigest [r debug digest]
}
test { DIGEST list the same after flushing cache } {
r flushall
r lpush testkey foo bar
set expectedDigest [r debug digest]
r flushall cache
assert_equal $expectedDigest [r debug digest]
}
test { DELETE of flushed set member persists after another flush } {
r flushall
r sadd set1 val1 val2 val3
assert_equal {3} [r scard set1]
r flushall cache
r srem set1 val1
r flushall cache
assert_equal {2} [r scard set1]
}
r flushall
# If a weak storage memory model is set, wait for any pending snapshot writes to finish
after 500
foreach policy {
allkeys-random allkeys-lru allkeys-lfu
} {
test "FLASH - is eviction working without data loss (successfully stored to flash)? (policy $policy)" {
# Get the current memory limit and calculate a new limit.
# Set limit to 100M.
set used [s used_memory]
set limit [expr {$used+60*1024*1024}]
r config set maxmemory $limit
r config set maxmemory-policy $policy
# Now add keys equivalent to 1024b until the limit is almost reached.
set numkeys 0
r set first val
while 1 {
r set $numkeys xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
incr numkeys
if {[s used_memory]+1024 >= $limit} {
break
}
}
# Add additional keys to force eviction
# should still be under the limit for maxmemory, however all keys set should still exist between flash and memory
# check same number of keys exist in addition to values of first and last keys
set err 0
set extra_keys [expr floor([expr ($limit * 0.4) / 1024])]
for {set j 0} {$j < $extra_keys} {incr j} {
catch {
r set p2$j xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
} err
assert {$err == {OK}}
}
if {[log_file_matches [srv 0 stdout] "*Failed to evict*"]} {
fail "Server did not evict cleanly (detected full flush)"
}
r set last val
set dbsize [r dbsize]
assert {[s used_memory] < ($limit*1.2)}
assert {$dbsize == $numkeys+$extra_keys+2}
assert {[r get first] == {val}}
assert {[r get last] == {val}}
r flushall
}
}
}
+9 -15
View File
@@ -49,13 +49,10 @@ start_server {tags {"defrag"} overrides {appendonly yes auto-aof-rewrite-percent
r config set maxmemory 100mb
r config set maxmemory-policy allkeys-lru
for {set i 0} {$i < 10} {incr i} {
populate 700000 asdf1 150
populate 170000 asdf2 300
after 120 ;# serverCron only updates the info once in 100ms
set frag [s allocator_frag_ratio]
if {$frag >= 1.4} {break}
}
populate 700000 asdf1 150
populate 170000 asdf2 300
after 120 ;# serverCron only updates the info once in 100ms
set frag [s allocator_frag_ratio]
if {$::verbose} {
puts "frag $frag"
}
@@ -510,14 +507,11 @@ start_server {tags {"defrag"} overrides {server-threads 1 active-replica yes} }
r config set active-defrag-ignore-bytes 2mb
r config set maxmemory 100mb
r config set maxmemory-policy allkeys-lru
for {set i 0} {$i < 10} {incr i} {
populate 700000 asdf1 150
populate 170000 asdf2 300
r ping ;# trigger eviction following the previous population
after 120 ;# serverCron only updates the info once in 100ms
set frag [s allocator_frag_ratio]
if {$frag >= 1.4} {break}
}
populate 700000 asdf1 150
populate 170000 asdf2 300
r ping ;# trigger eviction following the previous population
after 120 ;# serverCron only updates the info once in 100ms
set frag [s allocator_frag_ratio]
if {$::verbose} {
puts "frag $frag"
}
-15
View File
@@ -1,15 +0,0 @@
set testmodule [file normalize tests/modules/load.so]
if {$::flash_enabled} {
start_server [list tags [list "modules"] overrides [list storage-provider {flash ./rocks.db.master} databases 256 loadmodule $testmodule]] {
test "Module is notified of keys loaded from flash" {
r flushall
r set foo bar
r set bar foo
r set foobar barfoo
assert_equal [r load.count] 0
r debug reload
assert_equal [r load.count] 3
}
}
}
+3 -2
View File
@@ -111,7 +111,7 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
# Read nothing
set fd [$rd channel]
assert_equal {} [$rd rawread]
assert_equal {} [read $fd]
}
# Note: This test assumes that what's written with one write, will be read by redis in one read.
@@ -151,7 +151,8 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
assert_equal "PONG" [r ping]
set clients [r client list]
assert_no_match "*name=multicommands*" $clients
assert_equal {} [$rd2 rawread]
set fd [$rd2 channel]
assert_equal {} [read $fd]
}
test {Execute transactions completely even if client output buffer limit is enforced} {
-4
View File
@@ -43,10 +43,6 @@ start_server {tags {"scripting"}} {
r eval {return redis.call('get',KEYS[1])} 1 mykey
} {myval}
test { EVAL - keys command works? } {
r eval {return redis.call('keys', 'test')} 0
}
test {EVALSHA - Can we call a SHA1 if already defined?} {
r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
} {myval}
-141
View File
@@ -1,141 +0,0 @@
start_server {tags {"soft_shutdown"} } {
test {soft shutdown command replies} {
assert_equal [r shutdown soft] "OK"
}
test {soft shutdown errors on ping} {
catch {[r ping]} e
assert_match {SHUTDOWN PENDING} $e
}
}
start_server {tags {"soft_shutdown"} } {
test {soft shutdown prevents new connections} {
assert_equal [r shutdown soft] "OK"
# reconnect
set catch_res [catch {set rd [redis_deferring_client]} e]
if {$::tls} {
assert_equal $catch_res 1
} else {
assert_match {*SHUTDOWN*} $e
}
}
}
start_server {tags {"soft_shutdown"} } {
test {soft shutdown allows commands to execute while waiting} {
assert_equal [r shutdown soft] "OK"
r set test val
assert_equal [r get test] {val}
}
}
start_server {tags {"soft_shutdown"} } {
test {soft shutdown shuts down after all clients exit} {
assert_equal [r shutdown soft] "OK"
r close
after 500
catch {set rd [redis_deferring_client]} e
assert_match {*refused*} $e
}
}
start_server {tags {"soft_shutdown"} overrides {soft-shutdown yes} } {
test {soft shutdown triggered by SIGINT} {
exec kill -SIGINT [s process_id]
after 100
catch {[r ping]} e
assert_match {SHUTDOWN PENDING} $e
}
test {second SIGINT forces a shutdown during a soft shutdown} {
exec kill -SIGINT [s process_id]
after 100
catch {[r ping]} e
assert_match {*I/O*} $e
}
}
start_server {tags {"soft_shutdown"} } {
test {monitor does not prevent soft shutdown} {
set monitor [redis_deferring_client]
$monitor monitor
assert_equal [r shutdown soft] "OK"
r close
after 500
catch {set rd [redis_deferring_client]} e
assert_match {*refused*} $e
}
}
start_server {tags {"soft_shutdown"} } {
start_server {} {
set node_0 [srv 0 client]
set node_0_host [srv 0 host]
set node_0_port [srv 0 port]
set node_0_pid [srv 0 pid]
set node_1 [srv -1 client]
set node_1_host [srv -1 host]
set node_1_port [srv -1 port]
set node_1_pid [srv -1 pid]
$node_0 replicaof $node_1_host $node_1_port
wait_for_sync $node_0
test {soft shutdown works for with master} {
$node_1 shutdown soft
} {OK}
test {soft shutdown on master doesn't affect replica} {
assert_equal [$node_0 ping] {PONG}
}
test {soft shutdown on master updates ping response} {
catch {$node_1 ping} e
assert_equal $e {SHUTDOWN PENDING}
}
test {master prevents new connections with soft shutdown} {
set c1 [redis $node_1_host $node_1_port 1 $::tls]
set catch_res [catch {$c1 read} e]
if {$::tls} {
assert_equal $catch_res 1
} else {
assert_match {*SHUTDOWN*} $e
}
}
test {master soft shutdown works after all clients disconnect} {
$node_1 close
after 500
catch {set c1 [redis $node_1_host $node_1_port 1 $::tls]} e
assert_match {*refused*} $e
}
}
}
start_server {tags {"soft_shutdown"} } {
start_server {} {
set node_0 [srv 0 client]
set node_0_host [srv 0 host]
set node_0_port [srv 0 port]
set node_0_pid [srv 0 pid]
set node_1 [srv -1 client]
set node_1_host [srv -1 host]
set node_1_port [srv -1 port]
set node_1_pid [srv -1 pid]
$node_0 replicaof $node_1_host $node_1_port
wait_for_sync $node_0
test {soft shutdown on replica is not blocked by master} {
assert_equal [$node_0 shutdown soft] {OK}
$node_0 close
after 500
catch {set c0 [redis $node_0_host $node_0_port 1 $::tls]} e
assert_match {*refused*} $e
}
}
}
-11
View File
@@ -315,15 +315,4 @@ start_server {
}
}
}
test {SETRANGE with huge offset} {
r lpush L 2 1 0
# expecting a different outcome on 32 and 64 bit systems
foreach value {9223372036854775807 2147483647} {
catch {[r sort_ro L by a limit 2 $value]} res
if {![string match "2" $res] && ![string match "*out of range*" $res]} {
assert_not_equal $res "expecting an error or 2"
}
}
}
}
-5
View File
@@ -68,11 +68,6 @@ start_server {tags {"hash"}} {
r hrandfield myhash 0
} {}
test "HRANDFIELD count overflow" {
r hmset myhash a 1
assert_error {*value is out of range*} {r hrandfield myhash -9223372036854770000 withvalues}
} {}
test "HRANDFIELD with <count> against non existing key" {
r hrandfield nonexisting_key 100
} {}
-10
View File
@@ -574,14 +574,4 @@ start_server {tags {"string"}} {
test {LCS indexes with match len and minimum match len} {
dict get [r STRALGO LCS IDX KEYS virus1 virus2 WITHMATCHLEN MINMATCHLEN 5] matches
} {{{1 222} {13 234} 222}}
test {SETRANGE with huge offset} {
foreach value {9223372036854775807 2147483647} {
catch {[r setrange K $value A]} res
# expecting a different error on 32 and 64 bit systems
if {![string match "*string exceeds maximum allowed size*" $res] && ![string match "*out of range*" $res]} {
assert_equal $res "expecting an error"
}
}
}
}
-5
View File
@@ -1714,11 +1714,6 @@ start_server {tags {"zset"}} {
r zrandmember nonexisting_key 100
} {}
test "ZRANDMEMBER count overflow" {
r zadd myzset 0 a
assert_error {*value is out of range*} {r zrandmember myzset -9223372036854770000 withscores}
} {}
# Make sure we can distinguish between an empty array and a null response
r readraw 1