小さな非同期サーバーを実装する

このトピックを公開する目的は、Habrahabrのオーディエンスに、ほとんど裸のソケットを使用してPythonで記述された小さな非同期サーバーのコードを提示することです。

ネットワークサービスとして動作する非常に多くのアプリケーションを作成する必要がありました。 これらのサービスは、異なる負荷で異なる言語で記述されており、そのたびに新しいサービスの実装は以前のものと多少異なりました。 habrakatの下で、私の意見では、「トレーニング」サーバーの実装がかなり成功している例を示し、必要に応じてコメントをコードに添付しています。

導入語

そのため、最初にサーバーが実際に行うことについていくつかの言葉を紹介します。 これはトレーニングの例であるため、タスクは非常に簡単です。起動時に、ポートでリッスンするソケットを作成します。 クライアントが接続したら、情報の送信を開始します。 情報として、クライアントはサーバーで計算された数字Piの桁の無限ストリームを受け取ります。 サーバーは、少なくとも1つのクライアントが接続されている場合にのみ番号を生成し、既に生成されたシーケンスを記憶します。 すべてのクライアントが切断されると、生成が停止し、サーバーは新しい接続を待ちます。

サーバー実装はシングルスレッドです。 このアプローチの利点は、一般的に使用される他の手法を使用する場合に発生する、別個のスレッドまたは別個のアプリケーションインスタンスという形で追加のオーバーヘッドがないことです。

概して、この違いは決定的なものとはほとんど言えません。 Pythonは依然としてインタープリター言語であり、使用領域に制限がありますが、将来、アプリケーションでこのアーキテクチャに「保存」されたスレッドを使用して、並列コンピューティングタスクを実行できるようになります。

続行する前に、コードをダウンロードして流 readingに読むことを強くお勧めします 。 また、コードはshow codeでオンラインで表示できます 。
サーバーを起動するために、UNIXライクシステムのユーザーは次のコマンドを実行できます。

chmod + x ./pypi.py
./pypi.py

サーバーに接続するには、次のコマンドを使用できます。

telnet 127.0.0.1 45067

コード解析

アプリケーションの中核は、 asyncoreモジュールのディスパッチサイクルを開始するmain（）関数です。

def main （ ） ：
試してください ：
非同期 ループ （ 0.1 、 True 、サーバー（ '127.0.0.1' ） ）
KeyboardInterrupt を除く ：
' \ n Bye：-*'を 印刷します
sys 終了 （ 0 ）

__name__ == '__main__'の場合 ：
メイン（ ）

asyncoreモジュールは、4つのオプション引数を取るループ関数を提供します。（1）タイムアウト、（2）通常のselectよりもポーリングメカニズムの優先フラグ、（3）ソケット記述子辞書、（4）反復回数。 3番目のパラメーターは重要です。これにより、新しく作成されたサーバーオブジェクトを関数に渡しました。

「特別なPythonマジック」のおかげで、ServerクラスはasyncoreモジュールのDispatcherクラスとdict辞書クラスの両方から継承されます。これにより、接続されたすべてのクライアントのサーバーソケットとソケット記述子のリポジトリの両方として機能します。

サーバークラス宣言の開始は次のとおりです。

クラスサーバー（ディスパッチャー、 dict ） ：
書き込み可能= ラムダ x： False

def __init__ （ self 、host = None 、port = 0xB00B ） ：
ディスパッチャ。 __init__ （ self ）

自己 。 create_socket （ AF_INET、SOCK_STREAM ）
dict 。 __init__ （ self 、 { self。fileno （ ） ： self } ）

自己 。 set_reuse_addr （ ）
自己 。 bind （ （ホスト、ポート） ）
自己 。 聞く （ 0xA ）

自己 。 dataSource = PiGenerator （ ）

コンストラクターでは、オブジェクトは最初にサーバーソケットハンドラーとして初期化され、次に単一のレコード（サーバーオブジェクトを指すサーバー自体のソケット記述子）で構成される辞書として初期化されます。 辞書が初期化される前に create_socket関数によってソケットが作成されることが重要です。 ソケットを作成する前に、ハンドルを取得できませんでした。 次に、サーバーソケットが指定されたホストのポートにバインドされ、盗聴が開始され、Pi番号ジェネレーターが作成されます。これは、クライアントへのデータストリームの生成にさらに使用されます。

ディスパッチサイクルが開始された後、作業の大部分はasyncoreモジュールに分類されます。このモジュールは、新しいクライアントが接続されると、サーバーオブジェクトのhandle_acceptメソッドを呼び出して着信接続を処理します。

def handle_accept （ self ） ：
sock、 （ host、port ） = self 。 受け入れる （ ）
print 'new client from％s：％d' ％ （ host、port ）

stream = Stream （ self。dataSource ）
自己 [靴下。 fileno （ ） ] =クライアント（ sock、 self 、stream ）

ハンドラメソッド内では、accept関数を使用して新しいクライアントが直接受け入れられます。accept関数は、クライアントおよび接続が発生したホストポートのペアとの通信用に新しく作成されたソケットを返します。 サーバーは、クライアントソケットを受信すると、ジェネレーターからデータを読み取るための新しいデータストリーム（Streamクラスで実装）を作成します。 その後、新しいクライアントオブジェクトがクライアントのリストに追加され、新しく作成されたデータストリームによって初期化されます。

クライアントは、書き込み可能（）メソッド内のストリームからデータを読み取ります。

def書き込み可能（ self ） ：
data = self ストリーム 読む （ ）
データ== なしの 場合 ：
「クライアント終了読み取り」を 印刷し ます
自己 。 閉じる （ ）
偽を返す

自己 。 バッファ + =データ
return len （ self。buffer ） > 0

書き込み可能メソッドは、ディスパッチサイクルの次の反復の前に、各ソケットのasyncoreモジュールによって呼び出され、このソケットの書き込みアクセスをチェックする必要があるかどうかを調べます。 これを使用して、ストリームからデータを読み取ろうとし、ストリームにデータがある場合は書き込みの必要性を報告します。 ストリームがNoneを返す場合、それはそれ以上データがなく、ソケットが閉じられることを意味します。 この例では、このような状況は発生しません。なぜなら、 数は無限に生成されます。

クライアントソケットの書き込み操作の可用性について学習した後、asyncoreはhandle_write（）メソッドを呼び出します。このメソッドは、以前にストリームから読み取ったデータをソケット経由で送信します。

def handle_write （ self ） ：
sent = self 。 send （ self。buffer ）
自己 。 buffer = self バッファ [送信： ]

ジェネレーターとフローは密接に相互接続されており、 オブザーバーパターンを実現しています。 ジェネレーターは監視可能なオブジェクトとして機能し、イベントのサブスクライブとサブスクライブ解除のために、サブスクライブメソッドとサブスクライブ解除メソッドをそれぞれ提供します。

クラス PiGenerator （ リスト ） ：
def subscribe （ self 、obj ） ：
自己 。 ロック 取得 （ ）
試してください ：
自己 。 追加 （ obj ）
self ._notify （ obj = obj ）
最後に ：
自己 。 ロック リリース （ ）

自己で なければ 。 電卓 ：
自己 。 calculator = PiCalcThread （ self 、self。lock ）
自己 。 電卓 開始 （ ）
その他 ：
len （ self ） > 0の場合 ：
self ._resumeCalculator （ ）

def unsubscribe （ self 、obj ） ：
自己 。 ロック 取得 （ ）
自己 。 削除 （ obj ）
自己 。 ロック リリース （ ）

len （ self ） < = 0の場合 ：
self ._pauseCalculator （ ）

数値の直接生成は、個別のスレッドで計算を実行する個別のクラスPiCalcThreadによって実装されるため、同期メカニズムを使用して、サブスクリプションの追加と削除の操作を同期します。 同じブロッキングオブジェクトを使用して、サブスクライブされたスレッドがない場合、スレッドは中断されます。 _notify（）メソッドを使用して新しいストリームをサブスクライブすると、それが最初にサブスクライブされたストリームでない場合、その前に計算および保存された数値が送信されます。

_notify（）メソッドは署名されたストリームを実行し、更新メソッドを呼び出して新しい桁をストリームに渡します。

def _notify （ self 、digits = None 、obj = None ） ：
objs = [ obj ] if objまたはelse
digit = digits または selfです。 桁

objsのobjの場合：
obj。 更新 （数字）

ストリームのupdate（）メソッドは、既存のデータに新しいデータを追加するだけです：

def update （ self 、data ） ：
自己 。 データ + =データ

Pi桁生成ストリームのクラスは、Piの桁のUnbounded Spigot Algorithmで Jeremy Gibbonsによって提案されたアルゴリズムを使用します。

クラス PiCalcThread （スレッド） ：
def __init__ （ self 、buffer、lock ） ：
スレッド __init__ （ self ）
自己 。 バッファー =バッファー
自己 。 ロック =ロック

def run （ self ） ：
q、r、t、k、n、l = 1、0、1、1、3、3

Trueの場合 ：
4 * q + rt < n * tの場合：
自己 。 ロック 取得 （ ）
自己 。 バッファ newDigits （ str （ n ） ）
自己 。 ロック リリース （ ）

q、r、t、k、n、l = （ 10 * q、 10 * （ rn * t ） 、t、k、 （ 10 * （ 3 * q + r ） ） / t- 10 * n、l ）
その他 ：
q、r、t、k、n、l = （ q * k、 （ 2 * q + r ） * l、t * l、k + 1 、 （ q * （ 7 * k + 2 ） + r * l ） / （ t * l ） 、l + 2 ）

時間 。 睡眠 （ 0.001 ）

run（）メソッドはPi番号の次の桁を無限に計算し、再びロックを使用してジェネレータに報告し、ジェネレータにサブスクライブしているデータストリームに渡します。 サーバーが単位時間あたりに大量のデータフローを生成しないように、人工遅延が追加されました。

マテリアルの準備で構文を強調するために、リソースhttp://highlight.hohli.com/が使用されました 。 この記事が誰かにとって有用であることを本当に願っていますが、その説明はかなり大量に散らかっています。