osu/osu.Game/Database/UserLookupCache.cs

145 lines
5.3 KiB
C#

// Copyright (c) ppy Pty Ltd <contact@ppy.sh>. Licensed under the MIT Licence.
// See the LICENCE file in the repository root for full licence text.
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;
using osu.Game.Online.API.Requests;
using osu.Game.Users;
namespace osu.Game.Database
{
public class UserLookupCache : MemoryCachingComponent<int, User>
{
[Resolved]
private IAPIProvider api { get; set; }
/// <summary>
/// Perform an API lookup on the specified user, populating a <see cref="User"/> model.
/// </summary>
/// <param name="userId">The user to lookup.</param>
/// <param name="token">An optional cancellation token.</param>
/// <returns>The populated user, or null if the user does not exist or the request could not be satisfied.</returns>
[ItemCanBeNull]
public Task<User> GetUserAsync(int userId, CancellationToken token = default) => GetAsync(userId, token);
/// <summary>
/// Perform an API lookup on the specified users, populating a <see cref="User"/> model.
/// </summary>
/// <param name="userIds">The users to lookup.</param>
/// <param name="token">An optional cancellation token.</param>
/// <returns>The populated users. May include null results for failed retrievals.</returns>
public Task<User[]> GetUsersAsync(int[] userIds, CancellationToken token = default)
{
var userLookupTasks = new List<Task<User>>();
foreach (var u in userIds)
{
userLookupTasks.Add(GetUserAsync(u, token).ContinueWith(task =>
{
if (!task.IsCompletedSuccessfully)
return null;
return task.Result;
}, token));
}
return Task.WhenAll(userLookupTasks);
}
protected override async Task<User> ComputeValueAsync(int lookup, CancellationToken token = default)
=> await queryUser(lookup).ConfigureAwait(false);
private readonly Queue<(int id, TaskCompletionSource<User>)> pendingUserTasks = new Queue<(int, TaskCompletionSource<User>)>();
private Task pendingRequestTask;
private readonly object taskAssignmentLock = new object();
private Task<User> queryUser(int userId)
{
lock (taskAssignmentLock)
{
var tcs = new TaskCompletionSource<User>();
// Add to the queue.
pendingUserTasks.Enqueue((userId, 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 user IDs from userTasks, which is used to perform the lookup.
var userTasks = new Dictionary<int, List<TaskCompletionSource<User>>>();
// Grab at most 50 unique user IDs from the queue.
lock (taskAssignmentLock)
{
while (pendingUserTasks.Count > 0 && userTasks.Count < 50)
{
(int id, TaskCompletionSource<User> task) next = pendingUserTasks.Dequeue();
// Perform a secondary check for existence, in case the user was queried in a previous batch.
if (CheckExists(next.id, out var existing))
next.task.SetResult(existing);
else
{
if (userTasks.TryGetValue(next.id, out var tasks))
tasks.Add(next.task);
else
userTasks[next.id] = new List<TaskCompletionSource<User>> { next.task };
}
}
}
// Query the users.
var request = new GetUsersRequest(userTasks.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 users to query.
lock (taskAssignmentLock)
{
pendingRequestTask = null;
if (pendingUserTasks.Count > 0)
createNewTask();
}
List<User> foundUsers = request.Result?.Users;
if (foundUsers != null)
{
foreach (var user in foundUsers)
{
if (userTasks.TryGetValue(user.Id, out var tasks))
{
foreach (var task in tasks)
task.SetResult(user);
userTasks.Remove(user.Id);
}
}
}
// if any tasks remain which were not satisfied, return null.
foreach (var tasks in userTasks.Values)
{
foreach (var task in tasks)
task.SetResult(null);
}
}
private void createNewTask() => pendingRequestTask = Task.Run(performLookup);
}
}