feat(indexer/server): implement a task buffer to improve efficiency in managing backpressure issues #519
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
FrankLi123
changed the title
1 task
brucexc
requested changes
Sep 3, 2024
| ) | ||
|
|
||
| // TaskBuffer represents a fixed-size buffer, it operates as a FIFO buffer | ||
| type TaskBuffer struct { |
There was a problem hiding this comment.
Suggested change
| type TaskBuffer struct { | |
| This is a new data structure, please add sufficient test cases. |
There was a problem hiding this comment.
Comprehensive test cases have been added in task_buffer_test.go under directory/engine
| } | ||
|
|
||
| task := sw.tasks[0] | ||
| sw.tasks = sw.tasks[1:] |
There was a problem hiding this comment.
Suggested change
| sw.tasks = sw.tasks[1:] | |
| The slice operation sw.tasks = sw.tasks[1:] changes the starting position of the slice, but the underlying array remains unchanged. If this operation is performed frequently (e.g., frequent task retrieval), the underlying array of the slice may grow indefinitely without releasing the used memory, potentially causing a memory leak. | |
| So i think adding sw.tasks[0] = nil would be safer. |
There was a problem hiding this comment.
I agree with your point, and I have made the change to address it.
internal/node/indexer/server.go
Outdated
| return fmt.Errorf("an error occurred in the source: %w", err) | ||
| } | ||
|
|
||
| return nil |
There was a problem hiding this comment.
Suggested change
| return nil | |
| The return nil statement immediately after the if err != nil check is unnecessary because if err != nil is true, the function will return early. |
internal/node/indexer/server.go
Outdated
| if err := s.handleTasks(ctx, tasks); err != nil { | ||
| return fmt.Errorf("handle tasks: %w", err) | ||
| if err := s.handleTasks(ctx, task); err != nil { | ||
| errorChan <- fmt.Errorf("handle tasks error: %w", err) |
There was a problem hiding this comment.
Suggested change
| errorChan <- fmt.Errorf("handle tasks error: %w", err) | |
| I think the function retryableFunc should return the error instead of sending it to the errorChan. This approach can make the retry mechanism clearer and more straightforward |
…nncessary operations on error channel in Run()
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Implement an adaptive "Task Buffer" that efficiently manages task flow between the producer (data-source) and the consumer (worker).
Task Buffer
Performance
Checklist
Does this PR introduce a breaking change?
Other information