Compartir a través de

Proyecto de ejemplo del coordinador de actividades

  • Artículo

En este ejemplo sencillo para el Coordinador de actividades se muestra cómo se puede aprovechar la API para volver a entrenar un modelo en segundo plano cuando se cumplen las condiciones del sistema.

Introducción al proyecto de ejemplo

Vamos a considerar el caso de una aplicación de edición de música. Esta aplicación tiene tareas en segundo plano de alta prioridad que el usuario de servicio solicita, como publicar contenido en el almacenamiento en la nube. También hay tareas en segundo plano de prioridad baja que admiten la interacción del usuario, como proporcionar recomendaciones automáticas para mejorar una composición mientras se edita. Por último, hay un conjunto de tareas diferidas que no necesitan ocurrir en ningún momento específico sin la solicitud del usuario, que es nuestro enfoque en este ejemplo. En concreto, nos gustaría volver a entrenar periódicamente el modelo de recomendación cuando el impacto del usuario es mínimo. Podemos usar la API de coordinador de actividades para lograrlo.

En este escenario, nos gustaría volver a entrenar el modelo cuando el usuario no está presente. El flujo de trabajo de reentrenamiento en este escenario también consume GPU, por lo que también queremos que se ejecute cuando sea un buen momento para utilizar la GPU. Podemos especificar estos requisitos mediante directivas de coordinador de actividades. La API del coordinador de actividades utilizará nuestra política para determinar cuándo se cumplen los requisitos y enviar notificaciones para saber cuándo iniciar o detener la ejecución de nuestro trabajo.

En este caso, la plantilla de directiva GOOD satisface la mayoría de nuestras necesidades, ya que realiza un seguimiento de la CPU, la memoria, el disco del sistema, la energía y la inactividad del usuario. Simplemente necesitamos establecer explícitamente una condición para GPU. Es importante recordar que, aunque nuestra carga de trabajo usará principalmente la GPU, la ejecución de nuestra actividad consume inherentemente CPU, memoria, disco y energía. Nuestro impacto en estos recursos también puede variar considerablemente entre las configuraciones del sistema. Por ejemplo, una GPU más rápida podría dar lugar a que la CPU pase más tiempo alimentando la GPU con datos, lo que puede dar lugar a que se lean o guarden más datos en el disco. La velocidad de este disco también puede afectar al consumo de CPU de forma similar. Al configurar todos los recursos a los que afectamos, podemos estar seguros de que no interferimos inadvertidamente en la experiencia del usuario ni degradamos el rendimiento del sistema. Además, el propio trabajo se ha desglosado para que se produzca en fragmentos pequeños, de modo que podamos responder adecuadamente a las notificaciones de coordinación para evitar que se ejecuten fuera de las condiciones deseadas.

Para demostrar cómo los desarrolladores pueden cambiar o rebajar las directivas, también añadimos el requisito de que queremos que el reentrenamiento se complete en 48 horas. Las primeras 24 horas, nuestro plazo flexible, intentamos funcionar con nuestra directiva ideal, y las últimas 24 horas bajamos a una directiva inferior.

Código de proyecto de ejemplo

El siguiente código es la aplicación de ejemplo de edición de música. Aprovecha la API del coordinador de actividades para realizar tareas en segundo plano, como se describe en la información general.

#include <chrono>
#include <mutex>
#include <condition_variable>
#include <Windows.h>
#include <ActivityCoordinator.h>
#include <wil/resource.h>

// To use ActivityCoordinator, we must link to the OneCoreUAP library.

#pragma comment(lib, "OneCoreUAP.lib")

using namespace std;
using namespace chrono;
using namespace wil;

// Declare RAII wrappers for the Activity Coordinator policy and subscription.
// These behave like traditional smart pointers and will call their associated
// API cleanup functions when they go out of scope.

typedef wil::unique_any<
        ACTIVITY_COORDINATOR_POLICY,
        decltype(&DestroyActivityCoordinatorPolicy),
        DestroyActivityCoordinatorPolicy>
    unique_policy;

typedef wil::unique_any<
        ACTIVITY_COORDINATOR_SUBSCRIPTION,
        decltype(&UnsubscribeActivityCoordinatorPolicy),
        UnsubscribeActivityCoordinatorPolicy>
    unique_subscription;

struct WORKER_CONTEXT {
    mutex ContextLock;
    unique_threadpool_work Worker;
    bool ShouldRun;
    bool IsRunning;
    bool IsComplete;
    std::condition_variable CompletionSignal;
};

_Requires_lock_held_(workerContext->ContextLock)
void
ResumeWorker(
    _In_ WORKER_CONTEXT* workerContext
    )
{
    workerContext->ShouldRun = true;
    if (!workerContext->IsRunning && !workerContext->IsComplete) {

        // No active workers, so start a new one.

        workerContext->IsRunning = true;
        SubmitThreadpoolWork(workerContext->Worker.get());
    }
}

void
DeferredWorkEventCallback(
    _In_ ACTIVITY_COORDINATOR_NOTIFICATION notificationType,
    _In_ void* callbackContext
    )
{
    WORKER_CONTEXT* workerContext = reinterpret_cast<WORKER_CONTEXT*>(callbackContext);

    // Use this callback thread to dispatch notifications to a worker thread
    // about whether or not it should process the next chunk of deferred work.

    // Note: Do not use this thread to perform your activity's workload.

    lock_guard<mutex> scopedLock(workerContext->ContextLock);
    switch (notificationType) {
    case ACTIVITY_COORDINATOR_NOTIFICATION_RUN:

        // Allow deferred work to be processed.

        ResumeWorker(workerContext);

        break;

    case ACTIVITY_COORDINATOR_NOTIFICATION_STOP:

        // Stop processing deferred work.

        workerContext->ShouldRun = false;

        break;

    default:
        FAIL_FAST();
        break;
    }
}

bool
TrainNextModelChunk(
    )
{
    //
    // Returns true if all work is completed, or false if there is more work.
    //

    return false;
}

void
DeferredModelTrainingWorker(
    _Inout_ PTP_CALLBACK_INSTANCE callbackInstance,
    _Inout_opt_ PVOID callbackContext,
    _Inout_ PTP_WORK work
    )
{
    // Threadpool callback instance and work are not needed for this sample.

    UNREFERENCED_PARAMETER(callbackInstance);
    UNREFERENCED_PARAMETER(work);

    WORKER_CONTEXT* workerContext = reinterpret_cast<WORKER_CONTEXT*>(callbackContext);
    bool workComplete = false;

    // Keep processing work until being told to stop or all work has been completed.

    while (true) {
        {
            lock_guard<mutex> scopedLock(workerContext->ContextLock);

            if (workComplete) {
                workerContext->IsComplete = true;
            }

            if (!workerContext->ShouldRun || workerContext->IsComplete) {
                workerContext->IsRunning = false;
                break;
            }
        }

        // TrainNextModelChunk returns true when there is no more work to do.

        workComplete = TrainNextModelChunk();
    }

    workerContext->CompletionSignal.notify_all();
}

int
__cdecl
wmain(
    )
{
    WORKER_CONTEXT workerContext;
    workerContext.ShouldRun = false;
    workerContext.IsRunning = false;
    workerContext.IsComplete = false;

    // Create the worker that will be started by our subscription callback.

    workerContext.Worker.reset(CreateThreadpoolWork(
        DeferredModelTrainingWorker,
        &workerContext,
        nullptr));
    RETURN_LAST_ERROR_IF_NULL(workerContext.Worker);

    // Allocate a policy suited for tasks that are best run when unlikely
    // to cause impact to the user or system performance.

    unique_policy policy;
    RETURN_IF_FAILED(CreateActivityCoordinatorPolicy(
        ACTIVITY_COORDINATOR_POLICY_TEMPLATE_GOOD,
        &policy));

    // The model training in this sample consumes GPU.
    // The GOOD policy template doesn't currently include the GPU resource. We
    // therefore customize the policy to include good GPU conditions to minimize
    // the impact of running our work.

    RETURN_IF_FAILED(SetActivityCoordinatorPolicyResourceCondition(
        policy.get(),
        ACTIVITY_COORDINATOR_RESOURCE_GPU,
        ACTIVITY_COORDINATOR_CONDITION_GOOD));

    // Subscribe to the policy for coordination notifications.

    unique_subscription subscription;
    RETURN_IF_FAILED(SubscribeActivityCoordinatorPolicy(
        policy.get(),
        DeferredWorkEventCallback,
        &workerContext,
        &subscription));;

    // Destroy the policy because we no longer need it.

    policy.reset();

    // We want our task to complete within 48h, so we allocate 24h under our
    // ideal policy and before falling back to a downgraded policy.

    bool workerCompleted;

    {
        unique_lock<mutex> scopedLock(workerContext.ContextLock);
        workerCompleted = workerContext.CompletionSignal.wait_for(
            scopedLock,
            hours(24),
            [&workerContext] { return workerContext.IsComplete; });
    }

    if (workerCompleted) {

        // Since our work is complete, we should clean up our subscription by
        // unsubscribing. This would normally be handled quietly by our RAII
        // types, but we release them explicitly to demonstrate API flow for
        // developers manually managing resources.

        subscription.reset();
        return S_OK;
    }

    // We passed our soft deadline, so downgrade the policy and wait the
    // remaining 24h until our hard deadline has been reached. Since
    // Subscriptions and policies are independent of each other, we need to
    // create a new subscription with our downgraded policy to receive
    // notifications based on its configuration.
    // 
    // The downgraded policy uses medium conditions for all needed resources.
    // This gives us the best chance to run while helping to prevent us from
    // critically degrading the user experience, which we are more likely to do
    // when falling back to manual execution.

    RETURN_IF_FAILED(CreateActivityCoordinatorPolicy(
        ACTIVITY_COORDINATOR_POLICY_TEMPLATE_MEDIUM,
        &policy));

    RETURN_IF_FAILED(SetActivityCoordinatorPolicyResourceCondition(
        policy.get(),
        ACTIVITY_COORDINATOR_RESOURCE_GPU,
        ACTIVITY_COORDINATOR_CONDITION_MEDIUM));

    subscription.reset();
    RETURN_IF_FAILED(SubscribeActivityCoordinatorPolicy(
        policy.get(),
        DeferredWorkEventCallback,
        &workerContext,
        &subscription));

    {
        unique_lock<mutex> scopedLock(workerContext.ContextLock);
        workerCompleted = workerContext.CompletionSignal.wait_for(
            scopedLock,
            hours(24),
            [&workerContext] { return workerContext.IsComplete; });
    }

    // We passed our deadline, so unsubscribe and manually resume our task.

    subscription.reset();
    ResumeWorker(&workerContext);

    // We destroyed our subscription, so we wait indefinitely for completion as
    // there's nothing to pause execution of our task.

    unique_lock<mutex> scopedLock(workerContext.ContextLock);
    workerContext.CompletionSignal.wait(
        scopedLock,
        [&workerContext] { return workerContext.IsComplete; });

    return S_OK;
}

Información general sobre la API de coordinador de actividades

API y terminología del coordinador de actividades

Elección de la directiva de coordinador de actividades adecuada