peb iterator #34
peb iterator #34
Conversation
frairon
commented
May 31, 2024
frairon
commented
- delete telly
- implement disktail
| return nil, fmt.Errorf("error opening storage %s: %w", path, err) | ||
| } | ||
|
|
||
| ring := &DiskTail[T]{ |
There was a problem hiding this comment.
wouldnt call the tailer ring :)
| ).ErrorOrNil() | ||
| } | ||
|
|
||
| // Run starts the ring. It's only allowed to be called once per instance. |
disktail/disktail.go
Outdated
| itemHandler := handler | ||
|
|
||
| if skipDuplicates { | ||
| keys := make(map[string]struct{}) |
There was a problem hiding this comment.
This map could grow really big, but our main use case is iterating over a couple thousand users with a key size of 24, so this should not be a problem.
| time.Sleep(100 * time.Millisecond) | ||
| } | ||
| t.Fatalf("operation timed out") | ||
| } |
There was a problem hiding this comment.
would also add a test for the cleaner
| storedOffsets, err := r.getOffsets(parts) | ||
| if err != nil { | ||
| return fmt.Errorf("error reading local offsets: %w", err) | ||
| } |
There was a problem hiding this comment.
Can be moved above errgroup.WithContext(ctx)
disktail/disktail.go
Outdated
| } | ||
|
|
||
| type IterStats struct { | ||
| sync.Mutex |
There was a problem hiding this comment.
This Mutex is public, but it shouldn't.
A caller of (r *DiskTail[T]) Iterate could pass already locked IterStats, deadlocking the whole thing.
| itemsIterated int64 | ||
| // number of unique keys. Note that this value will be 0 if duplicates are included when | ||
| // iterating. | ||
| uniqueKeys int64 |
There was a problem hiding this comment.
Use atomic.Int64 ? It's there since 1.19.
| // track oldest timestamp for stats | ||
| timestamp, _, _ := decodeKey(key) | ||
| if stats.oldestItem.IsZero() || timestamp.Before(stats.oldestItem) { | ||
| stats.oldestItem = timestamp |
There was a problem hiding this comment.
The read (func (is *IterStats) OldestItem() time.Time) is protected by the lock.
But the write isn't?
|
|
||
| func (r *DiskTail[T]) Iterate(ctx context.Context, skipDuplicates bool, stats *IterStats, handler func(item *Item[T]) HandleResult) error { | ||
| if stats == nil { | ||
| stats = &IterStats{} |
There was a problem hiding this comment.
Locking on these stats doesn't make much sense? They're not visible to the caller.
disktail/disktail.go
Outdated
| // skip it | ||
| return Continue | ||
| } | ||
| keys[item.Key] = struct{}{} |
There was a problem hiding this comment.
The map is not protected by the lock?
Is there a difference between uniqueKeys and map regarding access?
| } | ||
| } | ||
|
|
||
| for { |
There was a problem hiding this comment.
| for { | |
| for ctx.Err() == nil { |
and remove the condition below
disktail/disktail.go
Outdated
| if !it.Prev() { | ||
| return nil | ||
| } else { | ||
| continue | ||
| } |
There was a problem hiding this comment.
I would remove the negation and swap then & else.
Fits the skip it comment better IMO.
