diff --git a/c/coli b/c/coli index 933f635..4e0fec9 100755 --- a/c/coli +++ b/c/coli @@ -317,10 +317,21 @@ class Spinner: if TTY: sys.stdout.write("\r\033[K"); sys.stdout.flush() def stream_turn(p, sentinel, on_bytes): - """legge fino alla sentinella; on_bytes riceve i chunk della risposta. Poi legge la riga STAT.""" - pend=b"" + """legge fino alla sentinella; on_bytes riceve i chunk della risposta. Poi legge la riga STAT. + 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: - b=p.stdout.read(1) + try: + 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 pend+=b if pend.endswith(sentinel): @@ -328,7 +339,9 @@ def stream_turn(p, sentinel, on_bytes): if rest: on_bytes(rest) line=p.stdout.readline().decode("utf-8","replace").strip() # STAT tok tps hit rss 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): out=pend[:-len(sentinel)]; pend=pend[-len(sentinel):] on_bytes(out) @@ -466,7 +479,7 @@ def cmd_chat(a): for chunk in textwrap.wrap(l, term_w()-4) or [l]: print(f" {C.dgray}{chunk}{C.r}") 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 def user_box(msg): """ri-disegna il messaggio dentro una box che si ADATTA su piu' righe: @@ -531,10 +544,13 @@ def cmd_chat(a): el=time.time()-t0 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}") - 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() else: + if st.get("interrupted"): print(f" {C.yel}⏹ interrupted{C.r}") print() except KeyboardInterrupt: print(f"\n {C.dim}interrupted{C.r}") diff --git a/c/glm.c b/c/glm.c index b3451c0..8c1c112 100644 --- a/c/glm.c +++ b/c/glm.c @@ -35,6 +35,7 @@ #include #include /* mlock: inchioda le pagine in RAM / wire pages into RAM */ #include /* fstat per mmap degli shard (COLI_MMAP) */ +#include /* SIGINT = stop morbido del turno in serve mode */ #endif #include "st.h" #include "tok.h" @@ -3610,12 +3611,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. * 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. */ +/* 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, void (*emit)(int,void*), void *ud, int *kv_out){ Cfg *c=&m->c; int V=c->vocab; int emitted=0, done=0; int draft[64]; if(g_draft>63) g_draft=63; int carry_ban=-1; /* token rifiutato dalla verifica: escluso dal resample */ - while(emitted=0 && next==eos) || is_stop(next)) break; emit(next,ud); all[kv]=next; emitted++; m->n_emit++; @@ -4321,12 +4344,17 @@ static void run_serve_mux(Model *m, const char *snap){ setvbuf(stdout, NULL, _IONBF, 0); #endif 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); hwinfo_emit(m); tiers_emit(m); emap_emit(m); int eof=0; 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;ilen) #define first (sc->first) 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); tiers_emit(m); 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(!strcmp(line,"\x02RESET")){ len=0; first=1; if(m->has_mtp) m->kv_start[m->c.n_layers]=-1; kv_disk_reset(m);