はじめに

本記事ではAtCoder ヒューリスティックコンテスト(AHC)のインタラクティブ問題で、デバッグ実行を行う方法について紹介します。実際に実装したものをツールとして公開しておりますので、ご利用いただければと思います。

課題

AtCoder ヒューリスティックコンテスト(AHC)ではインタラクティブ問題と呼ばれる種類の問題があります。以下に問題例を示します。

この種の問題で配布される公式のローカルテスタは、テスタが回答プログラムを起動し、テストから標準入出力経由で回答プログラムと入出力を行うことによってテストを行うことが可能になっています。

しかし、テスタが回答プログラムの起動を行うため、VSCode等を用いたデバッグ実行ができない(もしくは困難)という課題がありました。

本記事ではこの課題の解消を目指します。

解決のアプローチ

目指したいのは以下の要件を満たすことです。

テスタが持っている入出力のエミュレート機能は生かしたい
回答プログラムがテスタから起動されることを避けて、VSCode等からデバッグ実行したい

そこで、テスタから起動されるのは入出力の中継機能だけを持ったProxy的なプログラムにして、回答プログラムはその中継機能とやり取りして動作させることを考えます。この「やり取り」はプロセス間通信ができれば良いので、今回はソケット通信を使うことにしました。（なお、名前付きパイプでも実現できましたが、OS依存がありそうなので、ソケット通信版を公開しています）

ただ、このままだと回答プログラムがソケット通信をすることになり、本番実行時と異なるロジックを埋め込むことになるので、ソケット通信を行なって標準入出力に置き換えるWrapperを作ることを目指します。

実装

interactive proxy tool

proxy部分は回答プログラムの実装言語に依存しないので、Pythonで作成しました。Pythonにはasyncioという非同期のI/O処理を行うライブラリがあったため、これを活用しました。

コア部分は以下のようになっています。単純に左から右、右から左に送るだけのプログラムです。

# 標準入力→Socketへの出力、Socketからの入力→標準出力の共通処理
async def handle_reader(reader, writer):
  while True:
    data = await reader.read(4096)
    if not data:
      # 通信が切れた場合はループを抜けて終了する
      break
    writer.write(data)
    await writer.drain()
  # どちらかの通信が切れたら全部終了しに行く
  for t in asyncio.all_tasks():
    if t is not asyncio.current_task():
      t.cancel()

async def server_main(sock_reader, sock_writer):
  # aioconsoleライブラリを使用して標準入出力のasyncio版を取得
  stdin_reader, stdout_writer = await get_standard_streams()
  await asyncio.gather(
    # 両方向の処理を登録
    handle_reader(stdin_reader, sock_writer),
    handle_reader(sock_reader, stdout_writer),
  )

async def main(port):
  server = await asyncio.start_server(server_main, LOCAL_HOST, port)
  try:
    async with server:
      await server.serve_forever()
  except:
    pass

最終的に、回答プログラムが終了した時点でSocketが閉じることになるので、それを受けてこのツールが終了できるよう、handle_reader内にクローズを検知して他の処理を止めに行く処理を入れています。

Windowsでの実行について

Windowsで実行した場合は、このままではうまくいきませんでした。aioconsoleライブラリのget_standard_streams()で取得した標準入力のstreamがブロックしてしまうようだったので、以下の方法でreadするように置き換えています。

data = await self.loop.run_in_executor(None, sys.stdin.readline)

ただし、これは別スレッドを起動して実行している形になるので、最終的に回答プログラムが終了してproxy側を終了しようとした時に、このスレッドがブロックして終了できない状態になっています。現在、これはWindowsでの実行時の制限事項としています。
解決策をご存知の方は教えていただけると助かります。

通信用Wrapper

Wrapper機能は回答プログラムと同言語で、Wrapperから回答プログラムを(プロセス内で)関数呼び出し等によって呼ぶことで、VSCode等からWrapper経由での回答プログラムのデバッグを可能にしています。
JavaとPythonの2パターン作ってみました。

Java版の実装

主要部分は10行程度なので以下に載せます。

  public static void main(String[] args) throws IOException {
    int portNo = 8888;
    // Socket通信開始
    Socket socket = new Socket("localhost", portNo);
    InputStream sock_in = new BufferedInputStream(socket.getInputStream());
    PrintStream sock_out = new PrintStream(new BufferedOutputStream(socket.getOutputStream()), true);

    // 標準入出力を置き換え
    System.setIn(sock_in);
    System.setOut(sock_out);

    // プログラム本体の実行
    Main.main(new String[0]);

    sock_in.close();
    sock_out.close();
    socket.close();
  }

Socket通信のStreamを、System.in, System.outと置き換えて、回答プログラム(Mainクラスのmainメソッドと想定)を呼び出すだけの処理です。

これにより、回答プログラムであるMainクラス側は何も変更しなくても、Wrapperクラスを実行することでデバッグ実行できるようになります。

Python版の実装

Pythonの場合は、Socket通信を開始した後、reader, writerとして処理する方法がよくわからなかったので、SocketIOというクラスを自作して、それをTextIOWrapperを通すことで通信を実現しています。
これも主要部分のみ載せておきます。(SocketIOクラスの実装は省略)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
  try:
    sock.connect(('127.0.0.1', PORT))
  except:
    sys.exit(1)
  with io.TextIOWrapper(SocketIO(sock), 'utf-8') as sio:
    # 標準入出力を置き換え
    sys.stdin = sio
    sys.stdout = sio

    # プログラム本体の実行
    import main

これも最終的にはsys.stdin, sys.stdoutをSocket通信に置き換えて、回答プログラム(main.pyにトップレベルで書かれていると想定)を呼んでいます。

その他の言語の場合

その他の言語では実装していませんが、似たような形でSocket通信を開始し、そのreader, writer相当を標準入出力と置き換えて実行すれば可能だと思われます。

最悪、回答プログラムの本体的な関数の引数にreader, writer相当を追加して、回答プログラム単体で呼ばれるときは標準入出力を、Wrapperから呼ばれる場合はSocket通信の入出力を渡せば良いです。

おわりに

もしかしたら世の中にはこんな苦労をしなくてもデバッグ実行が可能になる仕組みがあるのかもしれませんが、よくわからなかったので自力実装してみました。
同様に苦労している人がいるかもしれないので、ツールとして公開しています。ぜひご利用いただければと思います。

AtCoder AHCのインタラクティブ問題でデバッグ実行を実現する