Streaming gateway WIP #4

kalabukdima · 2024-08-16T09:18:52Z

No description provided.

Send additional request when all ongoing ones wait for too long

- Update to the latest transport crate - Downgrade rustls to 0.23.10 to avoid cert error #2 - Use subsquid-datasets crate from archive-router repo

If a worker has responded with only a subset of blocks in a chunk, send continuation request for the rest of the chunk repeatedly until the full chunk is fetched

src/controller/stream.rs

Cargo.toml

src/controller/stream.rs

eldargab · 2024-08-20T12:54:59Z

src/controller/task_manager.rs

+
+use super::stream::StreamController;
+
+const MAX_PARALLEL_STREAMS: u8 = 5;


Definitely should be much larger than that and I suggest to remove it for now.

It is not clear yet on what principles this limit should exist. Most likely it is simply total_memory / memory_usage_per_stream.

Yes, the main reason for this is to limit memory usage. If we want to host a public gateway, we surely don't want to make it unlimited

eldargab · 2024-08-20T13:14:29Z

src/types/request.rs

+pub struct ClientRequest {
+    pub dataset_id: DatasetId,
+    pub query: ParsedQuery,
+    pub buffer_size: usize,


Everything below is definitely something not to be exposed even in the "secret mode".

While it's primarily designed for debugging, I don't see a reason to not allow users of their own gateway to set those parameters per request. They can set them in the config anyway.
But if we will host a public gateway, those should definitely be limited or removed at all.

I don't see a reason to not allow users of their own gateway to set those parameters per request

It is not polite to expose to users low level details, that could change anytime. Also, I believe, those kinds of settings could be set once and for all.

Okay, I will hide it from public API, but I don't want to hardcode it because it slows down the development process if you have to recompile and restart the gateway (it's more than a minute) to test a new value

src/controller/stream.rs

eldargab · 2024-08-29T17:18:53Z

src/controller/stream.rs

+            }
+        }
+
+        let index = self.buffer.first_index();


This value could be returned from .push_front() and .first_index() could be removed all together, to make idea behind SlidingArray more clear.

eldargab · 2024-08-29T17:27:24Z

src/types/request.rs

+pub struct ClientRequest {
+    pub dataset_id: DatasetId,
+    pub query: ParsedQuery,
+    pub buffer_size: usize,


I don't see a reason to not allow users of their own gateway to set those parameters per request

It is not polite to expose to users low level details, that could change anytime. Also, I believe, those kinds of settings could be set once and for all.

src/controller/timeouts.rs

eldargab · 2024-08-30T11:20:53Z

src/controller/timeouts.rs

+            if let Some(timeout) = timeout {
+                tokio::select! {
+                    _ = tokio::time::sleep_until(start + timeout) => break,
+                    recv = current_timeout.changed() => {


Hmm, such a pedantic tracking of a desired timeout only useful at the start of streaming, when no times are available.

Another way to bootstrap things:

For the first batch of requests issue each one with a 100 ms timeout.

Once 100 ms timeout triggers

2.1. If there are successful responses or it is a very first chunk in the stream, then try again.

2.2. Otherwise wait another 100 ms and go to 2.

Fixed. I've made it only get the timeout at the moment of sending a request, without adjusting it if it changes

eldargab · 2024-08-30T12:21:49Z

src/controller/timeouts.rs

+        let num_infs = self.num_infs.load(Ordering::Relaxed);
+
+        // TODO: optimize time complexity
+        let kth = ((durations.len() + num_infs) as f32 * self.quantile).floor() as usize;


What is the role of num_infs here? Why "dead" workers should affect the expected response times of healthy ones?

My suggestion is:

Use VecDeque to hold up to N (~50) last successful response times

Make sure, that response times for chunks that are far behind the current buffer never sneak in

When there is no enough entries to trim at the desired percentile, use the longest response time for timeout (possibly some reserve).

When no time info is available use something like 200ms

This algorithm ensures that the timeout_percentile of the slowest requests are retried on average. Its a generalization of what you proposed (wait until 90% of the requests complete, then retry the rest) but in the streaming manner - when the requests could start at different times. In this case I'm tracking the durations of completed requests and the current number of "ongoing" requests. The goal is to find a timeout which would "cut" the 1-timeout_quantile fraction of the remaining requests.

I totally agree that this algorithm is hard to comprehend. But I couldn't fast a better one with the same performance and the same good qualities, namely:

the requests are retried as soon as we know they fall into the given fraction of the slowest ones,

if less than the timeout_percentile requests have finished so far, we will continue waiting for any of them to complete,

a request which has been just retried should be also timed out with the current duration

Fixed, almost as you suggested, but instead of using the longest existing time I always use the hard-coded timeout until there are 50 observations. Otherwise, some unexpected durations (either too fast or too long) could lead to unpredictable consequences.

eldargab · 2024-08-30T13:22:34Z

src/network/priorities.rs

+    priorities: HashMap<PeerId, Priority>,
+}
+
+impl WorkersPool {


My concern regarding the chosen method of evaluation of worker performance is that it is hard to reason about impact of the distant past behaviour.

For example, a worker could have a hiccup, but yet to receive quite a bit of queries at that moment (e.g. because it hosts some low replicated data chunks, that got a sudden demand). In such case, the worker will be significantly de-prioritised, but for how long? Might be for a very long time, depending on a usage pattern and overall network performance...

I suggest to track workers with a set of windowed (by time) counters and to compute resulting priorities at query time with some formula.

A set of counters could be:

Number of "big" timeouts

Number of times the worker failed to respond within a retry threshold

Number of times worker responded within a 50% percentile

Number of Busy errors

Number of currently pending requests made to to this worker

These priorities are reset every epoch, which is 30mins currently, but I suspect it may get shorter later. This simple algo has already sped up the streaming about 2-4 times.

I like what you're suggesting, but how about getting back to it after the initial release? It looks like improving performance without ability to run real world benchmarks

- Report stream summary - Add canonical log lines for HTTP requests - Reorganize span fields - Support JSON log format

Determine the request timeout at the moment of sending it. Just return the default timeout if not enough data is available

- Remove backoff time — retry errors immediately or fail - Don't extend the buffer if no more queries could be sent

Get rid of S3 API dependency

Convert JSON lines to a plain JSON array

- Configure max number of parallel streams - Configure max number of chunks for a single request

kalabukdima added 14 commits July 30, 2024 13:58

Implement streaming gateway MVP

755741b

Add prometheus metrics

662aa32

Set app level timeouts

3036692

Send additional request when all ongoing ones wait for too long

Add transport metrics

3319411

Add dynamic worker priorities

d1ee272

Add sliding percentile timeouts

f14c87e

Update dependencies

535341e

- Update to the latest transport crate - Downgrade rustls to 0.23.10 to avoid cert error #2 - Use subsquid-datasets crate from archive-router repo

Filter workers by version

0212894

Fetch chunk with multiple requests if it's too big

6843bb3

If a worker has responded with only a subset of blocks in a chunk, send continuation request for the rest of the chunk repeatedly until the full chunk is fetched

Use block range from query

84fe82d

Update known chunks in background

5acc286

Add dockerfile

c763377

Structure source files

6fc2e0e

Merge branch 'streaming'

665646d

kalabukdima self-assigned this Aug 16, 2024

kalabukdima added 2 commits August 16, 2024 14:42

Track on-chain epochs

31c361c

Don't hold mutex across await points

2ac3091

kalabukdima force-pushed the streaming branch from 3e30849 to 2ac3091 Compare August 16, 2024 12:01

kalabukdima added 2 commits August 19, 2024 18:14

Change HTTP endpoint to adjust for the current SDK

8027a60

Set RUST_LOG in the network configs

572e510

kalabukdima assigned eldargab Aug 19, 2024

kalabukdima requested a review from Wiezzel August 19, 2024 15:36

kalabukdima unassigned eldargab Aug 19, 2024

kalabukdima requested a review from eldargab August 19, 2024 15:36

Wiezzel force-pushed the master branch from bf47441 to 03489bc Compare August 20, 2024 11:32

eldargab reviewed Aug 20, 2024

View reviewed changes

src/controller/stream.rs Outdated Show resolved Hide resolved

Wiezzel reviewed Aug 20, 2024

View reviewed changes

Cargo.toml Outdated Show resolved Hide resolved

eldargab reviewed Aug 20, 2024

View reviewed changes

kalabukdima added 2 commits August 26, 2024 14:36

Remove trial requests

f13758e

Handle partial responses with bounded memory usage

b56fed6

kalabukdima added 2 commits August 28, 2024 18:02

Make HTTP interface backward compatible

fd0b442

Rename crate to sqd-portal

903ae07

eldargab reviewed Aug 29, 2024

View reviewed changes

eldargab reviewed Aug 30, 2024

View reviewed changes

kalabukdima added 3 commits September 2, 2024 08:07

Improve logs

910e066

- Report stream summary - Add canonical log lines for HTTP requests - Reorganize span fields - Support JSON log format

Simplify timeout handling

808da8d

Determine the request timeout at the moment of sending it. Just return the default timeout if not enough data is available

Handle errors if the query couldn't be sent

803212a

- Remove backoff time — retry errors immediately or fail - Don't extend the buffer if no more queries could be sent

kalabukdima force-pushed the streaming branch 2 times, most recently from a3113f8 to f3f1fe2 Compare September 9, 2024 18:09

Read chunks list from a shared file

fa849f9

Get rid of S3 API dependency

kalabukdima force-pushed the streaming branch from f3f1fe2 to fa849f9 Compare September 17, 2024 13:24

kalabukdima added 10 commits September 19, 2024 14:16

Make /query responses compatible with the archives

8532dd1

Convert JSON lines to a plain JSON array

Add more config options for the public portal

42e799e

- Configure max number of parallel streams - Configure max number of chunks for a single request

Add EVM and Substrate datasets

6a9192d

Update to transport 1.0.25

d91ff27

Rewrite StreamController in a polling fashion

eb5597e

Implement better priorities

6b6b1d4

Minor fixes

e480616

Improve logging

6e03cea

Add more prometheus metrics

323171a

Remove unnecessary Arcs

4f94e04

kalabukdima force-pushed the streaming branch from e4c3869 to 4f94e04 Compare October 17, 2024 15:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Streaming gateway WIP #4

Streaming gateway WIP #4

kalabukdima commented Aug 16, 2024

eldargab Aug 20, 2024

kalabukdima Aug 21, 2024

eldargab Aug 20, 2024

kalabukdima Aug 21, 2024

eldargab Aug 29, 2024

kalabukdima Aug 30, 2024

eldargab Aug 29, 2024

eldargab Aug 29, 2024

eldargab Aug 30, 2024

kalabukdima Sep 6, 2024

eldargab Aug 30, 2024

kalabukdima Sep 2, 2024

kalabukdima Sep 6, 2024

eldargab Aug 30, 2024

kalabukdima Sep 2, 2024

kalabukdima Oct 21, 2024


		use super::stream::StreamController;

		const MAX_PARALLEL_STREAMS: u8 = 5;

Streaming gateway WIP #4

Are you sure you want to change the base?

Streaming gateway WIP #4

Conversation

kalabukdima commented Aug 16, 2024

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