2025年10月12日日曜日

DelphiからPythonを call

 


「Pythonコードを呼び出して、引数を渡し、実行する」Delphiコードを 2パターン 用意

  1. 外部プロセスとして python.exe を起動して実行(標準出力を取得)

  2. Python4Delphi(P4D)で CPython を埋め込み、関数呼び出し


1) 外部プロセスとして起動(CreateProcess + パイプで stdout を受け取る)

  • どの Delphi 環境でも動きやすい定番手法

  • 仮想環境(venv)や python.exe のフルパスを指定すれば OK

  • -u でバッファリングを無効化して即時に出力を受け取ります

unit PyRunner; interface uses System.SysUtils, System.Classes, Winapi.Windows; function RunPythonAndCapture( const PythonExe: string; // 例: 'C:\Python312\python.exe' や venv の python.exe パス const ScriptPath: string; // 例: 'C:\work\myscript.py' const Args: array of string; // 例: ['--name', 'Alice', '--n', '3'] const WorkDir: string = '' // 作業ディレクトリ(空なら ScriptPath のディレクトリ) ): string; implementation function QuoteIfNeeded(const S: string): string; begin if (S = '') or (S.IndexOfAny([' ', '"']) >= 0) then Result := '"' + StringReplace(S, '"', '\"', [rfReplaceAll]) + '"' else Result := S; end; function JoinArgs(const Items: array of string): string; var i: Integer; begin Result := ''; for i := 0 to High(Items) do begin if i > 0 then Result := Result + ' '; Result := Result + QuoteIfNeeded(Items[i]); end; end; function RunPythonAndCapture( const PythonExe: string; const ScriptPath: string; const Args: array of string; const WorkDir: string ): string; var SA: SECURITY_ATTRIBUTES; StdOutRd, StdOutWr: THandle; StartInfo: STARTUPINFOW; ProcInfo: PROCESS_INFORMATION; CmdLine: string; Buffer: array[0..8191] of Byte; BytesRead: DWORD; OK: BOOL; OldStdOutMode: Cardinal; CurDir: string; begin Result := ''; // パイプ作成(子プロセスの stdout を親側で読めるように) ZeroMemory(@SA, SizeOf(SA)); SA.nLength := SizeOf(SA); SA.bInheritHandle := TRUE; if not CreatePipe(StdOutRd, StdOutWr, @SA, 0) then raise EOSError.CreateFmt('CreatePipe failed: %d', [GetLastError]); try // 子プロセス側で StdOutWr を継承できるようにする if not SetHandleInformation(StdOutRd, HANDLE_FLAG_INHERIT, 0) then raise EOSError.CreateFmt('SetHandleInformation failed: %d', [GetLastError]); // コマンドライン作成 // 例: "C:\Python312\python.exe" -u "C:\work\myscript.py" --name "Alice" --n 3 CmdLine := QuoteIfNeeded(PythonExe) + ' -u ' + QuoteIfNeeded(ScriptPath); if Length(Args) > 0 then CmdLine := CmdLine + ' ' + JoinArgs(Args); ZeroMemory(@StartInfo, SizeOf(StartInfo)); StartInfo.cb := SizeOf(StartInfo); StartInfo.hStdOutput := StdOutWr; StartInfo.hStdError := StdOutWr; // エラーもまとめて受け取る StartInfo.dwFlags := STARTF_USESTDHANDLES; ZeroMemory(@ProcInfo, SizeOf(ProcInfo)); if WorkDir <> '' then CurDir := WorkDir else CurDir := ExtractFileDir(ScriptPath); // WideString → PWideChar if not CreateProcessW(nil, PWideChar(CmdLine), nil, nil, TRUE, CREATE_NO_WINDOW, nil, PWideChar(CurDir), StartInfo, ProcInfo) then raise EOSError.CreateFmt('CreateProcess failed: %d'#13#10'%s', [GetLastError, CmdLine]); try // 親側は書き込み口を閉じ、読み取りだけにする CloseHandle(StdOutWr); StdOutWr := 0; // 子プロセス標準出力を読み取り while True do begin OK := ReadFile(StdOutRd, Buffer, SizeOf(Buffer), BytesRead, nil); if (not OK) or (BytesRead = 0) then Break; Result := Result + TEncoding.UTF8.GetString(Buffer, 0, BytesRead); end; // 終了待ち WaitForSingleObject(ProcInfo.hProcess, INFINITE); finally if ProcInfo.hThread <> 0 then CloseHandle(ProcInfo.hThread); if ProcInfo.hProcess <> 0 then CloseHandle(ProcInfo.hProcess); end; finally if StdOutWr <> 0 then CloseHandle(StdOutWr); if StdOutRd <> 0 then CloseHandle(StdOutRd); end; end; end.

使い方(呼び出し側)

uses System.SysUtils, PyRunner; procedure TForm1.Button1Click(Sender: TObject); var OutText: string; begin OutText := RunPythonAndCapture( 'C:\Python312\python.exe', 'C:\work\myscript.py', ['--name', '山川', '--n', '2'], // ← 渡したい引数 '' // 作業ディレクトリ(省略可) ); Memo1.Lines.Text := OutText; end;

参考の Python 側(例)

# myscript.py import argparse p = argparse.ArgumentParser() p.add_argument('--name', required=True) p.add_argument('--n', type=int, default=1) a = p.parse_args() for i in range(a.n): print(f"hello {a.name} #{i+1}")

2) Python4Delphi(P4D)で埋め込み実行(関数に引数を渡す)

  • GUIアプリ内で Python 関数を直接呼びたい場合に便利

  • Delphi コンポーネント(TPythonEngine, TPythonGUIInputOutput など)をフォームに配置

  • 以下は ランタイム作成 版(コードだけで完結)。関数に文字列・整数引数を渡して戻り値(文字列)を受け取ります

unit PyEmbed; interface uses System.SysUtils, PythonEngine, VarPyth; function CallPythonFunc( const ModulePath: string; // 例: 'C:\work\mymodule.py' const FuncName: string; // 例: 'greet' const S: string; // 引数1(文字列) const N: Integer // 引数2(整数) ): string; implementation function CallPythonFunc( const ModulePath, FuncName, S: string; const N: Integer): string; var Eng: TPythonEngine; ModObj, FuncObj, RetVal: Variant; Dir, FileName, ModName: string; begin Result := ''; Eng := TPythonEngine.Create(nil); try // 必要なら Eng.DllPath, DllName を指定(例:埋め込む Python の DLL など) // Eng.DllPath := 'C:\Python312\'; // Eng.DllName := 'python312.dll'; Eng.LoadDll; // CPython を初期化 // sys.path にモジュールのディレクトリを追加 Dir := ExtractFileDir(ModulePath); FileName := ExtractFileName(ModulePath); ModName := ChangeFileExt(FileName, ''); // "mymodule.py" -> "mymodule" GetPythonEngine.EvalString(Format('import sys; sys.path.insert(0, r"%s")', [Dir])); // モジュールを import ModObj := ImportModule(ModName); // 関数を取得 FuncObj := ModObj.__getattr__(FuncName); // 呼び出し(引数 S と N を渡す) RetVal := FuncObj(S, N); // 返り値(文字列)を Delphi 文字列へ Result := VarToStr(RetVal); finally Eng.Free; end; end; end.

使い方(呼び出し側)

uses System.SysUtils, PyEmbed; procedure TForm1.Button2Click(Sender: TObject); var R: string; begin R := CallPythonFunc('C:\work\mymodule.py', 'greet', '山川', 3); ShowMessage(R); end;

参考の Python モジュール(mymodule.py

def greet(name: str, n: int) -> str: return "\n".join([f"hello {name} #{i+1}" for i in range(n)])

✅ 補足

  • P4D を使う場合は、開発環境に Python4Delphi を導入してください(TPythonEngine 等)。

  • 既存の Python 仮想環境を使う場合は、python3x.dll に合うように Eng.DllName, Eng.DllPath を設定します。

  • NumPy など C 拡張を使う場合は、DLL の互換性(bitness / バージョン)に注意してください。


どちらを選ぶ?

  • 手早く実行+出力取得方式1(外部プロセス) が簡単・堅牢

  • アプリ内に組み込み、関数を直接コール方式2(P4D) が便利(連続呼び出しや相互作用に強い)

用途や制約(配布形態、Python 環境、依存モジュール)を教えていただければ、最適化(例:環境変数設定、非同期実行、タイムアウト、stderr 分離、JSONでの双方向通信など)まで含めたコードを整えます!

あなた:
「Pythonコード=Pythonスクリプト」を 外部プロセスとして起動 して、引数を渡して実行するための 実用的な Delphi ユニット を用意しました。

標準出力/標準エラーの分離取得、作業ディレクトリ指定、環境変数の追加、標準入力の送信、タイムアウト、終了コードの取得にも対応しています。


Delphi: Pythonスクリプト実行ユニット

unit PyScriptRunner; interface uses System.SysUtils, System.Classes, Winapi.Windows; type EPyRunError = class(Exception); EPyRunTimeout = class(EPyRunError); // 実行オプション TPythonRunOptions = record PythonExe: string; // 例: 'C:\Python312\python.exe'(venvのpython.exeでもOK) ScriptPath: string; // 例: 'C:\work\myscript.py' Args: TArray<string>; // 例: ['--name','山川','--n','3'] WorkDir: string; // 作業ディレクトリ(空なら ScriptPath のディレクトリ) InputUTF8: UTF8String; // 標準入力に流す内容(UTF-8)。不要なら空 ExtraEnv: TArray<string>;// 追加の環境変数 'KEY=VALUE' 形式(PATH追加など) TimeoutMS: Cardinal; // タイムアウトms(0=無限) MergeStdErr: Boolean; // Trueでstderrをstdoutへ結合 Unbuffered: Boolean; // Trueで -u を付けて即時出力 end; TPythonRunResult = record ExitCode: Cardinal; // プロセスの終了コード StdOut: UTF8String; // 標準出力(UTF-8) StdErr: UTF8String; // 標準エラー(UTF-8) function StdOutText: string; function StdErrText: string; end; function RunPythonScript(const Opt: TPythonRunOptions): TPythonRunResult; implementation function TPythonRunResult.StdOutText: string; begin Result := TEncoding.UTF8.GetString(BytesOf(StdOut)); // Delphi文字列へ end; function TPythonRunResult.StdErrText: string; begin Result := TEncoding.UTF8.GetString(BytesOf(StdErr)); end; function QuoteIfNeeded(const S: string): string; begin if (S = '') or (S.IndexOfAny([' ', '"']) >= 0) then Result := '"' + StringReplace(S, '"', '\"', [rfReplaceAll]) + '"' else Result := S; end; function JoinArgs(const Items: TArray<string>): string; var i: Integer; begin Result := ''; for i := 0 to High(Items) do begin if i > 0 then Result := Result + ' '; Result := Result + QuoteIfNeeded(Items[i]); end; end; procedure AppendEnvironment(const EnvAdd: TArray<string>; var EnvBlock: string); var i: Integer; begin if Length(EnvAdd) = 0 then Exit; for i := 0 to High(EnvAdd) do EnvBlock := EnvBlock + EnvAdd[i] + #0; end; function BuildEnvironmentBlock(const ExtraEnv: TArray<string>): PChar; var SysEnv: TStringList; Block: string; i: Integer; begin // ベースは親プロセスの環境 SysEnv := TStringList.Create; try SysEnv.Sorted := True; SysEnv.Duplicates := dupIgnore; // 既存環境を拾う i := 0; while PChar(GetEnvironmentStrings)^ <> #0 do Break; // 使わず標準APIから文字列化してもよいが、簡便化のため空 // 簡易版: 追加分だけブロック化(既存環境の引き継ぎはCreateProcessデフォルトに任せる) Block := ''; AppendEnvironment(ExtraEnv, Block); Block := Block + #0; // 末尾ダブルNUL Result := StrNew(PChar(Block)); finally SysEnv.Free; end; end; function RunPythonScript(const Opt: TPythonRunOptions): TPythonRunResult; var SA: SECURITY_ATTRIBUTES; SI: STARTUPINFOW; PI: PROCESS_INFORMATION; // stdout OutRd, OutWr: THandle; // stderr ErrRd, ErrWr: THandle; // stdin InRd, InWr: THandle; CmdLine: string; Buffer: array[0..8191] of Byte; BytesRead: DWORD; OK: BOOL; StartTick: Cardinal; WaitRes: DWORD; CurDir: string; ProcFlags: DWORD; EnvBlock: PChar; Wrote: DWORD; begin ZeroMemory(@Result, SizeOf(Result)); // パイプ(継承可能ハンドルの作成) ZeroMemory(@SA, SizeOf(SA)); SA.nLength := SizeOf(SA); SA.bInheritHandle := TRUE; if not CreatePipe(OutRd, OutWr, @SA, 0) then raise EPyRunError.CreateFmt('CreatePipe(stdout) failed: %d', [GetLastError]); if not SetHandleInformation(OutRd, HANDLE_FLAG_INHERIT, 0) then raise EPyRunError.CreateFmt('SetHandleInformation(stdout) failed: %d', [GetLastError]); if not Opt.MergeStdErr then begin if not CreatePipe(ErrRd, ErrWr, @SA, 0) then raise EPyRunError.CreateFmt('CreatePipe(stderr) failed: %d', [GetLastError]); if not SetHandleInformation(ErrRd, HANDLE_FLAG_INHERIT, 0) then raise EPyRunError.CreateFmt('SetHandleInformation(stderr) failed: %d', [GetLastError]); end else begin ErrRd := 0; ErrWr := OutWr; // 結合 end; if not CreatePipe(InRd, InWr, @SA, 0) then raise EPyRunError.CreateFmt('CreatePipe(stdin) failed: %d', [GetLastError]); if not SetHandleInformation(InWr, HANDLE_FLAG_INHERIT, 0) then raise EPyRunError.CreateFmt('SetHandleInformation(stdin) failed: %d', [GetLastError]); try // コマンドライン // 例: "python.exe" -u "myscript.py" --name "山川" --n 3 CmdLine := QuoteIfNeeded(Opt.PythonExe) + ' '; if Opt.Unbuffered then CmdLine := CmdLine + '-u '; CmdLine := CmdLine + QuoteIfNeeded(Opt.ScriptPath); if Length(Opt.Args) > 0 then CmdLine := CmdLine + ' ' + JoinArgs(Opt.Args); // 作業ディレクトリ if Opt.WorkDir <> '' then CurDir := Opt.WorkDir else CurDir := ExtractFileDir(Opt.ScriptPath); // STARTUPINFO ZeroMemory(@SI, SizeOf(SI)); SI.cb := SizeOf(SI); SI.dwFlags := STARTF_USESTDHANDLES; SI.hStdInput := InRd; SI.hStdOutput := OutWr; SI.hStdError := ErrWr; ZeroMemory(@PI, SizeOf(PI)); // 親環境 + 追加環境(簡易:追加のみ上書き) EnvBlock := nil; if Length(Opt.ExtraEnv) > 0 then EnvBlock := BuildEnvironmentBlock(Opt.ExtraEnv); ProcFlags := CREATE_NO_WINDOW; // CreateProcessW if not CreateProcessW( nil, PWideChar(CmdLine), nil, nil, TRUE, ProcFlags, EnvBlock, PWideChar(CurDir), SI, PI) then raise EPyRunError.CreateFmt('CreateProcess failed: %d' + sLineBreak + '%s', [GetLastError, CmdLine]); try // 子側に不要なハンドルは閉じる(親側) CloseHandle(OutWr); OutWr := 0; if not Opt.MergeStdErr then begin CloseHandle(ErrWr); ErrWr := 0; end; CloseHandle(InRd); InRd := 0; // 標準入力へ書き込む(必要なら) if Length(Opt.InputUTF8) > 0 then begin if not WriteFile(InWr, Opt.InputUTF8[1], Length(Opt.InputUTF8), Wrote, nil) then raise EPyRunError.CreateFmt('Write to stdin failed: %d', [GetLastError]); end; // 入力終了 CloseHandle(InWr); InWr := 0; // 出力読み取り(非同期ループ + タイムアウト) StartTick := GetTickCount; // ループ:プロセスが生きている間、随時パイプを読む while True do begin // タイムアウト判定 if (Opt.TimeoutMS > 0) and (GetTickCount - StartTick >= Opt.TimeoutMS) then begin TerminateProcess(PI.hProcess, Cardinal(-1)); raise EPyRunTimeout.CreateFmt('Python script timed out (%d ms).', [Opt.TimeoutMS]); end; // stdout OK := ReadFile(OutRd, Buffer, SizeOf(Buffer), BytesRead, nil); if OK and (BytesRead > 0) then Result.StdOut := Result.StdOut + UTF8String(TEncoding.UTF8.GetString(Buffer, 0, BytesRead)); // stderr(分離時のみ) if (ErrRd <> 0) then begin if PeekNamedPipe(ErrRd, nil, 0, nil, @BytesRead, nil) and (BytesRead > 0) then begin if ReadFile(ErrRd, Buffer, SizeOf(Buffer), BytesRead, nil) and (BytesRead > 0) then Result.StdErr := Result.StdErr + UTF8String(TEncoding.UTF8.GetString(Buffer, 0, BytesRead)); end; end; // プロセス終了チェック(短い待機で負荷を抑える) WaitRes := WaitForSingleObject(PI.hProcess, 10); if WaitRes = WAIT_OBJECT_0 then begin // 残りの出力を読み切る while ReadFile(OutRd, Buffer, SizeOf(Buffer), BytesRead, nil) and (BytesRead > 0) do Result.StdOut := Result.StdOut + UTF8String(TEncoding.UTF8.GetString(Buffer, 0, BytesRead)); if (ErrRd <> 0) then while ReadFile(ErrRd, Buffer, SizeOf(Buffer), BytesRead, nil) and (BytesRead > 0) do Result.StdErr := Result.StdErr + UTF8String(TEncoding.UTF8.GetString(Buffer, 0, BytesRead)); Break; end; // 小休止 Sleep(1); end; // 終了コード GetExitCodeProcess(PI.hProcess, Result.ExitCode); finally if PI.hThread <> 0 then CloseHandle(PI.hThread); if PI.hProcess <> 0 then CloseHandle(PI.hProcess); end; finally if OutWr <> 0 then CloseHandle(OutWr); if OutRd <> 0 then CloseHandle(OutRd); if (ErrRd <> 0) and (ErrRd <> OutRd) then CloseHandle(ErrRd); if (ErrWr <> 0) and (ErrWr <> OutWr) then CloseHandle(ErrWr); if InWr <> 0 then CloseHandle(InWr); if InRd <> 0 then CloseHandle(InRd); if Assigned(EnvBlock) then StrDispose(EnvBlock); end; end; end.

使い方(呼び出し例)

uses System.SysUtils, PyScriptRunner; procedure TForm1.ButtonRunClick(Sender: TObject); var Opt: TPythonRunOptions; Res: TPythonRunResult; begin Opt.PythonExe := 'C:\Python312\python.exe'; // venvのpython.exeでもOK Opt.ScriptPath := 'C:\work\myscript.py'; Opt.Args := TArray<string>.Create('--name', '山川', '--n', '2'); Opt.WorkDir := ''; // 空ならScriptのフォルダ Opt.InputUTF8 := ''; // 標準入力に流したい文字列があればUTF-8で Opt.ExtraEnv := TArray<string>.Create('MYFLAG=1'); // 追加の環境変数があれば Opt.TimeoutMS := 15000; // 15秒でタイムアウト(0=無限) Opt.MergeStdErr:= False; // Falseでstderr分離 Opt.Unbuffered := True; // -u(即時出力) try Res := RunPythonScript(Opt); Memo1.Lines.Add('ExitCode: ' + Res.ExitCode.ToString); Memo1.Lines.Add('--- STDOUT ---'); Memo1.Lines.Add(Res.StdOutText); Memo1.Lines.Add('--- STDERR ---'); Memo1.Lines.Add(Res.StdErrText); except on E: EPyRunTimeout do ShowMessage('Timeout: ' + E.Message); on E: Exception do ShowMessage('Error: ' + E.Message); end; end;

Pythonスクリプト側(例)

# myscript.py import argparse, sys, time p = argparse.ArgumentParser() p.add_argument('--name', required=True) p.add_argument('--n', type=int, default=1) a = p.parse_args() for i in range(a.n): print(f"hello {a.name} #{i+1}", flush=True) time.sleep(0.5) print("done.", file=sys.stderr, flush=True)

よくあるポイント

  • venv を使う: Opt.PythonExe にその venv の ...\Scripts\python.exe を指定。

  • 長い出力/日本語: 本コードは UTF-8 前提で取り込み。Windows側コンソールコードページに依存しません。

  • JSONで引数をまとめる: 複雑なパラメータは --json "<json文字列>" として渡すのが安全です。

  • 大きな標準入力: Opt.InputUTF8 に入れて送れます(サイズが大きい場合は断片送信に変更可能)。

  • タイムアウト: TimeoutMS を設定すると、超過時にプロセスを kill して例外を投げます。

必要なら、

  • 非同期(スレッド / コールバック)版

  • PowerShell/WSL 経由での実行

  • macOS/Linux(TProcess 相当)対応

2025年10月11日土曜日

Nuitka Python コンパイラー

 


参照

https://blog.tsukumijima.net/article/python-nuitka-usage/


PyInstaller で作ったソフトは ファイルサイズがデカい・起動が遅い・ウイルス判定されやすい とちょっと微妙な点が多いです。

それでも比較的簡単に exe 化できるのでよく使われているわけですが、とはいえネイティブの exe ほどパフォーマンスは上がりませんし、むしろ普通に Python で実行したときより遅くなります。

ここで、PyInstaller の仕組みを簡潔に説明します。
PyInstaller は基本 --onefile オプションをつけて実行することが多いと思いますが、あれは exe の中に自己解凍機能が入っていて、実行すると Windows なら AppData\Local\Temp あたりに exe の中に入っていたパッケージ一式が解凍されます。その中に入っているスクリプトを実行することで、見た感じあたかも単一の exe ファイルで実行できるように見せかけている、というものです。
解凍されたパッケージ一式は終了時に削除されるため、ダブルクリックで実行した際にそうしたパッケージを毎回解凍することになり、それが起動時のオーバーヘッドになっています。

ほかに Python を exe 化するソフトとしてはたとえば py2exe という Windows 専用のツールなどもあったりしますが、Windows 専用な上に使い方が面倒くさく、あまり手軽とはいえません。

Sponsored Link

Nuitka の導入

GitHub - Nuitka/Nuitka: Nuitka is a Python compiler written in Python. It's fully compatible with Python 2.6, 2.7, 3.4-3.13. You feed it your Python app, it does a lot of clever things, and spits out an executable or extension module.
Nuitka is a Python compiler written in Python. It's fully compatible with Python 2.6, 2.7, 3.4-3.13. You feed it your Py...

Nuitka - Wikipedia

そこで Python の公式ドキュメントにはどう書いてあるのかな?と思い漁ってみると、「Nuitka」というソフトがいくつかある中の一番上で紹介されていました。

そもそもどういうツールなのか分かりにくいのですが、PyInstaller と同じ、Python プログラムの実行ファイル化ができるツールになります。なんと Python コードを一度C言語にトランスパイルし、それをさらに gcc などのコンパイラでコンパイルしたものを実行ファイルとして出力するというんだから驚きです。

今までまったく知らなかったのですが、調べると2013年頃にはすでに開発が始まっていたようで、それなりに老舗のツールのようでした。いまだに PyInstaller 一強になっているのが不思議なくらい…。
一度C言語にコンパイルするため(Python のコードでない)ネイティブな実行ファイルとして使える上に、C言語向けのパフォーマンスチューニングが行われるため、ファイルサイズが小さくなる上に実行速度も普通に Python を実行したときよりも速くなるんだそう。すごい。
ただ、日本だと VOICEVOX のビルドに使われていたりを見かけた以外は、知名度も利用例もあまりなさそうな印象でした。

とりあえず良さそうだし、まずはインストールして使ってみましょう。
今回は Windows 上で pipenv を使います。理由は PyInstaller と同様、グローバルの pip パッケージを使ってしまうとそこにインストールされている余計なパッケージも同梱されてしまうおそれがあったためです。
とはいえ、Nuitka は賢くて、 --follow-imports オプションをつければ Python ファイルを静的解析して自動的に使ってるライブラリだけを読み込んでくれるっぽいので、あまり関係なさそうだとは思います。

ここは PyInstaller でも概ね同じだと思うのですが、Nuitka は残念ながら OS をまたいだクロスコンパイルはできないみたいです。なので、Windows 用の .exe であれば Windows 上でビルドする必要があるし、Mac や Linux の場合も然りです。

pipenv install nuitka zstandard

pipenv で、nuitka と zstandard をインストールします。

ビルド時につまづいた点なのですが、後述する --onefile オプションで単一の実行ファイルにまとめる場合、Zstandard という圧縮形式(見慣れない圧縮形式だが、そこそこ圧縮率が高くてめちゃくちゃ解凍が速いのが特徴らしい)を使ってコード全体を圧縮するみたいです。

Nuitka も単一の実行ファイルにまとめる場合は、PyInstaller と同じく Temp フォルダに自己解凍してそれを実行するような形になっています。Zstandard の解凍が速いという特性は、前述した解凍時のオーバーヘッドを抑えつつ、ファイルサイズを圧縮するのに便利なのでしょう。
Zstandard が使えなくても単一の実行ファイルにはまとめられますが、ファイルサイズが圧縮する場合の2~3倍にまで膨れあがります…。

注意点として、Python のアーキテクチャと後述する MinGW64 などのコンパイラのアーキテクチャは一致している必要があるらしいです。64bit 環境なのに Python の 32bit 版をインストールしている場合は注意してください。
今どき 32bit の exe は多くのユースケースで不要だと思うので、切り捨てちゃっていいとは思います。

Nuitka の使い方

pipenv shell

事前に pipenv の仮想環境の中に入っておきます。

nuitka --mingw64 --follow-imports --onefile (エントリーポイントにしたい Python ファイル).py

あとは上記のコマンドを実行するだけです。

--mingw64 オプションでは明示的に C コンパイラに MinGW64 内の GCC を指定します。
Visual Studio が入っているなどで MSVC が使える状況にあればそちらが自動で利用されるのですが、どうやら MSVC の言語が日本語になっていると、CP932🤮諸々の文字コード問題で Nuitka 側がビルドに失敗してしまうようです。
MSVC (MSBuild とかそのへん) は日本語環境だとコンソールに CP932 (Shift-JIS) で出力するみたいなのですが、Nuitka は英語しか想定していないため、予想されていない文字コードという事で落ちてしまうようでした。MSVC を英語版で入れてねみたいな事が公式ドキュメントに書いてありましたが、さすがにそこまでのやる気もなく…。

そもそも MSVC が入ってない場合は自動で MinGW64 を使うようになっているので、そちらを明示的に指定します。MinGW64 というのは要は Windows 上で GCC を使うためのやつなんですが、Nuitka を使うために MinGW64 をインストールしておく必要はありません。

Nuitka はとても賢くて、ビルド環境に GCC がないと分かった場合、「GCC を AppData 以下にダウンロードするけどいいですか?一回限りでキャッシュされます」みたいな事を訊かれます。そこで YES を入力すると、Nuitka 側がちゃんと動作する MinGW64 と GCC を環境を汚さずインストールしてくれて、さらにそれを自動で使ってくれます。ありがたすぎる…。
他にも depends っていうやつも自動でインストールするか訊かれるので、それも YES と答えておきましょう。2回目以降は自動でダウンロードされた GCC や depends が使われます。

最初こちらで GCC 8.1.0 の MinGW64 をインストールしたりしたのですが、最新の Nuitka は GCC 11.2 以降にしか対応していないようで、使えないと判断されて自動で最新の GCC のインストールを求められたりしてちょっと嵌まりました。Nuitka にまかせておけば自動でいい感じに GCC を使ってくれるので、任せてしまうのが手っ取り早いと思います。

--follow-imports は Python ファイルを静的解析して、インポートしているファイルを再帰的にビルド対象にいれてくれるオプションみたいです。単一ファイルと標準ライブラリで収まるソフトでないならば、これをつけないとうまく動かなくて壊れます(そりゃそう)。

最後の --onefile は PyInstaller と同様に単一の実行ファイルにまとめるオプションです。これをつけないと、現在の環境にインストールされている Python を利用する exe ファイルが生成されてしまいます。

単一の実行ファイルにまとめなくてもいい場合は --standalone オプションを代わりに指定します。
何も出力フォルダを指定しない場合、デフォルトで (エントリーポイントの py ファイルの名前).dist/ フォルダ以下に出力されます。

--standalone の場合、Python のコードはすべて exe に統合されていますが、Python 本体や依存している Visual C++ のランタイム、Tcl/Tk などのネイティブ拡張に関してはファイルとしてそのまま配置されます。実行する場合はこれら全てをセットで配置する必要があるみたいです。
今回検証に利用した EDCBNotifier の場合、サイズはフォルダ全体で 36MB ほどでした。
内訳は本体が 18MB 、Python 本体と標準ライブラリ諸々が残りの 18MB を閉めています。ファイルに何も圧縮が掛かっていないので当然っちゃそう。結構細々としたファイルでぐちゃぐちゃになるので、それが嫌な人は --onefile を使う事になると思います。

--onefile オプションを指定した場合、--standalone のファイルを生成した後にそれを前述の Zstandard で圧縮して一つの exe にまとめます。PyInstaller 同様に一度 Temp フォルダに展開してから実行する点は代わりませんが、Zstandard のおかげでファイルサイズが --standalone と比較して2倍近く圧縮される上に、PyInstaller よりも起動がかなり速いのが大きな差です。
–onefile オプションを指定した場合は、実行ファイルが (エントリーポイントの py ファイルの名前).exe に生成されます。今回の場合ファイルサイズは 12MB 弱で、1/3 ほどに圧縮できている事になります。

--onefile オプションは --standalone オプションの動作を包含するため、--onefile で実行した場合にも (エントリーポイントの py ファイルの名前).dist/ に単一の実行ファイル化されていないファイルが出力されます。
大は小を兼ねると言いますし、基本的に --onefile で良いんじゃないでしょうか。

試しに python EDCBNotifier.py PostAddReserve と ./EDCBNotifier.exe PostAddReserve の差を測ってみたところ、前者が 2.0 秒、後者が 2.5 ~ 2.7 秒ほどでした。解凍に 0.5 秒ほどかかるようで直接 Python コードを実行するよりも起動は遅くなりますが、それでも PyInstaller よりかはかなり速くなるんじゃないかと思います。コードの規模にもよると思いますが、C 言語に変換した分コードも速くなっているはずなので、そのあたりの恩恵もあるかもしれません。

ちなみに、Nuitka にも PyInstaller と同様に、オプションからカスタムのアプリアイコンを設定することが可能です。さらに Windows の exe のファイルバージョンのメタデータなどもオプションから細かく指定することができたりなど、至れり尽くせりで助かります。

使ってみた感想とまとめ

動作感やパフォーマンスは PyInstaller より良好で、PyInstaller を使うならこっちを使った方がよりスマートになりそうだな、という感想を持ちました。動作自体にも今のところ特に問題はなさそうに見えます。

ただ、一番のネックは「とにかくビルドが重い」という事に尽きます。PyInstaller もそれなりに時間がかかりますが、Nuitka のビルド時間はそれの比ではありません。
そもそも PyInstaller は特にソースコードをコンパイルせず単一ファイルにまとめてそれを Temp に展開しているだけなので、わざわざ C のネイティブコードに変換してそれを更にビルドしている Nuitka と比較するほうが無理がありそうです…。

一度ビルドすると、ビルド済みの C コードが (エントリーポイントの py ファイルの名前).build/ 以下に、CCache というビルドキャッシュが AppData\Local\Nuitka 以下に生成されます。
変更されたコードだけを再ビルドして変更のないコード(ライブラリ)などはそのまま使うようにして一定の時間短縮を図っているようですが、コード量も依存ライブラリもそこまで多くない EDCBNotifier ですら、何もソースコードをいじらずビルドキャッシュが効いている状態なのにも関わらず、ビルドが終わるまでに 2 分 30 秒も掛かってしまいました。
たとえば VOICEVOX のようなクソデカソフトの場合、さらに機械学習系のクソデカライブラリもビルドする必要があるので、ビルド時間が恐ろしいことになるのは想像に難くありません。

ただ、逆を言えば遅いのはビルドするときだけで、それ以外では高速に動作します。当然ユーザーに配布する場合も高速に動作することが期待できますし、開発マシンのスペックがクソ重いとかでないならば、十二分に検討する価値はあると思います。

あと、私が確認した限りでは、--onefile オプションをつけて単一の実行ファイルにまとめると、なぜか Windows 7 で実行できなくなってしまっていました。Windows 7 がサポートされているはずの Python 3.8 でもダメだったため、おそらく Nuitka 側に何らかの問題があると考えられます。
ただ、Windows 7 はすでにサポート終了から2年近く経ちますし、もうサポートを切ってもいい頃合いだとは思います。

冒頭で上げた PyInstaller のデメリットのうち、「ファイルサイズがデカい」と「起動が遅い」に関しては Nuitka を使うことで大幅な改善が期待できると思います。「ウイルス判定されやすい」という点に関してはまだわかりませんが、すくなくとも Nuitka は C 言語にコンパイルしたものを exe にしているため、コードがそのまま載っている PyInstaller よりも誤検知されにくいことが期待できるはずです。

2022 年の Python の exe 化には Nuitka を使っていこう!!ということで、今年最後の記事を〆たいと思います。

RTX1210で2拠点間同一セグメント

 


RTX1200とインターネットを利用して2つの拠点間を同一セグメントにする方法を教えてください
ChatGPT:

RTX1200を使ってインターネット経由で2拠点を「同一セグメント」にする方法ですね。これはいわゆる L2TPv3(Ether/IP)やブリッジ接続を用いたVPN構成 になります。ポイントを整理すると以下の通りです。


方法の概要

通常のIPsec-VPNでは「拠点間をルータでルーティング」する形になるため、拠点ごとに異なるIPセグメントを割り当てます。
しかし「同一セグメントで使いたい」場合は、ルーティングではなく レイヤ2トンネル を張る必要があります。RTX1200では以下の手段が使えます。

  1. L2TPv3 (Ether over IP)

    • RTX同士でL2TPv3セッションを張り、Ethernetフレームをそのままカプセル化して転送。

    • これによりブリッジ接続が可能になり、2拠点を同一L2セグメントにできる。

  2. IPsecとの併用

    • L2TPv3トンネルを暗号化するためにIPsecを併用するのが一般的。


設定イメージ

例えば拠点Aと拠点Bを同一セグメント 192.168.10.0/24 で使いたい場合:

拠点A(RTX1200)

# WAN 側の設定(省略) # Ether over IP の定義 tunnel select 1 tunnel encapsulation l2tpv3 tunnel endpoint address <拠点BのグローバルIP> ipsec tunnel 1 exit # ブリッジ作成 bridge 1/1 bridge member lan1 bridge member tunnel1

拠点B(RTX1200)

tunnel select 1 tunnel encapsulation l2tpv3 tunnel endpoint address <拠点AのグローバルIP> ipsec tunnel 1 exit bridge 1/1 bridge member lan1 bridge member tunnel1

注意点

  • DHCPやブロードキャストも拠点間で流れるため、利用環境によっては帯域に影響が出ることがあります。

  • RTX1200は古い機種なので、最新ファームでの対応を確認してください。

  • L2TPv3はNAT環境をまたぐと不安定になることがあります。できればグローバルIP直収が望ましいです。

  • 運用面では「同一セグメント」にせず、通常のルーティングVPN構成を推奨されることが多いです(管理・拡張性の観点)。


👉 ご希望としては「DHCPサーバを1拠点にまとめたい」などでしょうか?
それによっては L2VPNではなくDHCPリレーやルーティング構成の方が安定・簡単に運用できるケースもあります。

必要であれば、具体的なユースケース(DHCP共有したい・ブロードキャストを通したい・単純に同じセグメントを維持したい等)を伺えれば、さらに最適な構成例をご提案できます。

あなた:
RTX1210とインターネットを利用して2つの拠点間を同一セグメントにする方法を教えてください
ChatGPT:

ご質問ありがとうございます!
RTX1210でも、基本的な考え方はRTX1200と同じで「レイヤ2トンネル(L2TPv3/EoIP)」を使う必要があります。つまり、通常のIPsec-VPNのようにルーティングではなく、Ethernetフレームをそのままカプセル化して転送する仕組みです。これにより、2拠点を同じブロードキャストドメイン(同一セグメント)にできます。


方法の概要

  1. L2TPv3 (Ether over IP) を使う

    • RTX1210はL2TPv3に対応しているため、これを使って拠点間をL2接続にします。

  2. IPsecで暗号化する

    • セキュリティを考慮し、トンネルはIPsecで暗号化します。

  3. ブリッジを構成する

    • LANインタフェースとトンネルインタフェースをブリッジに参加させ、同一セグメントにします。


設定例

前提

  • 拠点AのグローバルIP: 203.0.113.1

  • 拠点BのグローバルIP: 203.0.113.2

  • 共通セグメント: 192.168.10.0/24


拠点A(RTX1210)

# WAN側は省略 # トンネル定義 tunnel select 1 tunnel encapsulation l2tpv3 tunnel endpoint address 203.0.113.2 ipsec tunnel 1 exit # IPsec設定 ipsec auto refresh on ipsec ike keepalive log on ipsec ike keepalive use on ipsec ike pre-shared-key 1 text <共通鍵> ipsec ike local address 203.0.113.1 ipsec ike remote address 203.0.113.2 # ブリッジ設定 bridge 1/1 bridge member lan1 bridge member tunnel1

拠点B(RTX1210)

tunnel select 1 tunnel encapsulation l2tpv3 tunnel endpoint address 203.0.113.1 ipsec tunnel 1 exit ipsec auto refresh on ipsec ike keepalive log on ipsec ike keepalive use on ipsec ike pre-shared-key 1 text <共通鍵> ipsec ike local address 203.0.113.2 ipsec ike remote address 203.0.113.1 bridge 1/1 bridge member lan1 bridge member tunnel1

注意点

  • DHCPやブロードキャストも拠点間に流れるため、ネットワーク規模が大きいとトラフィックが膨らみやすいです。

  • NATを超えると動作が不安定になる場合があります。できれば両拠点ともグローバルIPを直接持たせる方が安定します。

  • 管理面では「同一セグメント」より「別セグメント+ルーティング(IPsec-VPN)」の方がシンプルでトラブルも少ないです。


👉 ちなみに、同一セグメント化をしたい理由は「DHCPを一元管理したい」「プリンタやNASを拠点間で透過的に使いたい」などでしょうか?
もしよければ利用目的を教えていただければ、L2VPNが本当にベストか、それともL3VPNで代替可能かもご提案できます。

2025年10月9日木曜日

Eth0->eth1転送

 


転送コード


以下は「片方のEthernetインターフェースで受け取ったフレームを、もう一方へそのまま転送する」最小のPython例です。Linux向け、root権限が必要です。手軽さ重視で Scapy 版と、標準ライブラリだけで動く raw socket(AF_PACKET) 版の2つを載せます。


1) Scapy版(簡単・分かりやすい)

#!/usr/bin/env python3 # bridge_scapy.py # 要: pip install scapy かつ root権限 from scapy.all import sniff, sendp import argparse import threading import signal import sys def forward_loop(in_if, out_if, bpf): # 受信 → そのまま送信 sniff( iface=in_if, prn=lambda pkt: sendp(pkt, iface=out_if, verbose=False), store=False, filter=bpf if bpf else None, promisc=True, ) def main(): ap = argparse.ArgumentParser(description="L2 frame forwarder (simple bridge) using Scapy") ap.add_argument("--if1", required=True, help="インターフェース1 (例: eth0)") ap.add_argument("--if2", required=True, help="インターフェース2 (例: eth1)") ap.add_argument("--bpf", default="", help="BPFフィルタ (例: 'arp or tcp')") args = ap.parse_args() # Ctrl+Cで両方止める def handle_sigint(signum, frame): sys.exit(0) signal.signal(signal.SIGINT, handle_sigint) t1 = threading.Thread(target=forward_loop, args=(args.if1, args.if2, args.bpf), daemon=True) t2 = threading.Thread(target=forward_loop, args=(args.if2, args.if1, args.bpf), daemon=True) t1.start(); t2.start() t1.join(); t2.join() if __name__ == "__main__": main()

使い方例:

sudo python3 bridge_scapy.py --if1 eth0 --if2 eth1 # 必要ならBPFで絞り込み sudo python3 bridge_scapy.py --if1 eth0 --if2 eth1 --bpf "arp or tcp"

ポイント

  • L2(イーサネット)フレームをそのまま転送します(ブリッジ動作)。

  • 速度より手軽さ重視。高スループットや低遅延が必要ならOSのブリッジ機能(ip link add br0 type bridge 等)やDPDK/ebpf等を検討してください。


2) 標準ライブラリだけ(AF_PACKET raw socket; Linux限定)

#!/usr/bin/env python3 # bridge_raw.py # 要: root権限 / Linux import socket import fcntl import struct import os import select import argparse import signal import sys SIOCGIFINDEX = 0x8933 ETH_P_ALL = 0x0003 # htons不要: socket作成で自動 def ifindex(sock, ifname: str) -> int: ifreq = struct.pack('16sH14s', ifname.encode('utf-8'), socket.AF_PACKET, b'\x00'*14) res = fcntl.ioctl(sock.fileno(), SIOCGIFINDEX, ifreq) return struct.unpack('16sH14s', res)[1] def bind_raw(ifname: str) -> socket.socket: s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4 * 1024 * 1024) idx = ifindex(s, ifname) s.bind((ifname, 0)) # プロミスキャス有効化(best-effort) PACKET_ADD_MEMBERSHIP = 1 PACKET_MR_PROMISC = 1 mreq = struct.pack("IHH8s", idx, PACKET_MR_PROMISC, 0, b"\x00" * 8) try: s.setsockopt(socket.SOL_PACKET, PACKET_ADD_MEMBERSHIP, mreq) except OSError: pass return s def main(): ap = argparse.ArgumentParser(description="L2 frame forwarder (raw AF_PACKET)") ap.add_argument("--if1", required=True) ap.add_argument("--if2", required=True) ap.add_argument("--mtu", type=int, default=65535, help="受信バッファ(1フレーム)サイズ") args = ap.parse_args() def handle_sigint(signum, frame): sys.exit(0) signal.signal(signal.SIGINT, handle_sigint) s1 = bind_raw(args.if1) s2 = bind_raw(args.if2) # selectで双方向中継 while True: rlist, _, _ = select.select([s1, s2], [], []) for rs in rlist: try: data = rs.recv(args.mtu) if not data: continue # 受けた側と反対へ送る (s2 if rs is s1 else s1).send(data) except (BlockingIOError, InterruptedError): continue except OSError as e: # 一時的なエラーは無視 if e.errno in (11, 35): # EAGAIN/EWOULDBLOCK continue raise if __name__ == "__main__": main()

使い方:

sudo python3 bridge_raw.py --if1 eth0 --if2 eth1

運用メモ

  • どちらの方法も root権限 が必要です(生パケットアクセスのため)。

  • 片方/両方のIFでIPスタックによる処理を避けたい場合、ifconfig ethX promiscip link set ethX promisc on を確認。

  • 既にOS側でブリッジ(br0)を組んでいる場合は、本スクリプトと競合し得ます。片方にしてください。

  • フィルタしたい場合はScapy版の --bpf が手軽(例: --bpf "vlan and tcp")。

  • 高スループットが必要なら:OSブリッジ + ethtool でオフロード設定、あるいはeBPF/DPDK/専用スイッチを推奨。

必要なら、VLANタグの保持/書き換え、特定MACのみ通す、簡易L2学習(MACテーブル)付きに拡張したサンプルも用意できます。

あなた:
長いメッセージを送れるようバッファーを大きくできるコード
ChatGPT:

「長いメッセージを送れるようにバッファを大きくしたい」というのは、Ethernet転送コードで 一度に受け取れるパケットのサイズや送信バッファ容量を増やしたい という意味ですね。
以下のように、socket.setsockopt() を使って受信・送信バッファサイズを大きく設定できます。


🔧 バッファを拡張した例(AF_PACKET版)

#!/usr/bin/env python3 # bridge_buffered.py import socket, fcntl, struct, os, select, argparse, signal, sys SIOCGIFINDEX = 0x8933 ETH_P_ALL = 0x0003 # all protocols def ifindex(sock, ifname): ifreq = struct.pack('16sH14s', ifname.encode(), socket.AF_PACKET, b'\x00'*14) res = fcntl.ioctl(sock.fileno(), SIOCGIFINDEX, ifreq) return struct.unpack('16sH14s', res)[1] def bind_raw(ifname: str, bufsize=8*1024*1024): """raw socketを指定インターフェースにバインドし、バッファ拡張""" s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, bufsize) # 受信バッファ s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, bufsize) # 送信バッファ idx = ifindex(s, ifname) s.bind((ifname, 0)) return s def main(): ap = argparse.ArgumentParser(description="Ethernet bridge with larger buffers") ap.add_argument("--if1", required=True) ap.add_argument("--if2", required=True) ap.add_argument("--buf", type=int, default=8*1024*1024, help="受信・送信バッファサイズ (bytes)") ap.add_argument("--mtu", type=int, default=65535) args = ap.parse_args() s1 = bind_raw(args.if1, args.buf) s2 = bind_raw(args.if2, args.buf) print(f"Bridge running between {args.if1}{args.if2}") print(f"Buffer size: {args.buf:,} bytes") def handle_sigint(signum, frame): sys.exit(0) signal.signal(signal.SIGINT, handle_sigint) while True: r, _, _ = select.select([s1, s2], [], []) for rs in r: try: data = rs.recv(args.mtu) if not data: continue (s2 if rs is s1 else s1).send(data) except OSError as e: continue if __name__ == "__main__": main()

✅ ポイント

  • SO_RCVBUF / SO_SNDBUF により OS 側のソケットバッファを拡張。

  • --buf オプションで任意サイズを指定可能(例:--buf 16777216 → 16MB)。

  • Ethernetフレーム自体は最大 1518 バイト程度ですが、スクリプトの受信・転送キューを深くしてロスを防げます。

  • 実際のMTU(最大転送単位)を超えるサイズの「長いメッセージ」を送るには、アプリケーション層で分割するか、Jumbo Frame をNIC設定で有効にする必要があります(例:ip link set eth0 mtu 9000)。


💡 Jumbo Frameを併用する場合

sudo ip link set eth0 mtu 9000 sudo ip link set eth1 mtu 9000 sudo python3 bridge_buffered.py --if1 eth0 --if2 eth1 --buf 16777216 --mtu 9000

これで、大きなEthernetフレームを通しつつ、受信/送信バッファも拡大してパケットドロップを減らすことができます。


ご希望に応じて、

  • TCP/UDPレベルでの長メッセージ送受信(ソケット通信)版

  • 複数スレッド/非同期転送でのバッファリング強化版

なども示せますが、どちらの方向を想定されていますか?
(Ethernetフレーム中継?それともアプリ層でのデータ転送?)

あなた:
#!/usr/bin/env python3 # l2bridge.py import argparse import os import select import signal import socket import struct import sys import time ETH_P_ALL = 0x0003 # すべてのEtherType SOL_PACKET = 263 # from linux/sol_packet.h PACKET_ADD_MEMBERSHIP = 1 PACKET_DROP_MEMBERSHIP = 2 PACKET_MR_PROMISC = 1 PACKET_OUTGOING = 4 RUNNING = True def set_promisc(sock, ifindex: int, enable: bool = True): mreq = struct.pack("IHH8s", ifindex, PACKET_MR_PROMISC, 0, b"\x00"*8) opt = PACKET_ADD_MEMBERSHIP if enable else PACKET_DROP_MEMBERSHIP sock.setsockopt(SOL_PACKET, opt, mreq) def if_nametoindex(ifname: str) -> int: return socket.if_nametoindex(ifname) def open_raw_socket(ifname: str) -> socket.socket: s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) # インターフェイスにバインド s.bind((ifname, 0)) # 受信バッファ拡大(任意) try: s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4 * 1024 * 1024) except OSError: pass return s def forward_loop(if1: str, if2: str, print_stats: bool = True): s1 = open_raw_socket(if1) s2 = open_raw_socket(if2) # プロミスキャス有効化(これをしないと一部フレームを拾えない環境がある) set_promisc(s1, if_nametoindex(if1), True) set_promisc(s2, if_nametoindex(if2), True) cnt12 = cnt21 = 0 last_report = time.time() def _cleanup(*_): global RUNNING RUNNING = False signal.signal(signal.SIGINT, _cleanup) signal.signal(signal.SIGTERM, _cleanup) print(f"[+] Bridging {if1} <-> {if2} (Ctrl-C to stop)") # selectで両方監視 while RUNNING: r, _, _ = select.select([s1, s2], [], [], 1.0) for rs in r: try: # data: bytes, addr: (proto, ifindex, pkttype, halen, addrbytes) data, addr = rs.recvfrom(65535) pkttype = addr[2] if len(addr) >= 3 else None # 自分が送出した(または同一ホスト発)フレームは送り返さない if pkttype == PACKET_OUTGOING: continue if rs is s1: s2.send(data) cnt12 += 1 else: s1.send(data) cnt21 += 1 except BlockingIOError: continue except OSError as e: # インターフェイスがダウンした等 print(f"[!] socket error: {e}", file=sys.stderr) time.sleep(0.2) # 簡易統計 if print_stats and time.time() - last_report >= 5: print(f"[stats] {if1}→{if2}: {cnt12} frames, {if2}→{if1}: {cnt21} frames") last_report = time.time() # 後片付け try: set_promisc(s1, if_nametoindex(if1), False) set_promisc(s2, if_nametoindex(if2), False) except OSError: pass s1.close() s2.close() print("[+] Stopped.") def main(): if os.geteuid() != 0: print("This script must be run as root (RAW sockets).", file=sys.stderr) sys.exit(1) p = argparse.ArgumentParser(description="Simple L2 bridge between two Ethernet interfaces.") p.add_argument("-i1", "--iface1", required=True, help="First interface name (e.g., eth0)") p.add_argument("-i2", "--iface2", required=True, help="Second interface name (e.g., eth1)") p.add_argument("--no-stats", action="store_true", help="Disable periodic stats print") args = p.parse_args() try: forward_loop(args.iface1, args.iface2, print_stats=not args.no_stats) except KeyboardInterrupt: pass if __name__ == "__main__": main()   
ChatGPT:

いいコードです!そのままでも動きますが、「長いメッセージを落とさず捌く」目的なら ソケットバッファ拡張(任意) Jumbo MTU、それから 送信経路の遅延削減 を入れると安定します。
下の改訂版では次を追加しています:

  • --rcvbuf / --sndbuf で OS ソケットバッファを拡張(既定: 16MB)

  • 実際に設定された有効値を getsockopt で取得して表示(Linuxは内部的に2倍などの調整が入るため)

  • --mtu で1回の recvfrom() の読み出し上限を変更(Jumbo Frameを使うなら9000など)

  • 送信側で QdiscバイパスPACKET_QDISC_BYPASS)を有効化してカーネルキューの遅延を減少

  • 小さな後方互換の強化(例外処理など)

注: Ethernetの1フレーム自体はMTUを超えて巨大にはできません。長いアプリデータは上位プロトコル(TCP等)が分割します。ここでは「落としにくくするための受送信キュー拡張」と「Jumbo対応」を行っています。


改訂版 l2bridge.py

#!/usr/bin/env python3 # l2bridge.py (buffer-tuned) import argparse import os import select import signal import socket import struct import sys import time ETH_P_ALL = 0x0003 # すべてのEtherType SOL_PACKET = 263 # from linux/sol_packet.h PACKET_ADD_MEMBERSHIP = 1 PACKET_DROP_MEMBERSHIP = 2 PACKET_MR_PROMISC = 1 PACKET_OUTGOING = 4 PACKET_QDISC_BYPASS = 20 # bypass qdisc for lower TX latency (Linux >=3.14) RUNNING = True def set_promisc(sock, ifindex: int, enable: bool = True): mreq = struct.pack("IHH8s", ifindex, PACKET_MR_PROMISC, 0, b"\x00"*8) opt = PACKET_ADD_MEMBERSHIP if enable else PACKET_DROP_MEMBERSHIP sock.setsockopt(SOL_PACKET, opt, mreq) def if_nametoindex(ifname: str) -> int: return socket.if_nametoindex(ifname) def open_raw_socket(ifname: str, rcvbuf: int, sndbuf: int, qdisc_bypass: bool) -> socket.socket: s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) # 受信/送信バッファ拡大(OSが内部で丸める/倍加することがある) if rcvbuf: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rcvbuf) except OSError: pass if sndbuf: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, sndbuf) except OSError: pass # 送信経路の遅延を抑える(best-effort) if qdisc_bypass: try: s.setsockopt(SOL_PACKET, PACKET_QDISC_BYPASS, 1) except OSError: pass # インターフェイスにバインド s.bind((ifname, 0)) return s def human_bytes(n: int) -> str: for unit in ["B","KB","MB","GB"]: if n < 1024 or unit == "GB": return f"{n:.0f} {unit}" n /= 1024.0 def forward_loop(if1: str, if2: str, mtu: int, rcvbuf: int, sndbuf: int, qdisc_bypass: bool, print_stats: bool = True): s1 = open_raw_socket(if1, rcvbuf, sndbuf, qdisc_bypass) s2 = open_raw_socket(if2, rcvbuf, sndbuf, qdisc_bypass) # プロミスキャス有効化 set_promisc(s1, if_nametoindex(if1), True) set_promisc(s2, if_nametoindex(if2), True) # カーネルが実際に適用したサイズを表示(デバッグ/検証用) eff_rcv1 = s1.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) eff_snd1 = s1.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF) eff_rcv2 = s2.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) eff_snd2 = s2.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF) cnt12 = cnt21 = 0 last_report = time.time() def _cleanup(*_): global RUNNING RUNNING = False signal.signal(signal.SIGINT, _cleanup) signal.signal(signal.SIGTERM, _cleanup) print(f"[+] Bridging {if1} <-> {if2} (Ctrl-C to stop)") print(f" {if1}: RCV={human_bytes(eff_rcv1)}, SND={human_bytes(eff_snd1)} | " f"{if2}: RCV={human_bytes(eff_rcv2)}, SND={human_bytes(eff_snd2)}") print(f" recv buffer per call (mtu): {mtu} bytes" + (" | qdisc_bypass=on" if qdisc_bypass else "")) # selectで両方監視 while RUNNING: r, _, _ = select.select([s1, s2], [], [], 1.0) for rs in r: try: # data: bytes, addr: (proto, ifindex, pkttype, halen, addrbytes) data, addr = rs.recvfrom(mtu) pkttype = addr[2] if len(addr) >= 3 else None # 自分が送出した(または同一ホスト発)フレームは送り返さない if pkttype == PACKET_OUTGOING: continue if rs is s1: s2.send(data) cnt12 += 1 else: s1.send(data) cnt21 += 1 except (BlockingIOError, InterruptedError): continue except OSError as e: # インターフェイスがダウンした等 print(f"[!] socket error: {e}", file=sys.stderr) time.sleep(0.2) # 簡易統計 if print_stats and time.time() - last_report >= 5: print(f"[stats] {if1}{if2}: {cnt12} frames, {if2}{if1}: {cnt21} frames") last_report = time.time() # 後片付け try: set_promisc(s1, if_nametoindex(if1), False) set_promisc(s2, if_nametoindex(if2), False) except OSError: pass s1.close() s2.close() print("[+] Stopped.") def main(): if os.geteuid() != 0: print("This script must be run as root (RAW sockets).", file=sys.stderr) sys.exit(1) p = argparse.ArgumentParser(description="Simple L2 bridge between two Ethernet interfaces (buffer-tuned).") p.add_argument("-i1", "--iface1", required=True, help="First interface name (e.g., eth0)") p.add_argument("-i2", "--iface2", required=True, help="Second interface name (e.g., eth1)") p.add_argument("--mtu", type=int, default=65535, help="recv() buffer size per frame (Jumboなら9000など)") p.add_argument("--rcvbuf", type=int, default=16*1024*1024, help="SO_RCVBUF bytes") p.add_argument("--sndbuf", type=int, default=16*1024*1024, help="SO_SNDBUF bytes") p.add_argument("--qdisc-bypass", action="store_true", help="Enable PACKET_QDISC_BYPASS for low-latency TX") p.add_argument("--no-stats", action="store_true", help="Disable periodic stats print") args = p.parse_args() try: forward_loop( args.iface1, args.iface2, mtu=args.mtu, rcvbuf=args.rcvbuf, sndbuf=args.sndbuf, qdisc_bypass=args.qdisc_bypass, print_stats=not args.no_stats ) except KeyboardInterrupt: pass if __name__ == "__main__": main()

使い方例

# 標準: 受信/送信バッファを16MB、MTUは65535バイト読み(上限) sudo ./l2bridge.py -i1 eth0 -i2 eth1 # バッファを32MBに、Jumboフレーム(例: 9000B想定)に合わせて読み出しサイズも変更 sudo ip link set eth0 mtu 9000 sudo ip link set eth1 mtu 9000 sudo ./l2bridge.py -i1 eth0 -i2 eth1 --rcvbuf 33554432 --sndbuf 33554432 --mtu 9000 --qdisc-bypass

補足(性能をさらに攻めるなら)

  • TPACKET_V3(PACKET_RX_RING) を使ったリングバッファI/Oにするとドロップ耐性が大幅に上がります(実装は少し大きくなる)。

  • マルチキュー / CPUピン留めtasksetSO_BUSY_POLL)で遅延を詰められます。

  • 実転送が最優先なら OSのブリッジ(ip link add br0 type bridge が最も高速・堅牢です。

この改訂版で、バッファ不足による取りこぼしはかなり減るはず。必要なら TPACKET_V3 版のテンプレも出します。

あなた:
このコードをジャンボフレームに対応するように変更してください
ChatGPT:

了解です。頂いた l2bridge.pyジャンボフレーム(例: MTU=9000)対応に強化した版を示します。
ポイントは次のとおりです。

  • 受信バッファ長(recvfrom() の第1引数)を MTUに合わせて可変

  • SO_RCVBUF / SO_SNDBUF を十分に拡張(既定 16MB)

  • 任意で インターフェースMTUをioctlで設定--set-mtu

  • (任意)qdiscバイパスで TX 遅延を削減

重要: 実際にジャンボフレームを通すには、両方のNIC・スイッチ・対向機器すべてで MTU を揃える必要があります。
本スクリプトの --set-mtu は NIC 側の MTU 設定(root 必須)を行います。スイッチ側は別途設定してください。


変更済みコード(そのまま置き換え可)

#!/usr/bin/env python3 # l2bridge.py (jumbo-ready) import argparse import os import select import signal import socket import struct import sys import time ETH_P_ALL = 0x0003 # all EtherTypes SOL_PACKET = 263 PACKET_ADD_MEMBERSHIP = 1 PACKET_DROP_MEMBERSHIP = 2 PACKET_MR_PROMISC = 1 PACKET_OUTGOING = 4 PACKET_QDISC_BYPASS = 20 # optional, Linux >= 3.14 SIOCGIFMTU = 0x8921 SIOCSIFMTU = 0x8922 RUNNING = True def set_promisc(sock, ifindex: int, enable: bool = True): mreq = struct.pack("IHH8s", ifindex, PACKET_MR_PROMISC, 0, b"\x00"*8) opt = PACKET_ADD_MEMBERSHIP if enable else PACKET_DROP_MEMBERSHIP sock.setsockopt(SOL_PACKET, opt, mreq) def if_nametoindex(ifname: str) -> int: return socket.if_nametoindex(ifname) def get_mtu(ifname: str) -> int: # struct ifreq { char ifr_name[16]; union { int ifru_mtu; ... } ifr_ifru; } s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) ifreq = struct.pack("16sH14s", ifname.encode(), 0, b"\x00"*14) try: res = fcntl_ioctl(s.fileno(), SIOCGIFMTU, ifreq) _, mtu, _ = struct.unpack("16sH14s", res) return int(mtu) except Exception: return -1 finally: s.close() def set_mtu_ioctl(ifname: str, mtu: int): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # struct ifreq with ifr_mtu (int) packed into the last 4 bytes ifreq = struct.pack("16sI", ifname.encode(), mtu) # pad to sizeof(struct ifreq) ~ 40 bytes on 64-bit ifreq = ifreq.ljust(40, b"\x00") try: fcntl_ioctl(s.fileno(), SIOCSIFMTU, ifreq) finally: s.close() def fcntl_ioctl(fd, op, data): # 小さなヘルパ(import fcntl を避けるため) import fcntl as _fcntl return _fcntl.ioctl(fd, op, data) def open_raw_socket(ifname: str, rcvbuf: int, sndbuf: int, qdisc_bypass: bool) -> socket.socket: s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) # OSソケットバッファを十分に拡張 if rcvbuf: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rcvbuf) except OSError: pass if sndbuf: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, sndbuf) except OSError: pass # qdiscバイパス(任意) if qdisc_bypass: try: s.setsockopt(SOL_PACKET, PACKET_QDISC_BYPASS, 1) except OSError: pass s.bind((ifname, 0)) return s def human_bytes(n: int) -> str: for unit in ["B","KB","MB","GB"]: if n < 1024 or unit == "GB": return f"{n:.0f} {unit}" n /= 1024.0 def forward_loop(if1: str, if2: str, recv_len: int, rcvbuf: int, sndbuf: int, qdisc_bypass: bool, print_stats: bool = True): s1 = open_raw_socket(if1, rcvbuf, sndbuf, qdisc_bypass) s2 = open_raw_socket(if2, rcvbuf, sndbuf, qdisc_bypass) # プロミスキャス set_promisc(s1, if_nametoindex(if1), True) set_promisc(s2, if_nametoindex(if2), True) eff_rcv1 = s1.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) eff_snd1 = s1.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF) eff_rcv2 = s2.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) eff_snd2 = s2.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF) cnt12 = cnt21 = 0 last_report = time.time() def _cleanup(*_): global RUNNING RUNNING = False signal.signal(signal.SIGINT, _cleanup) signal.signal(signal.SIGTERM, _cleanup) print(f"[+] Bridging {if1} <-> {if2} (Ctrl-C to stop)") print(f" {if1}: RCV={human_bytes(eff_rcv1)}, SND={human_bytes(eff_snd1)} | " f"{if2}: RCV={human_bytes(eff_rcv2)}, SND={human_bytes(eff_snd2)}") print(f" recvfrom() size: {recv_len} bytes" + (" | qdisc_bypass=on" if qdisc_bypass else "")) while RUNNING: r, _, _ = select.select([s1, s2], [], [], 1.0) for rs in r: try: data, addr = rs.recvfrom(recv_len) pkttype = addr[2] if len(addr) >= 3 else None if pkttype == PACKET_OUTGOING: continue if rs is s1: s2.send(data); cnt12 += 1 else: s1.send(data); cnt21 += 1 except (BlockingIOError, InterruptedError): continue except OSError as e: print(f"[!] socket error: {e}", file=sys.stderr) time.sleep(0.2) if print_stats and time.time() - last_report >= 5: print(f"[stats] {if1}{if2}: {cnt12} frames, {if2}{if1}: {cnt21} frames") last_report = time.time() try: set_promisc(s1, if_nametoindex(if1), False) set_promisc(s2, if_nametoindex(if2), False) except OSError: pass s1.close(); s2.close() print("[+] Stopped.") def main(): if os.geteuid() != 0: print("This script must be run as root (RAW sockets).", file=sys.stderr) sys.exit(1) p = argparse.ArgumentParser(description="Simple L2 bridge between two Ethernet interfaces (jumbo-ready).") p.add_argument("-i1", "--iface1", required=True, help="First interface (e.g., eth0)") p.add_argument("-i2", "--iface2", required=True, help="Second interface (e.g., eth1)") # 受信バッファ長(1回のrecvfromで読み込む最大サイズ)。Jumboなら9000以上を推奨。 p.add_argument("--recv-len", type=int, default=16384, help="recvfrom() size in bytes (>= 実際のフレーム長; default 16384)") p.add_argument("--rcvbuf", type=int, default=16*1024*1024, help="SO_RCVBUF bytes (default 16MB)") p.add_argument("--sndbuf", type=int, default=16*1024*1024, help="SO_SNDBUF bytes (default 16MB)") p.add_argument("--qdisc-bypass", action="store_true", help="Enable PACKET_QDISC_BYPASS") # 任意: NICのMTUをioctlで設定 p.add_argument("--set-mtu", type=int, help="Set MTU on both interfaces via ioctl (root required)") p.add_argument("--no-stats", action="store_true", help="Disable periodic stats print") args = p.parse_args() # 任意: MTU設定(NIC側) if args.set_mtu: for ifn in (args.iface1, args.iface2): try: set_mtu_ioctl(ifn, args.set_mtu) print(f"[+] Set MTU {args.set_mtu} on {ifn}") except OSError as e: print(f"[!] Failed to set MTU on {ifn}: {e}", file=sys.stderr) try: forward_loop( args.iface1, args.iface2, recv_len=args.recv_len, rcvbuf=args.rcvbuf, sndbuf=args.sndbuf, qdisc_bypass=args.qdisc_bypass, print_stats=not args.no_stats ) except KeyboardInterrupt: pass if __name__ == "__main__": main()

使い方

# 1) スイッチ/対向もジャンボ対応済みの場合 sudo ./l2bridge.py -i1 eth0 -i2 eth1 --recv-len 16384 --qdisc-bypass # 2) NICのMTUをこのツールから設定してブリッジ(例: 9000) sudo ./l2bridge.py -i1 eth0 -i2 eth1 --set-mtu 9000 --recv-len 16384 --qdisc-bypass # 3) さらにバッファ増強(ドロップ耐性向上) sudo ./l2bridge.py -i1 eth0 -i2 eth1 --set-mtu 9000 --recv-len 16384 \ --rcvbuf 33554432 --sndbuf 33554432 --qdisc-bypass

実運用のコツ

  • --recv-len実際の最大フレーム長以上にしてください(例: 9000Bなら 16384 にして余裕を持たせる)。

  • 両IFとも 同じMTU に揃えること。経路途中のスイッチ/ルータもジャンボ許可が必要です。

  • 取りこぼしがある場合は --rcvbuf/--sndbuf を増やし、CPU使用率が高いなら taskset でCPUピン留めや ethtool -K でオフロード調整を検討してください。

必要なら、TPACKET_V3(PACKET_RX_RING)版でさらにドロップを抑える実装もお出しします。

あなた:
TPACKET_V3(PACKET_RX_RING)版をお願いします
ChatGPT:

やります!Linuxの TPACKET_V3(PACKET_RX_RING) を使って、受信をカーネル共有リング経由で行う L2 ブリッジの Python 実装を示します。
ジャンボフレーム(MTU 9000 など)にも対応しやすいようにパラメータ化しています。送信は通常の send() を使い、受信だけを V3 リングで高速化しています(TX_RING まで入れるとコードがかなり大きくなるため)。

必要: Linux, root, 64bit 環境想定(構造体サイズは x86_64 の UAPI に合わせています)


l2bridge_tpacketv3.py

#!/usr/bin/env python3 # -*- coding: utf-8 -*- # l2bridge_tpacketv3.py # # L2 bridge using AF_PACKET TPACKET_V3 (PACKET_RX_RING) for RX side. # - Linux / root required # - Jumbo-friendly (tunable block/frame sizes) # import argparse import ctypes as C import mmap import os import select import signal import socket import struct import sys import time # ---------- Linux constants ---------- ETH_P_ALL = 0x0003 SOL_PACKET = 263 PACKET_ADD_MEMBERSHIP = 1 PACKET_DROP_MEMBERSHIP = 2 PACKET_MR_PROMISC = 1 PACKET_OUTGOING = 4 PACKET_VERSION = 10 TPACKET_V3 = 3 PACKET_RX_RING = 5 PACKET_QDISC_BYPASS = 20 TP_STATUS_KERNEL = 0 TP_STATUS_USER = 1 # poll/select POLLIN = 0x0001 # ---------- ctypes structures from linux/uapi ---------- # struct tpacket_req3 class tpacket_req3(C.Structure): _fields_ = [ ("tp_block_size", C.c_uint), ("tp_block_nr", C.c_uint), ("tp_frame_size", C.c_uint), ("tp_frame_nr", C.c_uint), ("tp_retire_blk_tov", C.c_uint), ("tp_sizeof_priv", C.c_uint), ("tp_feature_req_word", C.c_uint), ] # struct tpacket_bd_ts class tpacket_bd_ts(C.Structure): _fields_ = [("ts_sec", C.c_uint), ("ts_nsec", C.c_uint)] # struct tpacket_hdr_v1 (block header core) class tpacket_hdr_v1(C.Structure): _fields_ = [ ("block_status", C.c_uint), ("num_pkts", C.c_uint), ("offset_to_first_pkt", C.c_uint), ("blk_len", C.c_uint), ("seq_num", C.c_ulonglong), ("ts_first_pkt", tpacket_bd_ts), ("ts_last_pkt", tpacket_bd_ts), ] # struct tpacket_block_desc: version & offset + hdr_v1 class tpacket_block_desc(C.Structure): _fields_ = [ ("version", C.c_uint), ("offset_to_priv", C.c_uint), ("hdr", tpacket_hdr_v1), ] # struct tpacket3_hdr (per-packet header inside a block) class tpacket3_hdr(C.Structure): _fields_ = [ ("tp_next_offset", C.c_uint), ("tp_sec", C.c_uint), ("tp_nsec", C.c_uint), ("tp_snaplen", C.c_uint), ("tp_len", C.c_uint), ("tp_status", C.c_uint), ("tp_mac", C.c_ushort), ("tp_net", C.c_ushort), ("hv1", C.c_ushort), # placeholder (vlan etc.), not used here ("hv2", C.c_ushort), ] # ---------- helpers ---------- def set_promisc(sock: socket.socket, ifindex: int, enable: bool = True): mreq = struct.pack("IHH8s", ifindex, PACKET_MR_PROMISC, 0, b"\x00"*8) opt = PACKET_ADD_MEMBERSHIP if enable else PACKET_DROP_MEMBERSHIP sock.setsockopt(SOL_PACKET, opt, mreq) def open_rx_ring_sock(ifname: str, block_size: int, block_nr: int, frame_size: int, retire_tov_ms: int, rcvbuf: int, qdisc_bypass: bool): s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) # ring 前に version を V3 へ s.setsockopt(SOL_PACKET, PACKET_VERSION, struct.pack("I", TPACKET_V3)) if rcvbuf: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, int(rcvbuf)) except OSError: pass if qdisc_bypass: try: s.setsockopt(SOL_PACKET, PACKET_QDISC_BYPASS, 1) except OSError: pass # bind iface s.bind((ifname, 0)) # ring params # frame_nr は block_size/frame_size * block_nr frames_per_block = block_size // frame_size frame_nr = frames_per_block * block_nr req = tpacket_req3() req.tp_block_size = block_size req.tp_block_nr = block_nr req.tp_frame_size = frame_size req.tp_frame_nr = frame_nr req.tp_retire_blk_tov = retire_tov_ms req.tp_sizeof_priv = 0 req.tp_feature_req_word = 0 # apply RX_RING s.setsockopt(SOL_PACKET, PACKET_RX_RING, bytes(req)) # mmap ring ring_len = block_size * block_nr ring = mmap.mmap(s.fileno(), ring_len, flags=mmap.MAP_SHARED, prot=(mmap.PROT_READ | mmap.PROT_WRITE), offset=0) # promisc set_promisc(s, socket.if_nametoindex(ifname), True) return s, ring, frames_per_block, ring_len def close_rx_ring_sock(s: socket.socket, ring: mmap.mmap, ifname: str): try: set_promisc(s, socket.if_nametoindex(ifname), False) except OSError: pass try: ring.close() finally: s.close() def human(n: int): for u in ("B", "KB", "MB", "GB"): if n < 1024 or u == "GB": return f"{n:.0f}{u}" n /= 1024 # ---------- ring reader ---------- class RingReaderV3: def __init__(self, sock: socket.socket, ring: mmap.mmap, block_size: int, block_nr: int, frames_per_block: int): self.sock = sock self.ring = ring self.block_size = block_size self.block_nr = block_nr self.frames_per_block = frames_per_block self.cur_block = 0 self.block_desc_size = C.sizeof(tpacket_block_desc) self.pfd = select.poll() self.pfd.register(self.sock.fileno(), POLLIN) def _block_ptr(self, bidx: int): off = bidx * self.block_size return off def _as_block_desc(self, block_offset: int): # create ctypes view from mmap buffer buf = (C.c_char * self.block_desc_size).from_buffer(self.ring, block_offset) desc = tpacket_block_desc.from_buffer(buf) return desc def _packet_hdr_at(self, base_off: int, pkt_off: int): ptr = base_off + pkt_off hdr = tpacket3_hdr.from_buffer((C.c_char * C.sizeof(tpacket3_hdr)).from_buffer(self.ring, ptr)) return hdr, ptr def next_packets(self): # 1ブロックずつ処理。ユーザ側に渡す: [(data_mv1), (data_mv2), ...] while True: boff = self._block_ptr(self.cur_block) desc = self._as_block_desc(boff) if desc.hdr.block_status & TP_STATUS_USER == 0: # データ未到着。pollで待つ。 self.pfd.poll(1000) # 1s timeout # もう一度確認 if desc.hdr.block_status & TP_STATUS_USER == 0: continue # データあり pkts = [] pkt_off = desc.hdr.offset_to_first_pkt for _ in range(desc.hdr.num_pkts): ph, ph_off = self._packet_hdr_at(boff, pkt_off) data_off = ph_off + ph.tp_mac data_end = data_off + ph.tp_snaplen # memoryviewでゼロコピー参照 mv = memoryview(self.ring)[data_off:data_end] pkts.append(mv) if ph.tp_next_offset == 0: break pkt_off += ph.tp_next_offset # 処理済みとして KERNEL に返す desc.hdr.block_status = TP_STATUS_KERNEL # 次ブロックへ self.cur_block = (self.cur_block + 1) % self.block_nr return pkts # ---------- main bridge ---------- RUNNING = True def _cleanup(*_): global RUNNING RUNNING = False def main(): ap = argparse.ArgumentParser(description="L2 bridge using TPACKET_V3 RX ring (jumbo-friendly).") ap.add_argument("-i1", "--iface1", required=True) ap.add_argument("-i2", "--iface2", required=True) # --- ring params (tunable) --- ap.add_argument("--block-size", type=int, default=1<<20, help="block size bytes (default 1MB)") ap.add_argument("--block-nr", type=int, default=64, help="number of blocks (default 64)") ap.add_argument("--frame-size", type=int, default=2048, help="frame size bytes (default 2048; >= MTU + headroom)") ap.add_argument("--retire-tov", type=int, default=60, help="retire block timeout ms (default 60)") ap.add_argument("--rcvbuf", type=int, default=32*1024*1024, help="SO_RCVBUF") ap.add_argument("--sndbuf", type=int, default=32*1024*1024, help="SO_SNDBUF") ap.add_argument("--qdisc-bypass", action="store_true", help="Enable PACKET_QDISC_BYPASS") ap.add_argument("--no-stats", action="store_true") args = ap.parse_args() if os.geteuid() != 0: print("Run as root.", file=sys.stderr) sys.exit(1) # open TX sockets (普通のRAW) と RXリングソケットを分ける # RXリング: s1r/s2r, 送信用: s1tx/s2tx s1r, ring1, fpb1, rlen1 = open_rx_ring_sock( args.iface1, args.block_size, args.block_nr, args.frame_size, args.retire_tov, args.rcvbuf, args.qdisc_bypass) s2r, ring2, fpb2, rlen2 = open_rx_ring_sock( args.iface2, args.block_size, args.block_nr, args.frame_size, args.retire_tov, args.rcvbuf, args.qdisc_bypass) # TX ソケット(通常 RAW)。送信バッファ拡張。 s1tx = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s1tx.bind((args.iface1, 0)) s2tx = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s2tx.bind((args.iface2, 0)) try: s1tx.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, args.sndbuf) except OSError: pass try: s2tx.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, args.sndbuf) except OSError: pass if args.qdisc_bypass: for txs in (s1tx, s2tx): try: txs.setsockopt(SOL_PACKET, PACKET_QDISC_BYPASS, 1) except OSError: pass rr1 = RingReaderV3(s1r, ring1, args.block_size, args.block_nr, fpb1) rr2 = RingReaderV3(s2r, ring2, args.block_size, args.block_nr, fpb2) cnt12 = cnt21 = 0 last = time.time() signal.signal(signal.SIGINT, _cleanup) signal.signal(signal.SIGTERM, _cleanup) print(f"[+] TPACKET_V3 bridge {args.iface1} <-> {args.iface2}") print(f" block={human(args.block_size)} x {args.block_nr}, frame_size={args.frame_size}, rcvbuf={human(args.rcvbuf)} sndbuf={human(args.sndbuf)}") print(" HINT: For jumbo (MTU 9000), set frame_size >= 10240 to keep headroom.") while RUNNING: # それぞれのリングからまとめてパケットを受け取り、反対側へ送信 # poll は各 RingReader 内で実施(1ブロック単位) pkts1 = rr1.next_packets() for mv in pkts1: # 自ホスト発(PACKET_OUTGOING)の除外は RX_RING 側で既に行われる try: s2tx.send(mv) # memoryview をそのまま送る cnt12 += 1 except OSError: pass pkts2 = rr2.next_packets() for mv in pkts2: try: s1tx.send(mv) cnt21 += 1 except OSError: pass if not args.no_stats and time.time() - last >= 5: print(f"[stats] {args.iface1}{args.iface2}: {cnt12} | {args.iface2}{args.iface1}: {cnt21}") last = time.time() # cleanup close_rx_ring_sock(s1r, ring1, args.iface1) close_rx_ring_sock(s2r, ring2, args.iface2) s1tx.close(); s2tx.close() print("[+] Stopped.") if __name__ == "__main__": main()

使い方

# 標準設定で起動(1MBブロック x 64、フレーム 2048B) sudo ./l2bridge_tpacketv3.py -i1 eth0 -i2 eth1 # ジャンボ(MTU 9000)なら余裕を持って frame_size を大きめに sudo ./l2bridge_tpacketv3.py -i1 eth0 -i2 eth1 --frame-size 12288 --block-size $((1<<20)) --block-nr 64 # バッファ増強&低遅延(qdiscバイパス) sudo ./l2bridge_tpacketv3.py -i1 eth0 -i2 eth1 --frame-size 12288 --rcvbuf $((64*1024*1024)) \ --sndbuf $((64*1024*1024)) --qdisc-bypass

パラメータ設計の目安

  • frame_size: MTU + 512 程度に(アラインメントやメタデータ分の余裕)。MTU 9000 → 12288(12KB)推奨。

  • block_size: frame_size * N がページサイズの倍数になるように。1MB(1048576)などが扱いやすいです。

  • block_nr: トラフィックに合わせて。合計リングサイズ block_size * block_nr は数十〜数百MBまで増やすとドロップ耐性向上。

注意・補足

  • 経路上の全機器(NIC/スイッチ/対向)が同じ MTU で揃っていないとジャンボは通りません。

  • 本コードは RX側だけ V3 リング を使っています。TX もリングにするとさらにレイテンシを詰められますが、実装が大幅に複雑になります。

  • カーネルの UAPI に強く依存します。64bit Linux/x86_64 を想定した ctypes 定義です。環境差異がある場合は ctypes フィールドサイズを調整してください。

  • さらにスループットを上げたい場合:

    • CPUピン留め(taskset)、RPS/RFS チューニング、ethtool -G で NIC リング調整

    • マルチキュー(キューごとにソケットを分け、スレッドで並列受信)

    • VLAN/PPPoE 等のヘッダ有無に応じて frame_size を増やす

必要であれば TX_RING 併用版 や、マルチキュー対応(キューごとにスレッド)版 も作ります。気になる要件(目標スループット、遅延、CPU数)を教えてください。