Skip to content
This repository has been archived by the owner on Oct 9, 2023. It is now read-only.

Add item ID to the workqueue instead #165

Closed
wants to merge 18 commits into from
Closed

Add item ID to the workqueue instead #165

Changes from 6 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
bc466a1
test
pingsutw Sep 20, 2023
ef6b4ff
test
pingsutw Sep 20, 2023
c396231
test
pingsutw Sep 20, 2023
5a2399f
test
pingsutw Sep 20, 2023
bbce912
test
pingsutw Sep 20, 2023
b09b997
nit
pingsutw Sep 21, 2023
d3e61ed
nit
pingsutw Sep 25, 2023
9910612
test
pingsutw Sep 25, 2023
7b775a3
nit
pingsutw Sep 25, 2023
a923364
test
pingsutw Sep 26, 2023
23967d5
test
pingsutw Sep 26, 2023
bfcb747
test
pingsutw Sep 26, 2023
80d4cd8
test
pingsutw Sep 26, 2023
a3a8536
test
pingsutw Sep 26, 2023
5ea45af
nit
pingsutw Sep 26, 2023
d4ae3ad
nit
pingsutw Sep 26, 2023
d612b68
nit
pingsutw Sep 26, 2023
07e6f22
nit
pingsutw Sep 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions cache/auto_refresh.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,8 @@

for _, batch := range batches {
b := batch
w.workqueue.Add(&b)
logger.Debugf(ctx, "Enqueuing batch with id: %v", b[0].GetID())
w.workqueue.Add(b[0].GetID())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This assumes that batches only have one item... I don't think this is true in all usage of AutoRefresh cache... if it's, I would rather deprecate the whole batching mechanism...
This implementation won't work for any batches > 1 item.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In flytepropeller, batches always only have one item.

flytestdlib/cache/auto_refresh.go

Lines 133 to 140 in 1285ff3

func SingleItemBatches(_ context.Context, snapshot []ItemWrapper) (batches []Batch, err error) {
res := make([]Batch, 0, len(snapshot))
for _, item := range snapshot {
res = append(res, Batch{item})
}
return res, nil
}

Will clean up the code.

}

return nil
Expand Down Expand Up @@ -273,13 +274,21 @@
case <-ctx.Done():
return nil
default:
item, shutdown := w.workqueue.Get()
itemId, shutdown := w.workqueue.Get()

Check failure on line 277 in cache/auto_refresh.go

View workflow job for this annotation

GitHub Actions / Lint / Run Lint 

var `itemId` should be `itemID` (golint)
if shutdown {
return nil
}

t := w.metrics.SyncLatency.Start()
updatedBatch, err := w.syncCb(ctx, *item.(*Batch))
item, ok := w.lruMap.Get(itemId)
if !ok {
logger.Debugf(ctx, "item with id [%v] not found in cache", itemId)
return nil
}
updatedBatch, err := w.syncCb(ctx, Batch{itemWrapper{
id: itemId.(ItemID),
item: item.(Item),
}})

// Since we create batches every time we sync, we will just remove the item from the queue here
// regardless of whether it succeeded the sync or not.
Expand Down
Loading