feat(server): Pubsub updates with RCU #980

dranikpg · 2023-03-22T16:27:59Z

Implements RCU (read-copy-update) for updating the centralized channel store.

Contrary to old mechanism of sharding subscriber info across shards, a centralized store allows avoiding a hop for fetching subscribers. In general, it only slightly improves the latency, but in case of heavy traffic on one channel it allows "spreading" the load, as the single shard no longer is a bottleneck, thus increasing throughput by multiple times.

Benchmarks:

1. Dry run without subscribers

OLD
===================================================================================================
Type          Ops/sec    Avg. Latency     p50 Latency     p99 Latency   p99.9 Latency       KB/sec  
---------------------------------------------------------------------------------------------------
Publishs    955759.86         0.75294         0.71100         1.32700         4.57500     43709.23  
Totals      955759.86         0.75294         0.71100         1.32700         4.57500     43709.23

NEW
===================================================================================================
Type          Ops/sec    Avg. Latency     p50 Latency     p99 Latency   p99.9 Latency       KB/sec  
---------------------------------------------------------------------------------------------------
Publishs   1040942.05         0.69134         0.67900         1.07900         3.75900     47604.77  
Totals     1040942.05         0.69134         0.67900         1.07900         3.75900     47604.77 

2. Run with subscribers

OLD
===================================================================================================
Type          Ops/sec    Avg. Latency     p50 Latency     p99 Latency   p99.9 Latency       KB/sec  
---------------------------------------------------------------------------------------------------
Publishs    399898.33         1.79988         1.37500         7.00700        11.64700     18288.39  
Totals      399898.33         1.79988         1.37500         7.00700        11.64700     18288.39 

NEW
===================================================================================================
Type          Ops/sec    Avg. Latency     p50 Latency     p99 Latency   p99.9 Latency       KB/sec  
---------------------------------------------------------------------------------------------------
Publishs    418730.46         1.71866         1.21500         6.97500        10.81500     19149.67  
Totals      418730.46         1.71866         1.21500         6.97500        10.81500     19149.6

3. Single channel

OLD
===================================================================================================
Type          Ops/sec    Avg. Latency     p50 Latency     p99 Latency   p99.9 Latency       KB/sec  
---------------------------------------------------------------------------------------------------
Publishs     88210.54         9.06565         8.76700        15.10300        22.01500      4048.73  
Totals       88210.54         9.06565         8.76700        15.10300        22.01500      4048.73 

NEW
===================================================================================================
Type          Ops/sec    Avg. Latency     p50 Latency     p99 Latency   p99.9 Latency       KB/sec  
---------------------------------------------------------------------------------------------------
Publishs    236476.81         3.38045         1.13500        27.39100        35.32700     10853.92  
Totals      236476.81         3.38045         1.13500        27.39100        35.32700     10853.92

Signed-off-by: Vladislav Oleshko <[email protected]>

romange · 2023-03-22T16:38:28Z

Please adopt a habit of adding additional info to the commit description. It's not a one-liner PR.

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg · 2023-03-22T18:01:27Z

src/server/channel_store.h

+  ChannelMap* channels_;
+  ChannelMap* patterns_;
+  ControlBlock* control_block_;
+};


That's a lot of indirections as its easier to work it this way, but to flatten it we can:

Store the ChannelStore in threads by value (3x8)

Store the ChannelMaps without pointer and move them around with ugly memcpy's

Store a single ChannelStore and make the ChannelMap* pointers atomic. We still need to dispatch to the shard_set to allow deletes (as we don't have hazard pointers)

dranikpg · 2023-03-22T18:03:56Z

src/server/channel_store.h

+  // Wrapper around atomic pointer that allows copying and moving.
+  // Made to overcome restrictions of absl::flat_hash_map.
+  // Copy/Move don't need to be atomic with RCU.
+  struct UpdatablePointer {
+    UpdatablePointer(SubscribeMap* sm) : ptr{sm} {
+    }
+


That's a tricky part... flat_hash_map really doesn't allow full in-place construction, std does (it has std::piecewise_construct)

We can also add to helio:
https://github.com/romange/beeri/blob/master/base/atomic_wrapper.h
which should solve the problem

dranikpg · 2023-03-22T18:05:03Z

src/server/channel_store.h

+  // Centralized controller to prevent overlaping updates.
+  struct ControlBlock {
+    void Destroy();
+
+    ChannelStore* most_recent;
+    ::boost::fibers::mutex update_mu;  // locked during updates.
+  };


In the future, instead of plain locking, we can accumulate all changes and let all fulfill all writers requests with a single operation

Its a bit more difficult to synchronize it, besides we need to explicitly aggregate changes by key to tell whether we'll modify the map slots (replace/add on single value cancel each other)

src/server/channel_store.h

src/server/main_service.cc

dranikpg · 2023-03-22T18:11:12Z

src/server/channel_store.cc

+  // RCU update existing SubscribeMap entry.
+  DCHECK(it->second->size() > 0);
+  auto* replacement = new SubscribeMap{*it->second};
+  if (add)
+    replacement->emplace(cntx_, thread_id_);
+  else
+    replacement->erase(cntx_);
+
+  freelist_.push_back(it->second.Get());
+  it->second.Set(replacement);
+}


This is what no full copy RCU looks like

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg · 2023-03-24T18:07:51Z

src/server/main_service.cc

+    };
+    shard_set->pool()->DispatchBrief(std::move(cb));


TODO: Dispatch only on active threads

Signed-off-by: Vladislav Oleshko <[email protected]>

src/server/channel_store.h

romange · 2023-03-24T18:59:55Z

src/server/channel_store.cc

+
+  // RCU update existing SubscribeMap entry.
+  DCHECK(it->second->size() > 0);
+  auto* replacement = new SubscribeMap{*it->second};


can Modify run on multiple threads in parallel? I see it under mutex lock below so I am a bit confused
why do you need atomics?

Because the map slot is read by all the other reader threads, added a comment

Signed-off-by: Vladislav Oleshko <[email protected]>

romange

Looks very good. My only concern is around Apply() bottleneck

src/server/channel_store.h

romange · 2023-03-25T08:27:36Z

src/server/channel_store.h

+  // Wrapper around atomic pointer that allows copying and moving.
+  // Made to overcome restrictions of absl::flat_hash_map.
+  // Copy/Move don't need to be atomic with RCU.
+  struct UpdatablePointer {
+    UpdatablePointer(SubscribeMap* sm) : ptr{sm} {
+    }
+


We can also add to helio:
https://github.com/romange/beeri/blob/master/base/atomic_wrapper.h
which should solve the problem

src/server/channel_store.h

src/server/channel_store.cc

src/server/main_service.cc

romange · 2023-03-25T08:47:57Z

src/server/channel_store.h

+  std::pair<ChannelMap*, bool> GetTargetMap();
+
+  // Apply modify operation to target map.
+  void Modify(ChannelMap* target, std::string_view key);


src/server/channel_store.h

romange · 2023-03-25T08:59:24Z

src/server/channel_store.cc

+
+  // Update control block and unlock it.
+  cb.most_recent = replacement;
+  cb.update_mu.unlock();


food for thought: Since the critical section includes a hop, it set a low limit for the throughput capacity of this operation. maybe another way to improve it is to make most_recent atomic as well,
move Await call to after unlock and instead of passing "replacement", read directly from most_recent.

Good idea. I'm actually not sure how much of an improvement the thread local pointer has, we do a lot of atomic ops either way 🤷🏻‍♂️

src/server/channel_store.cc

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg · 2023-03-25T12:45:57Z

Fixed.

My only concern is around Apply() bottleneck

Yes, I didn't yet optimize it - we should squash parallel updates in the future.

feat(server): Pubsub updates with RCU

f174002

Signed-off-by: Vladislav Oleshko <[email protected]>

chore(server): Small rcu fixes

a9367a0

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg force-pushed the pubsub-rcu branch from fa856ec to a9367a0 Compare March 22, 2023 18:09

dranikpg commented Mar 22, 2023

View reviewed changes

dranikpg added 2 commits March 23, 2023 12:44

fix(server): Skip empty publish

5fbbbc7

Signed-off-by: Vladislav Oleshko <[email protected]>

fix(server): Small fixes

d624609

dranikpg marked this pull request as ready for review March 24, 2023 17:21

dranikpg requested a review from romange March 24, 2023 17:21

fix(server): Use DispatchBrief in Publish

923a88b

Signed-off-by: Vladislav Oleshko <[email protected]>

dranikpg force-pushed the pubsub-rcu branch from 354aea0 to 923a88b Compare March 24, 2023 17:41

dranikpg commented Mar 24, 2023

View reviewed changes

dranikpg added 2 commits March 24, 2023 21:08

Merge branch 'main' into pubsub-rcu

b5445bf

fix(server): Fix test

26c7bb3

Signed-off-by: Vladislav Oleshko <[email protected]>

romange reviewed Mar 24, 2023

View reviewed changes

fix(server): Small fixes

31c85b8

Signed-off-by: Vladislav Oleshko <[email protected]>

romange reviewed Mar 25, 2023

View reviewed changes

fix(server): More fixes

3d4baec

Signed-off-by: Vladislav Oleshko <[email protected]>

romange approved these changes Mar 25, 2023

View reviewed changes

dranikpg merged commit 139e56b into dragonflydb:main Mar 26, 2023

dranikpg deleted the pubsub-rcu branch April 1, 2023 16:32

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(server): Pubsub updates with RCU #980

feat(server): Pubsub updates with RCU #980

dranikpg commented Mar 22, 2023 •

edited

Loading

romange commented Mar 22, 2023

dranikpg Mar 22, 2023

dranikpg Mar 22, 2023

romange Mar 25, 2023

dranikpg Mar 22, 2023

dranikpg Mar 22, 2023

dranikpg Mar 22, 2023

dranikpg Mar 24, 2023

romange Mar 24, 2023

dranikpg Mar 25, 2023 •

edited

Loading

romange left a comment

romange Mar 25, 2023

romange Mar 25, 2023

romange Mar 25, 2023

dranikpg Mar 25, 2023

dranikpg commented Mar 25, 2023

feat(server): Pubsub updates with RCU #980

feat(server): Pubsub updates with RCU #980

Conversation

dranikpg commented Mar 22, 2023 • edited Loading

romange commented Mar 22, 2023

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

dranikpg Mar 25, 2023 • edited Loading

Choose a reason for hiding this comment

romange left a 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

dranikpg commented Mar 25, 2023

dranikpg commented Mar 22, 2023 •

edited

Loading

dranikpg Mar 25, 2023 •

edited

Loading