glm.c/coli: Ctrl-C soft-stops the current turn instead of killing the engine
In serve/chat mode, Ctrl-C during generation killed the whole engine, losing the loaded model and forcing a full reload. Now a SIGINT handler (armed only in run_serve / run_serve_mux) sets a flag that spec_decode's token loop treats exactly like hitting the NGEN cap: the turn ends through the normal path, so the END sentinel, STAT line, usage_save and KV append all run — and :more can continue the interrupted answer. The mux loop closes in-flight requests via the same mux_done path. One-shot ./glm runs and Windows keep default SIGINT (die). coli: stream_turn survives the first KeyboardInterrupt, forwards SIGINT to the engine (covers non-TTY), drains to the turn boundary, and reports it. A second Ctrl-C quits. Help line and per-turn footer updated. POSIX only (sigaction); no behaviour change on Windows. Verified end-to-end on Apple M4 + Metal: interrupt mid-decode, engine stays up, next prompt answers, :q exits 0. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -317,10 +317,21 @@ class Spinner:
|
|||||||
if TTY: sys.stdout.write("\r\033[K"); sys.stdout.flush()
|
if TTY: sys.stdout.write("\r\033[K"); sys.stdout.flush()
|
||||||
|
|
||||||
def stream_turn(p, sentinel, on_bytes):
|
def stream_turn(p, sentinel, on_bytes):
|
||||||
"""legge fino alla sentinella; on_bytes riceve i chunk della risposta. Poi legge la riga STAT."""
|
"""legge fino alla sentinella; on_bytes riceve i chunk della risposta. Poi legge la riga STAT.
|
||||||
pend=b""
|
Il PRIMO Ctrl-C durante lo stream non chiude la sessione: il motore (handler SIGINT)
|
||||||
|
chiude il turno per la via del tetto NGEN e noi dreniamo fino alla sentinella.
|
||||||
|
Un SECONDO Ctrl-C esce davvero."""
|
||||||
|
pend=b""; interrupted=False
|
||||||
while True:
|
while True:
|
||||||
|
try:
|
||||||
b=p.stdout.read(1)
|
b=p.stdout.read(1)
|
||||||
|
except KeyboardInterrupt:
|
||||||
|
if interrupted or p.poll() is not None: raise
|
||||||
|
interrupted=True
|
||||||
|
try: p.send_signal(signal.SIGINT) # non-TTY: il motore potrebbe non aver visto il Ctrl-C
|
||||||
|
except Exception: pass
|
||||||
|
print(f"\n {C.yel}⏹ stopping… (Ctrl-C again to quit){C.r}", flush=True)
|
||||||
|
continue
|
||||||
if b==b"": return None
|
if b==b"": return None
|
||||||
pend+=b
|
pend+=b
|
||||||
if pend.endswith(sentinel):
|
if pend.endswith(sentinel):
|
||||||
@@ -328,7 +339,9 @@ def stream_turn(p, sentinel, on_bytes):
|
|||||||
if rest: on_bytes(rest)
|
if rest: on_bytes(rest)
|
||||||
line=p.stdout.readline().decode("utf-8","replace").strip() # STAT tok tps hit rss
|
line=p.stdout.readline().decode("utf-8","replace").strip() # STAT tok tps hit rss
|
||||||
m=re.match(r"STAT (\S+) (\S+) (\S+) (\S+)", line)
|
m=re.match(r"STAT (\S+) (\S+) (\S+) (\S+)", line)
|
||||||
return {"tok":int(m.group(1)),"tps":float(m.group(2)),"hit":float(m.group(3)),"rss":float(m.group(4))} if m else {}
|
st={"tok":int(m.group(1)),"tps":float(m.group(2)),"hit":float(m.group(3)),"rss":float(m.group(4))} if m else {}
|
||||||
|
if interrupted: st["interrupted"]=True
|
||||||
|
return st
|
||||||
if len(pend)>len(sentinel):
|
if len(pend)>len(sentinel):
|
||||||
out=pend[:-len(sentinel)]; pend=pend[-len(sentinel):]
|
out=pend[:-len(sentinel)]; pend=pend[-len(sentinel):]
|
||||||
on_bytes(out)
|
on_bytes(out)
|
||||||
@@ -464,7 +477,7 @@ def cmd_chat(a):
|
|||||||
for chunk in textwrap.wrap(l, term_w()-4) or [l]:
|
for chunk in textwrap.wrap(l, term_w()-4) or [l]:
|
||||||
print(f" {C.dgray}{chunk}{C.r}")
|
print(f" {C.dgray}{chunk}{C.r}")
|
||||||
except Exception: pass
|
except Exception: pass
|
||||||
print(f" {C.dim}type and press Enter · :more continues · :reset clears memory · :q exits{C.r}\n")
|
print(f" {C.dim}type and press Enter · Ctrl-C stops the answer · :more continues · :reset clears memory · :q exits{C.r}\n")
|
||||||
w=term_w()-4
|
w=term_w()-4
|
||||||
def user_box(msg):
|
def user_box(msg):
|
||||||
"""ri-disegna il messaggio dentro una box che si ADATTA su piu' righe:
|
"""ri-disegna il messaggio dentro una box che si ADATTA su piu' righe:
|
||||||
@@ -529,10 +542,13 @@ def cmd_chat(a):
|
|||||||
el=time.time()-t0
|
el=time.time()-t0
|
||||||
if st.get("tok"):
|
if st.get("tok"):
|
||||||
print(f"\r {C.dgray}└─ {st['tok']} tok · {st['tps']:.2f} tok/s · hit {st['hit']:.0f}% · RSS {st['rss']:.1f} GB · {el:.0f}s{C.r}")
|
print(f"\r {C.dgray}└─ {st['tok']} tok · {st['tps']:.2f} tok/s · hit {st['hit']:.0f}% · RSS {st['rss']:.1f} GB · {el:.0f}s{C.r}")
|
||||||
if st["tok"]>=a.ngen:
|
if st.get("interrupted"):
|
||||||
|
print(f" {C.yel}⏹ interrupted; type :more to continue the response{C.r}")
|
||||||
|
elif st["tok"]>=a.ngen:
|
||||||
print(f" {C.yel}…stopped at --ngen ({a.ngen}); type :more to continue the response{C.r}")
|
print(f" {C.yel}…stopped at --ngen ({a.ngen}); type :more to continue the response{C.r}")
|
||||||
print()
|
print()
|
||||||
else:
|
else:
|
||||||
|
if st.get("interrupted"): print(f" {C.yel}⏹ interrupted{C.r}")
|
||||||
print()
|
print()
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
print(f"\n {C.dim}interrupted{C.r}")
|
print(f"\n {C.dim}interrupted{C.r}")
|
||||||
|
|||||||
@@ -35,6 +35,7 @@
|
|||||||
#include <sys/resource.h>
|
#include <sys/resource.h>
|
||||||
#include <sys/mman.h> /* mlock: inchioda le pagine in RAM / wire pages into RAM */
|
#include <sys/mman.h> /* mlock: inchioda le pagine in RAM / wire pages into RAM */
|
||||||
#include <sys/stat.h> /* fstat per mmap degli shard (COLI_MMAP) */
|
#include <sys/stat.h> /* fstat per mmap degli shard (COLI_MMAP) */
|
||||||
|
#include <signal.h> /* SIGINT = stop morbido del turno in serve mode */
|
||||||
#endif
|
#endif
|
||||||
#include "st.h"
|
#include "st.h"
|
||||||
#include "tok.h"
|
#include "tok.h"
|
||||||
@@ -3579,12 +3580,34 @@ static void stops_arm(const Cfg *c, int tok_eos){
|
|||||||
* all: storia token (capacita' >= kv+n_new+g_draft+2), kv = token gia' in KV.
|
* all: storia token (capacita' >= kv+n_new+g_draft+2), kv = token gia' in KV.
|
||||||
* logit = logits della posizione kv-1 (dal prefill); viene liberato qui.
|
* logit = logits della posizione kv-1 (dal prefill); viene liberato qui.
|
||||||
* emit(tok,ud) per ogni token emesso. Ritorna i token emessi; *kv_out = nuova kv. */
|
* emit(tok,ud) per ogni token emesso. Ritorna i token emessi; *kv_out = nuova kv. */
|
||||||
|
/* STOP MORBIDO (serve/chat): SIGINT chiude il turno CORRENTE per la stessa via
|
||||||
|
* del tetto NGEN (stats, usage_save, KV append, sentinella END tutti normali)
|
||||||
|
* invece di uccidere il motore; :more puo' continuare la risposta interrotta.
|
||||||
|
* Il flag e' armato solo nei serve-loop (intr_install): nei run one-shot e in
|
||||||
|
* validazione SIGINT resta il default (morte immediata). Solo POSIX: su
|
||||||
|
* Windows il comportamento di Ctrl-C non cambia.
|
||||||
|
* EN: soft stop (serve/chat): SIGINT ends the CURRENT turn through the same
|
||||||
|
* path as the NGEN cap — stats/usage/KV/END sentinel all normal — instead of
|
||||||
|
* killing the engine; :more can continue the interrupted answer. Armed only
|
||||||
|
* in the serve loops; one-shot runs keep default SIGINT. POSIX only. */
|
||||||
|
static volatile sig_atomic_t g_intr=0;
|
||||||
|
#if defined(__APPLE__) || defined(__linux__) || defined(__FreeBSD__)
|
||||||
|
static void intr_sig(int s){ (void)s; g_intr=1; }
|
||||||
|
static void intr_install(void){
|
||||||
|
struct sigaction sa; memset(&sa,0,sizeof(sa));
|
||||||
|
sa.sa_handler=intr_sig; sigemptyset(&sa.sa_mask);
|
||||||
|
sa.sa_flags=SA_RESTART; /* getline/pread non devono vedere EINTR */
|
||||||
|
sigaction(SIGINT,&sa,NULL);
|
||||||
|
}
|
||||||
|
#else
|
||||||
|
static void intr_install(void){}
|
||||||
|
#endif
|
||||||
static int spec_decode(Model *m, int *all, int kv, int n_new, int eos, float *logit,
|
static int spec_decode(Model *m, int *all, int kv, int n_new, int eos, float *logit,
|
||||||
void (*emit)(int,void*), void *ud, int *kv_out){
|
void (*emit)(int,void*), void *ud, int *kv_out){
|
||||||
Cfg *c=&m->c; int V=c->vocab; int emitted=0, done=0;
|
Cfg *c=&m->c; int V=c->vocab; int emitted=0, done=0;
|
||||||
int draft[64]; if(g_draft>63) g_draft=63;
|
int draft[64]; if(g_draft>63) g_draft=63;
|
||||||
int carry_ban=-1; /* token rifiutato dalla verifica: escluso dal resample */
|
int carry_ban=-1; /* token rifiutato dalla verifica: escluso dal resample */
|
||||||
while(emitted<n_new && !done){
|
while(emitted<n_new && !done && !g_intr){ /* g_intr: stessa uscita del tetto n_new */
|
||||||
int next=pick_tok(logit,V,carry_ban); carry_ban=-1; free(logit); logit=NULL;
|
int next=pick_tok(logit,V,carry_ban); carry_ban=-1; free(logit); logit=NULL;
|
||||||
if((eos>=0 && next==eos) || is_stop(next)) break;
|
if((eos>=0 && next==eos) || is_stop(next)) break;
|
||||||
emit(next,ud); all[kv]=next; emitted++; m->n_emit++;
|
emit(next,ud); all[kv]=next; emitted++; m->n_emit++;
|
||||||
@@ -4290,12 +4313,17 @@ static void run_serve_mux(Model *m, const char *snap){
|
|||||||
setvbuf(stdout, NULL, _IONBF, 0);
|
setvbuf(stdout, NULL, _IONBF, 0);
|
||||||
#endif
|
#endif
|
||||||
setvbuf(stdin,NULL,_IONBF,0);
|
setvbuf(stdin,NULL,_IONBF,0);
|
||||||
|
intr_install(); /* Ctrl-C = chiudi i turni in volo, non il processo */
|
||||||
printf("\x01\x01READY\x01\x01\nSTAT 0 0.00 0.0 %.2f\n",rss_gb()); fflush(stdout);
|
printf("\x01\x01READY\x01\x01\nSTAT 0 0.00 0.0 %.2f\n",rss_gb()); fflush(stdout);
|
||||||
hwinfo_emit(m);
|
hwinfo_emit(m);
|
||||||
tiers_emit(m);
|
tiers_emit(m);
|
||||||
emap_emit(m);
|
emap_emit(m);
|
||||||
int eof=0;
|
int eof=0;
|
||||||
for(;;){
|
for(;;){
|
||||||
|
if(g_intr){ g_intr=0; /* stop morbido: ogni request attiva finisce ORA per la
|
||||||
|
* via normale di mux_done (DONE+stats+KV coerenti) */
|
||||||
|
for(int i=0;i<nctx;i++) if(req[i].active) mux_done(m,&ctx[i],&req[i]);
|
||||||
|
}
|
||||||
int active=0; for(int i=0;i<nctx;i++) active+=req[i].active;
|
int active=0; for(int i=0;i<nctx;i++) active+=req[i].active;
|
||||||
/* Poll stdin for available input without blocking. On POSIX this is
|
/* Poll stdin for available input without blocking. On POSIX this is
|
||||||
* select(); on Windows, select() on a pipe handle routes to winsock
|
* select(); on Windows, select() on a pipe handle routes to winsock
|
||||||
@@ -4388,9 +4416,11 @@ static void run_serve(Model *m, const char *snap){
|
|||||||
#define len (sc->len)
|
#define len (sc->len)
|
||||||
#define first (sc->first)
|
#define first (sc->first)
|
||||||
char *line=NULL; size_t cap=0; ssize_t nr; char *buf=malloc(1<<16);
|
char *line=NULL; size_t cap=0; ssize_t nr; char *buf=malloc(1<<16);
|
||||||
|
intr_install(); /* Ctrl-C = fine turno, non fine processo */
|
||||||
printf("\x01\x01" "READY" "\x01\x01\n"); printf("STAT 0 0.00 0.0 %.2f\n", rss_gb()); fflush(stdout);
|
printf("\x01\x01" "READY" "\x01\x01\n"); printf("STAT 0 0.00 0.0 %.2f\n", rss_gb()); fflush(stdout);
|
||||||
tiers_emit(m);
|
tiers_emit(m);
|
||||||
while((nr=getline(&line,&cap,stdin))>0){
|
while((nr=getline(&line,&cap,stdin))>0){
|
||||||
|
g_intr=0; /* interruzioni arrivate tra i turni: stantie */
|
||||||
if(nr>0 && line[nr-1]=='\n') line[--nr]=0;
|
if(nr>0 && line[nr-1]=='\n') line[--nr]=0;
|
||||||
if(!strcmp(line,"\x02RESET")){ len=0; first=1; if(m->has_mtp) m->kv_start[m->c.n_layers]=-1;
|
if(!strcmp(line,"\x02RESET")){ len=0; first=1; if(m->has_mtp) m->kv_start[m->c.n_layers]=-1;
|
||||||
kv_disk_reset(m);
|
kv_disk_reset(m);
|
||||||
|
|||||||
Reference in New Issue
Block a user