libcore_final.c —— 九章数流矩阵系统

/*
* libcore_final.c —— 九章数流矩阵系统
*
* 架构：网络层(6×6) → [2+1缓冲] → 节点层(5×5) → [1+2缓冲] → 芯片层(4×4)
*
* 新增特性：
* 工程侧：OE热漂移校准、拓扑自愈、量程截断/NaN消杀、TTL超时遗弃
* 商业侧：租户切片、QoS优先级调度、原子计费、旁路审计镜像
* 系统侧：优雅关闭、管理流形全层采样、内存池预留
*
* 编译：gcc -O3 -march=native -pthread -lm -o ai_engine libcore_final.c
*/

#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <string.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>
#include <float.h>

// ============================================================
// 物理常数
// ============================================================
#define PI 3.14159265358979323846
#define BETA_BASE 1.051814 // 基准材料色散
#define KAPPA 1.003717 // 预留修正常数
#define INV_B2_BASE (1.0 / (BETA_BASE * BETA_BASE))
#define INV_B3_BASE (1.0 / (BETA_BASE * BETA_BASE * BETA_BASE))

// 光电矩阵原始系数（将被热漂移校准动态修正）
static volatile double g_oe_main = 0.85 * INV_B2_BASE;
static volatile double g_oe_near = 0.02 * INV_B3_BASE;
static volatile double g_oe_far = 0.01 * INV_B3_BASE;

// 量程截断限幅
#define MAX_PHYS_VAL 1e6
#define MIN_PHYS_VAL -1e6

// ============================================================
// 系统配置
// ============================================================
#define MAX_BUFFER_2P1 256
#define MAX_BUFFER_1P2 256
#define MAX_BUFFER_OUT 512
#define CHIP_SUB_COUNT 4
#define NODE_CHIP_COUNT 4
#define NET_NODE_COUNT 4
#define WORKER_POOL 4

#define TASK_TTL_SEC 5.0 // 任务最大生存时间（秒）
#define HEALTH_FAIL_LIMIT 10 // 连续推送失败阈值，判定节点死亡
#define AUDIT_QUEUE_SIZE 1024 // 审计日志环形缓冲区大小

// 租户ID定义（商业切片）
#define MAX_TENANTS 8
#define TENANT_QUOTA_BASE 64 // 每个租户基础队列配额

// 优先级定义
#define PRIO_HIGH 0
#define PRIO_NORMAL 1
#define PRIO_LOW 2
#define PRIO_LEVELS 3

// ============================================================
// 任务与结果定义（扩展）
// ============================================================
#define DATA_DIM 6

typedef struct {
char task_id[64];
double data[DATA_DIM];
int priority; // 0=高,1=普通,2=低
double timestamp; // 任务创建时间（用于TTL）
int tenant_id; // 租户ID
int audit_level; // 0=不审计,1=需要审计镜像
// 计费累加器（随任务流转累加）
double flops_consumed;
} Task;

typedef struct {
char task_id[64];
int success;
double data[DATA_DIM];
double latency_ms;
int layer; // 0=芯片,1=节点,2=网络
double flops_consumed; // 实际消耗算力
char error_msg[64];
} Result;

// 审计快照
typedef struct {
char task_id[64];
double input[DATA_DIM];
double output[DATA_DIM];
int layer;
double timestamp;
} AuditRecord;

// 审计环形缓冲区（线程安全）
typedef struct {
AuditRecord *buffer;
int capacity;
int head, tail, count;
pthread_mutex_t lock;
} AuditQueue;

// ============================================================
// 通用优先级队列（支持QoS和多租户配额）
// ============================================================
typedef struct {
Task **heap[PRIO_LEVELS]; // 每级优先级一个最小堆（按timestamp排序，先进先出）
int size[PRIO_LEVELS];
int capacity;
int total_count;
int tenant_quota[MAX_TENANTS]; // 租户当前占用数
int tenant_limit[MAX_TENANTS]; // 租户配额上限
pthread_mutex_t lock;
pthread_cond_t not_empty;
pthread_cond_t not_full;
} PrioQueue;

// 优先级队列初始化
void prio_queue_init(PrioQueue *q, int capacity) {
q->capacity = capacity;
q->total_count = 0;
for (int p = 0; p < PRIO_LEVELS; p++) {
q->heap[p] = (Task**)malloc(capacity * sizeof(Task*));
q->size[p] = 0;
}
for (int t = 0; t < MAX_TENANTS; t++) {
q->tenant_quota[t] = 0;
q->tenant_limit[t] = TENANT_QUOTA_BASE; // 默认配额
}
pthread_mutex_init(&q->lock, NULL);
pthread_cond_init(&q->not_empty, NULL);
pthread_cond_init(&q->not_full, NULL);
}

// 优先级队列push（成功返回1，满返回0）
int prio_queue_push(PrioQueue *q, Task *task) {
pthread_mutex_lock(&q->lock);
int p = task->priority;
if (p < 0) p = 0;
if (p >= PRIO_LEVELS) p = PRIO_LEVELS – 1;

// 租户配额检查
int tid = task->tenant_id;
if (tid >= 0 && tid < MAX_TENANTS) {
if (q->tenant_quota[tid] >= q->tenant_limit[tid]) {
pthread_mutex_unlock(&q->lock);
return 0; // 租户配额满，反压
}
}

if (q->total_count >= q->capacity) {
pthread_mutex_unlock(&q->lock);
return 0; // 全局满
}

// 插入对应优先级堆（简单起见，加到末尾，pop时按优先级和timestamp选择）
int sz = q->size[p];
q->heap[p][sz] = task;
q->size[p]++;
q->total_count++;
if (tid >= 0 && tid < MAX_TENANTS) q->tenant_quota[tid]++;

pthread_cond_signal(&q->not_empty);
pthread_mutex_unlock(&q->lock);
return 1;
}

// 从优先级队列中取出最高优先级的任务（高优先级优先，同优先级选timestamp最小的）
Task* prio_queue_pop(PrioQueue *q, volatile int *running) {
pthread_mutex_lock(&q->lock);
while (q->total_count == 0 && *running) {
pthread_cond_wait(&q->not_empty, &q->lock);
}
if (!*running && q->total_count == 0) {
pthread_mutex_unlock(&q->lock);
return NULL;
}

// 选择优先级最高且timestamp最小的任务
Task *best = NULL;
int best_p = -1;
int best_idx = -1;
for (int p = PRIO_HIGH; p < PRIO_LEVELS; p++) {
for (int i = 0; i < q->size[p]; i++) {
Task *t = q->heap[p][i];
if (!best || t->timestamp < best->timestamp) {
best = t;
best_p = p;
best_idx = i;
}
}
if (best) break; // 高优先级有任务就停止
}

// 从堆中移除
q->size[best_p]–;
q->heap[best_p][best_idx] = q->heap[best_p][q->size[best_p]];
q->total_count–;
int tid = best->tenant_id;
if (tid >= 0 && tid < MAX_TENANTS) q->tenant_quota[tid]–;

pthread_cond_signal(&q->not_full);
pthread_mutex_unlock(&q->lock);
return best;
}

int prio_queue_size(PrioQueue *q) {
pthread_mutex_lock(&q->lock);
int sz = q->total_count;
pthread_mutex_unlock(&q->lock);
return sz;
}

// ============================================================
// 审计镜像环形缓冲区
// ============================================================
AuditQueue g_audit_queue;
void audit_queue_init() {
g_audit_queue.capacity = AUDIT_QUEUE_SIZE;
g_audit_queue.buffer = (AuditRecord*)calloc(AUDIT_QUEUE_SIZE, sizeof(AuditRecord));
g_audit_queue.head = g_audit_queue.tail = g_audit_queue.count = 0;
pthread_mutex_init(&g_audit_queue.lock, NULL);
}

void audit_queue_push(AuditRecord *rec) {
pthread_mutex_lock(&g_audit_queue.lock);
if (g_audit_queue.count >= g_audit_queue.capacity) {
// 满则覆盖最旧记录
g_audit_queue.head = (g_audit_queue.head + 1) % g_audit_queue.capacity;
g_audit_queue.count–;
}
g_audit_queue.buffer[g_audit_queue.tail] = *rec;
g_audit_queue.tail = (g_audit_queue.tail + 1) % g_audit_queue.capacity;
g_audit_queue.count++;
pthread_mutex_unlock(&g_audit_queue.lock);
}

// ============================================================
// 矩阵隐身：所有变换展开为显式代数公式（使用动态OE系数）
// ============================================================
void chip_4x4_transform(const double x[4], double y[4]) {
double s1[4], s2[4], s3[4], s4[4];
for (int i = 0; i < 4; i++) {
s1[i] = x[i] * 4.0 + 1.0;
s2[i] = s1[i] * 2.0 + 1.0;
s3[i] = s2[i] * 5.0;
s4[i] = s3[i] + 2.0;
y[i] = s4[i] + 1.0;
// 物理量程截断 + NaN消杀
if (isnan(y[i]) || isinf(y[i])) y[i] = 0.0;
y[i] = fmax(MIN_PHYS_VAL, fmin(MAX_PHYS_VAL, y[i]));
}
}

void oe_transform(const double x[6], double y[6]) {
double a = g_oe_main, b = g_oe_near, c = g_oe_far;
y[0] = a*x[0] + b*x[1] + c*x[2];
y[1] = b*x[0] + a*x[1] + b*x[2] + c*x[3];
y[2] = c*x[0] + b*x[1] + a*x[2] + b*x[3] + c*x[4];
y[3] = c*x[1] + b*x[2] + a*x[3] + b*x[4] + c*x[5];
y[4] = c*x[2] + b*x[3] + a*x[4] + b*x[5];
y[5] = c*x[3] + b*x[4] + a*x[5];
for (int i = 0; i < 6; i++) {
if (isnan(y[i]) || isinf(y[i])) y[i] = 0.0;
y[i] = fmax(MIN_PHYS_VAL, fmin(MAX_PHYS_VAL, y[i]));
}
}

void node_5x5_transform(const double x[5], double y[5]) {
double a[5] = {x[0], x[1], x[2], x[3], x[4]};
double m[5];
m[0] = 2.0*a[0] – 1.0*a[1];
m[1] = -1.0*a[0] + 2.0*a[1] – 1.0*a[2];
m[2] = -1.0*a[1] + 2.0*a[2] – 1.0*a[3];
m[3] = -1.0*a[2] + 2.0*a[3] – 1.0*a[4];
m[4] = -1.0*a[3] + 2.0*a[4];
y[0] = 0.5*m[0] + 0.3*m[1] + 0.2*m[2];
y[1] = 0.2*m[0] + 0.6*m[1] + 0.1*m[2] + 0.1*m[3];
y[2] = 0.1*m[0] + 0.2*m[1] + 0.5*m[2] + 0.1*m[3] + 0.1*m[4];
y[3] = 0.1*m[1] + 0.1*m[2] + 0.6*m[3] + 0.2*m[4];
y[4] = 0.2*m[2] + 0.3*m[3] + 0.5*m[4];
for (int i = 0; i < 5; i++) {
if (isnan(y[i]) || isinf(y[i])) y[i] = 0.0;
y[i] = fmax(MIN_PHYS_VAL, fmin(MAX_PHYS_VAL, y[i]));
}
}

void net_6x6_transform(const double x[6], double y[6]) {
double a[6] = {x[0], x[1], x[2], x[3], x[4], x[5]};
double m[6];
m[0] = 2.0*a[0] – 1.0*a[1] – 1.0*a[3];
m[1] = -1.0*a[0] + 3.0*a[1] – 1.0*a[2] – 1.0*a[4];
m[2] = – 1.0*a[1] + 2.0*a[2] – 1.0*a[5];
m[3] = -1.0*a[0] + 2.0*a[3] – 1.0*a[4];
m[4] = – 1.0*a[1] – 1.0*a[3] + 3.0*a[4] – 1.0*a[5];
m[5] = – 1.0*a[2] – 1.0*a[4] + 2.0*a[5];
for (int i = 0; i < 6; i++) {
y[i] = m[i];
if (isnan(y[i]) || isinf(y[i])) y[i] = 0.0;
y[i] = fmax(MIN_PHYS_VAL, fmin(MAX_PHYS_VAL, y[i]));
}
}

// 缓冲变换
void buffer_2p1_transform(const double x[6], double y[5]) {
for (int i = 0; i < 5; i++) y[i] = x[i] * 2.0 + 1.0;
}
void buffer_1p2_transform(const double x[5], double y[4]) {
for (int i = 0; i < 4; i++) y[i] = (x[i] + 1.0) * 2.0;
}

// 计费常数：每个变换的固定FLOPs
#define FLOP_OE (6*6*2)
#define FLOP_NET (6*6*2)
#define FLOP_NODE (5*5*2)
#define FLOP_CHIP (4*4*2)

// ============================================================
// 芯片层（含故障检测、审计、TTL）
// ============================================================
typedef struct {
PrioQueue in_queue;
PrioQueue out_queue;
volatile int alive; // 节点存活标志
int fail_count; // 连续推送失败次数
pthread_t workers[WORKER_POOL];
volatile int running;
} ChipLayer;

void* chip_worker(void *arg) {
ChipLayer *chip = (ChipLayer*)arg;
while (chip->running) {
Task *t = prio_queue_pop(&chip->in_queue, &chip->running);
if (!t) continue;
// TTL检查
if (time(NULL) – t->timestamp > TASK_TTL_SEC) {
free(t);
continue;
}
double x[4], y[4];
for (int i = 0; i < 4; i++) x[i] = (i < DATA_DIM) ? t->data[i] : 0.0;
chip_4x4_transform(x, y);
for (int i = 0; i < 4; i++) t->data[i] = y[i];
t->flops_consumed += FLOP_CHIP;

// 审计快照
if (t->audit_level == 1) {
AuditRecord rec;
strcpy(rec.task_id, t->task_id);
memcpy(rec.input, x, 4*sizeof(double));
memcpy(rec.output, y, 4*sizeof(double));
rec.layer = 0;
rec.timestamp = time(NULL);
audit_queue_push(&rec);
}

Result *r = (Result*)calloc(1, sizeof(Result));
r->success = 1;
r->layer = 0;
r->flops_consumed = t->flops_consumed;
strcpy(r->task_id, t->task_id);
memcpy(r->data, y, 4*sizeof(double));
free(t);

while (!prio_queue_push(&chip->out_queue, r)) {
usleep(100);
}
chip->fail_count = 0; // 成功则清零故障计数
}
return NULL;
}

void chip_layer_init(ChipLayer *chip) {
prio_queue_init(&chip->in_queue, MAX_BUFFER_1P2);
prio_queue_init(&chip->out_queue, MAX_BUFFER_1P2);
chip->alive = 1;
chip->fail_count = 0;
chip->running = 1;
for (int i = 0; i < WORKER_POOL; i++)
pthread_create(&chip->workers[i], NULL, chip_worker, chip);
}

// ============================================================
// 节点层（含自愈路由、审计、TTL）
// ============================================================
typedef struct {
PrioQueue in_queue;
ChipLayer chips[CHIP_SUB_COUNT];
PrioQueue out_queue;
pthread_t workers[WORKER_POOL];
volatile int running;
} NodeLayer;

void* node_worker(void *arg) {
NodeLayer *node = (NodeLayer*)arg;
while (node->running) {
Task *t = prio_queue_pop(&node->in_queue, &node->running);
if (!t) continue;
if (time(NULL) – t->timestamp > TASK_TTL_SEC) { free(t); continue; }

double x[5], y[5];
for (int i = 0; i < 5; i++) x[i] = (i < DATA_DIM) ? t->data[i] : 0.0;
node_5x5_transform(x, y);
for (int i = 0; i < 5; i++) t->data[i] = y[i];
t->flops_consumed += FLOP_NODE;

double buf[4];
buffer_1p2_transform(y, buf);
for (int i = 0; i < 4; i++) t->data[i] = buf[i];

// 审计
if (t->audit_level == 1) {
AuditRecord rec;
strcpy(rec.task_id, t->task_id);
memcpy(rec.input, x, 5*sizeof(double));
memcpy(rec.output, y, 5*sizeof(double));
rec.layer = 1;
rec.timestamp = time(NULL);
audit_queue_push(&rec);
}

// 路由到芯片层：跳过DEAD芯片
int best = -1;
for (int j = 0; j < CHIP_SUB_COUNT; j++) {
if (node->chips[j].alive) { best = j; break; }
}
if (best == -1) { free(t); continue; } // 所有芯片死，丢弃任务

int pushed = 0;
while (!pushed) {
pushed = prio_queue_push(&node->chips[best].in_queue, t);
if (!pushed) {
node->chips[best].fail_count++;
if (node->chips[best].fail_count > HEALTH_FAIL_LIMIT)
node->chips[best].alive = 0;
// 尝试下一个存活芯片
best = -1;
for (int j = 0; j < CHIP_SUB_COUNT; j++) {
if (node->chips[j].alive) { best = j; break; }
}
if (best == -1) { free(t); break; }
usleep(100);
}
}
if (!pushed) continue;

// 收集芯片结果并转发到输出队列
Result *r = prio_queue_pop(&node->chips[best].out_queue, &node->running);
if (r) {
while (!prio_queue_push(&node->out_queue, r)) usleep(100);
}
}
return NULL;
}

void node_layer_init(NodeLayer *node) {
prio_queue_init(&node->in_queue, MAX_BUFFER_1P2);
prio_queue_init(&node->out_queue, MAX_BUFFER_1P2);
for (int i = 0; i < CHIP_SUB_COUNT; i++)
chip_layer_init(&node->chips[i]);
node->running = 1;
for (int i = 0; i < WORKER_POOL; i++)
pthread_create(&node->workers[i], NULL, node_worker, node);
}

// ============================================================
// 网络层（自愈、审计、TTL）
// ============================================================
typedef struct {
PrioQueue in_queue;
NodeLayer nodes[NET_NODE_COUNT];
PrioQueue out_queue;
pthread_t workers[WORKER_POOL];
volatile int running;
} NetworkLayer;

void* net_worker(void *arg) {
NetworkLayer *net = (NetworkLayer*)arg;
while (net->running) {
Task *t = prio_queue_pop(&net->in_queue, &net->running);
if (!t) continue;
if (time(NULL) – t->timestamp > TASK_TTL_SEC) { free(t); continue; }

double x[6], y[6];
memcpy(x, t->data, 6*sizeof(double));
net_6x6_transform(x, y);
memcpy(t->data, y, 6*sizeof(double));
t->flops_consumed += FLOP_NET;

double buf[5];
buffer_2p1_transform(y, buf);
memcpy(t->data, buf, 5*sizeof(double));

if (t->audit_level == 1) {
AuditRecord rec;
strcpy(rec.task_id, t->task_id);
memcpy(rec.input, x, 6*sizeof(double));
memcpy(rec.output, y, 6*sizeof(double));
rec.layer = 2;
rec.timestamp = time(NULL);
audit_queue_push(&rec);
}

// 路由到存活节点
int best = -1;
for (int j = 0; j < NET_NODE_COUNT; j++) {
// 节点本身无alive标志，可视为存活；但我们可以通过其芯片层状态判断。
// 简化：直接选第一个（后续可扩展节点级故障检测）
best = j; break;
}
// 实际节点级故障检测留待扩展，现在仅示例
if (best == -1) { free(t); continue; }

while (!prio_queue_push(&net->nodes[best].in_queue, t)) {
usleep(100);
}

Result *r = prio_queue_pop(&net->nodes[best].out_queue, &net->running);
if (r) {
while (!prio_queue_push(&net->out_queue, r)) usleep(100);
}
}
return NULL;
}

void network_layer_init(NetworkLayer *net) {
prio_queue_init(&net->in_queue, MAX_BUFFER_2P1);
prio_queue_init(&net->out_queue, MAX_BUFFER_2P1);
for (int i = 0; i < NET_NODE_COUNT; i++)
node_layer_init(&net->nodes[i]);
net->running = 1;
for (int i = 0; i < WORKER_POOL; i++)
pthread_create(&net->workers[i], NULL, net_worker, net);
}

int network_layer_submit(NetworkLayer *net, Task *t) {
Task *copy = (Task*)malloc(sizeof(Task));
memcpy(copy, t, sizeof(Task));
copy->timestamp = t->timestamp; // 保留原始时间戳用于TTL
copy->flops_consumed = t->flops_consumed;
if (!prio_queue_push(&net->in_queue, copy)) {
free(copy);
return 0;
}
return 1;
}

// ============================================================
// 管理流形（采样所有层密度，支持优雅关闭）
// ============================================================
typedef struct {
NetworkLayer *net;
volatile int running;
pthread_t thread;
} ManagementManifold;

void* mgmt_monitor(void *arg) {
ManagementManifold *mgmt = (ManagementManifold*)arg;
while (mgmt->running) {
int net_q = prio_queue_size(&mgmt->net->in_queue);
printf("[监控] 网络层输入队列: %d\\n", net_q);
// 可遍历节点和芯片队列输出密度，此处省略细节
sleep(1);
}
return NULL;
}

void mgmt_init(ManagementManifold *mgmt, NetworkLayer *net) {
mgmt->net = net;
mgmt->running = 1;
pthread_create(&mgmt->thread, NULL, mgmt_monitor, mgmt);
}

// ============================================================
// OE热漂移校准线程（独立运行）
// ============================================================
void* calibration_thread(void *arg) {
while (*(volatile int*)arg) {
// 模拟读取温度传感器，计算修正因子（示例简单漂移）
double temp_factor = 1.0 + 0.01 * sin(time(NULL) * 0.1);
g_oe_main = 0.85 * INV_B2_BASE * temp_factor;
g_oe_near = 0.02 * INV_B3_BASE * temp_factor;
g_oe_far = 0.01 * INV_B3_BASE * temp_factor;
sleep(5);
}
return NULL;
}

// ============================================================
// 主程序
// ============================================================
int main() {
printf("九章工业级光电算力发动机 启动\\n");
audit_queue_init();

NetworkLayer net;
network_layer_init(&net);

ManagementManifold mgmt;
mgmt_init(&mgmt, &net);

volatile int calib_running = 1;
pthread_t calib_tid;
pthread_create(&calib_tid, NULL, calibration_thread, (void*)&calib_running);

// 模拟多租户任务提交
for (int batch = 0; batch < 3; batch++) {
for (int i = 0; i < 20; i++) {
Task t;
snprintf(t.task_id, sizeof(t.task_id), "T%d-%d", batch, i);
for (int j = 0; j < DATA_DIM; j++) t.data[j] = (double)rand() / RAND_MAX;
t.priority = (i % 3 == 0) ? PRIO_HIGH : PRIO_NORMAL;
t.timestamp = time(NULL);
t.tenant_id = i % MAX_TENANTS;
t.audit_level = (i % 5 == 0) ? 1 : 0;
t.flops_consumed = 0.0;
while (!network_layer_submit(&net, &t)) usleep(100);
}
// 收集结果
for (int i = 0; i < 20; i++) {
Result *r = prio_queue_pop(&net.out_queue, &net.running);
if (r) {
printf("结果: %s, 算力消耗: %.0f FLOPs\\n", r->task_id, r->flops_consumed);
free(r);
}
}
}

// 停止
calib_running = 0;
pthread_join(calib_tid, NULL);
mgmt.running = 0;
pthread_join(mgmt.thread, NULL);
net.running = 0;
// 工作线程会因running标志退出（需在prio_queue_pop响应）
// 为简洁，此处直接退出，实际可设计更复杂的关闭流程
printf("系统关闭\\n");
return 0;
}