func()

in internal/gcsx/file_cache_reader.go [88:169]
67 lines of code
14 McCabe index (conditional complexity)

func (fc *FileCacheReader) tryReadingFromFileCache(ctx context.Context, p []byte, offset int64) (int, bool, error) {
	if fc.fileCacheHandler == nil {
		return 0, false, nil
	}

	// By default, consider read type random if the offset is non-zero.
	isSequential := offset == 0

	var handleID uint64
	if readOp, ok := ctx.Value(ReadOp).(*fuseops.ReadFileOp); ok {
		handleID = uint64(readOp.Handle)
	}
	requestID := uuid.New()
	logger.Tracef("%.13v <- FileCache(%s:/%s, offset: %d, size: %d, handle: %d)", requestID, fc.bucket.Name(), fc.object.Name, offset, len(p), handleID)

	startTime := time.Now()
	var bytesRead int
	var cacheHit bool
	var err error

	defer func() {
		executionTime := time.Since(startTime)
		var requestOutput string
		if err != nil {
			requestOutput = fmt.Sprintf("err: %v (%v)", err, executionTime)
		} else {
			if fc.fileCacheHandle != nil {
				isSequential = fc.fileCacheHandle.IsSequential(offset)
			}
			requestOutput = fmt.Sprintf("OK (isSeq: %t, cacheHit: %t) (%v)", isSequential, cacheHit, executionTime)
		}

		logger.Tracef("%.13v -> %s", requestID, requestOutput)

		readType := util.Random
		if isSequential {
			readType = util.Sequential
		}
		captureFileCacheMetrics(ctx, fc.metricHandle, readType, bytesRead, cacheHit, executionTime)
	}()

	// Create fileCacheHandle if not already.
	if fc.fileCacheHandle == nil {
		fc.fileCacheHandle, err = fc.fileCacheHandler.GetCacheHandle(fc.object, fc.bucket, fc.cacheFileForRangeRead, offset)
		if err != nil {
			switch {
			case errors.Is(err, lru.ErrInvalidEntrySize):
				logger.Warnf("tryReadingFromFileCache: while creating CacheHandle: %v", err)
				return 0, false, nil
			case errors.Is(err, cacheUtil.ErrCacheHandleNotRequiredForRandomRead):
				// Fall back to GCS if it is a random read, cacheFileForRangeRead is
				// false and there doesn't already exist file in cache.
				isSequential = false
				return 0, false, nil
			default:
				return 0, false, fmt.Errorf("tryReadingFromFileCache: GetCacheHandle failed: %w", err)
			}
		}
	}

	bytesRead, cacheHit, err = fc.fileCacheHandle.Read(ctx, fc.bucket, fc.object, offset, p)
	if err == nil {
		return bytesRead, cacheHit, nil
	}

	bytesRead = 0
	cacheHit = false

	if cacheUtil.IsCacheHandleInvalid(err) {
		logger.Tracef("Closing cacheHandle:%p for object: %s:/%s", fc.fileCacheHandle, fc.bucket.Name(), fc.object.Name)
		closeErr := fc.fileCacheHandle.Close()
		if closeErr != nil {
			logger.Warnf("tryReadingFromFileCache: close cacheHandle error: %v", closeErr)
		}
		fc.fileCacheHandle = nil
	} else if !errors.Is(err, cacheUtil.ErrFallbackToGCS) {
		return 0, false, fmt.Errorf("tryReadingFromFileCache: while reading via cache: %w", err)
	}
	err = nil

	return 0, false, nil
}