/** * 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 #include #include #include #include #include #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 "); MODULE_DESCRIPTION("ASOS Intent Router - Dynamic KAR Dispatch"); MODULE_VERSION("2.0.0");