// Copyright (c) ppy Pty Ltd . Licensed under the MIT Licence. // See the LICENCE file in the repository root for full licence text. using System; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using JetBrains.Annotations; using osu.Framework.Allocation; using osu.Game.Online.API; namespace osu.Game.Database { public abstract class OnlineLookupCache : MemoryCachingComponent where TLookup : IEquatable where TValue : class, IHasOnlineID where TRequest : APIRequest { [Resolved] private IAPIProvider api { get; set; } /// /// Creates an to retrieve the values for a given collection of s. /// /// The IDs to perform the lookup with. protected abstract TRequest CreateRequest(IEnumerable ids); /// /// Retrieves a list of s from a successful created by . /// [CanBeNull] protected abstract IEnumerable RetrieveResults(TRequest request); /// /// Perform a lookup using the specified , populating a . /// /// The ID to lookup. /// An optional cancellation token. /// The populated , or null if the value does not exist or the request could not be satisfied. [ItemCanBeNull] protected Task LookupAsync(TLookup id, CancellationToken token = default) => GetAsync(id, token); /// /// Perform an API lookup on the specified , populating a . /// /// The IDs to lookup. /// An optional cancellation token. /// The populated values. May include null results for failed retrievals. protected Task LookupAsync(TLookup[] ids, CancellationToken token = default) { var lookupTasks = new List>(); foreach (var id in ids) { lookupTasks.Add(LookupAsync(id, token).ContinueWith(task => { if (!task.IsCompletedSuccessfully) return null; return task.Result; }, token)); } return Task.WhenAll(lookupTasks); } // cannot be sealed due to test usages (see TestUserLookupCache). protected override async Task ComputeValueAsync(TLookup lookup, CancellationToken token = default) => await queryValue(lookup).ConfigureAwait(false); private readonly Queue<(TLookup id, TaskCompletionSource)> pendingTasks = new Queue<(TLookup, TaskCompletionSource)>(); private Task pendingRequestTask; private readonly object taskAssignmentLock = new object(); private Task queryValue(TLookup id) { lock (taskAssignmentLock) { var tcs = new TaskCompletionSource(); // Add to the queue. pendingTasks.Enqueue((id, tcs)); // Create a request task if there's not already one. if (pendingRequestTask == null) createNewTask(); return tcs.Task; } } private void performLookup() { // contains at most 50 unique IDs from tasks, which is used to perform the lookup. var nextTaskBatch = new Dictionary>>(); // Grab at most 50 unique IDs from the queue. lock (taskAssignmentLock) { while (pendingTasks.Count > 0 && nextTaskBatch.Count < 50) { (TLookup id, TaskCompletionSource task) next = pendingTasks.Dequeue(); // Perform a secondary check for existence, in case the value was queried in a previous batch. if (CheckExists(next.id, out var existing)) next.task.SetResult(existing); else { if (nextTaskBatch.TryGetValue(next.id, out var tasks)) tasks.Add(next.task); else nextTaskBatch[next.id] = new List> { next.task }; } } } if (nextTaskBatch.Count == 0) return; // Query the values. var request = CreateRequest(nextTaskBatch.Keys.ToArray()); // rather than queueing, we maintain our own single-threaded request stream. // todo: we probably want retry logic here. api.Perform(request); // Create a new request task if there's still more values to query. lock (taskAssignmentLock) { pendingRequestTask = null; if (pendingTasks.Count > 0) createNewTask(); } var foundValues = RetrieveResults(request); if (foundValues != null) { foreach (var value in foundValues) { if (nextTaskBatch.TryGetValue(value.OnlineID, out var tasks)) { foreach (var task in tasks) task.SetResult(value); nextTaskBatch.Remove(value.OnlineID); } } } // if any tasks remain which were not satisfied, return null. foreach (var tasks in nextTaskBatch.Values) { foreach (var task in tasks) task.SetResult(null); } } private void createNewTask() => pendingRequestTask = Task.Run(performLookup); } }