Add basic retry functionality #1119

dryajov · 2025-02-19T15:37:18Z

No description provided.

marcinczenko

I have started the review, and I will be working on by adding more comments today and over next days if possible (will try to look up a couple of things over the weekend as well). No need to wait for me, if someone is faster. My objective is to have some observations confirmed and some questions answered. Because I have documented the whole flow recently, I just want to take advantage of this more substantial change to improve the docs and perhaps acting as sounding board...

codex/blockexchange/engine/engine.nim

marcinczenko · 2025-02-21T14:00:14Z

codex/blockexchange/engine/engine.nim

+  let peerCtx = self.peers.get(peer)

  if peerCtx != nil:
-    await b.payForBlocks(peerCtx, blocksDelivery)
+    await self.payForBlocks(peerCtx, blocksDelivery)
    ## shouldn't we remove them from the want-list instead of this:
    peerCtx.cleanPresence(blocksDelivery.mapIt(it.address))


Just above the highlighted lines we are calling resolveBlocks. resolveBlocks will call cancelBlocks which will take care for removing the relevant presence entries, right?

If this is the case, perhaps, we could move the payment there as well?

gmega

Great first pass. I can see the improvements in the benchmarks.

gmega · 2025-02-21T12:02:39Z

Makefile

+	$(NPH) *.nim
+	$(NPH) codex/
+	$(NPH) tests/
+


ehehe good easter egg... 😃

gmega · 2025-02-21T12:16:02Z

codex/blockexchange/engine/discovery.nim

@@ -143,8 +143,6 @@ proc start*(b: DiscoveryEngine) {.async.} =
    asyncSpawn fut

  b.discoveryLoop = b.discoveryQueueLoop()
-  b.trackedFutures.track(b.discoveryLoop)
-  asyncSpawn b.discoveryLoop


Hm... why are you excluding this? This will cause the discovery loop not to stop when someone calls .stop() on the discovery service. Or rather, it will stop cause Codex will exit anyway, but won't receive a cancel signal.

No, not really - this doesn't affect stopping since we call cancel the tracked futures on stop. The asyncSpawn call tries to do some logging and re-throw any future error as a Defect, but that was really a workaround because we lacked checked exceptions on async code. I do intend to make tracked futures only accept Future.Raises([]) for that matter.

Wait but you're not adding the discoveryLoop future to the trackedFutures, so that one won't stop.

gmega · 2025-02-21T12:18:59Z

codex/blockexchange/engine/engine.nim

+  self.blockexcRunning = true
+  for i in 0 ..< self.concurrentTasks:
+    let fut = self.blockexcTaskRunner()
+    self.trackedFutures.track(fut)


OK I guess you don't need asyncspawn here cause we're already keeping a ref to the future in trackedFutures?

gmega · 2025-02-21T12:19:46Z

codex/blockexchange/engine/engine.nim

    warn "Stopping blockexc without starting it"
    return

-  b.blockexcRunning = false
-  await b.trackedFutures.cancelTracked()
+  self.blockexcRunning = false


eheh good catch...

gmega · 2025-02-21T12:21:31Z

codex/blockexchange/engine/engine.nim

-  except CatchableError as exc:
-    warn "Error block handle, disconnecting peer", address, exc = exc.msg, peerId
+proc randomPeer(peers: seq[BlockExcPeerCtx]): BlockExcPeerCtx =
+  Rng.instance.sample(peers)


Yep that's much better. The previous approach was not random at all, you'd keep pounding the same peer if you're trying to fetch the same block.

yeah, that was quite annoying when I realized what was going on.

gmega · 2025-02-21T19:03:38Z

codex/blockexchange/network/network.nim


 logScope:
  topics = "codex blockexcnetwork"

 const
  Codec* = "/codex/blockexc/1.0.0"
-  MaxInflight* = 100
+  DefaultMaxInflight* = 100


For reference, Deluge allows 1500 pending block requests per peer in its most aggressive profiles.

Good to know, we can definitely adjust.

gmega · 2025-02-21T19:05:05Z

codex/blockexchange/network/network.nim

  ## Perform protocol initialization
  ##

  proc peerEventHandler(peerId: PeerId, event: PeerEvent) {.async.} =
    if event.kind == PeerEventKind.Joined:
-      b.setupPeer(peerId)
+      self.setupPeer(peerId)
    else:


Hm... I see there are three PeerEventKinds:

PeerEventKind* {.pure.} = enum Left Joined Identified

as it is now, if we get Indentified, we'll drop the peer. I'm assuming you're aware of this but pointing it out anyhow.

Oh, not really, thats a new event 🤔. Good catch!

gmega · 2025-02-21T19:10:16Z

codex/conf.nim

@@ -251,15 +251,15 @@ type
      desc:
        "Time interval in seconds - determines frequency of block " &
        "maintenance cycle: how often blocks are checked " & "for expiration and cleanup",
-      defaultValue: DefaultBlockMaintenanceInterval,
-      defaultValueDesc: $DefaultBlockMaintenanceInterval,
+      defaultValue: DefaultBlockInterval,


Not sure this is improving what this constant contains...

The SuperLongSelfDescriptingValueIdentifierSuperConstant is quite annoying tho 😁. Matter of taste, but I can live with it if we find it useful.

gmega · 2025-02-21T19:16:44Z

codex/node.nim


-    if err =? (await erasureJob()).errorOption:
-      return failure(err)
+  self.trackedFutures.track(prefetch())


OK I see why you're doing this, but now streamEntireDataset will effectively continue downloading even if the client stops streaming. Anyhow, I guess we're gonna drop streaming soon enough anyway.

Not sure why you think that streaming is going to be dropped? Otherwise, good observation.

gmega · 2025-02-21T19:40:09Z

codex/rng.nim

+    raise newException(RngSampleError, "Limit cannot be larger than sample!")
+
+  for _ in 0 ..< min(sample.len, limit):
+    result.add(rng.sample(sample, result))


These sample procedures aren't great, I'm going to do a PR on rng.nim.

marcinczenko · 2025-02-22T12:52:06Z

codex/blockexchange/engine/discovery.nim

@@ -143,8 +143,6 @@ proc start*(b: DiscoveryEngine) {.async.} =
    asyncSpawn fut

  b.discoveryLoop = b.discoveryQueueLoop()
-  b.trackedFutures.track(b.discoveryLoop)
-  asyncSpawn b.discoveryLoop

 proc stop*(b: DiscoveryEngine) {.async.} =


Since you are changing b => self all over the places, maybe you want to do it here as well?

marcinczenko · 2025-02-22T13:01:32Z

codex/blockexchange/engine/engine.nim


  trace "NetworkStore stopped"

 proc sendWantHave(
-    b: BlockExcEngine, addresses: seq[BlockAddress], peers: seq[BlockExcPeerCtx]
+    self: BlockExcEngine, addresses: seq[BlockAddress], peers: seq[BlockExcPeerCtx]
 ): Future[void] {.async.} =
  for p in peers:
    let toAsk = addresses.filterIt(it notin p.peerHave)


I know you did not touch that line, but it would be cleaner to have sendWantHave not to do any filtering (I mean, this should happen elsewhere, where it is really relevant). Especially that sendWantBlock is nice and clean without side-effected (the caller expects the request to be just sent)... Maybe a small separate PR...

marcinczenko · 2025-02-22T15:53:25Z

codex/blockexchange/engine/engine.nim

+    self: BlockExcEngine, address: BlockAddress
+): Future[?!Block] {.async: (raises: [CancelledError]).} =
+  if address notin self.pendingBlocks:
+    self.trackedFutures.track(self.downloadInternal(address))


So, what are we really tracking here in:

self.trackedFutures.track(self.downloadInternal(address))

Which future will the downloadInternal return? If I read it well it will be the result of await (handle or sleepAsync(self.pendingBlocks.retryInterval)), which returns a new future which will complete once either handle or sleepAsync complete, which could well be sleepAsync. I think we always want to track handle.

marcinczenko · 2025-02-22T18:23:16Z

codex/blockexchange/engine/engine.nim

+    if address in peerHave and not self.pendingBlocks.retriesExhausted(address) and
+        not self.pendingBlocks.isInFlight(address):
+      self.pendingBlocks.setInFlight(address, true)
+      self.pendingBlocks.decRetries(address)


It is convenient to do it here, but I never like having side effects in filter... it also feels a bit out of context here when you read the code. We could add setInFlight that takes list of addresses. The same could be done for self.pendingBlocks.decRetries. We could also consider if self.pendingBlocks.setInFlight(address, true) should be part of sendWantBlock (and then self.self.pendingBlocks.setInFlight(address, false) to be inside sendWantHave (but yeah, also said in other comment earlier that I like those two to be as focused as possible (still as long as they do not filter our blocks I am kind of fine with it).

Of course it is a bit faster to do it here, but , that could be a bit premature optimization....

marcinczenko · 2025-02-22T18:50:23Z

codex/blockexchange/engine/engine.nim

      )
-    )
-  ).filterIt(it.failed)
+      peerCtx.cleanPresence(addrs)


And not peerCtx.cleanPresence(blocks)?

marcinczenko · 2025-02-22T18:54:59Z

codex/blockexchange/engine/engine.nim

+        price = @(self.pricing.get(Pricing(price: 0.u256)).price.toBytesBE)
+
+      if e.cancel:
+        trace "Received cancelation for untracked block, skipping", address = e.address


Suggested change

trace "Received cancelation for untracked block, skipping", address = e.address

trace "Received cancellation for untracked block, skipping", address = e.address

marcinczenko · 2025-02-22T19:27:40Z

codex/blockexchange/engine/engine.nim

-    # it doesn't mean that we're never talking to this peer again.
-    # TODO: we need a lot more work around peer selection and
-    # prioritization
+proc downloadInternal(


Just a thought. I am wondering if downloadInternal does not better belong to pendingblocks. It would perhaps drag with it other things that should be there. For example RetriesExhaustedError is defined in pendingblocks, yet never used there but here.

dryajov added 27 commits February 19, 2025 09:37

adding basic retry functionality

3f030e3

avoid duplicate requests and batch them

25c0c16

fix cancelling blocks

62dc92c

properly resolve blocks

94926ad

minor cleanup - use self

d399654

avoid useless asyncSpawn

cce0b17

track retries

5507e88

limit max inflight and set libp2p maxIncomingStreams

6b4e587

cleanup

2a44d5d

add basic yield in readLoop

255edf8

use tuple instead of object

4cf466e

cleanup imports and logs

5611cf4

increase defaults

ad8340a

wip

2671587

fix prefetch batching

f455022

cleanup

b090162

decrease timeouts to speedup tests

f7dc789

remove outdated test

a806be0

add retry tests

616672d

should track retries

ac27292

remove useless test

7701e75

use correct block address (index was off by 1)

732de6c

remove duplicate noop proc

45af78a

add BlockHandle type

aeb3ac6

Use BlockHandle type

83bd438

add fetchLocal to control batching from local store

b619482

add format target

1c98b31

dryajov force-pushed the feat/add-retry-functionality branch from 74d689c to 1c98b31 Compare February 19, 2025 17:22

dryajov added 2 commits February 19, 2025 11:54

revert deps

999ed6e

adjust quotaMaxBytes

2e97fe8

dryajov mentioned this pull request Feb 20, 2025

Prevent duplicate block transmission #1093

Closed

dryajov marked this pull request as ready for review February 20, 2025 15:12

Merge branch 'master' into feat/add-retry-functionality

18b5d6c

dryajov requested review from gmega, munna0908 and marcinczenko February 20, 2025 18:31

gmega mentioned this pull request Feb 21, 2025

BlockExchange Performance and reliability #974

Open

marcinczenko reviewed Feb 21, 2025

View reviewed changes

codex/blockexchange/engine/engine.nim Show resolved Hide resolved

marcinczenko reviewed Feb 21, 2025

View reviewed changes

codex/blockexchange/engine/engine.nim Show resolved Hide resolved

marcinczenko reviewed Feb 21, 2025

View reviewed changes

codex/blockexchange/engine/engine.nim Show resolved Hide resolved

marcinczenko reviewed Feb 21, 2025

View reviewed changes

gmega approved these changes Feb 21, 2025

View reviewed changes

marcinczenko reviewed Feb 22, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add basic retry functionality #1119

Add basic retry functionality #1119

dryajov commented Feb 19, 2025

marcinczenko left a comment •

edited

Loading

marcinczenko Feb 21, 2025 •

edited

Loading

gmega left a comment

gmega Feb 21, 2025

gmega Feb 21, 2025

dryajov Feb 21, 2025

gmega Feb 21, 2025 •

edited

Loading

gmega Feb 21, 2025

gmega Feb 21, 2025

gmega Feb 21, 2025

dryajov Feb 21, 2025

gmega Feb 21, 2025

dryajov Feb 21, 2025

gmega Feb 21, 2025

dryajov Feb 21, 2025

gmega Feb 21, 2025

dryajov Feb 21, 2025

gmega Feb 21, 2025

dryajov Feb 21, 2025

gmega Feb 21, 2025

marcinczenko Feb 22, 2025

marcinczenko Feb 22, 2025

marcinczenko Feb 22, 2025

marcinczenko Feb 22, 2025

marcinczenko Feb 22, 2025

marcinczenko Feb 22, 2025

marcinczenko Feb 22, 2025 •

edited

Loading

	trace "Received cancelation for untracked block, skipping", address = e.address
	trace "Received cancellation for untracked block, skipping", address = e.address

Add basic retry functionality #1119

Are you sure you want to change the base?

Add basic retry functionality #1119

Conversation

dryajov commented Feb 19, 2025

marcinczenko left a comment • edited Loading

Choose a reason for hiding this comment

marcinczenko Feb 21, 2025 • edited Loading

Choose a reason for hiding this comment

gmega 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

gmega Feb 21, 2025 • 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

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

marcinczenko Feb 22, 2025 • edited Loading

Choose a reason for hiding this comment

marcinczenko left a comment •

edited

Loading

marcinczenko Feb 21, 2025 •

edited

Loading

gmega Feb 21, 2025 •

edited

Loading

marcinczenko Feb 22, 2025 •

edited

Loading