feat: cleanup worker #126
feat: cleanup worker #126
Conversation
integration/singularity/options.go
Outdated
| @@ -39,6 +40,7 @@ type ( | |||
| totalDealSize string | |||
| maxPendingDealSize string | |||
| maxPendingDealNumber int | |||
| localCleanupInterval time.Duration | |||
There was a problem hiding this comment.
The only cleanup is local right? Remove local prefix in variable names?
integration/singularity/reader.go
Outdated
| } | ||
|
|
||
| func (r *SingularityReader) Read(p []byte) (int, error) { | ||
| logger.Infof("buffer size: %v", len(p)) |
There was a problem hiding this comment.
- Change to debug level as this can be noisy?
- Use
wvariation instead of format for nicer JSON log formatting.
There was a problem hiding this comment.
oops, this was supposed to be removed. i will remove it.
integration/singularity/reader.go
Outdated
| logger.Infof("buffer size: %v", len(p)) | ||
|
|
||
| buf := bytes.NewBuffer(p) | ||
| buf.Reset() |
There was a problem hiding this comment.
I am curious why buffer is being reset? it's a fresh buffer right?
There was a problem hiding this comment.
the buffer is constructed from the byte array passed into the function, but creating a new buffer like that starts the cursor at the end of the byte array, so buf.Reset() puts the cursor back at the start so the array will be overwritten instead of extended from the end
| buf := bytes.NewBuffer(p) | ||
| buf.Reset() | ||
|
|
||
| if r.offset >= r.size { |
There was a problem hiding this comment.
Move checks earlier in the function to fail fast?
integration/singularity/store.go
Outdated
| } | ||
| }() | ||
| case <-s.closing: | ||
| break cleanupLoop |
There was a problem hiding this comment.
| break cleanupLoop | |
| return |
integration/singularity/store.go
Outdated
| }() | ||
|
|
||
| ticker := time.NewTicker(s.localCleanupInterval) | ||
| cleanupLoop: |
integration/singularity/store.go
Outdated
|
|
||
| func (s *SingularityStore) cleanup(ctx context.Context) error { | ||
| if s.cleanupActive.Load() { | ||
| return errCleanupAlreadyRunning |
There was a problem hiding this comment.
Returning error here means if a cycle is running, the next one that may get started will be skipped. In the worst case, it means it can take up to 2X the configured interval for cleanup to run.
We should document this behaviour on the options, or alternatively use a mutex instead that locks and unlocks the cycle, such that the next cycle starts immediately. i.e. average cleanup interval will converge to the configured value.
There was a problem hiding this comment.
actually, this made me realize that a mutex is not even necessary, since the worker is already inside a goroutine
|
|
||
| logger.Infof("Starting local store cleanup...") | ||
|
|
||
| dir, err := os.ReadDir(s.local.Dir()) |
There was a problem hiding this comment.
I am curious if using filepath.Walk would make the search logic simpler?
There was a problem hiding this comment.
it's only scanning one layer of a directory, so i don't think so? unless i misunderstood how the local store is laid out
integration/test/integration_test.go
Outdated
| wantBlob := []byte("fish") | ||
| buf := bytes.NewBuffer(wantBlob) | ||
| buf := new(bytes.Buffer) | ||
| for i := 0; i < 10000000; i++ { |
There was a problem hiding this comment.
Use rand.Read to generate random bytes of desired length?
elijaharita
force-pushed
the
feat/cleanup-process
branch
from
b547e01
to
5deacc1
Compare
elijaharita
force-pushed
the
feat/cleanup-process
branch
from
df06a70
to
6621ba9
Compare
|
merging |
branched off feat/singularity-retrieval, will rebase off main when it merges
i'm not really sure how to test this. any recommendations @hannahhoward?
closes #112