Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
df7fff0ce1 |
@@ -86,8 +86,8 @@ func TestService_WaitForCountWrites(t *testing.T) {
|
||||
ended, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
svc.writeCount(ended, func(ctx context.Context) {
|
||||
svc.cache.IncrementStats(ctx, false, 0)
|
||||
svc.writeCount(ended, func(writeCtx context.Context) {
|
||||
svc.cache.IncrementStats(writeCtx, false, 0)
|
||||
})
|
||||
|
||||
shortCtx, cancelShort := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
|
||||
@@ -165,8 +165,8 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
|
||||
s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err)
|
||||
// Fall through to re-process
|
||||
} else {
|
||||
s.writeCount(ctx, func(ctx context.Context) {
|
||||
s.cache.IncrementStats(ctx, true, 0)
|
||||
s.writeCount(ctx, func(writeCtx context.Context) {
|
||||
s.cache.IncrementStats(writeCtx, true, 0)
|
||||
})
|
||||
|
||||
return &ImageResponse{
|
||||
@@ -184,8 +184,8 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
|
||||
// failed or the request context has ended meanwhile
|
||||
response, err := s.processOrWait(ctx, req)
|
||||
|
||||
s.writeCount(ctx, func(ctx context.Context) {
|
||||
s.cache.IncrementStats(ctx, false, 0)
|
||||
s.writeCount(ctx, func(writeCtx context.Context) {
|
||||
s.cache.IncrementStats(writeCtx, false, 0)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
@@ -234,6 +234,34 @@ func (s *Service) WaitForCountWrites(ctx context.Context) bool {
|
||||
}
|
||||
}
|
||||
|
||||
// writeCount runs write, which adds to a count in the database, in a
|
||||
// goroutine of its own, and waits for it, though not past ctx's deadline.
|
||||
// The write gets ctx without its cancellation or deadline, so an ended
|
||||
// request is still counted. A request past its deadline thus does not wait
|
||||
// for the database connection, which every request shares; its count is
|
||||
// written after it returns.
|
||||
func (s *Service) writeCount(ctx context.Context, write func(context.Context)) {
|
||||
written := make(chan struct{})
|
||||
|
||||
s.countWrites.Go(func() {
|
||||
defer close(written)
|
||||
|
||||
write(context.WithoutCancel(ctx))
|
||||
})
|
||||
|
||||
deadline, hasDeadline := ctx.Deadline()
|
||||
if !hasDeadline {
|
||||
<-written
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case <-written:
|
||||
case <-time.After(time.Until(deadline)):
|
||||
}
|
||||
}
|
||||
|
||||
// ValidateRequest validates the request signature if required.
|
||||
func (s *Service) ValidateRequest(req *ImageRequest) error {
|
||||
// Check if host is allowed (no signature required)
|
||||
@@ -280,38 +308,6 @@ func (s *Service) GenerateSignedURL(
|
||||
baseURL, path, sig, exp, req.Quality, req.FitMode), nil
|
||||
}
|
||||
|
||||
// writeCount runs write, which adds to a count in the database, in a
|
||||
// goroutine of its own. The write gets ctx without its cancellation or
|
||||
// deadline, so an ended request is still counted. writeCount waits for it
|
||||
// until ctx's deadline, even if ctx is cancelled first, so a request
|
||||
// returns with its count written unless its deadline has passed; past
|
||||
// that, it does not wait for the database connection, which every request
|
||||
// shares.
|
||||
func (s *Service) writeCount(ctx context.Context, write func(context.Context)) {
|
||||
written := make(chan struct{})
|
||||
|
||||
s.countWrites.Go(func() {
|
||||
defer close(written)
|
||||
|
||||
write(context.WithoutCancel(ctx))
|
||||
})
|
||||
|
||||
deadline, hasDeadline := ctx.Deadline()
|
||||
if !hasDeadline {
|
||||
<-written
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
deadlineTimer := time.NewTimer(time.Until(deadline))
|
||||
defer deadlineTimer.Stop()
|
||||
|
||||
select {
|
||||
case <-written:
|
||||
case <-deadlineTimer.C:
|
||||
}
|
||||
}
|
||||
|
||||
// errPanicked is returned when processing a variant panicked.
|
||||
var errPanicked = errors.New("panic while processing image")
|
||||
|
||||
@@ -510,8 +506,8 @@ func (s *Service) fetchAndProcess(
|
||||
sourceData, err := io.ReadAll(fetchResult.Content)
|
||||
fetchBytes := int64(len(sourceData))
|
||||
|
||||
s.writeCount(ctx, func(ctx context.Context) {
|
||||
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
|
||||
s.writeCount(ctx, func(writeCtx context.Context) {
|
||||
s.cache.IncrementUpstreamFetch(writeCtx, fetchBytes)
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user