Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions app/lavik.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -391,11 +391,14 @@ int main(int argc, char** argv) {

constexpr std::size_t kMiB = 1024 * 1024;
constexpr std::size_t kKiB = 1024;
if (registered_buffer_mb > std::numeric_limits<std::size_t>::max() / kMiB ||
replication_publish_queue_mb >
if (static_cast<std::size_t>(registered_buffer_mb) >
std::numeric_limits<std::size_t>::max() / kMiB ||
storage_read_buffer_kb > std::numeric_limits<std::size_t>::max() / kKiB ||
flush_size_kb > std::numeric_limits<std::size_t>::max() / kKiB) {
static_cast<std::size_t>(replication_publish_queue_mb) >
std::numeric_limits<std::size_t>::max() / kMiB ||
static_cast<std::size_t>(storage_read_buffer_kb) >
std::numeric_limits<std::size_t>::max() / kKiB ||
static_cast<std::size_t>(flush_size_kb) >
std::numeric_limits<std::size_t>::max() / kKiB) {
return 2;
}
options.registered_buffer_bytes_ =
Expand Down
2 changes: 1 addition & 1 deletion include/lavik/storage/detail/grouped/hash.h
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ struct HashGroupSnapshot {
bool retired_ = false;
HashValue value_;
// Mutually exclusive with value_; the encoder checks this invariant.
std::optional<PreparedHashGroupPayload> prepared_;
std::optional<PreparedHashGroupPayload> prepared_ = std::nullopt;
std::size_t field_count() const noexcept {
return prepared_ ? prepared_->count() : value_.entries_.size();
}
Expand Down
12 changes: 7 additions & 5 deletions include/lavik/storage/engine.h
Original file line number Diff line number Diff line change
Expand Up @@ -890,7 +890,7 @@ struct StreamTrimRequest {
struct StreamTrimResult {
std::uint64_t removed_ = 0;
std::uint64_t length_ = 0;
std::optional<std::array<std::uint64_t, 2>> first_id_;
std::optional<std::array<std::uint64_t, 2>> first_id_ = std::nullopt;
};

struct CompactValueUpdate {
Expand All @@ -903,6 +903,8 @@ struct CompactValueUpdate {
std::uint64_t logical_size_ = 0;
// nullopt preserves the current deadline (or persistence for a new key).
std::optional<std::uint64_t> expire_at_ms_;
// Non-Stream aggregate callers explicitly use {} for the Stream-only state.
// GCC 13 emits redundant stores with default member initializers here.
std::optional<StreamTrimRequest> stream_trim_;
// Called once after planning, before publication, to canonicalize replication
// and the reply from the exact boundary. It must not access storage;
Expand All @@ -924,7 +926,7 @@ struct StreamRangeAccess {
bool reverse_ = false;
// Pinned reply sources may select sorted, unique IDs known to exist under
// the caller's intent. Empty means an ordinary contiguous ID range.
std::vector<std::array<std::uint64_t, 2>> selected_ids_;
std::vector<std::array<std::uint64_t, 2>> selected_ids_{};
};

// A writable callback sees only the named messages and may remove them. Storage
Expand All @@ -943,7 +945,7 @@ struct StreamAckAccess {
// additionally load the corresponding messages; deleted messages stay absent.
struct StreamPendingAccess {
StreamRangeAccess range_;
std::optional<std::string> consumer_;
std::optional<std::string> consumer_ = std::nullopt;
std::uint64_t min_idle_ms_ = 0;
std::uint64_t now_ms_ = 0;
bool load_entries_ = false;
Expand All @@ -958,13 +960,13 @@ struct StreamGroupAccess {
std::vector<std::array<std::uint64_t, 2>> pending_ids_{};
std::vector<std::array<std::uint64_t, 2>> entry_ids_{};
std::optional<std::uint64_t> read_new_count_ = std::nullopt;
std::optional<StreamPendingAccess> pending_scan_;
std::optional<StreamPendingAccess> pending_scan_ = std::nullopt;
// Delivery callbacks need IDs/counters, while replies read immutable fields
// through a pinned range source after publication under the same key intent.
bool entry_ids_only_ = false;
bool create_ = false;
bool destroy_ = false;
std::optional<std::string> remove_consumer_;
std::optional<std::string> remove_consumer_ = std::nullopt;
};

struct StreamInspectAccess {
Expand Down
5 changes: 3 additions & 2 deletions src/config.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -806,8 +806,9 @@ absl::Status ValidateServerOptions(const ServerOptions& options) {
"storage-read-buffer-kb must be a 4 KiB multiple greater than 8 KiB "
"and fit the 32-bit slot size");
}
if (options.shard_count_ > std::numeric_limits<std::size_t>::max() /
storage::kStorageBlockBytes ||
if (static_cast<std::size_t>(options.shard_count_) >
std::numeric_limits<std::size_t>::max() /
storage::kStorageBlockBytes ||
options.replication_options_.backlog_size_bytes_ <
static_cast<std::size_t>(options.shard_count_) *
storage::kStorageBlockBytes) {
Expand Down
2 changes: 1 addition & 1 deletion src/redis/backup.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ class RdbOutputQueue {
return true;
}

void EndEntry(unsigned owner) {
void EndEntry([[maybe_unused]] unsigned owner) {
std::lock_guard lock(mutex_);
assert(entry_owner_ == owner);
entry_owner_ = UINT_MAX;
Expand Down
3 changes: 1 addition & 2 deletions src/redis/command.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -6580,8 +6580,7 @@ Task<std::string> ExecuteExecSequentialSort(
std::vector<SortExecKey> sort_keys;
sort_keys.reserve(keys.size());
for (const ExecKey& key : keys) {
sort_keys.push_back(SortExecKey{
.digest_ = key.digest_, .owner_ = key.owner_, .arg_ = key.arg_});
sort_keys.push_back(SortExecKey{.arg_ = key.arg_});
}
co_return co_await ExecuteSortCommandLocked(command, sort_keys, tx_writes,
deterministic_set_order);
Expand Down
2 changes: 1 addition & 1 deletion src/redis/rdb_record_spool.h
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ class RecordSpool {
public:
struct Record {
std::optional<MemoryReservation> reservation_;
std::string key_, value_;
std::string key_{}, value_{};
};
absl::Status Add(std::string_view key, std::string_view value) {
if (finished_)
Expand Down
6 changes: 3 additions & 3 deletions src/redis/sort_command.h
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,10 @@ namespace lavik {
class ReplyBuilder;

struct SortExecKey {
storage::Digest digest_;
std::uint16_t owner_ = 0;
// EXEC supplies a command argument; scripts supply a declared key name.
// The SORT adapter derives the digest and owner from that selected name.
std::uint16_t arg_ = 0;
std::string_view name_;
std::string_view name_{};
};

void InitSortCommandStorage(storage::StorageEngine* engine);
Expand Down
19 changes: 12 additions & 7 deletions src/redis/stream_command.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ struct Group {
std::int64_t entries_read_ = -1;
std::vector<Consumer> consumers_;
std::vector<Pending> pending_;
std::optional<storage::StreamGroupSummary> summary_;
std::optional<storage::StreamGroupSummary> summary_ = std::nullopt;
};

struct Stream {
Expand Down Expand Up @@ -781,10 +781,15 @@ storage::CompactValueUpdate NoChange() { return {}; }
absl::StatusOr<storage::CompactValueUpdate> Changed(Stream stream) {
auto encoded = Encode(stream);
if (!encoded.ok()) return encoded.status();
return storage::CompactValueUpdate{.changed_ = true,
.encoded_ = std::move(*encoded),
.logical_size_ = stream.entries_.size(),
.expire_at_ms_ = std::nullopt};
return storage::CompactValueUpdate{
.changed_ = true,
.encoded_ = std::move(*encoded),
.logical_size_ = stream.entries_.size(),
.expire_at_ms_ = std::nullopt,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
}

Group* FindGroup(Stream* stream, std::string_view name) {
Expand Down Expand Up @@ -1378,9 +1383,9 @@ struct ReadOneResult {
Id id_;
std::optional<Entry> entry_;
};
std::vector<Item> entries_;
std::vector<Item> entries_{};
Id cursor_;
std::shared_ptr<StreamRangeReplyState> stream_;
std::shared_ptr<StreamRangeReplyState> stream_{};
};

Task<absl::StatusOr<ReadOneResult>> ReadOneLocal(
Expand Down
46 changes: 34 additions & 12 deletions src/redis/string_command.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -637,6 +637,9 @@ bycorf::Task<std::string> RunBitmapLocked(
.encoded_ = std::move(next),
.logical_size_ = logical_size,
.expire_at_ms_ = std::nullopt,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
};

Expand Down Expand Up @@ -805,19 +808,27 @@ bycorf::Task<std::string> RunStringLocked(
.encoded_ = std::move(encoded),
.logical_size_ = size,
.expire_at_ms_ = expire_at,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
};

if (request.kind_ == CommandKind::kGetDel) {
reply = exists ? EncodeBulkString(old)
: EncodeSemanticNull(request.resp_version_);
if (exists) captured_args = {"DEL", args[1]};
return exists ? storage::CompactValueUpdate{.changed_ = true,
.erase_ = true,
.encoded_ = {},
.logical_size_ = 0,
.expire_at_ms_ = std::nullopt}
: storage::CompactValueUpdate{};
if (!exists) return storage::CompactValueUpdate{};
captured_args = {"DEL", args[1]};
return storage::CompactValueUpdate{
.changed_ = true,
.erase_ = true,
.encoded_ = {},
.logical_size_ = 0,
.expire_at_ms_ = std::nullopt,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
}

if (request.kind_ == CommandKind::kGetEx) {
Expand All @@ -836,6 +847,9 @@ bycorf::Task<std::string> RunStringLocked(
.encoded_ = {},
.logical_size_ = old.size(),
.expire_at_ms_ = 0,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
}
const bool ex = RedisEqualsIgnoreCase(args[2], "ex");
Expand All @@ -849,18 +863,26 @@ bycorf::Task<std::string> RunStringLocked(
captured_args = {"PEXPIREAT", args[1], std::to_string(*getex_deadline)};
if (replication.has_value()) replication->args_ = captured_args;
if (*getex_deadline <= RedisUnixTimeMillis()) {
return storage::CompactValueUpdate{.changed_ = true,
.erase_ = true,
.encoded_ = {},
.logical_size_ = 0,
.expire_at_ms_ = std::nullopt};
return storage::CompactValueUpdate{
.changed_ = true,
.erase_ = true,
.encoded_ = {},
.logical_size_ = 0,
.expire_at_ms_ = std::nullopt,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
}
return storage::CompactValueUpdate{
.changed_ = true,
.reuse_encoded_ = true,
.encoded_ = {},
.logical_size_ = old.size(),
.expire_at_ms_ = *getex_deadline,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
}

Expand Down
17 changes: 11 additions & 6 deletions src/redis/zset_command.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -771,7 +771,7 @@ std::uint64_t Interleave26(std::uint32_t lon, std::uint32_t lat) {

std::uint64_t GeoScore(double lon, double lat) {
constexpr double scale = static_cast<double>(std::uint64_t{1} << 26);
auto normalized = [scale](double value, double minimum, double maximum) {
auto normalized = [](double value, double minimum, double maximum) {
double n = (value - minimum) / (maximum - minimum);
n = std::clamp(n, 0.0, std::nextafter(1.0, 0.0));
return static_cast<std::uint32_t>(n * scale);
Expand All @@ -796,7 +796,7 @@ std::pair<double, double> GeoDecode(std::uint64_t hash) {
std::string GeoHashString(double lon, double lat) {
static constexpr char alphabet[] = "0123456789bcdefghjkmnpqrstuvwxyz";
constexpr double scale = static_cast<double>(std::uint64_t{1} << 26);
auto normalized = [scale](double value, double minimum, double maximum) {
auto normalized = [](double value, double minimum, double maximum) {
double n = (value - minimum) / (maximum - minimum);
n = std::clamp(n, 0.0, std::nextafter(1.0, 0.0));
return static_cast<std::uint32_t>(n * scale);
Expand Down Expand Up @@ -2299,10 +2299,15 @@ Task<absl::Status> ReplaceMultiDestination(
return absl::InternalError(
"sorted-set STORE destination was not replaced");
}
return storage::CompactValueUpdate{.changed_ = true,
.encoded_ = std::move(*encoded),
.logical_size_ = context->output_.size(),
.expire_at_ms_ = std::nullopt};
return storage::CompactValueUpdate{
.changed_ = true,
.encoded_ = std::move(*encoded),
.logical_size_ = context->output_.size(),
.expire_at_ms_ = std::nullopt,
.stream_trim_ = {},
.stream_trim_complete_ = {},
.stream_pending_removed_ = {},
};
};
co_return co_await g_storage->ExecuteCompactLocked(
request.db_id_, destination, digest, storage::ValueType::kSortedSet,
Expand Down
2 changes: 1 addition & 1 deletion src/replication/redis_export.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,7 @@ class RedisRdbStreamQueue
available, owner, std::memory_order_acquire, std::memory_order_relaxed);
}

bool TryPush(std::string* fragment, unsigned owner) {
bool TryPush(std::string* fragment, [[maybe_unused]] unsigned owner) {
assert(entry_owner_.load(std::memory_order_relaxed) == owner);
if (aborted_.load(std::memory_order_relaxed)) return false;
const std::size_t bytes = fragment->size();
Expand Down
3 changes: 2 additions & 1 deletion src/storage/engine/compact_api.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -491,7 +491,8 @@ Task<absl::Status> StorageEngine::Impl::ExecuteCompactLocked(
// Preserve explicit identical-value/metadata mutation semantics from
// the callback. A single page suffices; EXPIRE/PERSIST themselves use
// the separate root-only metadata path.
const auto& first = directory.groups().front();
const auto groups = directory.groups();
const auto& first = groups.front();
OrderedGroupSnapshot page{.kind_ = directory.root().kind_,
.incarnation_ = directory.root().incarnation_,
.id_ = first.id_,
Expand Down
3 changes: 2 additions & 1 deletion src/storage/engine/grouped/list.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -423,6 +423,7 @@ Task<absl::StatusOr<ListResult>> StorageEngine::Impl::ExecuteGroupedListLocked(
store, partition, db_id, key, digest, object,
directory.groups()[begin_page].id_, offset, remaining));
} else {
const auto groups = directory.groups();
for (std::size_t i = begin_page; i < end_page;) {
ListReadBatch batch(store.worker_);
for (std::size_t j = 0; j < ListReadBatch::kWidth && i + j < end_page;
Expand All @@ -435,7 +436,7 @@ Task<absl::StatusOr<ListResult>> StorageEngine::Impl::ExecuteGroupedListLocked(
"OOM grouped List read batch admission");
break;
}
const auto& route = directory.groups()[i + j];
const auto& route = groups[i + j];
const auto take =
std::min<std::uint64_t>(remaining, route.item_count_ - offset);
batch.Add(LoadOrderedListRange(store, partition, db_id, key, digest,
Expand Down
5 changes: 4 additions & 1 deletion src/storage/engine/grouped/stream.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -913,7 +913,10 @@ Task<absl::Status> StorageEngine::Impl::ExecuteGroupedStreamInspect(
if (!status.ok()) co_return status;
std::string group_header(*value);
StreamGroupSummary summary{.name_ = name,
.consumers_ = Count(*value, value->size() - 4)};
.consumers_ = Count(*value, value->size() - 4),
.first_pending_ = {},
.last_pending_ = {},
.consumer_pending_ = {}};
auto count = co_await cache.Required(prefix + '\2');
if (!count.ok()) co_return count.status();
auto count_value = StreamRecordPayload(*count);
Expand Down
Loading