Implement concurrent partitions - #246
Conversation
|
@JasoonS this is working well on the hosted service and I've added tests. Could you review? |
| fn is_valid_release_version_number(version: &str) -> bool { | ||
| let re_version_pattern = Regex::new(r"^\d+\.\d+\.\d+(-rc\.\d+)?$") | ||
| .expect("version regex pattern should be valid regex"); | ||
| re_version_pattern.is_match(version) || version.starts_with("0.0.0-main-") |
There was a problem hiding this comment.
Now starts with 2.0.0 so just checking that it contains "-main"
There was a problem hiding this comment.
Interesting. So this code actually won't, for example, match on release. It will only match on releases or those with RC on them?
Eg this won't match: 2.0.0-release-concurrent-partitions-20241004091855-c02f47f9
There was a problem hiding this comment.
This is just for contract import, I think we can improve it, it was a bit of a hack #244
| let currentWrite = ref(Promise.resolve()) | ||
| let schedule = ref(() => Promise.resolve()) | ||
|
|
||
| let saveToCacheFile = data => { | ||
| let json = data->S.serializeToJsonStringOrRaiseWith(Data.schema) | ||
| NodeJsLocal.Fs.Promises.writeFile(~filepath=cacheFilePath, ~content=json)->ignore | ||
| let write = () => { | ||
| let json = data->S.serializeToJsonStringOrRaiseWith(Data.schema) | ||
| NodeJsLocal.Fs.Promises.writeFile(~filepath=cacheFilePath, ~content=json) | ||
| } | ||
| schedule := write | ||
| let _ = currentWrite.contents->Promise.thenResolve(_ => { | ||
| currentWrite := schedule.contents() | ||
| }) |
There was a problem hiding this comment.
Stops concurrent writes/corrupted files.
| } | ||
|
|
||
| let addBlockRangeFetched = ( | ||
| ~stats: ChainWorker.blockRangeFetchStats, |
There was a problem hiding this comment.
Caused dependency cycle
| let maxPartitionConcurrency = | ||
| envSafe->EnvSafe.get("ENVIO_MAX_PARTITION_CONCURRENCY", S.int, ~fallback=10) |
There was a problem hiding this comment.
Allows multiple partitions to fire off hs queries simultaneously. We can experiment with increasing/decreasing this.
| /** | ||
| Retrieves an array of partitions that are most behind with a max number based on | ||
| the max number of queries with the context of the partitions currently fetching. | ||
|
|
||
| The array could be shorter than the max number of queries if the partitions are | ||
| at the max queue size. | ||
| */ | ||
| let getMostBehindPartitions = ( |
There was a problem hiding this comment.
Read this description to help give context to the function
There was a problem hiding this comment.
Hmm, this function is definitely a little bit challenging to follow, but starting to get it. At least there are tests 😬
There was a problem hiding this comment.
Sorry it's probably because of the loop fn. The main reason to do that over just iterating is that you can model an early exit for control flow. Let me add some more comments to it.
| let getNextQueriesOrThrow = ( | ||
| self: t, | ||
| ~eventFilters=?, | ||
| ~currentBlockHeight, | ||
| ~maxPerChainQueueSize, | ||
| ~partitionsCurrentlyFetching, | ||
| ) => { |
There was a problem hiding this comment.
This now returns an array of queries to make based on max concurrency, partitions currently fetching and which partition is most behind
d649fd1 to
c02f47f
Compare
c02f47f to
f39bd58
Compare
JasoonS
left a comment
JasoonS
left a comment
There was a problem hiding this comment.
Looks good to me - I want to fiddle a bit with the getNextQueriesOrThrow function a bit more, but shouldn't block this
* Add benchmark for partition count * Schedule async fs writes with benchmarks * Implement concurrency between partitions * Fix tests * include aarch64-unknown-linux-gnu in non release builds for hosted service * Upgrade version of dev releases and main releases * Fix over-filled partitions * Update benchmarks to separate fetchtime by partition id * Add test for partitioned fetch state getMostBehindPartitions * Add description to getMostBehindPartitions function
2 participants
This to allow indexers with lots of partitions to do concurrent requests to hypersync.
Currently it's experimental and seems like it could contain a memory leak so the idea is to have a side release that I can test a deployment.