char-fork/llama.cpp

git clone https://git.t4t.associates/char-fork/llama.cpp

Max Krasnyanskyhexagon: L2 cache handling rework (dirty bit tracking with lazy flushing) and more MUL_MAT updates (#25762)b2dd28a3b

char/aliceai-t5-moe
brancheschar/aliceai-t5-moeall refs →
4.8 KiB170 linesraw
1#pragma clang diagnostic ignored "-Wunused-function"
2
3#include <stdbool.h>
4#include <stdlib.h>
5#include <string.h>
6
7#include <qurt_thread.h>
8#include <qurt_futex.h>
9#include <qurt_hvx.h>
10
11#include <HAP_compute_res.h>
12
13#include "hmx-queue.h"
14
15#define QURT_LOWEST_PRIO (254)
16
17static inline void hmx_lock(hmx_queue_t q)
18{
19    if (!q->hmx_locked) {
20        HAP_compute_res_hmx_lock(q->hap_rctx);
21        q->hmx_locked = true;
22    }
23}
24
25static inline void hmx_unlock(hmx_queue_t q)
26{
27    if (q->hmx_locked) {
28        HAP_compute_res_hmx_unlock(q->hap_rctx);
29        q->hmx_locked = false;
30    }
31}
32
33static inline void hmx_queue_process(hmx_queue_t q, bool* killed) {
34    unsigned int ir = atomic_load(&q->idx_read);
35
36    while (ir != atomic_load(&q->idx_write)) {
37        struct hmx_queue_desc *d = &q->desc[ir];
38        if (!d->done) {
39            FARF(HIGH, "hmx-queue-process: ir %u func %p data %p", ir, d->func, d->data);
40
41            uintptr_t sig = (uintptr_t) d->func;
42            switch (sig) {
43                case HMX_QUEUE_NOOP:    /* noop */;     break;
44                case HMX_QUEUE_KILL:    *killed = true; break;
45                case HMX_QUEUE_SUSPEND: hmx_unlock(q);  break;
46                case HMX_QUEUE_WAKEUP:  hmx_lock(q);    break;
47                default:
48                    hmx_lock(q);
49                    htp_trace_event_start(q->trace, HTP_TRACE_EVT_HMX_COMP, ir);
50                    d->func(d->data);
51                    htp_trace_event_stop(q->trace, HTP_TRACE_EVT_HMX_COMP, ir);
52                    break;
53            }
54
55            atomic_fetch_add(&d->done, 1);
56        }
57
58        ir = (ir + 1) & q->idx_mask;
59        atomic_store(&q->idx_read, ir);
60    }
61}
62
63static void hmx_queue_thread(void * arg) {
64    hmx_queue_t q = (hmx_queue_t) arg;
65
66    FARF(HIGH, "hmx-queue-thread: started");
67
68    bool killed = false;
69
70    unsigned int poll_cnt  = HMX_QUEUE_POLL_COUNT;
71    unsigned int prev_seqn = 0;
72    while (!killed) {
73        unsigned int seqn = atomic_load(&q->seqn);
74        if (seqn == prev_seqn) {
75            // drop HVX context while spinning
76            if (poll_cnt > 1 && poll_cnt == HMX_QUEUE_POLL_COUNT) {
77                qurt_hvx_unlock();
78            }
79            if (--poll_cnt) { hex_pause(); continue; }
80            FARF(HIGH, "hmx-queue-thread: sleeping");
81            qurt_futex_wait(&q->seqn, prev_seqn);
82            poll_cnt = HMX_QUEUE_POLL_COUNT;
83            continue;
84        }
85        prev_seqn = seqn;
86        poll_cnt  = HMX_QUEUE_POLL_COUNT;
87
88        FARF(HIGH, "hmx-queue-thread: new work");
89
90        hmx_queue_process(q, &killed);
91    }
92
93    FARF(HIGH, "hmx-queue-thread: stopped");
94}
95
96size_t hmx_queue_sizeof(size_t capacity, uint32_t stack_size) {
97    capacity = hex_ceil_pow2(capacity);
98    size_t size_q = hex_align_up(sizeof(struct hmx_queue_s), HEX_L2_LINE_SIZE);
99    size_t size_desc = hex_align_up(capacity * sizeof(struct hmx_queue_desc), HEX_L2_LINE_SIZE);
100    size_t size_stack = stack_size;
101    return size_q + size_desc + size_stack;
102}
103
104size_t hmx_queue_alignof(void) {
105    return HEX_L2_LINE_SIZE;
106}
107
108hmx_queue_t hmx_queue_init(void * ptr, size_t capacity, uint32_t stack_size, uint32_t hap_rctx, struct htp_thread_trace * trace) {
109    capacity = hex_ceil_pow2(capacity);
110    size_t size_q = hex_align_up(sizeof(struct hmx_queue_s), HEX_L2_LINE_SIZE);
111    size_t size_desc = hex_align_up(capacity * sizeof(struct hmx_queue_desc), HEX_L2_LINE_SIZE);
112
113    uint8_t * block = (uint8_t *) ptr;
114
115    hmx_queue_t q = (hmx_queue_t) block; block += size_q;
116    memset(q, 0, sizeof(struct hmx_queue_s));
117
118    q->capacity = capacity;
119    q->idx_mask = capacity - 1;
120    q->hap_rctx = hap_rctx;
121    q->external_mem = true;
122
123    q->desc = (struct hmx_queue_desc *) block; block += size_desc;
124    memset(q->desc, 0, capacity * sizeof(struct hmx_queue_desc));
125
126    q->stack = block;
127    memset(q->stack, 0, stack_size);
128
129    q->trace = trace;
130
131    // Match caller thread priority (same pattern as worker-pool.c).
132    int prio = qurt_thread_get_priority(qurt_thread_get_id());
133    if (prio < 1) {
134        prio = 1;
135    }
136    if (prio > QURT_LOWEST_PRIO) {
137        prio = QURT_LOWEST_PRIO;
138    }
139
140    qurt_thread_attr_t attr;
141    qurt_thread_attr_init(&attr);
142    qurt_thread_attr_set_stack_addr(&attr, q->stack);
143    qurt_thread_attr_set_stack_size(&attr, stack_size);
144    qurt_thread_attr_set_priority(&attr, prio);
145    qurt_thread_attr_set_name(&attr, "hmx-queue");
146
147    int err = qurt_thread_create(&q->thread, &attr, hmx_queue_thread, q);
148    if (err) {
149        FARF(ERROR, "hmx-worker: thread create failed (%d)", err);
150        return NULL;
151    }
152
153    FARF(HIGH, "hmx-queue: capacity %u\n", capacity);
154
155    return q;
156}
157
158void hmx_queue_free(hmx_queue_t q) {
159    if (!q) {
160        return;
161    }
162
163    // Tell the worker to exit.
164    hmx_queue_flush(q);
165    hmx_queue_signal(q, HMX_QUEUE_KILL);
166    hmx_queue_flush(q);
167
168    int status;
169    qurt_thread_join(q->thread, &status);
170}