Concepto y Propósito de Singleflight

El patrón Singleflight se utiliza para garantizar que múltiples solicitudes concurrentes para el mismo recurso costoso (como una consulta de base de datos o un cálculo complejo) ejecuten la operación subyacente solo una vez. Las solicitudes posteriores que llegan mientras la primera está en proceso esperan y reciben el mismo resultado, evitando así la sobrecarga del sistema. Esta implementación está inspirada en el paquete singleflight de Go, adaptado al paradigma de C++.

Estructura de Datos y Gestión de Estado

#include <map>
#include <memory>
#include <mutex>
#include <condition_variable>

template<typename KeyType, typename ValueType>
class ConcurrentCallGuard {
    struct OperationResult {
        bool completed;
        std::mutex result_mutex;
        std::condition_variable completion_signal;
        ValueType output;
    };

    std::mutex registry_mutex;
    std::map<KeyType, std::shared_ptr<OperationResult>> operation_registry;

public:
    template<typename Function, typename... Arguments>
    ValueType execute(KeyType identifier, Function&& target_operation, Arguments&&... operands) {
        std::unique_lock<std::mutex> registry_lock(registry_mutex);
        auto existing_entry = operation_registry.find(identifier);

        if (existing_entry != operation_registry.end()) {
            std::shared_ptr<OperationResult> result_container = existing_entry->second;
            registry_lock.unlock();
            std::unique_lock<std::mutex> result_lock(result_container->result_mutex);
            result_container->completion_signal.wait(result_lock,
                [result_container]() { return result_container->completed; });
            return result_container->output;
        }

        std::shared_ptr<OperationResult> new_result = std::make_shared<OperationResult>();
        new_result->completed = false;
        operation_registry[identifier] = new_result;
        registry_lock.unlock();

        try {
            new_result->output = target_operation(std::forward<Arguments>(operands)...);
        } catch (...) {
            std::lock_guard<std::mutex> guard(new_result->result_mutex);
            new_result->completed = true;
            new_result->completion_signal.notify_all();
            registry_lock.lock();
            operation_registry.erase(identifier);
            throw;
        }

        {
            std::lock_guard<std::mutex> guard(new_result->result_mutex);
            new_result->completed = true;
        }
        new_result->completion_variable.notify_all();

        registry_lock.lock();
        operation_registry.erase(identifier);
        return new_result->output;
    }
};

Casos de Prueba y Ejemplo de Uso

#include "concurrent_call_guard.hpp"
#include <string>
#include <thread>
#include <vector>
#include <chrono>
#include <iostream>

std::string get_current_time() {
    auto now = std::chrono::system_clock::now();
    auto time = std::chrono::system_clock::to_time_t(now);
    char buffer[80];
    std::strftime(buffer, sizeof(buffer), "%H:%M:%S", std::localtime(&time));
    return buffer;
}

void resource_intensive_operation(int param_a, int param_b, std::vector<uint8_t>& data_buffer) {
    std::cout << get_current_time() << " Operación iniciada\n";
    std::this_thread::sleep_for(std::chrono::seconds(2));
    for (size_t i = 0; i < 10; ++i) {
        data_buffer[i] = '0' + i;
    }
    std::cout << get_current_time() << " Operación finalizada\n";
}

int main() {
    ConcurrentCallGuard<std::string, std::shared_ptr<std::vector<uint8_t>>> guard;

    for (int i =長 = 0; i < 8; ++i) {
        std::thread([&guard]() {
            auto computed_result = guard.execute(
                "resource_key",
                [](int x, int y) -> std::shared_ptr<std::vector<uint8_t>> {
                    std::vector<uint8_t> buffer(100);
                    resource_intensive_operation(x, y, buffer);
                    return std::make_shared<std::vector<uint8_t>>(buffer);
                },
                200, 150);

            std::cout << "Resultado: ";
            for (int j = 0; j < 10; ++j) {
                std::cout << (*computed_result)[j];
            }
            std::cout << std::endl;
        }).detach();
    }

    std::this_thread::sleep_for(std::chrono::seconds(4));
    return 0;
}

Esta prueba demuestra que solo una llamada ejecuta la operación costosa, mientras que los otros siete hilos reciben el resultado calculado de forma simultánea.

Consideraciones sobre el Manejo de Excepciones

Es fundamental manejar excepciones dentro de la operación objetivo. Si ocurre una excepción no capturada, se debe aseguarr que los hilos en espera sean notificados y el registro se limpie para evitar bloqueos permanentes. La implementación incluye un bloque try-catch que notifica a los esperantes y re-lanza la excepción para su manejo externo.