From 689dbedcbc80b3d8a343edab8af4d1ba6b62ca8c Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Wed, 21 Oct 2026 22:07:32 -0300 Subject: [PATCH 01/16] =?UTF-8?q?Testes=20do=20Ctrl-C=20que=20grava=20o=20?= =?UTF-8?q?checkpoint=20nos=20dois=20la=C3=A7os?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O Ctrl-C é simulado com SIGINT ao thread principal: no sequencial ele corta a chamada em curso, e no paralelo chega enquanto o thread principal espera as linhas. O caso da issue, 12 linhas, batch_size=5 e Ctrl-C na 9ª chamada, roda com entrada pandas e polars nos dois modos e confere que o checkpoint tem as linhas pagas e que a retomada não as chama de novo. O teste da execução recusada passa a interromper B depois de uma linha: com o limite de uso e com o Ctrl-C, B grava o próprio checkpoint e C retoma dele. --- tests/test_interrupcao.py | 214 ++++++++++++++++++++++++++++++++ tests/test_retomada_assinada.py | 28 +++-- 2 files changed, 233 insertions(+), 9 deletions(-) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 5ce3d54d..2c95069f 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -1,5 +1,7 @@ """Falha que impede as linhas seguintes interrompe a execução sem gastá-las.""" +import signal +import threading from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch @@ -206,3 +208,215 @@ def test_interrupcao_com_texto_ilegivel_ainda_interrompe(paralelo): saida = _roda(df, llm, parallel_requests=paralelo) assert saida["_dataframeit_status"].isna().all() + + +# ============================================================================= +# Ctrl-C +# ============================================================================= + + +def _ctrl_c(): + """Manda SIGINT ao thread principal, como o Ctrl-C do terminal. + + No modo sequencial a chamada roda no thread principal, e o + `KeyboardInterrupt` sai do próprio `pthread_kill`, antes de a chamada + terminar. No paralelo, ele chega ao thread principal enquanto este espera as + linhas, e a chamada do worker termina e é paga. + """ + assert signal.getsignal(signal.SIGINT) is signal.default_int_handler + principal = threading.main_thread().ident + assert principal is not None + signal.pthread_kill(principal, signal.SIGINT) + + +def _llm_com_ctrl_c(na_chamada: int | None = None): + """Devolve a lista dos textos cujas chamadas terminaram (pagas) e o falso LLM.""" + trava = threading.Lock() + iniciadas: list[str] = [] + pagas: list[str] = [] + + def llm(texto, *args, **kwargs): + with trava: + iniciadas.append(texto) + vez = len(iniciadas) + if vez == na_chamada: + _ctrl_c() + with trava: + pagas.append(texto) + return {"data": {"campo1": f"r-{texto}"}, "usage": {}} + + return pagas, llm + + +def _entrada(biblioteca: str, textos: list[str]): + if biblioteca == "polars": + pl = pytest.importorskip("polars") + return pl.DataFrame({"texto": textos}) + return pd.DataFrame({"texto": textos}) + + +requer_pthread_kill = pytest.mark.skipif( + not hasattr(signal, "pthread_kill"), reason="Ctrl-C simulado com signal.pthread_kill" +) + + +@requer_pthread_kill +@pytest.mark.parametrize("paralelo", [1, 3]) +@pytest.mark.parametrize("biblioteca", ["pandas", "polars"]) +def test_ctrl_c_grava_as_linhas_pagas_e_a_retomada_nao_as_chama_de_novo( + tmp_path, biblioteca, paralelo +): + """12 linhas, batch_size=5 e Ctrl-C durante a 9ª chamada. + + No sequencial, a 9ª chamada não termina, e 8 linhas estão pagas. Com 3 + workers, a 9ª é a última do terceiro lote, e as três do lote terminam + enquanto o executor as espera. A gravação a cada 5 linhas deixaria as + últimas 3 ou 4 de fora. + """ + textos = [f"t{i:02d}" for i in range(1, 13)] + entrada = _entrada(biblioteca, textos) + ckpt = tmp_path / "ckpt.csv" + pagas, llm = _llm_com_ctrl_c(na_chamada=9) + + with pytest.raises(KeyboardInterrupt): + _roda( + entrada, + llm, + parallel_requests=paralelo, + batch_size=5, + checkpoint_path=ckpt, + ) + + assert len(pagas) == (8 if paralelo == 1 else 9) + gravado = pd.read_csv(ckpt) + processadas = gravado.loc[gravado["_dataframeit_status"].eq("processed"), "texto"] + assert sorted(processadas) == sorted(pagas) + + retomada, llm_ok = _llm_com_ctrl_c() + saida = _roda( + _entrada(biblioteca, textos), + llm_ok, + parallel_requests=paralelo, + batch_size=5, + checkpoint_path=ckpt, + ) + + assert sorted(retomada) == sorted(set(textos) - set(pagas)) + assert list(saida["campo1"]) == [f"r-{texto}" for texto in textos] + + +@requer_pthread_kill +@pytest.mark.parametrize("paralelo", [1, 3]) +def test_falha_ao_gravar_no_ctrl_c_avisa_e_mantem_o_keyboard_interrupt(tmp_path, paralelo): + """A falha vira aviso, como nas gravações a cada batch_size.""" + _, llm = _llm_com_ctrl_c(na_chamada=5) + + with ( + patch("dataframeit.core._save_checkpoint", side_effect=OSError("disco cheio")) as grava, + pytest.warns(UserWarning, match="Falha ao gravar o checkpoint.*OSError: disco cheio"), + pytest.raises(KeyboardInterrupt), + ): + _roda( + pd.DataFrame({"texto": [f"t{i}" for i in range(9)]}), + llm, + parallel_requests=paralelo, + batch_size=100, + checkpoint_path=tmp_path / "ckpt.csv", + ) + + assert grava.call_count == 1 + + +@requer_pthread_kill +def test_segundo_ctrl_c_grava_so_as_linhas_concluidas(tmp_path): + """O segundo Ctrl-C corta a espera do executor, e a chamada que não terminou fica pendente.""" + textos = ["lenta", "ctrl-c"] + ckpt = tmp_path / "ckpt.csv" + liberada = threading.Event() + + def llm(texto, *args, **kwargs): + if texto == "lenta": + liberada.wait(timeout=10) + else: + _ctrl_c() + return {"data": {"campo1": f"r-{texto}"}, "usage": {}} + + saida_do_executor = ThreadPoolExecutor.__exit__ + + def segundo_ctrl_c_na_espera(self, tipo, valor, rastro): + # A saída por Ctrl-C espera as threads; o segundo Ctrl-C chega nessa + # espera, depois de o executor ter sido encerrado sem esperar. + if tipo is KeyboardInterrupt: + self.shutdown(wait=False) + _ctrl_c() + return saida_do_executor(self, tipo, valor, rastro) + + try: + with ( + patch.object(ThreadPoolExecutor, "__exit__", segundo_ctrl_c_na_espera), + pytest.raises(KeyboardInterrupt), + ): + _roda( + pd.DataFrame({"texto": textos}), + llm, + parallel_requests=2, + batch_size=100, + checkpoint_path=ckpt, + ) + gravado = pd.read_csv(ckpt) + finally: + liberada.set() + + processadas = set(gravado.loc[gravado["_dataframeit_status"].eq("processed"), "texto"]) + assert "lenta" not in processadas + + retomada, llm_ok = _llm_com_ctrl_c() + _roda( + pd.DataFrame({"texto": textos}), + llm_ok, + parallel_requests=2, + batch_size=100, + checkpoint_path=ckpt, + ) + assert sorted(retomada) == sorted(set(textos) - processadas) + + +@pytest.mark.parametrize("paralelo", [1, 3]) +def test_ctrl_c_antes_de_terminar_uma_linha_nao_reescreve_o_checkpoint(tmp_path, paralelo): + """Sem linha registrada desde a última gravação, o arquivo fica como estava.""" + ckpt = tmp_path / "ckpt.csv" + df = pd.DataFrame({"texto": ["a", "b", "c"]}) + _roda( + df.copy(), + _llm_com_ctrl_c()[1], + model_kwargs={"temperature": 0.5}, + batch_size=10, + checkpoint_path=ckpt, + ) + anterior = ckpt.read_bytes() + + def interrompe(*args, **kwargs): + raise KeyboardInterrupt + + with pytest.warns(UserWarning, match="outra configuração"), pytest.raises(KeyboardInterrupt): + _roda( + df.copy(), + interrompe, + parallel_requests=paralelo, + batch_size=10, + checkpoint_path=ckpt, + ) + + assert ckpt.read_bytes() == anterior + + +@requer_pthread_kill +@pytest.mark.parametrize("paralelo", [1, 3]) +def test_ctrl_c_sem_checkpoint_deixa_as_linhas_pagas_na_entrada_pandas(paralelo): + df = pd.DataFrame({"texto": [f"t{i}" for i in range(9)]}) + pagas, llm = _llm_com_ctrl_c(na_chamada=5) + + with pytest.raises(KeyboardInterrupt): + _roda(df, llm, parallel_requests=paralelo) + + assert sorted(df.loc[df["_dataframeit_status"].eq("processed"), "texto"]) == sorted(pagas) diff --git a/tests/test_retomada_assinada.py b/tests/test_retomada_assinada.py index 01bbe546..d7954fec 100644 --- a/tests/test_retomada_assinada.py +++ b/tests/test_retomada_assinada.py @@ -73,21 +73,31 @@ def test_execucao_recusada_e_interrompida_nao_adota_o_checkpoint_antigo(tmp_path df = pd.DataFrame({"texto": ["a", "b", "c"]}) _roda(df, _llm(100, SystemExit())[1], prompt="P1 {texto}", batch_size=10, checkpoint_path=ckpt) + aviso_do_limite = ( + contextlib.nullcontext() + if isinstance(interrupcao, KeyboardInterrupt) + else pytest.warns(UserWarning, match="Execução interrompida") + ) with ( pytest.warns(UserWarning, match="outra configuração"), + aviso_do_limite, contextlib.suppress(KeyboardInterrupt), ): - _roda(df, _llm(0, interrupcao)[1], prompt="P2 {texto}", batch_size=10, checkpoint_path=ckpt) + _roda( + df, + _llm(1, interrupcao, prefixo="B")[1], + prompt="P2 {texto}", + batch_size=10, + checkpoint_path=ckpt, + ) - # Com o limite de uso, B grava o próprio checkpoint, com as linhas pendentes, - # e C retoma dele; com Ctrl-C, o arquivo ainda é o de A, e C recomeça. - chamadas, llm = _llm(100, SystemExit(), prefixo="P2-") - with warnings.catch_warnings(): - warnings.simplefilter("ignore") - saida = _roda(df, llm, prompt="P2 {texto}", batch_size=10, checkpoint_path=ckpt) + # Nas duas interrupções, B grava o próprio checkpoint, com a linha que pagou e + # as pendentes, e C retoma dele, sem nenhuma resposta de A. + chamadas, llm = _llm(100, SystemExit(), prefixo="C") + saida = _roda(df, llm, prompt="P2 {texto}", batch_size=10, checkpoint_path=ckpt) - assert len(chamadas) == 3 - assert saida["campo1"].tolist() == ["P2-1", "P2-2", "P2-3"] + assert len(chamadas) == 2 + assert saida["campo1"].tolist() == ["B1", "C1", "C2"] @pytest.mark.parametrize("outra", [["x", "y", "z"], ["x", "y"]]) From 6fe535e69c976009ae588a270754354010c4cfc6 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Wed, 21 Oct 2026 22:07:38 -0300 Subject: [PATCH 02/16] =?UTF-8?q?Ctrl-C=20grava=20no=20checkpoint=20as=20l?= =?UTF-8?q?inhas=20conclu=C3=ADdas=20e=20relan=C3=A7a=20a=20interrup=C3=A7?= =?UTF-8?q?=C3=A3o?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Os dois laços passam a tratar o KeyboardInterrupt num envoltório: gravam as linhas registradas desde a última gravação e relançam. Sem linha nova, o arquivo fica como estava. No paralelo, a saída do executor espera as chamadas em andamento, e a cópia sob a trava leva só as que terminaram; a gravação passa pelo _SnapshotWriter, que não regride o arquivo. A falha de gravação vira aviso em _try_save_checkpoint e não troca a interrupção. O tratamento de cada linha não muda; o diff reindenta o corpo dos laços. --- src/dataframeit/core.py | 247 ++++++++++++++++++++++------------------ 1 file changed, 136 insertions(+), 111 deletions(-) diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index 09019e12..e46aeefe 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -2456,7 +2456,7 @@ def _progress_suffix( return "" -def _process_rows( # noqa: PLR0913 (estado da execução repassado por dataframeit) +def _process_rows( # noqa: C901, PLR0913 (estado da execução repassado por dataframeit, e um ramo por desfecho da linha) df: pd.DataFrame, *, text_column: str, @@ -2508,78 +2508,91 @@ def _process_rows( # noqa: PLR0913 (estado da execução repassado por datafram rows_saved = 0 # Processar cada linha - for i, (idx, row) in enumerate(tqdm(df.iterrows(), total=len(df), desc=desc)): - # Verificar se linha já foi processada - row_already_processed = pd.notna(row[status_col]) and row[status_col] == "processed" + try: + for i, (idx, row) in enumerate(tqdm(df.iterrows(), total=len(df), desc=desc)): + # Verificar se linha já foi processada + row_already_processed = pd.notna(row[status_col]) and row[status_col] == "processed" - # Com reprocess_columns, todas as linhas são processadas. - if not reprocess_columns and not is_pending[i]: - continue + # Com reprocess_columns, todas as linhas são processadas. + if not reprocess_columns and not is_pending[i]: + continue - success = False - if _is_missing_text(row[text_column]): - _record_missing_text( - df, - idx, - status_col=status_col, - row_already_processed=row_already_processed, - reprocess_columns=reprocess_columns, - ) - else: - try: - result = _invoke_row( - backend, - str(row[text_column]), - row, - reprocess_columns if row_already_processed else None, - expected_columns, - nan_columns=nan_columns, - ) - _record_success( - df, - idx, - result, - status_col=status_col, - expected_columns=expected_columns, - config=config, - track_tokens=track_tokens, - row_already_processed=row_already_processed, - reprocess_columns=reprocess_columns, - trace_mode=trace_mode, - token_stats=token_stats, - ) - success = True - except ProviderAbortError as e: - token_stats["abort"] = _error_text(e) - _mark_unreached( - df, - [df.index[j] for j in range(i, len(df)) if reprocess_columns or is_pending[j]], - status_col=status_col, - reprocess_columns=reprocess_columns, - ) - break - except Exception as e: # noqa: BLE001 (qualquer falha da linha vira status 'error') - _record_error_safely( + success = False + if _is_missing_text(row[text_column]): + _record_missing_text( df, idx, - e, status_col=status_col, - config=config, row_already_processed=row_already_processed, reprocess_columns=reprocess_columns, - token_stats=token_stats, ) + else: + try: + result = _invoke_row( + backend, + str(row[text_column]), + row, + reprocess_columns if row_already_processed else None, + expected_columns, + nan_columns=nan_columns, + ) + _record_success( + df, + idx, + result, + status_col=status_col, + expected_columns=expected_columns, + config=config, + track_tokens=track_tokens, + row_already_processed=row_already_processed, + reprocess_columns=reprocess_columns, + trace_mode=trace_mode, + token_stats=token_stats, + ) + success = True + except ProviderAbortError as e: + token_stats["abort"] = _error_text(e) + _mark_unreached( + df, + [ + df.index[j] + for j in range(i, len(df)) + if reprocess_columns or is_pending[j] + ], + status_col=status_col, + reprocess_columns=reprocess_columns, + ) + break + except Exception as e: # noqa: BLE001 (qualquer falha da linha vira status 'error') + _record_error_safely( + df, + idx, + e, + status_col=status_col, + config=config, + row_already_processed=row_already_processed, + reprocess_columns=reprocess_columns, + token_stats=token_stats, + ) - rows_processed_this_run += 1 - if ( - checkpoint - and rows_processed_this_run % checkpoint.batch_size == 0 - and _try_save_checkpoint(df, checkpoint) - ): - rows_saved = rows_processed_this_run - - if success and config.rate_limit_delay > 0: - time.sleep(config.rate_limit_delay) + rows_processed_this_run += 1 + if ( + checkpoint + and rows_processed_this_run % checkpoint.batch_size == 0 + and _try_save_checkpoint(df, checkpoint) + ): + rows_saved = rows_processed_this_run + + if success and config.rate_limit_delay > 0: + time.sleep(config.rate_limit_delay) + except KeyboardInterrupt: + # Ctrl-C: grava as linhas registradas desde a última gravação, que a + # retomada pagaria de novo, e relança a interrupção. A linha cuja chamada + # foi cortada não chegou a ser registrada e fica pendente. A falha de + # gravação vira aviso em _try_save_checkpoint e não troca a interrupção. + if checkpoint and rows_processed_this_run > rows_saved: + _try_save_checkpoint(df, checkpoint) + raise # Save final: a cauda (< batch_size), o que uma gravação que falhou deixou de # fora e as marcas de uma interrupção, que não contam como linha processada. @@ -2589,7 +2602,7 @@ def _process_rows( # noqa: PLR0913 (estado da execução repassado por datafram return token_stats -def _process_rows_parallel( # noqa: C901, PLR0913, PLR0915 (estado compartilhado entre threads) +def _process_rows_parallel( # noqa: C901, PLR0912, PLR0913, PLR0915 (estado compartilhado entre threads) df: pd.DataFrame, *, text_column: str, @@ -2786,54 +2799,66 @@ def process_single_row(row_data: tuple) -> dict: return {"success": True, "idx": idx} # Processar com ThreadPoolExecutor - with tqdm(total=len(rows_to_process), desc=desc) as pbar: - # Usar abordagem iterativa para permitir ajuste dinâmico de workers - pending_rows = list(rows_to_process) + try: + with tqdm(total=len(rows_to_process), desc=desc) as pbar: + # Usar abordagem iterativa para permitir ajuste dinâmico de workers + pending_rows = list(rows_to_process) - while pending_rows and not abort_event.is_set(): - # Pegar batch com número atual de workers - with lock: - worker_batch = min(current_workers, len(pending_rows)) - batch = pending_rows[:worker_batch] - pending_rows = pending_rows[worker_batch:] - - with ThreadPoolExecutor(max_workers=worker_batch) as executor: - futures = {} - for row in batch: - # A linha que não foi despachada nunca rodou: falhar ao - # despachar interrompe a execução e a deixa sem status, para - # a retomada. Como 'error', o resume=True a preservaria. - try: - futures[executor.submit(process_single_row, row)] = row - except Exception as e: # noqa: BLE001 (ver o comentário acima) - with lock: - token_stats.setdefault("abort", _error_text(e)) - abort_event.set() - break - - for future in as_completed(futures): - try: - outcome = future.result() - except Exception as e: # noqa: BLE001 (falha fora da linha não para a execução) - # Escapou do tratamento da própria linha; sem isto ela - # ficaria sem status e sumiria da saída. - _i, idx, _row = futures[future] - if idx in counted: - warnings.warn( - f"Erro inesperado no executor depois de registrar a linha " - f"{idx}, que mantém o status gravado: {_error_text(e)}", - stacklevel=1, - ) - else: + while pending_rows and not abort_event.is_set(): + # Pegar batch com número atual de workers + with lock: + worker_batch = min(current_workers, len(pending_rows)) + batch = pending_rows[:worker_batch] + pending_rows = pending_rows[worker_batch:] + + with ThreadPoolExecutor(max_workers=worker_batch) as executor: + futures = {} + for row in batch: + # A linha que não foi despachada nunca rodou: falhar ao + # despachar interrompe a execução e a deixa sem status, para + # a retomada. Como 'error', o resume=True a preservaria. + try: + futures[executor.submit(process_single_row, row)] = row + except Exception as e: # noqa: BLE001 (ver o comentário acima) with lock: - _record_unrecorded_error(df, idx, e, status_col=status_col) - snapshot = _count_row(idx) - _save_snapshot(snapshot) - recorded.add(idx) - else: - if outcome["success"] or outcome["error"] is not None: - recorded.add(outcome["idx"]) - pbar.update(1) + token_stats.setdefault("abort", _error_text(e)) + abort_event.set() + break + + for future in as_completed(futures): + try: + outcome = future.result() + except Exception as e: # noqa: BLE001 (falha fora da linha não para a execução) + # Escapou do tratamento da própria linha; sem isto ela + # ficaria sem status e sumiria da saída. + _i, idx, _row = futures[future] + if idx in counted: + warnings.warn( + f"Erro inesperado no executor depois de registrar a linha " + f"{idx}, que mantém o status gravado: {_error_text(e)}", + stacklevel=1, + ) + else: + with lock: + _record_unrecorded_error(df, idx, e, status_col=status_col) + snapshot = _count_row(idx) + _save_snapshot(snapshot) + recorded.add(idx) + else: + if outcome["success"] or outcome["error"] is not None: + recorded.add(outcome["idx"]) + pbar.update(1) + except KeyboardInterrupt: + # Ctrl-C: a saída do executor espera as linhas em andamento, e elas chegam + # aqui registradas. Um segundo Ctrl-C durante essa espera deixa threads + # rodando: a cópia sob `lock` leva só as linhas que terminaram, e as + # demais ficam pendentes para a retomada. Pelo rótulo, o _SnapshotWriter + # não regrava o arquivo sem linha contada desde o último snapshot gravado. + if writer: + with lock: + frame, label = df.copy(), checkpoint_counter + writer.save(frame, label) + raise if abort_event.is_set(): _mark_unreached( From 5ed10ba4b4147026835c829906510c105445d736 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Wed, 21 Oct 2026 22:07:38 -0300 Subject: [PATCH 03/16] FAQ e CHANGELOG: o que o Ctrl-C faz com o lote em andamento --- CHANGELOG.md | 1 + docs/en/guides/faq.md | 2 ++ docs/guides/faq.md | 2 ++ 3 files changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa4c28b0..f63fe3b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,7 @@ e este projeto adere ao [Versionamento Semântico](https://semver.org/lang/pt-BR ### Corrigido +- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas em andamento que terminam enquanto o executor as espera; as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A falha ao gravar vira aviso e não troca a interrupção (#213). - No provider `codex`, a thread daemon que pede o `thread/unsubscribe` da linha abortada pela morte do app-server deixa de ficar presa para sempre. O processo morto ainda aceita a escrita enquanto termina, e o SDK registra a espera da resposta depois de o leitor do stdout ter acordado os pedidos em voo; agora o fechamento do cliente acorda essa espera (#250). - `read_df` com argumento de leitura que faz o pandas devolver outra coisa que um DataFrame, como `chunksize`, `iterator`, `sheet_name=None` ou lista e o `chunksize` do JSON em linhas, devolve o `TextFileReader`, o `JsonReader` ou o dict de abas como o pandas o devolve, sem normalizar, com ou sem modelo. Antes, sem modelo, ou com modelo que tem campo de lista, dict ou modelo aninhado, e com `normalize=True`, o padrão, a função levantava `AttributeError` ao procurar as colunas no objeto do leitor (#249). - Com entrada polars, a saída deixa de trocar sem aviso o valor que o modelo devolveu quando o pyarrow tipa a coluna pelo primeiro valor e converte os demais. Sai `Object`, com o aviso das colunas que o polars não reúne, o campo cuja anotação junta `timedelta` a outro tipo, `float` a `bool` ou `bytes` a `str`, e a coluna cujos valores trazem essas combinações ou as de data e hora, também em `Any`, onde o escalar do numpy, como o `np.int64`, conta como o número do Python, nos itens de lista e no campo de modelo aninhado, além da coluna cujos dicts do mesmo campo têm chaves diferentes e da que traz `datetime` com e sem fuso, `time` com fuso ou `Decimal` NaN. O NaN de campo `float` sai NaN. Antes, em `Union[timedelta, int]` o 5 virava cinco microssegundos, em `Union[float, bool]` o `True` virava `1.0`, o `datetime` sem fuso ganhava o fuso do primeiro, o `time` perdia o fuso, o `Decimal` NaN e o NaN de campo `float` viravam nulo, `{}` ao lado de `{"a": 1}` virava `{"a": None}`, e em `Union[NewType sobre date, int]` o 5 virava `1970-01-06`. Com `resume=True`, a coluna do campo cuja anotação é `int`, um `Literal` com inteiro ou uma união com um deles é lida com os inteiros do polars, e o inteiro acima de 2**53 deixa de voltar arredondado em `Float64`. O `Decimal` NaN e o NaN de campo `float` de uma linha processada deixam de ser acusados como incompatíveis na retomada, também com entrada pandas, e o reprocessamento campo a campo os repassa às condições como valor; na coluna float64, como a do checkpoint relido, o NaN não se distingue do nulo e segue contando como ausente. O `NewType` e o apelido `type`, inclusive o apelido genérico usado com argumentos, como `Opcional[int]` de `type Opcional[T] = T | None`, passam a contar pelo tipo a que se referem na saída polars, na retomada e em `read_df`: o `NewType` sobre `dict` sai em texto JSON, como o campo `dict`, e é retomado, e o campo `NewType` sobre `str` ou sobre `int` é relido do CSV como texto cru ou como inteiro exato (#241). diff --git a/docs/en/guides/faq.md b/docs/en/guides/faq.md index 6a2f9a0d..a1f0b88c 100644 --- a/docs/en/guides/faq.md +++ b/docs/en/guides/faq.md @@ -62,3 +62,5 @@ Run a sample first (`df.sample(30)`) with `track_tokens=True`. The summary at th ### A long run was interrupted Use `batch_size` and `checkpoint_path` to save progress during the run, and resume by running the same call again, which continues from the file; see [Checkpoints for Long Runs](performance.md#checkpoints-for-long-runs). + +A Ctrl-C saves to the checkpoint the rows completed since the last save before stopping, and resuming does not call them again. The row whose call the Ctrl-C cut off stays pending. With `parallel_requests`, the interruption waits for the calls in progress to finish, and they go into the save; a second Ctrl-C cuts that wait short, and the calls that did not finish stay pending. diff --git a/docs/guides/faq.md b/docs/guides/faq.md index d3364cfa..3ec3a095 100644 --- a/docs/guides/faq.md +++ b/docs/guides/faq.md @@ -62,3 +62,5 @@ Rode primeiro uma amostra (`df.sample(30)`) com `track_tokens=True`. O resumo ao ### Uma execução longa foi interrompida Use `batch_size` e `checkpoint_path` para gravar o progresso durante a execução, e retome rodando a mesma chamada de novo, que continua do arquivo; ver [Checkpoints em Execuções Longas](performance.md#checkpoints-em-execucoes-longas). + +Um Ctrl-C grava no checkpoint as linhas concluídas desde a última gravação antes de interromper, e a retomada não as chama de novo. A linha cuja chamada o Ctrl-C cortou fica pendente. Com `parallel_requests`, a interrupção espera as chamadas em andamento terminarem, e elas entram na gravação; um segundo Ctrl-C corta essa espera, e as chamadas que não terminaram ficam pendentes. From e3b1c716c2ab03437af182d61d4f0dfe0ca344e5 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Wed, 21 Oct 2026 22:14:26 -0300 Subject: [PATCH 04/16] =?UTF-8?q?Teste=20da=20execu=C3=A7=C3=A3o=20recusad?= =?UTF-8?q?a=20confere=20os=20avisos=20pelo=20registro?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No pytest 7.0, o piso do extra dev que o job lowest-direct instala, um pytest.warns dentro do outro engole os avisos do de fora, e o aviso de outra configuração não chegava ao pytest.warns que o esperava. --- tests/test_retomada_assinada.py | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/tests/test_retomada_assinada.py b/tests/test_retomada_assinada.py index d7954fec..495cf82d 100644 --- a/tests/test_retomada_assinada.py +++ b/tests/test_retomada_assinada.py @@ -73,16 +73,10 @@ def test_execucao_recusada_e_interrompida_nao_adota_o_checkpoint_antigo(tmp_path df = pd.DataFrame({"texto": ["a", "b", "c"]}) _roda(df, _llm(100, SystemExit())[1], prompt="P1 {texto}", batch_size=10, checkpoint_path=ckpt) - aviso_do_limite = ( - contextlib.nullcontext() - if isinstance(interrupcao, KeyboardInterrupt) - else pytest.warns(UserWarning, match="Execução interrompida") - ) - with ( - pytest.warns(UserWarning, match="outra configuração"), - aviso_do_limite, - contextlib.suppress(KeyboardInterrupt), - ): + # Os avisos se conferem pelo registro: um pytest.warns dentro do outro engole + # os avisos do de fora no pytest 7, e o pytest 8 reemite o que não casa. + with warnings.catch_warnings(record=True) as avisos, contextlib.suppress(KeyboardInterrupt): + warnings.simplefilter("always") _roda( df, _llm(1, interrupcao, prefixo="B")[1], @@ -90,6 +84,11 @@ def test_execucao_recusada_e_interrompida_nao_adota_o_checkpoint_antigo(tmp_path batch_size=10, checkpoint_path=ckpt, ) + mensagens = [str(aviso.message) for aviso in avisos] + assert any("outra configuração" in mensagem for mensagem in mensagens) + assert any("Execução interrompida" in mensagem for mensagem in mensagens) == isinstance( + interrupcao, ProviderUsageLimitError + ) # Nas duas interrupções, B grava o próprio checkpoint, com a linha que pagou e # as pendentes, e C retoma dele, sem nenhuma resposta de A. From bf7b7ccbe19f4e38ef228a509b7c739393815676 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Wed, 21 Oct 2026 22:34:03 -0300 Subject: [PATCH 05/16] =?UTF-8?q?La=C3=A7o=20sequencial:=20o=20try=20do=20?= =?UTF-8?q?Ctrl-C=20envolve=20s=C3=B3=20o=20la=C3=A7o,=20e=20o=20registro?= =?UTF-8?q?=20da=20linha=20vai=20para=20process=5Frow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O try em volta do laço inteiro reindentava o registro da linha, e qualquer mudança dentro dele conflitaria com este ramo. O registro volta à indentação da main numa função interna, e só o laço que pula, conta, grava e espera fica dentro do try. A gravação segue no frame de _process_rows, e o aviso de falha continua apontando a chamada do usuário. --- src/dataframeit/core.py | 123 ++++++++++++++++++++-------------------- 1 file changed, 63 insertions(+), 60 deletions(-) diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index e46aeefe..e1d71650 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -2507,73 +2507,76 @@ def _process_rows( # noqa: C901, PLR0913 (estado da execução repassado por da rows_processed_this_run = 0 rows_saved = 0 - # Processar cada linha - try: - for i, (idx, row) in enumerate(tqdm(df.iterrows(), total=len(df), desc=desc)): - # Verificar se linha já foi processada - row_already_processed = pd.notna(row[status_col]) and row[status_col] == "processed" - - # Com reprocess_columns, todas as linhas são processadas. - if not reprocess_columns and not is_pending[i]: - continue + def process_row(i: int, idx: Hashable, row: pd.Series) -> bool: + """Registra o desfecho da linha e devolve se a chamada deu certo.""" + # Verificar se linha já foi processada + row_already_processed = pd.notna(row[status_col]) and row[status_col] == "processed" - success = False - if _is_missing_text(row[text_column]): - _record_missing_text( + success = False + if _is_missing_text(row[text_column]): + _record_missing_text( + df, + idx, + status_col=status_col, + row_already_processed=row_already_processed, + reprocess_columns=reprocess_columns, + ) + else: + try: + result = _invoke_row( + backend, + str(row[text_column]), + row, + reprocess_columns if row_already_processed else None, + expected_columns, + nan_columns=nan_columns, + ) + _record_success( df, idx, + result, status_col=status_col, + expected_columns=expected_columns, + config=config, + track_tokens=track_tokens, row_already_processed=row_already_processed, reprocess_columns=reprocess_columns, + trace_mode=trace_mode, + token_stats=token_stats, ) - else: - try: - result = _invoke_row( - backend, - str(row[text_column]), - row, - reprocess_columns if row_already_processed else None, - expected_columns, - nan_columns=nan_columns, - ) - _record_success( - df, - idx, - result, - status_col=status_col, - expected_columns=expected_columns, - config=config, - track_tokens=track_tokens, - row_already_processed=row_already_processed, - reprocess_columns=reprocess_columns, - trace_mode=trace_mode, - token_stats=token_stats, - ) - success = True - except ProviderAbortError as e: - token_stats["abort"] = _error_text(e) - _mark_unreached( - df, - [ - df.index[j] - for j in range(i, len(df)) - if reprocess_columns or is_pending[j] - ], - status_col=status_col, - reprocess_columns=reprocess_columns, - ) - break - except Exception as e: # noqa: BLE001 (qualquer falha da linha vira status 'error') - _record_error_safely( - df, - idx, - e, - status_col=status_col, - config=config, - row_already_processed=row_already_processed, - reprocess_columns=reprocess_columns, - token_stats=token_stats, - ) + success = True + except ProviderAbortError as e: + token_stats["abort"] = _error_text(e) + _mark_unreached( + df, + [df.index[j] for j in range(i, len(df)) if reprocess_columns or is_pending[j]], + status_col=status_col, + reprocess_columns=reprocess_columns, + ) + except Exception as e: # noqa: BLE001 (qualquer falha da linha vira status 'error') + _record_error_safely( + df, + idx, + e, + status_col=status_col, + config=config, + row_already_processed=row_already_processed, + reprocess_columns=reprocess_columns, + token_stats=token_stats, + ) + return success + + # Processar cada linha. A gravação do checkpoint fica neste laço, e não em + # process_row, para que o aviso de falha aponte a chamada do usuário (ver o + # stacklevel de _try_save_checkpoint). + try: + for i, (idx, row) in enumerate(tqdm(df.iterrows(), total=len(df), desc=desc)): + # Com reprocess_columns, todas as linhas são processadas. + if not reprocess_columns and not is_pending[i]: + continue + success = process_row(i, idx, row) + if "abort" in token_stats: + break rows_processed_this_run += 1 if ( From b122947caa4fd35253b57de63d389594ba473153 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Wed, 21 Oct 2026 22:37:27 -0300 Subject: [PATCH 06/16] =?UTF-8?q?La=C3=A7o=20paralelo:=20o=20Ctrl-C=20entr?= =?UTF-8?q?a=20no=20with=20do=20tqdm,=20sem=20reindentar=20o=20la=C3=A7o?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O try em volta do with do tqdm reindentava o despacho e a coleta das linhas. Um gerenciador de contexto, _on_ctrl_c, entra no with que já existia e chama a gravação no Ctrl-C antes de relançá-lo. Ele fica fora do tqdm, que fecha antes da gravação, como no try. A gravação vira a função interna save_registered_rows, com o comentário que estava no except, e o PLR0912 que o try exigia sai do noqa. --- src/dataframeit/core.py | 126 ++++++++++++++++++++++------------------ 1 file changed, 69 insertions(+), 57 deletions(-) diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index e1d71650..6395bff7 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -2605,7 +2605,17 @@ def process_row(i: int, idx: Hashable, row: pd.Series) -> bool: return token_stats -def _process_rows_parallel( # noqa: C901, PLR0912, PLR0913, PLR0915 (estado compartilhado entre threads) +@contextmanager +def _on_ctrl_c(callback: Callable[[], None]) -> Iterator[None]: + """Chama `callback` quando um KeyboardInterrupt atravessa o bloco, e o relança.""" + try: + yield + except KeyboardInterrupt: + callback() + raise + + +def _process_rows_parallel( # noqa: C901, PLR0913, PLR0915 (estado compartilhado entre threads) df: pd.DataFrame, *, text_column: str, @@ -2801,67 +2811,69 @@ def process_single_row(row_data: tuple) -> dict: time.sleep(config.rate_limit_delay) return {"success": True, "idx": idx} - # Processar com ThreadPoolExecutor - try: - with tqdm(total=len(rows_to_process), desc=desc) as pbar: - # Usar abordagem iterativa para permitir ajuste dinâmico de workers - pending_rows = list(rows_to_process) + def save_registered_rows() -> None: + """Grava o checkpoint no Ctrl-C. - while pending_rows and not abort_event.is_set(): - # Pegar batch com número atual de workers - with lock: - worker_batch = min(current_workers, len(pending_rows)) - batch = pending_rows[:worker_batch] - pending_rows = pending_rows[worker_batch:] - - with ThreadPoolExecutor(max_workers=worker_batch) as executor: - futures = {} - for row in batch: - # A linha que não foi despachada nunca rodou: falhar ao - # despachar interrompe a execução e a deixa sem status, para - # a retomada. Como 'error', o resume=True a preservaria. - try: - futures[executor.submit(process_single_row, row)] = row - except Exception as e: # noqa: BLE001 (ver o comentário acima) - with lock: - token_stats.setdefault("abort", _error_text(e)) - abort_event.set() - break - - for future in as_completed(futures): - try: - outcome = future.result() - except Exception as e: # noqa: BLE001 (falha fora da linha não para a execução) - # Escapou do tratamento da própria linha; sem isto ela - # ficaria sem status e sumiria da saída. - _i, idx, _row = futures[future] - if idx in counted: - warnings.warn( - f"Erro inesperado no executor depois de registrar a linha " - f"{idx}, que mantém o status gravado: {_error_text(e)}", - stacklevel=1, - ) - else: - with lock: - _record_unrecorded_error(df, idx, e, status_col=status_col) - snapshot = _count_row(idx) - _save_snapshot(snapshot) - recorded.add(idx) - else: - if outcome["success"] or outcome["error"] is not None: - recorded.add(outcome["idx"]) - pbar.update(1) - except KeyboardInterrupt: - # Ctrl-C: a saída do executor espera as linhas em andamento, e elas chegam - # aqui registradas. Um segundo Ctrl-C durante essa espera deixa threads - # rodando: a cópia sob `lock` leva só as linhas que terminaram, e as - # demais ficam pendentes para a retomada. Pelo rótulo, o _SnapshotWriter - # não regrava o arquivo sem linha contada desde o último snapshot gravado. + A saída do executor espera as linhas em andamento, e elas chegam aqui + registradas. Um segundo Ctrl-C durante essa espera deixa threads + rodando: a cópia sob `lock` leva só as linhas que terminaram, e as + demais ficam pendentes para a retomada. Pelo rótulo, o _SnapshotWriter + não regrava o arquivo sem linha contada desde o último snapshot gravado. + """ if writer: with lock: frame, label = df.copy(), checkpoint_counter writer.save(frame, label) - raise + + # Processar com ThreadPoolExecutor + with _on_ctrl_c(save_registered_rows), tqdm(total=len(rows_to_process), desc=desc) as pbar: + # Usar abordagem iterativa para permitir ajuste dinâmico de workers + pending_rows = list(rows_to_process) + + while pending_rows and not abort_event.is_set(): + # Pegar batch com número atual de workers + with lock: + worker_batch = min(current_workers, len(pending_rows)) + batch = pending_rows[:worker_batch] + pending_rows = pending_rows[worker_batch:] + + with ThreadPoolExecutor(max_workers=worker_batch) as executor: + futures = {} + for row in batch: + # A linha que não foi despachada nunca rodou: falhar ao + # despachar interrompe a execução e a deixa sem status, para + # a retomada. Como 'error', o resume=True a preservaria. + try: + futures[executor.submit(process_single_row, row)] = row + except Exception as e: # noqa: BLE001 (ver o comentário acima) + with lock: + token_stats.setdefault("abort", _error_text(e)) + abort_event.set() + break + + for future in as_completed(futures): + try: + outcome = future.result() + except Exception as e: # noqa: BLE001 (falha fora da linha não para a execução) + # Escapou do tratamento da própria linha; sem isto ela + # ficaria sem status e sumiria da saída. + _i, idx, _row = futures[future] + if idx in counted: + warnings.warn( + f"Erro inesperado no executor depois de registrar a linha " + f"{idx}, que mantém o status gravado: {_error_text(e)}", + stacklevel=1, + ) + else: + with lock: + _record_unrecorded_error(df, idx, e, status_col=status_col) + snapshot = _count_row(idx) + _save_snapshot(snapshot) + recorded.add(idx) + else: + if outcome["success"] or outcome["error"] is not None: + recorded.add(outcome["idx"]) + pbar.update(1) if abort_event.is_set(): _mark_unreached( From b0f2f977ac73043b045e5c75e80c69249ec13e77 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 12:19:40 -0300 Subject: [PATCH 07/16] =?UTF-8?q?La=C3=A7o=20paralelo=20espera=20as=20linh?= =?UTF-8?q?as=20em=20fatias=20curtas,=20e=20o=20segundo=20Ctrl-C=20corta?= =?UTF-8?q?=20a=20espera=20com=20o=20polars=20importado?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O polars, ao ser importado, troca o handler C do SIGINT por um com SA_RESTART. Com ele, o kernel reinicia a espera sem prazo do as_completed e do join na saída do with ThreadPoolExecutor, e o KeyboardInterrupt só saía quando a linha lenta terminava. A espera agora usa wait com prazo curto, e o thread principal volta ao interpretador a cada fatia. O teste do segundo Ctrl-C deixa de trocar o __exit__ do executor: põe SA_RESTART no SIGINT, como o polars, e manda o segundo Ctrl-C enquanto a interrupção espera a chamada lenta. --- src/dataframeit/core.py | 53 +++++++++++++++++++++++++++++---- tests/test_interrupcao.py | 62 +++++++++++++++++++++++++++++---------- 2 files changed, 93 insertions(+), 22 deletions(-) diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index 6395bff7..0deb500d 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -14,7 +14,7 @@ import threading import time import warnings -from concurrent.futures import ThreadPoolExecutor, as_completed +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait from contextlib import contextmanager from dataclasses import dataclass from decimal import Decimal @@ -73,6 +73,7 @@ if TYPE_CHECKING: from collections.abc import Callable, Collection, Hashable, Iterator, Sequence + from concurrent.futures import Future from polars import DataFrame as PolarsDataFrame from polars import Series as PolarsSeries @@ -157,6 +158,10 @@ def _install_fallback_handler(logger: logging.Logger) -> None: # Chamadas de busca por campo acima das quais vale avisar sobre rate limit. _PER_FIELD_SEARCH_WARNING_CALLS = 100 +# Prazo de cada espera do laço paralelo pelas linhas em andamento; ver +# _completed_in_slices. +_WAIT_SLICE_SECONDS = 0.1 + # Contadores de uso somados de cada linha para o resumo. _TOKEN_STAT_KEYS = ( "input_tokens", @@ -2615,6 +2620,40 @@ def _on_ctrl_c(callback: Callable[[], None]) -> Iterator[None]: raise +def _completed_in_slices(futures: Collection[Future]) -> Iterator[Future]: + """Entrega os futuros à medida que terminam, esperando em fatias curtas. + + O polars, ao ser importado, instala um handler de SIGINT com SA_RESTART, e + com ele o kernel reinicia a espera sem prazo de um lock em vez de + interrompê-la. O `as_completed` e a saída do `with ThreadPoolExecutor` + esperam sem prazo, e com eles o Ctrl-C só vira `KeyboardInterrupt` quando + uma linha termina. Ao fim de cada fatia, o thread principal volta ao + interpretador, que levanta o `KeyboardInterrupt` pendente. + """ + pending = set(futures) + while pending: + done, pending = wait(pending, timeout=_WAIT_SLICE_SECONDS, return_when=FIRST_COMPLETED) + yield from done + + +@contextmanager +def _closing_in_slices( + executor: ThreadPoolExecutor, futures: Collection[Future] +) -> Iterator[ThreadPoolExecutor]: + """Na saída, encerra o executor e espera os futuros em fatias. + + Faz as vezes do `with ThreadPoolExecutor`, cuja saída espera as threads sem + prazo: com a espera em fatias, um segundo Ctrl-C a corta; ver + _completed_in_slices. `futures` é lido na saída, com o que o bloco despachou. + """ + try: + yield executor + finally: + executor.shutdown(wait=False) + for _future in _completed_in_slices(futures): + pass + + def _process_rows_parallel( # noqa: C901, PLR0913, PLR0915 (estado compartilhado entre threads) df: pd.DataFrame, *, @@ -2814,8 +2853,8 @@ def process_single_row(row_data: tuple) -> dict: def save_registered_rows() -> None: """Grava o checkpoint no Ctrl-C. - A saída do executor espera as linhas em andamento, e elas chegam aqui - registradas. Um segundo Ctrl-C durante essa espera deixa threads + O lote espera as linhas em andamento antes de chegar aqui, e elas + chegam registradas. Um segundo Ctrl-C durante essa espera deixa threads rodando: a cópia sob `lock` leva só as linhas que terminaram, e as demais ficam pendentes para a retomada. Pelo rótulo, o _SnapshotWriter não regrava o arquivo sem linha contada desde o último snapshot gravado. @@ -2837,8 +2876,10 @@ def save_registered_rows() -> None: batch = pending_rows[:worker_batch] pending_rows = pending_rows[worker_batch:] - with ThreadPoolExecutor(max_workers=worker_batch) as executor: - futures = {} + futures = {} + with _closing_in_slices( + ThreadPoolExecutor(max_workers=worker_batch), futures + ) as executor: for row in batch: # A linha que não foi despachada nunca rodou: falhar ao # despachar interrompe a execução e a deixa sem status, para @@ -2851,7 +2892,7 @@ def save_registered_rows() -> None: abort_event.set() break - for future in as_completed(futures): + for future in _completed_in_slices(futures): try: outcome = future.result() except Exception as e: # noqa: BLE001 (falha fora da linha não para a execução) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 2c95069f..7914a123 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -1,7 +1,9 @@ """Falha que impede as linhas seguintes interrompe a execução sem gastá-las.""" import signal +import sys import threading +import time from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch @@ -327,33 +329,58 @@ def test_falha_ao_gravar_no_ctrl_c_avisa_e_mantem_o_keyboard_interrupt(tmp_path, assert grava.call_count == 1 +@pytest.fixture +def sigint_reinicia_a_espera(): + """Põe SA_RESTART no SIGINT, como o handler que o polars instala ao ser importado. + + Com SA_RESTART, o kernel reinicia a espera sem prazo de um lock quando o + SIGINT chega, e o thread principal só levanta o `KeyboardInterrupt` quando a + espera termina. Sem isso, o SIGINT interromperia também a espera sem prazo, + e o teste não a distinguiria da espera em fatias. Com o polars importado, o + SA_RESTART já estava lá e fica. + """ + com_polars = "polars" in sys.modules + signal.siginterrupt(signal.SIGINT, False) + yield + if not com_polars: + signal.siginterrupt(signal.SIGINT, True) + + @requer_pthread_kill -def test_segundo_ctrl_c_grava_so_as_linhas_concluidas(tmp_path): - """O segundo Ctrl-C corta a espera do executor, e a chamada que não terminou fica pendente.""" +@pytest.mark.usefixtures("sigint_reinicia_a_espera") +def test_segundo_ctrl_c_corta_a_espera_e_a_chamada_lenta_fica_pendente(tmp_path): + """O segundo Ctrl-C chega enquanto a interrupção espera a chamada lenta. + + A espera acaba sem a chamada terminar, o checkpoint leva só a linha + concluída, e a retomada chama a lenta de novo. + """ textos = ["lenta", "ctrl-c"] ckpt = tmp_path / "ckpt.csv" + encerrando = threading.Event() liberada = threading.Event() + terminou = threading.Event() + encerra = ThreadPoolExecutor.shutdown + + def encerra_observado(self, *args, **kwargs): + encerrando.set() + return encerra(self, *args, **kwargs) def llm(texto, *args, **kwargs): if texto == "lenta": + # O primeiro Ctrl-C encerra o executor antes de esperar esta + # chamada; a folga deixa a espera começar antes do segundo. + encerrando.wait(timeout=10) + time.sleep(0.2) + _ctrl_c() liberada.wait(timeout=10) + terminou.set() else: _ctrl_c() return {"data": {"campo1": f"r-{texto}"}, "usage": {}} - saida_do_executor = ThreadPoolExecutor.__exit__ - - def segundo_ctrl_c_na_espera(self, tipo, valor, rastro): - # A saída por Ctrl-C espera as threads; o segundo Ctrl-C chega nessa - # espera, depois de o executor ter sido encerrado sem esperar. - if tipo is KeyboardInterrupt: - self.shutdown(wait=False) - _ctrl_c() - return saida_do_executor(self, tipo, valor, rastro) - try: with ( - patch.object(ThreadPoolExecutor, "__exit__", segundo_ctrl_c_na_espera), + patch.object(ThreadPoolExecutor, "shutdown", encerra_observado), pytest.raises(KeyboardInterrupt), ): _roda( @@ -363,12 +390,15 @@ def segundo_ctrl_c_na_espera(self, tipo, valor, rastro): batch_size=100, checkpoint_path=ckpt, ) + cortada = not terminou.is_set() gravado = pd.read_csv(ckpt) finally: liberada.set() - processadas = set(gravado.loc[gravado["_dataframeit_status"].eq("processed"), "texto"]) - assert "lenta" not in processadas + assert cortada + status = dict(zip(gravado["texto"], gravado["_dataframeit_status"], strict=True)) + assert status["ctrl-c"] == "processed" + assert pd.isna(status["lenta"]) retomada, llm_ok = _llm_com_ctrl_c() _roda( @@ -378,7 +408,7 @@ def segundo_ctrl_c_na_espera(self, tipo, valor, rastro): batch_size=100, checkpoint_path=ckpt, ) - assert sorted(retomada) == sorted(set(textos) - processadas) + assert retomada == ["lenta"] @pytest.mark.parametrize("paralelo", [1, 3]) From 600ef00fd1b0f7a8e3fb55dfd9aee6d4941c0873 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 12:19:40 -0300 Subject: [PATCH 08/16] CHANGELOG: a espera do Ctrl-C no paralelo corre em fatias curtas por causa do handler de SIGINT do polars --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f63fe3b1..d5f5260c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,7 +40,7 @@ e este projeto adere ao [Versionamento Semântico](https://semver.org/lang/pt-BR ### Corrigido -- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas em andamento que terminam enquanto o executor as espera; as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A falha ao gravar vira aviso e não troca a interrupção (#213). +- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas em andamento que terminam enquanto a interrupção as espera; as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A espera pelas linhas corre em fatias curtas, porque o handler de SIGINT que o polars instala ao ser importado faz o kernel reiniciar a espera sem prazo: com o polars instalado, o Ctrl-C do paralelo só chegava quando uma linha terminava, e o segundo não cortava a espera. A falha ao gravar vira aviso e não troca a interrupção (#213). - No provider `codex`, a thread daemon que pede o `thread/unsubscribe` da linha abortada pela morte do app-server deixa de ficar presa para sempre. O processo morto ainda aceita a escrita enquanto termina, e o SDK registra a espera da resposta depois de o leitor do stdout ter acordado os pedidos em voo; agora o fechamento do cliente acorda essa espera (#250). - `read_df` com argumento de leitura que faz o pandas devolver outra coisa que um DataFrame, como `chunksize`, `iterator`, `sheet_name=None` ou lista e o `chunksize` do JSON em linhas, devolve o `TextFileReader`, o `JsonReader` ou o dict de abas como o pandas o devolve, sem normalizar, com ou sem modelo. Antes, sem modelo, ou com modelo que tem campo de lista, dict ou modelo aninhado, e com `normalize=True`, o padrão, a função levantava `AttributeError` ao procurar as colunas no objeto do leitor (#249). - Com entrada polars, a saída deixa de trocar sem aviso o valor que o modelo devolveu quando o pyarrow tipa a coluna pelo primeiro valor e converte os demais. Sai `Object`, com o aviso das colunas que o polars não reúne, o campo cuja anotação junta `timedelta` a outro tipo, `float` a `bool` ou `bytes` a `str`, e a coluna cujos valores trazem essas combinações ou as de data e hora, também em `Any`, onde o escalar do numpy, como o `np.int64`, conta como o número do Python, nos itens de lista e no campo de modelo aninhado, além da coluna cujos dicts do mesmo campo têm chaves diferentes e da que traz `datetime` com e sem fuso, `time` com fuso ou `Decimal` NaN. O NaN de campo `float` sai NaN. Antes, em `Union[timedelta, int]` o 5 virava cinco microssegundos, em `Union[float, bool]` o `True` virava `1.0`, o `datetime` sem fuso ganhava o fuso do primeiro, o `time` perdia o fuso, o `Decimal` NaN e o NaN de campo `float` viravam nulo, `{}` ao lado de `{"a": 1}` virava `{"a": None}`, e em `Union[NewType sobre date, int]` o 5 virava `1970-01-06`. Com `resume=True`, a coluna do campo cuja anotação é `int`, um `Literal` com inteiro ou uma união com um deles é lida com os inteiros do polars, e o inteiro acima de 2**53 deixa de voltar arredondado em `Float64`. O `Decimal` NaN e o NaN de campo `float` de uma linha processada deixam de ser acusados como incompatíveis na retomada, também com entrada pandas, e o reprocessamento campo a campo os repassa às condições como valor; na coluna float64, como a do checkpoint relido, o NaN não se distingue do nulo e segue contando como ausente. O `NewType` e o apelido `type`, inclusive o apelido genérico usado com argumentos, como `Opcional[int]` de `type Opcional[T] = T | None`, passam a contar pelo tipo a que se referem na saída polars, na retomada e em `read_df`: o `NewType` sobre `dict` sai em texto JSON, como o campo `dict`, e é retomado, e o campo `NewType` sobre `str` ou sobre `int` é relido do CSV como texto cru ou como inteiro exato (#241). From 528c799a6f0b493198e16b57d13e86bff422796a Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 13:01:17 -0300 Subject: [PATCH 09/16] =?UTF-8?q?Ctrl-C=20no=20paralelo=20espera=20as=20th?= =?UTF-8?q?reads=20do=20executor,=20e=20a=20linha=20alcan=C3=A7ada=20dentr?= =?UTF-8?q?o=20do=20submit=20entra=20no=20checkpoint=20(#213)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O Ctrl-C pode sair de dentro do executor.submit depois de o item entrar na fila e de a thread começar, antes de o laço guardar o futuro. A linha rodava e era paga, mas a saída do lote esperava só os futuros guardados, gravava o checkpoint sem ela, e a retomada a chamava de novo. A saída do lote agora encerra o executor e espera, em fatias com prazo, as threads dele, achadas pelo prefixo de nome do executor em threading.enumerate(). Cada futuro guardado roda numa dessas threads, e a espera pelos futuros sai. O teste novo manda o Ctrl-C ao fim do _adjust_thread_count da última linha e confere que ela fica processed no checkpoint e que a retomada não a chama. --- CHANGELOG.md | 2 +- src/dataframeit/core.py | 43 ++++++++++++++++++++--------- tests/test_interrupcao.py | 57 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d5f5260c..3224a6e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,7 +40,7 @@ e este projeto adere ao [Versionamento Semântico](https://semver.org/lang/pt-BR ### Corrigido -- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas em andamento que terminam enquanto a interrupção as espera; as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A espera pelas linhas corre em fatias curtas, porque o handler de SIGINT que o polars instala ao ser importado faz o kernel reiniciar a espera sem prazo: com o polars instalado, o Ctrl-C do paralelo só chegava quando uma linha terminava, e o segundo não cortava a espera. A falha ao gravar vira aviso e não troca a interrupção (#213). +- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas já despachadas que terminam enquanto a interrupção espera as threads do executor, também a que o Ctrl-C alcança dentro do despacho; as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A espera pelas linhas corre em fatias curtas, porque o handler de SIGINT que o polars instala ao ser importado faz o kernel reiniciar a espera sem prazo: com o polars instalado, o Ctrl-C do paralelo só chegava quando uma linha terminava, e o segundo não cortava a espera. A falha ao gravar vira aviso e não troca a interrupção (#213). - No provider `codex`, a thread daemon que pede o `thread/unsubscribe` da linha abortada pela morte do app-server deixa de ficar presa para sempre. O processo morto ainda aceita a escrita enquanto termina, e o SDK registra a espera da resposta depois de o leitor do stdout ter acordado os pedidos em voo; agora o fechamento do cliente acorda essa espera (#250). - `read_df` com argumento de leitura que faz o pandas devolver outra coisa que um DataFrame, como `chunksize`, `iterator`, `sheet_name=None` ou lista e o `chunksize` do JSON em linhas, devolve o `TextFileReader`, o `JsonReader` ou o dict de abas como o pandas o devolve, sem normalizar, com ou sem modelo. Antes, sem modelo, ou com modelo que tem campo de lista, dict ou modelo aninhado, e com `normalize=True`, o padrão, a função levantava `AttributeError` ao procurar as colunas no objeto do leitor (#249). - Com entrada polars, a saída deixa de trocar sem aviso o valor que o modelo devolveu quando o pyarrow tipa a coluna pelo primeiro valor e converte os demais. Sai `Object`, com o aviso das colunas que o polars não reúne, o campo cuja anotação junta `timedelta` a outro tipo, `float` a `bool` ou `bytes` a `str`, e a coluna cujos valores trazem essas combinações ou as de data e hora, também em `Any`, onde o escalar do numpy, como o `np.int64`, conta como o número do Python, nos itens de lista e no campo de modelo aninhado, além da coluna cujos dicts do mesmo campo têm chaves diferentes e da que traz `datetime` com e sem fuso, `time` com fuso ou `Decimal` NaN. O NaN de campo `float` sai NaN. Antes, em `Union[timedelta, int]` o 5 virava cinco microssegundos, em `Union[float, bool]` o `True` virava `1.0`, o `datetime` sem fuso ganhava o fuso do primeiro, o `time` perdia o fuso, o `Decimal` NaN e o NaN de campo `float` viravam nulo, `{}` ao lado de `{"a": 1}` virava `{"a": None}`, e em `Union[NewType sobre date, int]` o 5 virava `1970-01-06`. Com `resume=True`, a coluna do campo cuja anotação é `int`, um `Literal` com inteiro ou uma união com um deles é lida com os inteiros do polars, e o inteiro acima de 2**53 deixa de voltar arredondado em `Float64`. O `Decimal` NaN e o NaN de campo `float` de uma linha processada deixam de ser acusados como incompatíveis na retomada, também com entrada pandas, e o reprocessamento campo a campo os repassa às condições como valor; na coluna float64, como a do checkpoint relido, o NaN não se distingue do nulo e segue contando como ausente. O `NewType` e o apelido `type`, inclusive o apelido genérico usado com argumentos, como `Opcional[int]` de `type Opcional[T] = T | None`, passam a contar pelo tipo a que se referem na saída polars, na retomada e em `read_df`: o `NewType` sobre `dict` sai em texto JSON, como o campo `dict`, e é retomado, e o campo `NewType` sobre `str` ou sobre `int` é relido do CSV como texto cru ou como inteiro exato (#241). diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index 0deb500d..193a3f28 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -6,6 +6,7 @@ import datetime import hashlib import importlib.util +import itertools import json import logging import numbers @@ -162,6 +163,10 @@ def _install_fallback_handler(logger: logging.Logger) -> None: # _completed_in_slices. _WAIT_SLICE_SECONDS = 0.1 +# Numera os executores do laço paralelo, para que o nome das threads de cada um +# não se confunda com o das de outro; ver _closing_in_slices. +_executor_ids = itertools.count() + # Contadores de uso somados de cada linha para o resumo. _TOKEN_STAT_KEYS = ( "input_tokens", @@ -2637,21 +2642,37 @@ def _completed_in_slices(futures: Collection[Future]) -> Iterator[Future]: @contextmanager -def _closing_in_slices( - executor: ThreadPoolExecutor, futures: Collection[Future] -) -> Iterator[ThreadPoolExecutor]: - """Na saída, encerra o executor e espera os futuros em fatias. +def _closing_in_slices(max_workers: int) -> Iterator[ThreadPoolExecutor]: + """Abre um executor e, na saída, o encerra e espera as threads dele em fatias. Faz as vezes do `with ThreadPoolExecutor`, cuja saída espera as threads sem - prazo: com a espera em fatias, um segundo Ctrl-C a corta; ver - _completed_in_slices. `futures` é lido na saída, com o que o bloco despachou. + prazo: com prazo em cada espera, um segundo Ctrl-C a corta; ver + _completed_in_slices. A espera é pelas threads, e não pelos futuros que o + bloco guardou, porque o Ctrl-C pode sair de dentro do `submit` depois de o + item entrar na fila. A linha roda e é paga sem que o futuro dela chegue ao + bloco, e a gravação do checkpoint precisa encontrá-la registrada. Cada + futuro guardado roda numa dessas threads, que só saem depois de esvaziar a + fila. + + As threads são achadas pelo nome, com o prefixo deste executor, em + `threading.enumerate()`. O conjunto `_threads` do executor é privado e só + recebe a thread depois de `Thread.start()` voltar; `enumerate` a lista + desde antes. A lista se refaz a cada fatia, para pegar a thread que começa + a rodar durante a espera, e `is_alive` deixa de fora a que o Ctrl-C + impediu de começar, que nunca terminaria. """ + prefix = f"dataframeit-lote{next(_executor_ids)}" + executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix=prefix) try: yield executor finally: executor.shutdown(wait=False) - for _future in _completed_in_slices(futures): - pass + while alive := [ + thread + for thread in threading.enumerate() + if thread.name.startswith(f"{prefix}_") and thread.is_alive() + ]: + alive[0].join(_WAIT_SLICE_SECONDS) def _process_rows_parallel( # noqa: C901, PLR0913, PLR0915 (estado compartilhado entre threads) @@ -2876,10 +2897,8 @@ def save_registered_rows() -> None: batch = pending_rows[:worker_batch] pending_rows = pending_rows[worker_batch:] - futures = {} - with _closing_in_slices( - ThreadPoolExecutor(max_workers=worker_batch), futures - ) as executor: + with _closing_in_slices(worker_batch) as executor: + futures = {} for row in batch: # A linha que não foi despachada nunca rodou: falhar ao # despachar interrompe a execução e a deixa sem status, para diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 7914a123..7e101034 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -411,6 +411,63 @@ def llm(texto, *args, **kwargs): assert retomada == ["lenta"] +@requer_pthread_kill +def test_ctrl_c_dentro_do_submit_espera_a_linha_que_ja_entrou_na_fila(tmp_path): + """O Ctrl-C chega dentro do `submit` da última linha, que já entrou na fila. + + A linha roda e é paga, embora o laço não tenha guardado o futuro dela. A + interrupção espera as threads do executor, e a linha entra no checkpoint. + """ + textos = ["a", "b", "lenta"] + ckpt = tmp_path / "ckpt.csv" + pagas, llm_rapido = _llm_com_ctrl_c() + + def llm(texto, *args, **kwargs): + if texto == "lenta": + # Termina depois da gravação que esperasse só os futuros guardados. + time.sleep(0.5) + return llm_rapido(texto, *args, **kwargs) + + ajusta = ThreadPoolExecutor._adjust_thread_count + despachos: list[int] = [] + + def ajusta_e_interrompe(self): + # O submit põe o item na fila e chama este método, que inicia a thread + # quando falta uma; o KeyboardInterrupt do pthread_kill sai do submit + # antes de o laço guardar o futuro. + ajusta(self) + despachos.append(1) + if len(despachos) == len(textos): + _ctrl_c() + + with ( + patch.object(ThreadPoolExecutor, "_adjust_thread_count", ajusta_e_interrompe), + pytest.raises(KeyboardInterrupt), + ): + _roda( + pd.DataFrame({"texto": textos}), + llm, + parallel_requests=3, + batch_size=100, + checkpoint_path=ckpt, + ) + + gravado = pd.read_csv(ckpt) + status = dict(zip(gravado["texto"], gravado["_dataframeit_status"], strict=True)) + assert status == dict.fromkeys(textos, "processed") + assert sorted(pagas) == sorted(textos) + + retomada, llm_ok = _llm_com_ctrl_c() + _roda( + pd.DataFrame({"texto": textos}), + llm_ok, + parallel_requests=3, + batch_size=100, + checkpoint_path=ckpt, + ) + assert retomada == [] + + @pytest.mark.parametrize("paralelo", [1, 3]) def test_ctrl_c_antes_de_terminar_uma_linha_nao_reescreve_o_checkpoint(tmp_path, paralelo): """Sem linha registrada desde a última gravação, o arquivo fica como estava.""" From 106f37913c4741fcc87d016eaea04db6be56cbb4 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 13:03:53 -0300 Subject: [PATCH 10/16] Teste do Ctrl-C com todas as linhas do lote lentas: o lote sai sem esperar uma terminar (#213) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No teste do segundo Ctrl-C, a linha que manda o primeiro termina logo depois, e a espera do lote devolve sem precisar de prazo. Voltar o laço do lote ao as_completed passava a suíte e trazia de volta o defeito com o polars: os dois Ctrl-C se fundiam e o KeyboardInterrupt só saía quando uma linha terminava. O caso novo deixa as três linhas em andamento até 5 s, manda os dois Ctrl-C com o SA_RESTART da fixture e afirma que o lote sai antes disso, sem linha terminada. Os Ctrl-C saem de _ctrl_c_depois, que cancela o envio que falta e espera o que está em curso na saída do bloco. --- tests/test_interrupcao.py | 84 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 7e101034..9bb1215c 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -4,7 +4,9 @@ import sys import threading import time +from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager from unittest.mock import patch import pandas as pd @@ -250,6 +252,42 @@ def llm(texto, *args, **kwargs): return pagas, llm +@contextmanager +def _ctrl_c_depois(gatilho: threading.Event, *folgas: float) -> Iterator[None]: + """Manda um Ctrl-C a cada folga, contada desde `gatilho`, enquanto o bloco roda. + + Na saída do bloco, o Ctrl-C que falta é cancelado e o envio em curso é + esperado: o `KeyboardInterrupt` cai dentro do teste, nunca num teste adiante. + """ + saiu = threading.Event() + trava = threading.Lock() + + def envia() -> None: + while not gatilho.wait(0.05): + if saiu.is_set(): + return + inicio = time.monotonic() + for folga in folgas: + if saiu.wait(max(0.0, inicio + folga - time.monotonic())): + return + with trava: + if saiu.is_set(): + return + _ctrl_c() + + remetente = threading.Thread(target=envia, daemon=True) + remetente.start() + try: + yield + finally: + # `saiu` vem antes da trava: um KeyboardInterrupt que chegue enquanto o + # thread principal espera a trava já não deixa outro envio para depois. + saiu.set() + with trava: + pass + remetente.join() + + def _entrada(biblioteca: str, textos: list[str]): if biblioteca == "polars": pl = pytest.importorskip("polars") @@ -411,6 +449,52 @@ def llm(texto, *args, **kwargs): assert retomada == ["lenta"] +@requer_pthread_kill +@pytest.mark.usefixtures("sigint_reinicia_a_espera") +def test_ctrl_c_com_todas_as_linhas_lentas_sai_sem_esperar_uma_terminar(tmp_path): + """Os dois Ctrl-C chegam com as três linhas em andamento e nenhuma termina. + + O primeiro vira `KeyboardInterrupt` sem esperar uma linha terminar, e o + segundo corta a espera da saída: o lote sai antes da duração das linhas. + """ + duracao = 5.0 + textos = ["a", "b", "c"] + trava = threading.Lock() + comecadas: list[str] = [] + terminadas: list[str] = [] + todas_comecaram = threading.Event() + liberada = threading.Event() + + def llm(texto, *args, **kwargs): + with trava: + comecadas.append(texto) + if len(comecadas) == len(textos): + todas_comecaram.set() + liberada.wait(timeout=duracao) + with trava: + terminadas.append(texto) + return {"data": {"campo1": f"r-{texto}"}, "usage": {}} + + inicio = time.monotonic() + try: + with _ctrl_c_depois(todas_comecaram, 0.0, 0.3), pytest.raises(KeyboardInterrupt): + _roda( + pd.DataFrame({"texto": textos}), + llm, + parallel_requests=3, + batch_size=100, + checkpoint_path=tmp_path / "ckpt.csv", + ) + saida = time.monotonic() - inicio + with trava: + terminadas_na_saida = list(terminadas) + finally: + liberada.set() + + assert saida < duracao + assert terminadas_na_saida == [] + + @requer_pthread_kill def test_ctrl_c_dentro_do_submit_espera_a_linha_que_ja_entrou_na_fila(tmp_path): """O Ctrl-C chega dentro do `submit` da última linha, que já entrou na fila. From 6a8271dcea95d13727462388b9818841a8c8d216 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 13:15:15 -0300 Subject: [PATCH 11/16] Teste da linha em andamento no Ctrl-C do paralelo, e o segundo Ctrl-C do teste nunca dispara fora dele (#213) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tirar a espera da saída do lote não reprovava asserção nenhuma: os testes do Ctrl-C no paralelo usam chamadas que terminam antes da gravação, e a suíte só caía porque o segundo Ctrl-C do teste chegava depois do fim dele e interrompia outro teste. O caso novo deixa duas linhas lentas em andamento quando a terceira manda o Ctrl-C. Elas só terminam depois de o executor ser encerrado, e o teste afirma que entram como processed no checkpoint, que a retomada não as chama e que nenhuma thread não daemon criada pelo lote sobrevive à interrupção. O teste do segundo Ctrl-C passa a mandá-lo por _ctrl_c_depois, que cancela o envio na saída do bloco. --- tests/test_interrupcao.py | 70 ++++++++++++++++++++++++++++++++++++--- 1 file changed, 65 insertions(+), 5 deletions(-) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 9bb1215c..0bb8a6a6 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -405,11 +405,6 @@ def encerra_observado(self, *args, **kwargs): def llm(texto, *args, **kwargs): if texto == "lenta": - # O primeiro Ctrl-C encerra o executor antes de esperar esta - # chamada; a folga deixa a espera começar antes do segundo. - encerrando.wait(timeout=10) - time.sleep(0.2) - _ctrl_c() liberada.wait(timeout=10) terminou.set() else: @@ -417,8 +412,11 @@ def llm(texto, *args, **kwargs): return {"data": {"campo1": f"r-{texto}"}, "usage": {}} try: + # O primeiro Ctrl-C encerra o executor antes de esperar a chamada + # lenta; a folga deixa a espera começar antes do segundo. with ( patch.object(ThreadPoolExecutor, "shutdown", encerra_observado), + _ctrl_c_depois(encerrando, 0.2), pytest.raises(KeyboardInterrupt), ): _roda( @@ -495,6 +493,68 @@ def llm(texto, *args, **kwargs): assert terminadas_na_saida == [] +@requer_pthread_kill +def test_ctrl_c_espera_a_linha_em_andamento_e_nao_deixa_thread_do_lote(tmp_path): + """As linhas lentas em andamento no Ctrl-C terminam na espera e entram no checkpoint. + + Elas só terminam depois de o executor ser encerrado, e nenhuma thread do + lote sobrevive à interrupção. + """ + textos = ["lenta1", "ctrl-c", "lenta2"] + ckpt = tmp_path / "ckpt.csv" + antes = set(threading.enumerate()) + trava = threading.Lock() + comecadas: list[str] = [] + todas_comecaram = threading.Event() + encerrando = threading.Event() + encerra = ThreadPoolExecutor.shutdown + + def encerra_observado(self, *args, **kwargs): + encerrando.set() + return encerra(self, *args, **kwargs) + + def llm(texto, *args, **kwargs): + with trava: + comecadas.append(texto) + if len(comecadas) == len(textos): + todas_comecaram.set() + if texto == "ctrl-c": + todas_comecaram.wait(timeout=10) + _ctrl_c() + else: + encerrando.wait(timeout=10) + time.sleep(0.2) + return {"data": {"campo1": f"r-{texto}"}, "usage": {}} + + with ( + patch.object(ThreadPoolExecutor, "shutdown", encerra_observado), + pytest.raises(KeyboardInterrupt), + ): + _roda( + pd.DataFrame({"texto": textos}), + llm, + parallel_requests=3, + batch_size=100, + checkpoint_path=ckpt, + ) + sobreviventes = [t for t in set(threading.enumerate()) - antes if not t.daemon] + + gravado = pd.read_csv(ckpt) + status = dict(zip(gravado["texto"], gravado["_dataframeit_status"], strict=True)) + assert status == dict.fromkeys(textos, "processed") + assert sobreviventes == [] + + retomada, llm_ok = _llm_com_ctrl_c() + _roda( + pd.DataFrame({"texto": textos}), + llm_ok, + parallel_requests=3, + batch_size=100, + checkpoint_path=ckpt, + ) + assert retomada == [] + + @requer_pthread_kill def test_ctrl_c_dentro_do_submit_espera_a_linha_que_ja_entrou_na_fila(tmp_path): """O Ctrl-C chega dentro do `submit` da última linha, que já entrou na fila. From e2ead46cb5e47fcbcbd062db6882503baa4a1c0d Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 13:15:25 -0300 Subject: [PATCH 12/16] =?UTF-8?q?Docstring=20de=20=5Fcompleted=5Fin=5Fslic?= =?UTF-8?q?es:=20o=20que=20corta=20a=20espera=20=C3=A9=20haver=20prazo,=20?= =?UTF-8?q?n=C3=A3o=20o=20fim=20da=20fatia=20(#213)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No Linux, a espera com prazo de um lock não é reiniciada nem com SA_RESTART e volta com EINTR assim que o SIGINT chega; com fatia de 30 s, o KeyboardInterrupt sai no mesmo instante que com 0,1 s. A docstring dizia que o thread voltava ao interpretador ao fim de cada fatia. --- src/dataframeit/core.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index 193a3f28..3a91731f 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -2632,8 +2632,11 @@ def _completed_in_slices(futures: Collection[Future]) -> Iterator[Future]: com ele o kernel reinicia a espera sem prazo de um lock em vez de interrompê-la. O `as_completed` e a saída do `with ThreadPoolExecutor` esperam sem prazo, e com eles o Ctrl-C só vira `KeyboardInterrupt` quando - uma linha termina. Ao fim de cada fatia, o thread principal volta ao - interpretador, que levanta o `KeyboardInterrupt` pendente. + uma linha termina. O que corta a espera é haver prazo, e não a duração da + fatia: no Linux, a espera com prazo de um semáforo não é reiniciada nem + com SA_RESTART (ver signal(7)) e volta com EINTR assim que o sinal chega, e + o interpretador levanta o `KeyboardInterrupt` na hora. A fatia curta só + limita a demora numa plataforma que reinicie também a espera com prazo. """ pending = set(futures) while pending: From f91f3b808d811dc0141a435c89052196258ad16d Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 14:15:21 -0300 Subject: [PATCH 13/16] Ctrl-C dentro do Thread.start() do paralelo: o encerramento cancela a linha da fila, que fica para a retomada (#213) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sem o polars, o SIGINT interrompe o _started.wait() do Thread.start() que o submit chama para a primeira thread do lote. A thread já foi lançada, mas is_alive() a dá como parada, e a saída do lote não a esperava: o checkpoint era gravado, e logo depois a thread começava, pegava a linha da fila e a pagava fora dele. O encerramento passa a usar cancel_futures=True. A fila é esvaziada antes de a thread começar, que encontra o sentinela e sai sem rodar a linha; a linha cancelada não é paga e fica sem status, para a retomada. As linhas que uma thread já tirou da fila terminam e entram no checkpoint, como antes. Fora do Ctrl-C a saída chega com a fila vazia, porque o bloco espera todos os futuros que guardou. O teste novo manda o Ctrl-C de dentro do _bootstrap_inner da primeira thread do lote, sem SA_RESTART, e afirma que nenhuma linha paga fica fora do checkpoint e que a retomada chama as demais. O teste do Ctrl-C dentro do submit passa a mandá-lo depois de a thread começar a linha lenta, que sem isso podia ser cancelada ainda na fila. --- CHANGELOG.md | 2 +- src/dataframeit/core.py | 27 ++++++---- tests/test_interrupcao.py | 106 ++++++++++++++++++++++++++++++++++---- 3 files changed, 114 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3224a6e8..53638561 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,7 +40,7 @@ e este projeto adere ao [Versionamento Semântico](https://semver.org/lang/pt-BR ### Corrigido -- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas já despachadas que terminam enquanto a interrupção espera as threads do executor, também a que o Ctrl-C alcança dentro do despacho; as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A espera pelas linhas corre em fatias curtas, porque o handler de SIGINT que o polars instala ao ser importado faz o kernel reiniciar a espera sem prazo: com o polars instalado, o Ctrl-C do paralelo só chegava quando uma linha terminava, e o segundo não cortava a espera. A falha ao gravar vira aviso e não troca a interrupção (#213). +- Com `checkpoint_path`, um Ctrl-C (`KeyboardInterrupt`) grava no checkpoint as linhas concluídas desde a última gravação e relança a interrupção, nos modos sequencial e paralelo. Antes, o arquivo ficava na última gravação de `batch_size`, e a retomada pagava de novo as linhas do lote em andamento. No paralelo, entram as chamadas já em andamento, que terminam enquanto a interrupção espera as threads do executor, também quando o Ctrl-C sai de dentro do despacho; a linha que ainda está na fila, sem thread que a tenha começado, é cancelada e fica para a retomada, e as que um segundo Ctrl-C deixa sem terminar ficam pendentes. A espera pelas linhas corre em fatias curtas, porque o handler de SIGINT que o polars instala ao ser importado faz o kernel reiniciar a espera sem prazo: com o polars instalado, o Ctrl-C do paralelo só chegava quando uma linha terminava, e o segundo não cortava a espera. A falha ao gravar vira aviso e não troca a interrupção (#213). - No provider `codex`, a thread daemon que pede o `thread/unsubscribe` da linha abortada pela morte do app-server deixa de ficar presa para sempre. O processo morto ainda aceita a escrita enquanto termina, e o SDK registra a espera da resposta depois de o leitor do stdout ter acordado os pedidos em voo; agora o fechamento do cliente acorda essa espera (#250). - `read_df` com argumento de leitura que faz o pandas devolver outra coisa que um DataFrame, como `chunksize`, `iterator`, `sheet_name=None` ou lista e o `chunksize` do JSON em linhas, devolve o `TextFileReader`, o `JsonReader` ou o dict de abas como o pandas o devolve, sem normalizar, com ou sem modelo. Antes, sem modelo, ou com modelo que tem campo de lista, dict ou modelo aninhado, e com `normalize=True`, o padrão, a função levantava `AttributeError` ao procurar as colunas no objeto do leitor (#249). - Com entrada polars, a saída deixa de trocar sem aviso o valor que o modelo devolveu quando o pyarrow tipa a coluna pelo primeiro valor e converte os demais. Sai `Object`, com o aviso das colunas que o polars não reúne, o campo cuja anotação junta `timedelta` a outro tipo, `float` a `bool` ou `bytes` a `str`, e a coluna cujos valores trazem essas combinações ou as de data e hora, também em `Any`, onde o escalar do numpy, como o `np.int64`, conta como o número do Python, nos itens de lista e no campo de modelo aninhado, além da coluna cujos dicts do mesmo campo têm chaves diferentes e da que traz `datetime` com e sem fuso, `time` com fuso ou `Decimal` NaN. O NaN de campo `float` sai NaN. Antes, em `Union[timedelta, int]` o 5 virava cinco microssegundos, em `Union[float, bool]` o `True` virava `1.0`, o `datetime` sem fuso ganhava o fuso do primeiro, o `time` perdia o fuso, o `Decimal` NaN e o NaN de campo `float` viravam nulo, `{}` ao lado de `{"a": 1}` virava `{"a": None}`, e em `Union[NewType sobre date, int]` o 5 virava `1970-01-06`. Com `resume=True`, a coluna do campo cuja anotação é `int`, um `Literal` com inteiro ou uma união com um deles é lida com os inteiros do polars, e o inteiro acima de 2**53 deixa de voltar arredondado em `Float64`. O `Decimal` NaN e o NaN de campo `float` de uma linha processada deixam de ser acusados como incompatíveis na retomada, também com entrada pandas, e o reprocessamento campo a campo os repassa às condições como valor; na coluna float64, como a do checkpoint relido, o NaN não se distingue do nulo e segue contando como ausente. O `NewType` e o apelido `type`, inclusive o apelido genérico usado com argumentos, como `Opcional[int]` de `type Opcional[T] = T | None`, passam a contar pelo tipo a que se referem na saída polars, na retomada e em `read_df`: o `NewType` sobre `dict` sai em texto JSON, como o campo `dict`, e é retomado, e o campo `NewType` sobre `str` ou sobre `int` é relido do CSV como texto cru ou como inteiro exato (#241). diff --git a/src/dataframeit/core.py b/src/dataframeit/core.py index 3a91731f..2df8d7d4 100644 --- a/src/dataframeit/core.py +++ b/src/dataframeit/core.py @@ -2650,26 +2650,33 @@ def _closing_in_slices(max_workers: int) -> Iterator[ThreadPoolExecutor]: Faz as vezes do `with ThreadPoolExecutor`, cuja saída espera as threads sem prazo: com prazo em cada espera, um segundo Ctrl-C a corta; ver - _completed_in_slices. A espera é pelas threads, e não pelos futuros que o - bloco guardou, porque o Ctrl-C pode sair de dentro do `submit` depois de o - item entrar na fila. A linha roda e é paga sem que o futuro dela chegue ao - bloco, e a gravação do checkpoint precisa encontrá-la registrada. Cada - futuro guardado roda numa dessas threads, que só saem depois de esvaziar a - fila. + _completed_in_slices. O encerramento cancela as linhas que ainda estão na + fila, e nenhuma thread começa outra depois dele: a linha cancelada não é + paga e fica sem status, para a retomada. Isso cobre o Ctrl-C que sai de + dentro do `Thread.start()`, com a thread já lançada e ainda sem marcar o + início; sem o cancelamento, ela começaria depois da gravação do checkpoint + e pagaria fora dele a linha da fila. Fora do Ctrl-C, a saída chega com a + fila vazia, porque o bloco espera todos os futuros que guardou. + + A espera é pelas threads, e não pelos futuros que o bloco guardou, porque o + Ctrl-C pode sair de dentro do `submit` depois de uma thread tirar a linha + da fila. A linha roda e é paga sem que o futuro dela chegue ao bloco, e a + gravação do checkpoint precisa encontrá-la registrada. As threads são achadas pelo nome, com o prefixo deste executor, em `threading.enumerate()`. O conjunto `_threads` do executor é privado e só recebe a thread depois de `Thread.start()` voltar; `enumerate` a lista - desde antes. A lista se refaz a cada fatia, para pegar a thread que começa - a rodar durante a espera, e `is_alive` deixa de fora a que o Ctrl-C - impediu de começar, que nunca terminaria. + desde antes. `is_alive` deixa de fora a thread que ainda não marcou o + início, que `join` recusaria: ela nunca começa ou, ao começar, encontra a + fila vazia e sai sem rodar linha. A lista se refaz a cada fatia e pega + também a thread que começa durante a espera. """ prefix = f"dataframeit-lote{next(_executor_ids)}" executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix=prefix) try: yield executor finally: - executor.shutdown(wait=False) + executor.shutdown(wait=False, cancel_futures=True) while alive := [ thread for thread in threading.enumerate() diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 0bb8a6a6..14699756 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -367,6 +367,18 @@ def test_falha_ao_gravar_no_ctrl_c_avisa_e_mantem_o_keyboard_interrupt(tmp_path, assert grava.call_count == 1 +def _sigint_com_restart(reinicia: bool) -> Iterator[None]: + """Liga ou desliga o SA_RESTART do SIGINT durante o teste. + + Na volta, o SA_RESTART fica como o processo o tinha: ligado com o polars + importado, que instala um handler com ele, e desligado sem o polars. + """ + com_polars = "polars" in sys.modules + signal.siginterrupt(signal.SIGINT, not reinicia) + yield + signal.siginterrupt(signal.SIGINT, not com_polars) + + @pytest.fixture def sigint_reinicia_a_espera(): """Põe SA_RESTART no SIGINT, como o handler que o polars instala ao ser importado. @@ -374,14 +386,20 @@ def sigint_reinicia_a_espera(): Com SA_RESTART, o kernel reinicia a espera sem prazo de um lock quando o SIGINT chega, e o thread principal só levanta o `KeyboardInterrupt` quando a espera termina. Sem isso, o SIGINT interromperia também a espera sem prazo, - e o teste não a distinguiria da espera em fatias. Com o polars importado, o - SA_RESTART já estava lá e fica. + e o teste não a distinguiria da espera em fatias. """ - com_polars = "polars" in sys.modules - signal.siginterrupt(signal.SIGINT, False) - yield - if not com_polars: - signal.siginterrupt(signal.SIGINT, True) + yield from _sigint_com_restart(reinicia=True) + + +@pytest.fixture +def sigint_interrompe_a_espera(): + """Tira o SA_RESTART do SIGINT, como fica o processo sem o polars importado. + + Sem SA_RESTART, o SIGINT interrompe também a espera sem prazo do + `Thread.start()` pelo início da thread nova, e o `KeyboardInterrupt` sai de + dentro dele. + """ + yield from _sigint_com_restart(reinicia=False) @requer_pthread_kill @@ -556,8 +574,8 @@ def llm(texto, *args, **kwargs): @requer_pthread_kill -def test_ctrl_c_dentro_do_submit_espera_a_linha_que_ja_entrou_na_fila(tmp_path): - """O Ctrl-C chega dentro do `submit` da última linha, que já entrou na fila. +def test_ctrl_c_dentro_do_submit_espera_a_linha_que_a_thread_ja_comecou(tmp_path): + """O Ctrl-C chega dentro do `submit` da última linha, que a thread já começou. A linha roda e é paga, embora o laço não tenha guardado o futuro dela. A interrupção espera as threads do executor, e a linha entra no checkpoint. @@ -565,9 +583,11 @@ def test_ctrl_c_dentro_do_submit_espera_a_linha_que_ja_entrou_na_fila(tmp_path): textos = ["a", "b", "lenta"] ckpt = tmp_path / "ckpt.csv" pagas, llm_rapido = _llm_com_ctrl_c() + lenta_comecou = threading.Event() def llm(texto, *args, **kwargs): if texto == "lenta": + lenta_comecou.set() # Termina depois da gravação que esperasse só os futuros guardados. time.sleep(0.5) return llm_rapido(texto, *args, **kwargs) @@ -578,10 +598,12 @@ def llm(texto, *args, **kwargs): def ajusta_e_interrompe(self): # O submit põe o item na fila e chama este método, que inicia a thread # quando falta uma; o KeyboardInterrupt do pthread_kill sai do submit - # antes de o laço guardar o futuro. + # antes de o laço guardar o futuro. O Ctrl-C espera a thread tirar a + # linha da fila: a que ainda estivesse nela seria cancelada. ajusta(self) despachos.append(1) if len(despachos) == len(textos): + assert lenta_comecou.wait(timeout=10) _ctrl_c() with ( @@ -612,6 +634,70 @@ def ajusta_e_interrompe(self): assert retomada == [] +@requer_pthread_kill +@pytest.mark.usefixtures("sigint_interrompe_a_espera") +def test_ctrl_c_dentro_do_start_da_thread_deixa_a_linha_da_fila_pendente(tmp_path): + """O Ctrl-C sai do `Thread.start()` da primeira thread do lote, antes de ela começar. + + A thread já foi lançada, mas ainda não marcou o início, e a linha que ela + pegaria está na fila. O encerramento cancela a linha, e a thread, ao + começar, não a roda: a linha não é paga e a retomada a chama. Nenhuma + linha é paga fora do checkpoint. + """ + textos = ["a", "b", "c"] + ckpt = tmp_path / "ckpt.csv" + pagas, llm = _llm_com_ctrl_c() + inicia = threading.Thread._bootstrap_inner # ty: ignore[unresolved-attribute] + lancadas: list[threading.Thread] = [] + saiu = threading.Event() + + def inicia_depois_do_ctrl_c(self): + # Roda na thread nova, antes de ela marcar o início: o thread principal + # está no `_started.wait()` do `start()`, e o Ctrl-C o interrompe ali. + if not self.name.startswith("dataframeit-lote") or lancadas: + inicia(self) + return + lancadas.append(self) + _ctrl_c() + time.sleep(0.2) + try: + inicia(self) + finally: + saiu.set() + + with ( + patch.object(threading.Thread, "_bootstrap_inner", inicia_depois_do_ctrl_c), + pytest.raises(KeyboardInterrupt), + ): + _roda( + pd.DataFrame({"texto": textos}), + llm, + parallel_requests=3, + batch_size=100, + checkpoint_path=ckpt, + ) + assert len(lancadas) == 1 + # A thread ainda não começou, e `join` a recusaria. + assert saiu.wait(timeout=10) + + # Sem linha registrada, a interrupção não grava o arquivo. + processadas = set() + if ckpt.exists(): + gravado = pd.read_csv(ckpt) + processadas = set(gravado.loc[gravado["_dataframeit_status"].eq("processed"), "texto"]) + assert set(pagas) <= processadas + + retomada, llm_ok = _llm_com_ctrl_c() + _roda( + pd.DataFrame({"texto": textos}), + llm_ok, + parallel_requests=3, + batch_size=100, + checkpoint_path=ckpt, + ) + assert sorted(retomada) == sorted(set(textos) - set(pagas)) + + @pytest.mark.parametrize("paralelo", [1, 3]) def test_ctrl_c_antes_de_terminar_uma_linha_nao_reescreve_o_checkpoint(tmp_path, paralelo): """Sem linha registrada desde a última gravação, o arquivo fica como estava.""" From e58ecccf676c5cd1ffb34cd780b45c89b3e468d9 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 14:18:01 -0300 Subject: [PATCH 14/16] =?UTF-8?q?Teste=20de=20duas=20chamadas=20concorrent?= =?UTF-8?q?es:=20a=20sa=C3=ADda=20do=20lote=20n=C3=A3o=20espera=20as=20thr?= =?UTF-8?q?eads=20do=20executor=20da=20outra=20(#213)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A saída de cada lote acha as threads do próprio executor pelo prefixo de nome, numerado por _executor_ids. Filtrar pelo prefixo de todo lote, numerar todo executor igual ou tirar o "_" do fim do prefixo passava a suíte; com qualquer um deles, uma chamada curta concorrente com outra longa ficava presa até a longa terminar. O caso novo roda a chamada longa num thread do usuário, com as linhas presas até a curta voltar, e afirma que a curta volta em menos de 2 s sem que nenhuma linha longa tenha terminado. Uma segunda variante fixa os números dos executores em 10 e 1, para que o nome das threads do lote 10 comece com o prefixo do lote 1. --- tests/test_interrupcao.py | 76 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 75 insertions(+), 1 deletion(-) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 14699756..84d4faf7 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -6,7 +6,7 @@ import time from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor -from contextlib import contextmanager +from contextlib import contextmanager, nullcontext from unittest.mock import patch import pandas as pd @@ -698,6 +698,80 @@ def inicia_depois_do_ctrl_c(self): assert sorted(retomada) == sorted(set(textos) - set(pagas)) +@pytest.mark.parametrize( + "numeros", + [None, (10, 1)], + ids=["numeracao_do_modulo", "lote_1_ao_lado_do_lote_10"], +) +def test_lote_nao_espera_as_threads_de_outra_chamada_concorrente(numeros): + """Duas chamadas ao mesmo tempo: a curta termina com a longa em andamento. + + A saída de cada lote espera só as threads do próprio executor. A longa roda + num thread do usuário, com as linhas presas até a curta voltar. Com os + números fixados, a longa abre o lote 10 e a curta o lote 1, e o nome + `dataframeit-lote10_0` começa com `dataframeit-lote1`. + """ + prazo = 10.0 + trava = threading.Lock() + longas = [f"longa{i}" for i in range(3)] + comecadas: list[str] = [] + terminadas: list[str] = [] + longas_comecaram = threading.Event() + liberada = threading.Event() + + def llm(texto, *args, **kwargs): + if texto in longas: + with trava: + comecadas.append(texto) + if len(comecadas) == len(longas): + longas_comecaram.set() + liberada.wait(timeout=prazo) + with trava: + terminadas.append(texto) + return {"data": {"campo1": f"r-{texto}"}, "usage": {}} + + saidas: list[pd.DataFrame] = [] + + def roda_longa() -> None: + saidas.append( + dataframeit( + pd.DataFrame({"texto": longas}), + questions=Modelo, + prompt="{texto}", + parallel_requests=3, + ) + ) + + numeracao = patch("dataframeit.core._executor_ids", iter(numeros)) if numeros else nullcontext() + longa = threading.Thread(target=roda_longa) + with ( + patch("dataframeit.core.call_langchain", side_effect=llm), + patch("dataframeit.core.validate_provider_dependencies"), + numeracao, + ): + try: + longa.start() + assert longas_comecaram.wait(timeout=prazo) + inicio = time.monotonic() + curta = dataframeit( + pd.DataFrame({"texto": ["a", "b"]}), + questions=Modelo, + prompt="{texto}", + parallel_requests=2, + ) + duracao = time.monotonic() - inicio + with trava: + terminadas_na_volta = list(terminadas) + finally: + liberada.set() + longa.join(timeout=prazo) + + assert duracao < 2.0 + assert terminadas_na_volta == [] + assert list(curta["campo1"]) == ["r-a", "r-b"] + assert [list(saida["campo1"]) for saida in saidas] == [[f"r-{t}" for t in longas]] + + @pytest.mark.parametrize("paralelo", [1, 3]) def test_ctrl_c_antes_de_terminar_uma_linha_nao_reescreve_o_checkpoint(tmp_path, paralelo): """Sem linha registrada desde a última gravação, o arquivo fica como estava.""" From 34f58f363f39091b7d970f793f847b3133e7d5f3 Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 14:22:10 -0300 Subject: [PATCH 15/16] Teste das linhas lentas: o primeiro Ctrl-C espera 0,1 s, e o teste reprova sempre a volta do as_completed (#213) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Com folga 0,0, o primeiro Ctrl-C saía assim que a terceira linha começava, muitas vezes antes de o thread principal entrar na espera do lote, e virava KeyboardInterrupt em qualquer espera. A volta do as_completed ao corpo do lote reprovava o teste só em parte das rodadas. Com as folgas 0,1 e 0,4, o mutante reprova 10 de 10 nos dois ambientes, e o teste passa 10 de 10. --- tests/test_interrupcao.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index 84d4faf7..9316e58c 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -493,7 +493,10 @@ def llm(texto, *args, **kwargs): inicio = time.monotonic() try: - with _ctrl_c_depois(todas_comecaram, 0.0, 0.3), pytest.raises(KeyboardInterrupt): + # A folga do primeiro Ctrl-C deixa o thread principal entrar na espera + # do lote: o que chega antes dela vira KeyboardInterrupt em qualquer + # espera, e o teste não distinguiria a espera sem prazo. + with _ctrl_c_depois(todas_comecaram, 0.1, 0.4), pytest.raises(KeyboardInterrupt): _roda( pd.DataFrame({"texto": textos}), llm, From 8d4b2393b77339ea793f332f4d8e17b019854e9f Mon Sep 17 00:00:00 2001 From: Bruno Oliveira Date: Thu, 29 Oct 2026 15:09:03 -0300 Subject: [PATCH 16/16] =?UTF-8?q?Testes=20do=20Ctrl-C=20contados=20pelas?= =?UTF-8?q?=20linhas=20em=20voo=20do=20executor=20=C3=BAnico=20(#213)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Com a fila contínua, o Ctrl-C na 9ª chamada já não cai no fim de uma onda de três linhas. O teste prende a 7ª e a 8ª chamadas até o encerramento do executor: as três em voo terminam e entram na gravação, e as linhas 10 a 12, que nenhuma thread começou, ficam para a retomada. Os testes deixam de falar em lote: o prefixo das threads é dataframeit-paralelo, e o das chamadas concorrentes fixa os executores 10 e 1. Dois testes novos cobrem o on_row_done: no paralelo, a espera do Ctrl-C cobre o callback que ainda roda na thread do worker; no sequencial, o Ctrl-C dentro do callback grava a linha entregue. Os context managers dos testes passam a ser anotados com Generator, como pede o ty. --- tests/test_interrupcao.py | 171 +++++++++++++++++++++++++++++++++----- 1 file changed, 150 insertions(+), 21 deletions(-) diff --git a/tests/test_interrupcao.py b/tests/test_interrupcao.py index d967832f..6316021b 100644 --- a/tests/test_interrupcao.py +++ b/tests/test_interrupcao.py @@ -4,7 +4,7 @@ import sys import threading import time -from collections.abc import Iterator +from collections.abc import Generator from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager, nullcontext from unittest.mock import patch @@ -275,7 +275,7 @@ def llm(texto, *args, **kwargs): @contextmanager -def _ctrl_c_depois(gatilho: threading.Event, *folgas: float) -> Iterator[None]: +def _ctrl_c_depois(gatilho: threading.Event, *folgas: float) -> Generator[None, None, None]: """Manda um Ctrl-C a cada folga, contada desde `gatilho`, enquanto o bloco roda. Na saída do bloco, o Ctrl-C que falta é cancelado e o envio em curso é @@ -331,16 +331,40 @@ def test_ctrl_c_grava_as_linhas_pagas_e_a_retomada_nao_as_chama_de_novo( """12 linhas, batch_size=5 e Ctrl-C durante a 9ª chamada. No sequencial, a 9ª chamada não termina, e 8 linhas estão pagas. Com 3 - workers, a 9ª é a última do terceiro lote, e as três do lote terminam - enquanto o executor as espera. A gravação a cada 5 linhas deixaria as - últimas 3 ou 4 de fora. + workers, a 9ª chamada começa com a 7ª e a 8ª em voo, presas até o executor + ser encerrado: as três terminam enquanto a interrupção espera as threads, e + as linhas 10 a 12, que nenhuma thread começou, ficam pendentes. A gravação + a cada 5 linhas deixaria as últimas 3 ou 4 de fora. """ textos = [f"t{i:02d}" for i in range(1, 13)] entrada = _entrada(biblioteca, textos) ckpt = tmp_path / "ckpt.csv" - pagas, llm = _llm_com_ctrl_c(na_chamada=9) + pagas, llm_base = _llm_com_ctrl_c() + trava = threading.Lock() + comecadas: list[str] = [] + encerrando = threading.Event() + encerra = ThreadPoolExecutor.shutdown - with pytest.raises(KeyboardInterrupt): + def encerra_observado(self, *args, **kwargs): + encerrando.set() + return encerra(self, *args, **kwargs) + + def llm(texto, *args, **kwargs): + # A 7ª e a 8ª chamadas seguram as threads delas até o encerramento, e + # com as três em voo o laço não despacha a 10ª linha antes do Ctrl-C. + with trava: + comecadas.append(texto) + vez = len(comecadas) + if vez == 9: + _ctrl_c() + elif paralelo > 1 and vez in {7, 8}: + assert encerrando.wait(timeout=10) + return llm_base(texto, *args, **kwargs) + + with ( + patch.object(ThreadPoolExecutor, "shutdown", encerra_observado), + pytest.raises(KeyboardInterrupt), + ): _roda( entrada, llm, @@ -349,7 +373,7 @@ def test_ctrl_c_grava_as_linhas_pagas_e_a_retomada_nao_as_chama_de_novo( checkpoint_path=ckpt, ) - assert len(pagas) == (8 if paralelo == 1 else 9) + assert sorted(pagas) == textos[: 8 if paralelo == 1 else 9] gravado = pd.read_csv(ckpt) processadas = gravado.loc[gravado["_dataframeit_status"].eq("processed"), "texto"] assert sorted(processadas) == sorted(pagas) @@ -389,7 +413,7 @@ def test_falha_ao_gravar_no_ctrl_c_avisa_e_mantem_o_keyboard_interrupt(tmp_path, assert grava.call_count == 1 -def _sigint_com_restart(reinicia: bool) -> Iterator[None]: +def _sigint_com_restart(reinicia: bool) -> Generator[None, None, None]: """Liga ou desliga o SA_RESTART do SIGINT durante o teste. Na volta, o SA_RESTART fica como o processo o tinha: ligado com o polars @@ -493,7 +517,7 @@ def test_ctrl_c_com_todas_as_linhas_lentas_sai_sem_esperar_uma_terminar(tmp_path """Os dois Ctrl-C chegam com as três linhas em andamento e nenhuma termina. O primeiro vira `KeyboardInterrupt` sem esperar uma linha terminar, e o - segundo corta a espera da saída: o lote sai antes da duração das linhas. + segundo corta a espera da saída: a execução sai antes da duração das linhas. """ duracao = 5.0 textos = ["a", "b", "c"] @@ -516,7 +540,7 @@ def llm(texto, *args, **kwargs): inicio = time.monotonic() try: # A folga do primeiro Ctrl-C deixa o thread principal entrar na espera - # do lote: o que chega antes dela vira KeyboardInterrupt em qualquer + # das linhas: o que chega antes dela vira KeyboardInterrupt em qualquer # espera, e o teste não distinguiria a espera sem prazo. with _ctrl_c_depois(todas_comecaram, 0.1, 0.4), pytest.raises(KeyboardInterrupt): _roda( @@ -537,11 +561,11 @@ def llm(texto, *args, **kwargs): @requer_pthread_kill -def test_ctrl_c_espera_a_linha_em_andamento_e_nao_deixa_thread_do_lote(tmp_path): +def test_ctrl_c_espera_a_linha_em_andamento_e_nao_deixa_thread_do_executor(tmp_path): """As linhas lentas em andamento no Ctrl-C terminam na espera e entram no checkpoint. Elas só terminam depois de o executor ser encerrado, e nenhuma thread do - lote sobrevive à interrupção. + executor sobrevive à interrupção. """ textos = ["lenta1", "ctrl-c", "lenta2"] ckpt = tmp_path / "ckpt.csv" @@ -662,7 +686,7 @@ def ajusta_e_interrompe(self): @requer_pthread_kill @pytest.mark.usefixtures("sigint_interrompe_a_espera") def test_ctrl_c_dentro_do_start_da_thread_deixa_a_linha_da_fila_pendente(tmp_path): - """O Ctrl-C sai do `Thread.start()` da primeira thread do lote, antes de ela começar. + """O Ctrl-C sai do `Thread.start()` da primeira thread do executor, antes de ela começar. A thread já foi lançada, mas ainda não marcou o início, e a linha que ela pegaria está na fila. O encerramento cancela a linha, e a thread, ao @@ -679,7 +703,7 @@ def test_ctrl_c_dentro_do_start_da_thread_deixa_a_linha_da_fila_pendente(tmp_pat def inicia_depois_do_ctrl_c(self): # Roda na thread nova, antes de ela marcar o início: o thread principal # está no `_started.wait()` do `start()`, e o Ctrl-C o interrompe ali. - if not self.name.startswith("dataframeit-lote") or lancadas: + if not self.name.startswith("dataframeit-paralelo") or lancadas: inicia(self) return lancadas.append(self) @@ -726,15 +750,15 @@ def inicia_depois_do_ctrl_c(self): @pytest.mark.parametrize( "numeros", [None, (10, 1)], - ids=["numeracao_do_modulo", "lote_1_ao_lado_do_lote_10"], + ids=["numeracao_do_modulo", "executor_1_ao_lado_do_executor_10"], ) -def test_lote_nao_espera_as_threads_de_outra_chamada_concorrente(numeros): +def test_execucao_nao_espera_as_threads_de_outra_chamada_concorrente(numeros): """Duas chamadas ao mesmo tempo: a curta termina com a longa em andamento. - A saída de cada lote espera só as threads do próprio executor. A longa roda - num thread do usuário, com as linhas presas até a curta voltar. Com os - números fixados, a longa abre o lote 10 e a curta o lote 1, e o nome - `dataframeit-lote10_0` começa com `dataframeit-lote1`. + A saída de cada execução espera só as threads do próprio executor. A longa + roda num thread do usuário, com as linhas presas até a curta voltar. Com os + números fixados, a longa abre o executor 10 e a curta o 1, e o nome + `dataframeit-paralelo10_0` começa com `dataframeit-paralelo1`. """ prazo = 10.0 trava = threading.Lock() @@ -836,3 +860,108 @@ def test_ctrl_c_sem_checkpoint_deixa_as_linhas_pagas_na_entrada_pandas(paralelo) _roda(df, llm, parallel_requests=paralelo) assert sorted(df.loc[df["_dataframeit_status"].eq("processed"), "texto"]) == sorted(pagas) + + +@requer_pthread_kill +def test_ctrl_c_espera_o_on_row_done_em_andamento_e_grava_a_linha(tmp_path): + """O Ctrl-C chega com o `on_row_done` de uma linha ainda rodando na thread do worker. + + O callback não recebe o KeyboardInterrupt, e a linha já está contada quando + ele começa: a interrupção espera a thread dele, o callback termina antes de + a interrupção sair da chamada, e a linha entra no checkpoint. A linha que + não foi despachada fica pendente. + """ + textos = ["callback", "ctrl-c", "fila"] + ckpt = tmp_path / "ckpt.csv" + callback_comecou = threading.Event() + encerrando = threading.Event() + trava = threading.Lock() + entregues: list[str] = [] + encerra = ThreadPoolExecutor.shutdown + + def encerra_observado(self, *args, **kwargs): + encerrando.set() + return encerra(self, *args, **kwargs) + + def llm(texto, *args, **kwargs): + if texto == "ctrl-c": + assert callback_comecou.wait(timeout=10) + _ctrl_c() + # Termina só depois do encerramento, como a linha em andamento. + assert encerrando.wait(timeout=10) + time.sleep(0.2) + return {"data": {"campo1": f"r-{texto}"}, "usage": {}} + + def ao_terminar(resultado): + if resultado.index == 0: + callback_comecou.set() + # Termina depois de a interrupção começar a esperar as threads. + encerrando.wait(timeout=10) + time.sleep(0.2) + with trava: + entregues.append(textos[resultado.index]) + + with ( + patch.object(ThreadPoolExecutor, "shutdown", encerra_observado), + pytest.raises(KeyboardInterrupt), + ): + _roda( + pd.DataFrame({"texto": textos}), + llm, + parallel_requests=2, + batch_size=100, + checkpoint_path=ckpt, + on_row_done=ao_terminar, + ) + with trava: + entregues_na_saida = sorted(entregues) + + assert entregues_na_saida == ["callback", "ctrl-c"] + gravado = pd.read_csv(ckpt) + status = dict(zip(gravado["texto"], gravado["_dataframeit_status"], strict=True)) + assert status["callback"] == "processed" + assert status["ctrl-c"] == "processed" + assert pd.isna(status["fila"]) + + retomada, llm_ok = _llm_com_ctrl_c() + _roda( + pd.DataFrame({"texto": textos}), + llm_ok, + parallel_requests=2, + batch_size=100, + checkpoint_path=ckpt, + ) + assert retomada == ["fila"] + + +def test_ctrl_c_dentro_do_on_row_done_no_sequencial_grava_a_linha_entregue(tmp_path): + """No sequencial, o callback roda no thread principal e pode receber o Ctrl-C. + + A linha entregue ao callback já está gravada no DataFrame e contada, e a + gravação do Ctrl-C a leva, também quando é a primeira linha da execução. A + retomada não a chama de novo. + """ + textos = ["a", "b", "c"] + ckpt = tmp_path / "ckpt.csv" + pagas, llm = _llm_com_ctrl_c() + + def interrompe(resultado): + raise KeyboardInterrupt + + with pytest.raises(KeyboardInterrupt): + _roda( + pd.DataFrame({"texto": textos}), + llm, + batch_size=100, + checkpoint_path=ckpt, + on_row_done=interrompe, + ) + + assert pagas == ["a"] + gravado = pd.read_csv(ckpt) + assert gravado["_dataframeit_status"].tolist()[0] == "processed" + assert gravado["_dataframeit_status"].isna().tolist()[1:] == [True, True] + + retomada, llm_ok = _llm_com_ctrl_c() + _roda(pd.DataFrame({"texto": textos}), llm_ok, batch_size=100, checkpoint_path=ckpt) + assert retomada == ["b", "c"]