バッチ処理

RPC の設計方針では、クライアントは呼び出しメッセージを送信して、サーバーがそれに応答するのを待ちます。すなわち、サーバーが要求を処理する間、クライアントは停止していることになります。これは、クライアントが各メッセージへの応答を待つ必要がないときには非効率です。

RPC のバッチ処理を使用すると、クライアントは非同期に処理を進めることができます。RPC メッセージは呼び出しパイプラインに入れてサーバーに送られます。バッチ処理では次のことが必要になります。

• サーバーはどのような中間メッセージにも応答しない

• 呼び出しパイプラインは、信頼性の高いトランスポート (たとえば、TCP) で伝送される

• 呼び出し時に指定する、戻り値に対する XDR ルーチンは NULL である

• RPC 呼び出しのタイムアウト値はゼロである

サーバーはそれぞれの呼び出しに対しては応答しないので、クライアントは、サーバーが前の呼び出しを処理している間に平行して次の呼び出しを送信できます。トランスポートは複数の呼び出しメッセージをバッファリングし、システムコール write() で一度にサーバーに送信します。このバッファリングにより、プロセス間通信のオーバヘッドが減少し、一連の呼び出しに要する総時間が短縮されます。クライアントは終了前に、パイプラインをフラッシュする呼び出しをバッチにしないで実行します。

次に、バッチ処理を使用しないクライアント側プログラムを示します。文字配列 buf を走査して文字列を順に取り出し、1 つずつサーバーに送信します。

例 5–4 バッチ処理を使用しないクライアントプログラム

#include <stdio.h>

#include <rpc/rpc.h>
#include "windows.h"
 
main(argc, argv)
	int argc;
	char **argv;
{
	struct timeval total_timeout;
	register CLIENT *client;
	enum clnt_stat clnt_stat;
	char buf[1000], *s = buf;
 
	if ((client = clnt_create( argv[1], WINDOWPROG, WINDOWVERS,
								"circuit_v")) == (CLIENT *) NULL) {
		clnt_pcreateerror("clnt_create");
		exit(1);
	}
 
	total_timeout.tv_sec = 20;
	total_timeout.tv_usec = 0;
	while (scanf( "%s", s ) != EOF) {
		if (clnt_call(client, RENDERSTRING, xdr_wrapstring, &s,
		   xdr_void, (caddr_t) NULL, total_timeout) != RPC_SUCCESS) {
			clnt_perror(client, "rpc");
			exit(1);
		}
	}
 
	clnt_destroy( client );
	exit(0);
}

次に、バッチ処理を使用したクライアント側プログラムを示します。各文字列を送信した後に応答は待ちません。サーバーからの終了応答だけを待ちます。

例 5–5 バッチ処理を使用するクライアントプログラム

#include <stdio.h>

#include <rpc/rpc.h>
#include "windows.h"
 
main(argc, argv)
	int argc;
	char **argv;
{
	struct timeval total_timeout;
	register CLIENT *client;
	enum clnt_stat clnt_stat;
	char buf[1000], *s = buf;
 
	if ((client = clnt_create( argv[1], WINDOWPROG, WINDOWVERS,
									"circuit_v")) == (CLIENT *) NULL) {
		clnt_pcreateerror("clnt_create");
		exit(1);
	}
	timerclear(&total_timeout);
	while (scanf("%s", s) != EOF)
		clnt_call(client, RENDERSTRING_BATCHED, xdr_wrapstring,
		           &s, xdr_void, (caddr_t) NULL, total_timeout);
	/* Now flush the pipeline */
	total_timeout.tv_sec = 20;
	clnt_stat = clnt_call(client, NULLPROC, xdr_void,
	         (caddr_t) NULL, xdr_void, (caddr_t) NULL,
total_timeout);
	if (clnt_stat != RPC_SUCCESS) {
		clnt_perror(client, "rpc");
		exit(1);
	}
	clnt_destroy(client);
	exit(0);
}

次に、バッチ処理を使用した場合のサーバーのディスパッチ部分を示します。サーバーは、メッセージを送信しないので、クライアント側は、失敗に気付きません。

例 5–6 バッチ処理を行うサーバー

#include <stdio.h>
#include <rpc/rpc.h>
#include "windows.h"
 
void
windowdispatch(rqstp, transp)
	struct svc_req *rqstp;
	SVCXPRT *transp;
{
 char    *s = NULL;
 
	switch(rqstp->rq_proc) {
 	case NULLPROC:
			if (!svc_sendreply( transp, xdr_void, NULL))
 			fprintf(stderr, "can't reply to RPC call\n");
			return;
 	case RENDERSTRING:
			if (!svc_getargs( transp, xdr_wrapstring, &s)) {
				fprintf(stderr, "can't decode arguments\n");
 			/* 呼び出し側にエラーを通知 */
				svcerr_decode(transp);
 			break;
			}
			/* 文字列 s を処理するコード */
 		if (!svc_sendreply( transp, xdr_void, (caddr_t) NULL))
	 		fprintf( stderr, "can't reply to RPC call\n");
			break;
 	case RENDERSTRING_BATCHED:
			if (!svc_getargs(transp, xdr_wrapstring, &s)) {
				fprintf(stderr, "can't decode arguments\n");
 			/* プロトコルエラーのため何も返さない */
				break;
			}
 		/* 文字列 s を処理するコード。ただし応答はしない。 */
			break;
 	default:
			svcerr_noproc(transp);
			return;
 }
	/* 引数の復号化で割り当てた文字列を解放 */
	svc_freeargs(transp, xdr_wrapstring, &s);
}

バッチ処理によるパフォーマンスの向上を調べるために、例 5–4、例 5–6 で 25144 行のファイルを処理しました。このサービスは、ファイルの各行を引き渡すだけの簡単なサービスです。バッチ処理を使用した方が、使用しない場合の 4 倍の速さで終了しました。