ZeroMQ。第2章：ソケットの紹介

みなさんこんにちは！

「ZeroMQ.ZeroMQを使用し、さまざまなメッセージパターンを適用する方法を学ぶ」という本の無料翻訳を続けます。 続編をあまり長く公開しなかったことを事前に謝罪しますが、彼らが言うように：「怠azineは私たちの前に生まれました...」。 さて、歌詞はさておきましょう。



内容

• 第1章：はじめに
• 第2章：ソケットの紹介
• 第3章：ソケットトポロジの使用
• 第4章：追加のパターン

前の章でZeroMQの基本構造を調べた後、これでソケットを検討します。

• クライアント/サーバーパターン（パブリッシュ/サブスクライブ）
• パイプラインパターン

パブリッシュ/サブスクライブパターン

最初に、サーバーが特定のクライアントのリストにメッセージを送信するときの配信の性質によって一方向である、クライアントサーバーパターン（パブリッシュサブスクライブ）という古典的なパターンを紹介しましょう。 これは1対多のモデルです。 このパターンの主なアイデアは、サーバーがメッセージを送信し、接続されたクライアントがこのメッセージを受信する一方で、切断されたクライアントは単にそれをスキップすることです。 サーバーはクライアントに緩やかに接続されており、クライアントが存在するかどうかはまったく気にしません。 これは、テレビチャンネルやラジオ局の仕組みに似ています。 テレビチャンネルは常にテレビ番組を放送し、視聴者のみがこの放送を受け入れるかどうかを決定します。 適切な時間を逃すと、お気に入りのテレビ番組を見ることができなくなります（TiVoなどがない場合は、録画が発明されていない世界でスクリプトが行われると仮定しましょう）。 パブリッシュ/サブスクライブテンプレートの利点は、動的なネットワークトポロジを提供することです。

クライアント/サーバーモデルは、次の主な側面から表示できます。

• 公開：投稿は作成者によって投稿されます
• 通知：クライアントにメッセージの公開が通知されます
• サブスクライブ：新しい署名がクライアントに発行されます
• 登録解除：クライアントは既存の署名を削除します

状況を明確にするために例を見てみましょう。 証券取引プログラムを作成するシナリオを考えてみましょう。 ブローカーがいて、彼らは市場でどのような行動が起こっているのか知りたいと思っています。 私たちのサーバーは株式市場になり、ブローカーは顧客になります。

実際の株式市場価格の代わりに、数個の数値を生成するだけです。

コードに移る前に、まず、クライアントサーバーモデルがどのように見えるかを見てみましょう。







作成者（サーバー）のコードは次のとおりです。

/* * Stock Market Server * Binds PUB socket to tcp://*:4040 * Publishes random stock values of random companies */ #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* publisher = zmq_socket(context, ZMQ_PUB); printf("Starting server...\n"); int conn = zmq_bind(publisher, "tcp://*:4040"); const char* companies[2] = {"Company1", "Company2"}; int count = 0; for(;;) { int price = count % 2; int which_company = count % 2; int index = strlen(companies[0]); char update[12]; snprintf(update, sizeof update, "%s", companies[which_company]); zmq_msg_t message; zmq_msg_init_size(&message, index); memcpy(zmq_msg_data(&message), update, index); zmq_msg_send(&message, publisher, 0); zmq_msg_close(&message); count++; } zmq_close(publisher); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

クライアントコードは次のとおりです。

 /* * Stock Market Client * Connects SUB socket to tcp://localhost:4040 * Collects stock exchange values */ #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, 0, 0); int i; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

SUBソケットを使用する場合は常に、 zmq_setsockopt ()



およびsubscribeis



を使用して署名を設定する必要があります。そうしないと、メッセージを受信しません。 これは非常によくある間違いです。

クライアントは、更新がいずれかのサブスクリプションと一致した場合に受信するメッセージのいずれかに多くの署名を設定できます。 また、特定のサブスクリプションを拒否する場合があります。 サブスクリプションは固定長です。

クライアントはzmq_msg_recv



（）を使用してメッセージを受信します。 zmq_msg_recv



（）は、メッセージを受信して​​保存します。 前のメッセージがあれば、アンロードされます。

 int zmq_msg_recv (zmq_msg_t *msg, void *socket, int flags);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

オプションフラグは、 ZMQ_DONTWAIT



1つの値のみを取ることができます。 フラグがZMQ_DONTWAIT



場合、操作は非ブロックモードで実行されます。 メッセージを受信すると、 zmq_msg_recv



（）はメッセージのサイズをバイト単位で返します。それ以外の場合は-1とエラーメッセージフラグが返されます。

クライアント/サーバーモデルは非同期であり、 SUB



ソケットにメッセージを送信するとエラーが発生します。 zmq_msg_send



（）を呼び出してメッセージを送信できますが、PUBソケットでzmq_msg_recv



（）を呼び出さないでください。

以下は、クライアント側の出力の例です。

 Company2 570 Company2 878 Company2 981 Company2 783 Company1 855 Company1 524 Company2 639 Company1 984 Company1 158 Company2 145
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

サーバーは、クライアントがなくても常にメッセージを送信します。 それをテストし、結果を見ることができます。 これを行うと、次のようなものが表示されます。

 Sending... Company2 36 Sending... Company2 215 Sending... Company2 712 Sending... Company2 924 Sending... Company2 721 Sending... Company1 668 Sending... Company2 83 Sending... Company2 209 Sending... Company1 450 Sending... Company1 940 Sending... Company1 57 Sending... Company2 3 Sending... Company1 100 Sending... Company2 947
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Company1、または名前を引数として指定する別の会社の結果を取得するとします。 この場合、クライアントプログラムを次のように変更する必要があります。

 // // Stock Market Client // Connects SUB socket to tcp://localhost:4040 // Collects stock exchange values // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); const char* filter; if(argc > 1) { filter = argv[1]; } else { filter = "Company1"; } printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); int i = 0; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

出力には次のようなものが含まれます。

 Company1 575 Company1 504 Company1 513 Company1 584 Company1 444 Company1 1010 Company1 524 Company1 963 Company1 929 Company1 718
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

メッセージフィルタリング

メインの証券取引所アプリケーションは、顧客にメッセージを送信します。 すべてのメッセージは期待どおりに配信されるようです。 残念ながら、ありません。

サーバーコードを次のように変更しましょう。

 // // Stock Market Server // Binds PUB socket to tcp://*:4040 // Publishes random stock values of random companies // #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* publisher = zmq_socket(context, ZMQ_PUB); int conn = zmq_bind(publisher, "tcp://*:4040"); const char* companies[3] = {"Company1", "Company10", "Company101"}; int count = 0; for(;;) { int price = count % 17; int which_company = count % 3; int index = strlen(companies[which_company]); char update[64]; snprintf(update, sizeof update, "%s", companies[which_company]); zmq_msg_t message; zmq_msg_init_size(&message, index); memcpy(zmq_msg_data(&message), update, index); zmq_msg_send(&message, publisher, 0); zmq_msg_close(&message); count++; } zmq_close(publisher); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

クライアントコードを次のように変更しましょう。

 // // Stock Market Client // Connects SUB socket to tcp://localhost:4040 // Collects stock exchange values // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); const char* filter; if(argc > 1) { filter = argv[1]; } else { filter = "Company1"; } printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); int i = 0; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

この場合、出力は次のようになります。

 Collecting stock information from the server. Company101 950 Company10 707 Company101 55 Company101 343 Company10 111 Company1 651 Company10 287 Company101 8 Company1 889 Company101 536
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

私たちのクライアントコードは、Company1の結果を見たいとはっきりと言っています。 ただし、サーバーはCompany10とCompany101の結果を再度送信します。 もちろん、これは私たちが望むものではありません。 この小さな問題を解決する必要があります。

少しハックして目的のものを取得できますが、セパレータを使用する方が簡単です。

クライアントコードとサーバーコードの両方でいくつかの変更を行う必要があります。セパレータを使用して会社名をフィルタリングします。

以下は、以前の問題を修正する更新されたサーバーコードです。 強調表示された行に注意してください。これらは、セパレータを使用して顧客にメッセージを送信する方法を示しています。

 // // Stock Market Server // Binds PUB socket to tcp://*:4040 // Publishes random stock values of random companies // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* publisher = zmq_socket(context, ZMQ_PUB); int conn = zmq_bind(publisher, "tcp://*:4040"); conn = zmq_bind(publisher, "ipc://stock.ipc"); const char* companies[3] = {"Company1", "Company10", "Company101"}; for(;;) { int price = count % 17; int which_company = count % 3; int index = strlen(companies[which_company]); char update[64]; sprintf(update, "%s| %d", companies[which_company], price); zmq_msg_t message; zmq_msg_init_size(&message, index); memcpy(zmq_msg_data(&message), update, index); zmq_msg_send(&message, publisher, 0); zmq_msg_close(&message); count++; } zmq_close(publisher); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

更新されたクライアントコードを見て、結果をフィルタリングします。

 // // Stock Market Client // Connects SUB socket to tcp://localhost:4040 // Collects stock exchange values // #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* subscriber = zmq_socket(context, ZMQ_SUB); const char* filter; filter = "Company1|"; printf("Collecting stock information from the server.\n"); int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); int i = 0; for(i = 0; i < 10; i++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, subscriber, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); printf("%s\n", value); free(value); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

クライアントとサーバーのコードに加えられた変更の後、予想された結果を正確に見ることができます。

ソケットオプション

クライアントサーバーモデルを使用するため、 ZMQ_SUBSCRIBE



という名前のパラメーターを使用します。

 int conn = zmq_connect(subscriber, "tcp://localhost:4040"); conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, option_value, strlen(option_value));
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ソケットオプションは、zmq_setsockopt（）関数で設定されます。 次の4つのパラメーターが必要です。

• ソケット
• オプション名
• オプション値
• オプションサイズ

これは、次の行から確認できます。

 int zmq_setsockopt (void *socket, int option_name, const void *option_ value, size_t option_len);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

購読する

ZMQ_SUBSCRIBE



は、 ZMQ_SUB



ソケットに新しいメッセージを作成します。 option_value



引数option_value



空でない場合、 option_value



で始まるすべてのメッセージをサブスクライブします。 1つのZMQ_SUB



ソケットに複数のフィルターを構成できます。

登録解除

ZMQ_UNSUBSCRIBE



は、 ZMQ_SUB



ソケットからメッセージを削除します。 複数のフィルターが構成されている場合でも、1つのメッセージのみを削除します。

クライアントとサーバーのソケットについて学習しなければならない主なことは、クライアントがメッセージの受信を開始するタイミングがわからないことです。 この場合、最初にクライアントを起動し、次にサーバーを起動することをお勧めします。 クライアントは常に最初のメッセージをサーバーへの接続として認識するため、多くの時間がかかり、サーバーはこの時点ですでにメッセージを送信している可能性があるためです。

ただし、クライアントが接続されていない場合はメッセージを送信しないため、サーバーとクライアントを同期する方法について説明します。

クライアントサーバーモデルノート

クライアント/サーバーモデルでは、次の点に注意する必要があります。

• TCPを使用すると、メッセージはサーバー側でキューに入れられ、クライアントはメッセージの受信が遅すぎます。 これからアプリケーションを保護する方法を示します。
• クライアントは複数のサーバーに接続できます。 データはフェアキュー戦略を使用して送信されます
• サーバーは、すべてのメッセージをすべてのクライアントに送信し、上記の為替市場プログラムに示すように、フィルタリングはクライアント側で実行されます

第4章では、クライアントサーバーモデルに戻り、より複雑な例を検討し、「遅い」クライアントに対処する方法を示します。

パイプラインパターン

パイプラインモデルについて考えてみましょう。 パイプラインパターンは、パイプライン内の順序付けられたノード間でデータを転送します。 データは継続的に送信され、各ステップでパイプはいくつかのノードのいずれかに接続されます。 ノード間では、周期的なデータ転送戦略が使用されます。 これは、要求/応答なしのモデルに少し似ています。

戦略を分割して征服する

この戦略を分割して征服してください。プログラムするときに逃げ道はありません。 プログラミングの勉強を始めたばかりで、教師がそれをほとんどマージソートで使用し、1週間後にグループの半分がクラスへの参加を停止したことを思い出してください。 このすべてがよく記憶されていると確信しています。 そしてここでも、分割して征服しましょう！

ZeroMQで何か平行して書きましょう。 乱数を生成するジェネレーターがあるシナリオを考えます。 ニュートン法を使用してこれらの数値の平方根を見つけるワーカーがいます。 また、労働者から結果を収集するコレクターがいます。

以下はサーバーコードです。

 #include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <string.h> #include <sys/time.h> #include <time.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); // This is the socket that we send messages. void* socket = zmq_socket(context, ZMQ_PUSH); zmq_bind(socket, "tcp://*:4040"); // This is the socket that we send batch message. void* connector = zmq_socket(context, ZMQ_PUSH); zmq_connect(connector, "tcp://localhost:5050"); printf("Please press enter when workers are ready..."); getchar(); printf("Sending tasks to workers...\n"); // The first message. It's also the signal start of batch. int length = strlen("-1"); zmq_msg_t message; zmq_msg_init_size(&message, length); memcpy(zmq_msg_data(&message), "-1", length); zmq_msg_send(&message, connector, 0); zmq_msg_close(&message); // Generate some random numbers. srandom((unsigned) time(NULL)); // Send the tasks. int count; int msec = 0; for(count = 0; count < 100; count++) { int load = (int) ((double) (100) * random () / RAND_MAX); msec += load; char string[10]; sprintf(string, "%d", load); } printf("Total: %d msec\n", msec); sleep(1); zmq_close(connector); zmq_close(socket); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

また、Newtonメソッドを使用して数値の平方根を計算するためのいくつかの計算を行う従業員コードを見てみましょう。

 #include <stdlib.h> #include <string.h> #include <unistd.h> #include "zmq.h" double square(double x) { return x * x; } double average(double x, double y) { return (x + y) / 2.0; } double good_enough(double guess, double x) { return abs(square(guess) - x) < 0.000001; } double improve(double guess, double x) { return average(guess, x / guess); } double sqrt_inner(double guess, double x) { if(good_enough(guess, x)) return guess; else return sqrt_inner(improve(guess, x), x); } double newton_sqrt(double x) { return sqrt_inner(1.0, x); } int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); // Let's initialize a socket to receive messages. void* receiver = zmq_socket(context, ZMQ_PULL); zmq_connect(receiver, "tcp://localhost:4040"); // Let's initialize a socket to send the messages. void* sender = zmq_socket(context, ZMQ_PUSH); zmq_connect(sender, "tcp://localhost:5050"); for(;;) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, receiver, 0); int length = zmq_msg_size(&reply); char* msg = malloc(length + 1); memcpy(msg, zmq_msg_data(&reply), length); zmq_msg_close(&reply); fflush(stdout); double val = atof(msg); printf("%.1f: %.1f\n", val, newton_sqrt(val)); sleep(1); free(msg); zmq_msg_t message; char* ssend = "T"; int t_length = strlen(ssend); zmq_msg_init_size(&message, t_length); memcpy(zmq_msg_data(&message), ssend, t_length); zmq_msg_send(&message, receiver, 0); zmq_msg_close(&message); } zmq_close(receiver); zmq_close(sender); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コードとコレクターは次のとおりです。

 #include <stdlib.h> #include <string.h> #include "zmq.h" int main (int argc, char const *argv[]) { void* context = zmq_ctx_new(); void* receiver = zmq_socket(context, ZMQ_PULL); zmq_bind(receiver, "tcp://*:5050"); // We receive the first message and discard it since it's the // signal start of batch which is -1. zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, receiver, 0); int length = zmq_msg_size(&reply); char* msg = malloc(length + 1); memcpy(msg, zmq_msg_data(&reply), length); zmq_msg_close(&reply); free(msg); int count; for(count = 0; count < 100; count++) { zmq_msg_t reply; zmq_msg_init(&reply); zmq_msg_recv(&reply, receiver, 0); int length = zmq_msg_size(&reply); char* value = malloc(length + 1); memcpy(value, zmq_msg_data(&reply), length); zmq_msg_close(&reply); free(value); if(count / 10 == 0) printf("10 Tasks have been processed."); fflush(stdout); } zmq_close(receiver); zmq_ctx_destroy(context); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次の図は、上記のコードを示しています。







私たちは何を持っています：

• 最初に、従業員が仕事を始める時間を同期する必要があります。 前述のように、接続プロセスには時間がかかります。 同期しない場合、最初の従業員はメッセージを受信しますが、他の従業員は接続段階にあります。 これを防ぐには、作業の開始を同期して、すべてが並行して動作するようにする必要があります。
• コレクターのPULLソケットは、開発されたフェアキューを使用して結果を受け取ります（ 最初の部分で詳細に説明しました）。
• PUSHサーバーソケットはタスクをワーカーに均等に送信します
• ワーカーはサーバーとコレクターの両方に接続されています。 必要に応じて、より多くの従業員を雇用できます。

ワーカーはサーバーとコレクターの両方に接続されていると述べました。 これらのリンクをさらに詳しく見てみましょう。

従業員コードから次の行を見てみましょう。

 // Let's initialize a socket to receive messages. void* receiver = zmq_socket(context, ZMQ_PULL); zmq_connect(receiver, "tcp://localhost:4040");
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ZMQ_PULLソケット

入力からノードへのデータを取得する場合、 ZMQ_PULL



を使用しZMQ_PULL



。 ソケットタイプZMQ_PULL



、パイプラインの上流ノードからメッセージを受信するために使用されます。 前述したように、このプロセスは公平なキュー計画を使用して実行されます。

ZMQ_PUSHソケット

下位ノードと通信したい場合、 ZMQ_PUSH



を使用しZMQ_PUSH



。 ソケットタイプZMQ_PUSH



、パイプラインの次のノードにメッセージを送信するために使用されます。

ZMQ_PUSH



はメッセージを破棄しません。 上流ノードが下流ノードにメッセージを送信する準備ができているが、後者がメッセージを受信して​​処理する準備ができていない場合、 zmq_send



（）を使用して送信されたすべてのメッセージは、少なくとも1つのノードがメッセージを受信できるようになるまでブロックされます。

ZeroMQコンテキストの取得

ほとんどの場合、前に示したすべての例がzmq_ctx_new（）で始まっていることに気づいたでしょう。 ZeroMQアプリケーションは、常にコンテキストの作成から始まります。 すべてのソケットは、1つのプロセスでスレッドを接続する最も速い方法であるため、ソケットの作成プロセスに関係するコンテキストを使用して、1つのプロセス内で作成されます。 ZeroMQコンテキストはスレッドセーフであるため、スレッド間で簡単に転送できます。

ZeroMQコンテキストを作成できない場合、NULLが返されます。

個別のZeroMQアプリケーションと見なされる複数のコンテキストを作成できるという事実にもかかわらず、最良のアイデアは、1つのコンテキストを作成し、それを他のスレッドに転送することです。

ZeroMQコンテキストデストラクタ

各アプリケーションの最後に、 zmq_ctx_destroy



（）を呼び出して作成したコンテキストを破棄する必要があります。 zmq_ctx_destroy



（）を呼び出した後、すべてのプロセスはエラーコード（ ETERM



）をzmq_ctx_destroy



、 zmq_ctx_destroy



（）はソケットを開く呼び出しをブロックし、 zmq_ctx_destroy



（）を呼び出してそれらを閉じます。

クリーニング

PyhtonやJavaなどのプログラミング言語でプログラミングする場合、これらの言語にはガベージコレクターが組み込まれているため、メモリ管理について心配する必要はありません。

たとえば、Pyhtonは参照カウントを使用します。カウンターがゼロになると、メモリは自動的に解放されます。 したがって、PyhtonでZeroMQアプリケーションを作成するときに、オブジェクトの参照カウントがゼロになるとすぐに自動的に閉じられるため、明示的に接続を閉じないでください。 ただし、これはJython、PyPy、IronPythonでは機能しないことに注意してください。 とにかく、Pythonドキュメントで十分な情報を見つけることができます。 メインタスクに戻りましょう。

Cで記述する場合、メモリ管理は完全にあなたの責任です。 そうしないと、メモリリークが発生する不安定なアプリケーションになります。

ソケットのクローズ、メッセージとZeroMQコンテキストの破壊は自分で行う必要があります。 アプリケーションを正常に完了するために考慮すべきことがいくつかあります。

• 前述のように、アプリケーションを閉じるには、 zmq_ctx_destroy
  
  
  
  （）を呼び出してZeroMQコンテキストを破棄する必要があります。 ただし、開いているソケットがある場合、 zmq_ctx_destroy
  
  
  
  はソケットが無期限に閉じるのを待つことができます。 したがって、最初にすべてのソケットを閉じてから、 zmq_ctx_destroy
  
  
  
  （）を呼び出してコンテキストを破棄する必要があります。
• zmq_ctx_destroy
  
  
  
  （）は、開いている接続がある場合、または送信するキューにメッセージがある場合、無期限に待機します。
• メッセージの処理が終了したら、 zmq_msg_close
  
  
  
  （）を呼び出してすぐにメッセージを閉じる必要があります。そうしないと、アプリケーションでメモリリークが発生します。
• 多くのソケットを開かないでください。 これを行う場合、それはあなたが何か間違ったことをしていることを意味し、アプリケーションをゼロから設計することを意味します。

スペルが間違っている場合、特にマルチスレッドの場合、アプリケーションで何が起こるかを見て驚くかもしれません。 この場合、エラーをキャッチすることは非常に困難です。

メモリリーク検出

メモリ管理は完全にプログラマーの責任であるため、CまたはC ++で記述されたアプリケーションには、適切に作成されたメモリマネージャが必要です。 これを行うには、Valgrindと呼ばれるすばらしいLinuxツールを使用します。 ソースコードの分析に役立つ他の多くの機能の中でも、このツールはメモリリークの検出に使用できます。

次のセクションは、Valgrindの小さなチュートリアルです。このチュートリアルでは、ZeroMQでアプリケーションを作成するときにValgrindを使用する方法を詳しく見ていきます。

Valgrindの概要

-gオプションを使用してアプリケーションをコンパイルし、デバッグ情報を表示できます。 この場合、エラーメッセージには正確な行番号が含まれます。

次の例を考えてみましょう。

 #include <stdio.h> #include <stdlib.h> int main(int argc, char const *argv[]) { char* a = malloc(4); int b; printf("b = %d\n", b); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

gcc –g –o test test.c



入力して、gccでコンパイルしましょう。 次に、Valgrindを実行してメモリリークをチェックします。 次のコマンドを実行してみましょう。

 valgrind --leak-check=full --show-reachable=yes test
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

前のコマンドを入力した後、Valgrindはmemcheckツールを使用してメモリエラーのコードのチェックを開始します。 tool = memcheckを実行して個別に呼び出すことができますが、memcheckがデフォルトのツールであるため、それは無意味です。 出力は次のようになります。

 ==98190== Conditional jump or move depends on uninitialised value(s) ==98190== at 0x2D923: __vfprintf ==98190== by 0x4AC5A: vfprintf_l ==98190== by 0x952BE: printf ==98190== by 0x1F5E: main (test.c:8) ==98190== 4 bytes in 1 blocks are definitely lost in loss record 1 of 5 ==98190== at 0xF656: malloc (vg_replace_malloc.c:195) ==98190== by 0x1F46: main (test.c:6) ==98190== LEAK SUMMARY: ==98190== definitely lost: 4 bytes in 1 blocks ==98190== indirectly lost: 0 bytes in 0 blocks ==98190== possibly lost: 0 bytes in 0 blocks
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここで、以前の出力について少し説明しましょう。

• これはプロセスIDであるため、98190については説明しません。
• Conditional jump or move depends on uninitialised value(s)
  
  
  
  ます。これは、コードの初期化が成功したことを意味します
• definitely lost
  
  
  
  ということは、メモリリークがあり、修正する必要があることを意味します
• indirectly lost
  
  
  
  れると、 indirectly lost
  
  
  
  れる可能性のあるブロックを示します
• possibly lost
  
  
  
  は、メモリリークの可能possibly lost
  
  
  
  ことを意味します

デフォルトでは、Valgrindは$PREFIX/lib/valgrind/default.supp



ます。 ただし、ZeroMQで使用する独自のファイルを作成する必要があります。次のようになります。

 { <socketcall_sendto> Memcheck:Param socketcall.sendto(msg) fun:send ... } { <socketcall_sendto> Memcheck:Param socketcall.send(msg) fun:send ... }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、必要な引数を次のように使用してValgrindを実行できます。

 valgrind --leak-check=full --show-reachable=yes --suppressions=zeromq.supp server
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

おわりに

この章では、ソケットについて説明し、2つの新しいモデル、つまりクライアントサーバーモデルとパイプラインパターンを紹介しました。 また、これらのテンプレートを使用して特定の問題を解決する方法について説明し、簡単な例を使用してこれを示しました。また、Valgrindのメモリリークを見つけるための優れたツールも検討しました。



ご清聴ありがとうございました。PMの翻訳に関するコメントをお送りください。