kernel-module/intent_router.c

/**
 * intent_router.c
 * موجه النوايا - نظام التوجيه الديناميكي
 * Intent Router - Dynamic Routing System
 *
 * Core routing logic that maps intents to KAR agents based on:
 * - Intent type (allocate, I/O, compute, seal)
 * - Priority level (background → critical)
 * - Current system load (thermal, memory, CPU)
 * - Agent availability
 *
 * Author: Ahmad Ali Parr
 * License: Sovereign Source
 */

#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/kthread.h>
#include <linux/delay.h>
#include <linux/atomic.h>

#include "intent_schema.h"
#include "intent_router.h"
#include "ring_buffer.h"

/* ═══════════════════════════════════════════════════════════════
 * Router State | حالة الموجه
 * ═══════════════════════════════════════════════════════════════ */

static struct {
    struct IntentRing *ring;         /* Ring buffer being monitored */
    struct RouterConfig config;      /* Configuration */
    struct task_struct *poller_task; /* Ring polling thread */
    atomic_t running;                /* Router active flag */

    /* Statistics */
    atomic64_t total_routed;
    atomic64_t routing_errors;
    atomic64_t queue_full_rejects;

} router_state;

/* Worker pool (Phase 2 implementation) */
static struct task_struct **worker_threads = NULL;

/* ═══════════════════════════════════════════════════════════════
 * Routing Policy | سياسة التوجيه
 * ═══════════════════════════════════════════════════════════════ */

/**
 * select_worker_for_intent - Choose worker thread for intent
 * @intent: Intent to route
 *
 * Returns: Worker ID (0 to num_workers-1)
 *
 * Routing strategy:
 * - CRITICAL priority: Always worker 0 (dedicated)
 * - HIGH priority: Workers 0-1 (round-robin)
 * - NORMAL/BACKGROUND: All workers (load-balanced)
 */
static unsigned int select_worker_for_intent(const struct SystemIntent *intent)
{
    static atomic_t round_robin_counter = ATOMIC_INIT(0);
    unsigned int worker_id;

    switch (intent->priority) {
        case PRIORITY_CRITICAL:
            /* Dedicated worker for critical intents */
            worker_id = 0;
            break;

        case PRIORITY_HIGH:
            /* Round-robin between workers 0-1 */
            worker_id = atomic_inc_return(&round_robin_counter) % 2;
            break;

        case PRIORITY_NORMAL:
        case PRIORITY_BACKGROUND:
        default:
            /* Load-balanced across all workers */
            worker_id = atomic_inc_return(&round_robin_counter) %
                        router_state.config.num_workers;
            break;
    }

    return worker_id;
}

/**
 * select_agent_for_intent - Choose KAR agent for intent type
 * @intent: Intent to route
 *
 * Returns: Agent ID
 *
 * Agent assignment:
 * - ALLOCATE → Agent 1 (Memory Manager)
 * - IO_READ/WRITE → Agent 2 (Storage Manager)
 * - NET_SEND/RECV → Agent 3 (Network Manager)
 * - COMPUTE → Agent 4 (Compute Offload)
 * - QUERY_STATE → Agent 5 (Observer)
 * - SEAL_COMMIT → Agent 6 (Cryptographic Sealer)
 */
static unsigned int select_agent_for_intent(const struct SystemIntent *intent)
{
    switch (intent->intent_type) {
        case INTENT_ALLOCATE:
            return 1; /* Memory Manager */

        case INTENT_IO_READ:
        case INTENT_IO_WRITE:
            return 2; /* Storage Manager */

        case INTENT_NET_SEND:
        case INTENT_NET_RECV:
            return 3; /* Network Manager */

        case INTENT_COMPUTE:
            return 4; /* Compute Offload */

        case INTENT_QUERY_STATE:
            return 5; /* System Observer */

        case INTENT_SEAL_COMMIT:
            return 6; /* Cryptographic Sealer */

        default:
            pr_warn("Router: unknown intent type %u, routing to agent 0\n",
                    intent->intent_type);
            return 0; /* Default agent */
    }
}

/**
 * estimate_processing_cost - Estimate intent processing time
 * @intent: Intent to estimate
 *
 * Returns: Estimated time in nanoseconds
 *
 * Cost model (simplified for Phase 2):
 * - I/O operations: payload_len * 100ns/byte
 * - Network: payload_len * 50ns/byte
 * - Compute: Fixed 1ms (actual depends on model)
 * - Other: Fixed 100µs
 */
static unsigned int estimate_processing_cost(const struct SystemIntent *intent)
{
    unsigned int cost_ns;

    switch (intent->intent_type) {
        case INTENT_IO_READ:
        case INTENT_IO_WRITE:
            /* Storage I/O cost */
            cost_ns = intent->payload_len * 100;
            break;

        case INTENT_NET_SEND:
        case INTENT_NET_RECV:
            /* Network cost */
            cost_ns = intent->payload_len * 50;
            break;

        case INTENT_COMPUTE:
            /* NPU/compute cost (fixed 1ms for now) */
            cost_ns = 1000000;
            break;

        default:
            /* Default overhead */
            cost_ns = 100000; /* 100µs */
            break;
    }

    return cost_ns;
}

/* ═══════════════════════════════════════════════════════════════
 * Routing Decision Engine | محرك قرار التوجيه
 * ═══════════════════════════════════════════════════════════════ */

int router_route_intent(const struct SystemIntent *intent,
                        struct RoutingDecision *decision_out)
{
    if (!intent || !decision_out)
        return -EINVAL;

    /* Select worker thread */
    decision_out->worker_id = select_worker_for_intent(intent);

    /* Select KAR agent */
    decision_out->agent_id = select_agent_for_intent(intent);

    /* Check if compute offload should use accelerator */
    decision_out->use_accelerator = (intent->intent_type == INTENT_COMPUTE);

    /* Estimate processing cost */
    decision_out->estimated_cost_ns = estimate_processing_cost(intent);

    pr_debug("Router: intent type=%u → worker=%u agent=%u cost=%uns\n",
             intent->intent_type,
             decision_out->worker_id,
             decision_out->agent_id,
             decision_out->estimated_cost_ns);

    atomic64_inc(&router_state.total_routed);
    return 0;
}

/* ═══════════════════════════════════════════════════════════════
 * Worker Thread Management | إدارة خيوط العمال
 * ═══════════════════════════════════════════════════════════════ */

/**
 * kar_worker_thread - Worker thread function
 * @data: Worker ID (void *cast)
 *
 * Each worker thread polls for intents assigned to it and
 * dispatches them to the appropriate KAR agent.
 */
static int kar_worker_thread(void *data)
{
    unsigned int worker_id = (unsigned long)data;
    struct SystemIntent intent;
    struct RoutingDecision decision;
    int ret;

    pr_info("KAR Worker %u: started\n", worker_id);

    while (!kthread_should_stop()) {
        /* Try to consume intent from ring */
        ret = ring_buffer_consume(router_state.ring, &intent);
        if (ret == -EAGAIN) {
            /* Ring empty, sleep briefly */
            msleep(10);
            continue;
        }

        if (ret < 0) {
            pr_err("KAR Worker %u: consume error %d\n", worker_id, ret);
            continue;
        }

        /* Route the intent */
        ret = router_route_intent(&intent, &decision);
        if (ret < 0) {
            pr_err("KAR Worker %u: routing failed, error=%d\n", worker_id, ret);
            atomic64_inc(&router_state.routing_errors);
            continue;
        }

        /* Check if this worker should handle it */
        if (decision.worker_id != worker_id) {
            /* Not for this worker, put it back (TODO: proper queue) */
            continue;
        }

        /* TODO Phase 3: Dispatch to Haskell policy engine
         *
         * For now, just log and mark as processed.
         */
        pr_info("KAR Worker %u: processing intent type=%u agent=%u\n",
                worker_id, intent.intent_type, decision.agent_id);

        /* Simulate processing delay */
        if (decision.estimated_cost_ns > 0) {
            usleep_range(decision.estimated_cost_ns / 1000,
                        decision.estimated_cost_ns / 1000 + 100);
        }

        /* Mark intent complete (Phase 1 ring_buffer_complete) */
        struct IntentResult result = {
            .status = 0,
            .kar_action = intent.intent_type,
            .resource_handle = 0,
            .bytes_transferred = intent.payload_len,
            .execution_time_ns = decision.estimated_cost_ns,
            .kar_agent_id = decision.agent_id,
        };

        ring_buffer_complete(router_state.ring, &result);
    }

    pr_info("KAR Worker %u: stopped\n", worker_id);
    return 0;
}

/* ═══════════════════════════════════════════════════════════════
 * Router Lifecycle | دورة حياة الموجه
 * ═══════════════════════════════════════════════════════════════ */

int router_init(struct IntentRing *ring, const struct RouterConfig *config)
{
    unsigned int i;
    int ret = 0;

    if (!ring || !config)
        return -EINVAL;

    pr_info("Router: initializing with %u workers\n", config->num_workers);

    /* Store config */
    router_state.ring = ring;
    memcpy(&router_state.config, config, sizeof(*config));
    atomic_set(&router_state.running, 1);

    /* Initialize statistics */
    atomic64_set(&router_state.total_routed, 0);
    atomic64_set(&router_state.routing_errors, 0);
    atomic64_set(&router_state.queue_full_rejects, 0);

    /* Allocate worker thread array */
    worker_threads = kzalloc(config->num_workers * sizeof(struct task_struct *),
                             GFP_KERNEL);
    if (!worker_threads) {
        pr_err("Router: failed to allocate worker array\n");
        return -ENOMEM;
    }

    /* Spawn worker threads */
    for (i = 0; i < config->num_workers; i++) {
        worker_threads[i] = kthread_run(kar_worker_thread,
                                        (void *)(unsigned long)i,
                                        "kar_worker_%u", i);
        if (IS_ERR(worker_threads[i])) {
            ret = PTR_ERR(worker_threads[i]);
            pr_err("Router: failed to create worker %u, error=%d\n", i, ret);

            /* Stop already-created workers */
            while (i > 0) {
                i--;
                kthread_stop(worker_threads[i]);
            }

            kfree(worker_threads);
            worker_threads = NULL;
            return ret;
        }
    }

    pr_info("Router: initialized successfully (%u workers active)\n",
            config->num_workers);

    return 0;
}

void router_shutdown(void)
{
    unsigned int i;

    pr_info("Router: shutting down...\n");

    atomic_set(&router_state.running, 0);

    /* Stop all worker threads */
    if (worker_threads) {
        for (i = 0; i < router_state.config.num_workers; i++) {
            if (worker_threads[i]) {
                kthread_stop(worker_threads[i]);
            }
        }
        kfree(worker_threads);
        worker_threads = NULL;
    }

    pr_info("Router: shutdown complete\n");
}

/* ═══════════════════════════════════════════════════════════════
 * Statistics | الإحصائيات
 * ═══════════════════════════════════════════════════════════════ */

int router_get_stats(struct RouterStats *stats_out)
{
    if (!stats_out)
        return -EINVAL;

    stats_out->total_routed = atomic64_read(&router_state.total_routed);
    stats_out->routing_errors = atomic64_read(&router_state.routing_errors);
    stats_out->queue_full_rejects = atomic64_read(&router_state.queue_full_rejects);
    stats_out->avg_routing_time_ns = 5000; /* TODO: real measurement */
    stats_out->current_queue_depth = 0;    /* TODO: real queue depth */

    return 0;
}

MODULE_LICENSE("Proprietary");
MODULE_AUTHOR("Ahmad Ali Parr <ahmedparr93@gmail.com>");
MODULE_DESCRIPTION("ASOS Intent Router - Dynamic KAR Dispatch");
MODULE_VERSION("2.0.0");