feat: add an option to flush serialized entries on threshold limit #3241

kostasrim · 2024-06-28T17:18:24Z

Resolves #3223

serialize big slots in chunks
allow preemption on large slots
disable big entries serialization for RDB files
add test

There is also a separate issue with compression. Previously, we only called FlushToSink in SaveEpilogue and before SaveBody (to flush previous entries). However, with the current changes, we call FlushToSink when we serialize X number of bytes. This doesn't work with the protocol implemented in rdb_load because we now split let's say a list in multiple chunks and LoadKeyValue on RdbLoader expects to read a string (for the values) but it gets a compressed blob and fails with unrecognized rdb type. I created this separate issue for that: #3324

romange · 2024-07-07T08:56:42Z

what is the related issue for this PR?

kostasrim · 2024-07-08T10:33:57Z

what is the related issue for this PR?

I will edit the description soon, it's still in draft and incomplete. Plz don't review

kostasrim · 2024-07-16T15:01:09Z

src/server/rdb_save.cc

@@ -430,6 +432,7 @@ error_code RdbSerializer::SaveListObject(const PrimeValue& pv) {
        RETURN_ON_ERR(SaveLzfBlob(Bytes{reinterpret_cast<uint8_t*>(data), compress_len}, node->sz));
      } else {
        RETURN_ON_ERR(SaveString(node->entry, node->sz));
+        MaybeApplyFlushFunc();


This is not complete but enough to reduce the memory footprint of the most use cases. We only serialize our own data structures (Which are the default) but not the Valkey one. Also json serializes to a string blob and it's rather difficult to make it work (unless we provide our own wrapper or something).

chakaz · 2024-07-17T06:05:32Z

src/server/db_slice.cc

+class FetchedItemsRestorer {
+ public:
+  using RestoreType = absl::flat_hash_set<CompactObjectView>;
+  explicit FetchedItemsRestorer(RestoreType* dst) : dst_to_restore_(dst) {


Is it safe to clear fetched_items_ while we yield? It looks like other parts of the code may use it, like our Find APIs

fetched_items are cleraed when OnCbFinish gets called which is at the end of a transaction (in fact I think we have a small bug there, I will ping internally). When the fiber preempts we should cache this and when it resumes we need to bring it back. The code above provides this guarantee.

Also if Find is called it will be a part of another transaction which will either conclude and call OnCbFinish or will preempt, cache this value and bring it back when the fiber resumes preserving the correctness

chakaz · 2024-07-17T06:34:53Z

src/server/rdb_save.h

  // for the dump command - thus it is public function
+  // Also this function might preempt if flush_fun_ is used


Suggested change

// for the dump command - thus it is public function

// Also this function might preempt if flush_fun_ is used

// for the dump command - thus it is public function.

// This function might preempt if flush_fun_ is used.

chakaz · 2024-07-17T06:35:29Z

src/server/rdb_save.h

@@ -238,8 +241,12 @@ class RdbSerializer : public SerializerBase {
  std::error_code SaveStreamPEL(rax* pel, bool nacks);
  std::error_code SaveStreamConsumers(streamCG* cg);

+  // Might preempt
+  bool MaybeApplyFlushFunc();


Suggested change

bool MaybeApplyFlushFunc();

bool FlushIfNeeded();

chakaz · 2024-07-17T06:39:16Z

src/server/snapshot.cc

@@ -20,6 +21,8 @@
 #include "server/tiered_storage.h"
 #include "util/fibers/synchronization.h"

+ABSL_FLAG(size_t, flush_big_entries_threshold, 0, "Total bytes before flushing big entries");


Suggested change

ABSL_FLAG(size_t, flush_big_entries_threshold, 0, "Total bytes before flushing big entries");

ABSL_FLAG(size_t, serialization_max_chunk_size, 10'000'000, "Total bytes before flushing big entries");

I'd rename it to be clearer, but also provide a default of 10mb. Any reason to disable this feature be default?

Cause if we got any issues that we did not forsee it will make DF unstable by default. Furthermore, what we do here is a trade off between memory, latency and throughput. By serializing smaller chunks we reduce the memory footprint and provide a little better latency per packet (since we send them at smaller intervals). However, we reduce throughput because now we yield and we go back and forth. I would not like to have this defaulted. We could if we think it's best to have this on, add it on the release after this release

Cause if we got any issues that we did not forsee it will make DF unstable by default

If it's disabled, we won't know if it's unstable. We should build it the best that we can, and if there are bugs, we'll fix them :)

Furthermore, what we do here is a trade off between memory, latency and throughput

One of the advantages of this feature is that actually Dragonfly will become more responsive to requests coming in while it is serializing. Today, when we serialize a huge entry, the entire thread is blocked, so for the duration of copying it (think 100mb, 1gb, even 10gb) we won't advance in the tx queue, won't even reply to pings coming from connections on this thread.
Latency of serialization is way less critical compared to this.

chakaz · 2024-07-17T06:42:37Z

src/server/snapshot.h

@@ -67,7 +67,7 @@ class SliceSnapshot {

  // Initialize snapshot, start bucket iteration fiber, register listeners.
  // In journal streaming mode it needs to be stopped by either Stop or Cancel.
-  void Start(bool stream_journal, const Cancellation* cll);
+  void Start(bool stream_journal, const Cancellation* cll, bool allow_flush_fun = false);


I'm not a fan of boolean arguments to functions. They tend to pile up and also they are unreadable at the call site.
I'd add an enum for this (enum class SnapshotFlush { kAllow, kDisallow };).
However, I hope we can simply enable flushing in all modes, and make the RDB reader aware of when chunks end.

+1 for adding an enum but as I explained above I am not a big fan of defaulting this

Why is defaulting related to bool vs enum?

I meant for defaulting the flag. Sorry my bad here :)

chakaz · 2024-07-17T06:53:47Z

src/server/snapshot.cc

+  if (flush_threshold != 0 && allow_flush_fun) {
+    flush_fun = [this, flush_threshold](size_t bytes_serialized) {
+      if (bytes_serialized > flush_threshold) {
+        auto serialized = Serialize();


Why not call PushSerializedToChannel() here?
BTW this is another place where if we allowed to flush also in RDB mode, it will make our code simpler (see line 310 below, we can remove delayed_entries_ and its usage)

I think we also found a bug. We can't send delayed entries because it could be that we are in a middle of a chunk (and if we serialize those delayed entries we just interleaved the channel breaking our invariants). The same thing applies for PushSerializedToChannel (and I patched this accordingly now)

chakaz · 2024-07-17T07:32:39Z

src/server/snapshot.h

Here's a proposal:
Add another bit to DbRecord, say bool record_in_progress;.
Set this to false normally, but when you today call flush function, set it to true.
Now we can apply the following changes, which in my opinion would simplify the code and remove edge cases:

We can always flush, we don't need the special case of RDB and to pass whether or not it is allowed

Only the RDB consumer of the channel will care about this - it will read from the same stream as long as this bit is set, and when it will be unset it will move to the next stream

wdyt?

Let me think about this one I will ping

I think this might be problematic because we have multiple producers on single consumer channel. So all of these DbRecords that are being serialized by separate shards might end up being interleaved (because all of these shards are not synchronized when pushing to the channel, so it could be the case that k1 from shard 1 sends a DbRecord that contains a half chunk only for shard 2 to send another DbRecord that contains another half chunk of k2) breaking the ordering requirements.

And saying this I think this is also problematic for any flow where we have multiple producers over a single channel which I think applies for df replication when the replica does not have the same amount of shard threads as the master. WDYT?

This pr #2474 introduced iterator protection by tracking which keys where bumped up during the transaction operation. This was done by managing keys view set. However, this can be simplified using fingerprints. Also, fingerprints do not require that the original keys exist. In addition, this #3241 PR introduces FetchedItemsRestorer that tracks bumped set and saves it to protect against fiber context switch. My claim is that it's redundant. Since we only keep the auto-laundering iterators, when a fiber preempts these iterators recognize it (see IteratorT::LaunderIfNeeded) and refresh themselves anyway. To summarize: fetched_items_ protect us from iterator invalidation during atomic scenarios, and auto-laundering protects us from everything else, so fetched_items_ can be cleared in that case. Signed-off-by: Roman Gershman <roman@dragonflydb.io>

* chore: simplify BumpUps deduplication This pr #2474 introduced iterator protection by tracking which keys where bumped up during the transaction operation. This was done by managing keys view set. However, this can be simplified using fingerprints. Also, fingerprints do not require that the original keys exist. In addition, this #3241 PR introduces FetchedItemsRestorer that tracks bumped set and saves it to protect against fiber context switch. My claim is that it's redundant. Since we only keep the auto-laundering iterators, when a fiber preempts these iterators recognize it (see IteratorT::LaunderIfNeeded) and refresh themselves anyway. To summarize: fetched_items_ protect us from iterator invalidation during atomic scenarios, and auto-laundering protects us from everything else, so fetched_items_ can be cleared in that case. --------- Signed-off-by: Roman Gershman <roman@dragonflydb.io>

kostasrim added 4 commits June 26, 2024 10:02

feat: yield when serialization is in progress

96c6aef

replace bool with mutex

babfc79

update restore streamer

de9b143

feat: add an option to flush serialized entries in SaveEntry

1f77383

kostasrim self-assigned this Jun 28, 2024

kostasrim changed the base branch from main to yield_on_db_change June 28, 2024 17:18

Base automatically changed from yield_on_db_change to main July 11, 2024 13:55

kostasrim added 3 commits July 15, 2024 16:13

Merge branch 'main' into chunks

7a3c5d9

refactor and serialize/split objects

9c27ae3

add test

5796bb7

kostasrim marked this pull request as ready for review July 16, 2024 14:59

kostasrim requested review from chakaz and adiholden July 16, 2024 14:59

kostasrim commented Jul 16, 2024

View reviewed changes

kostasrim added 2 commits July 16, 2024 18:11

clean up

5eef741

fixes

a5f1c13

kostasrim changed the title ~~feat: add an option to flush serialized entries in SaveEntryChunks~~ feat: add an option to flush serialized entries on threshold limit Jul 16, 2024

chakaz requested changes Jul 17, 2024

View reviewed changes

chakaz mentioned this pull request Jul 17, 2024

Support compression on serialization of big values #3324

Closed

poish and address some of the comments

654cf30

chakaz approved these changes Jul 18, 2024

View reviewed changes

kostasrim enabled auto-merge (squash) July 18, 2024 09:57

kostasrim disabled auto-merge July 18, 2024 09:57

kostasrim enabled auto-merge (squash) July 18, 2024 10:02

kostasrim merged commit bfa5df5 into main Jul 18, 2024
10 checks passed

kostasrim deleted the chunks branch July 18, 2024 10:15

romange mentioned this pull request Nov 9, 2024

chore: simplify BumpUps deduplication #4098

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add an option to flush serialized entries on threshold limit #3241

feat: add an option to flush serialized entries on threshold limit #3241

kostasrim commented Jun 28, 2024 •

edited

Loading

romange commented Jul 7, 2024

kostasrim commented Jul 8, 2024 •

edited

Loading

kostasrim Jul 16, 2024 •

edited

Loading

chakaz Jul 17, 2024

kostasrim Jul 17, 2024 •

edited

Loading

chakaz Jul 17, 2024

chakaz Jul 17, 2024

chakaz Jul 17, 2024

chakaz Jul 17, 2024

kostasrim Jul 17, 2024

chakaz Jul 17, 2024

chakaz Jul 17, 2024

kostasrim Jul 17, 2024

chakaz Jul 17, 2024

kostasrim Jul 17, 2024

chakaz Jul 17, 2024

kostasrim Jul 17, 2024

chakaz Jul 17, 2024

kostasrim Jul 17, 2024

kostasrim Jul 17, 2024 •

edited

Loading

		// for the dump command - thus it is public function
		// Also this function might preempt if flush_fun_ is used

	ABSL_FLAG(size_t, flush_big_entries_threshold, 0, "Total bytes before flushing big entries");
	ABSL_FLAG(size_t, serialization_max_chunk_size, 10'000'000, "Total bytes before flushing big entries");

feat: add an option to flush serialized entries on threshold limit #3241

feat: add an option to flush serialized entries on threshold limit #3241

Conversation

kostasrim commented Jun 28, 2024 • edited Loading

romange commented Jul 7, 2024

kostasrim commented Jul 8, 2024 • edited Loading

kostasrim Jul 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kostasrim Jul 17, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kostasrim Jul 17, 2024 • edited Loading

Choose a reason for hiding this comment

kostasrim commented Jun 28, 2024 •

edited

Loading

kostasrim commented Jul 8, 2024 •

edited

Loading

kostasrim Jul 16, 2024 •

edited

Loading

kostasrim Jul 17, 2024 •

edited

Loading

kostasrim Jul 17, 2024 •

edited

Loading