マルチスレッド RPC プログラミング

$ cc rstat.c -lnsl -lthread

/* @(#)rstat.c 2.3 93/11/30 4.0 RPCSRC */
/*
 * w コマンドと同様の形式で、遠隔ホストのステータスを表示する簡単な
 * プログラム
 */
 
#include <thread.h>		/* スレッドインタフェースの定義 */
#include <synch.h>	/* 相互排他的ロックの定義 */
#include <stdio.h>
#include <sys/param.h>
#include <rpc/rpc.h>
#include <rpcsvc/rstat.h>
#include <errno.h>
 
mutex_t tty;					/* printf のための tty の制御 */
cond_t cv_finish;
int count = 0;
 
main(argc, argv)
int argc;
char **argv;
{
 int i;
	thread_t tid;
	void *do_rstat();
 
	if (argc < 2) {
		fprintf(stderr, "usage: %s ¥"host¥" [...]¥n", argv[0]);
		exit(1);
	}
 
	mutex_lock(&tty);
 
	for (i = 1; i < argc; i++) {
		if (thr_create(NULL, 0, do_rstat, argv[i], 0, &tid) < 0) {
			fprintf(stderr, "thr_create failed: %d¥n", i);
			exit(1);
		} else
 		fprintf(stderr, "tid: %d¥n", tid);
	}
 
	while (count < argc-1) {
		printf("argc = %d, count = %d¥n", argc-1, count);
		cond_wait(&cv_finish, &tty);
 
	}
 
	exit(0);
}
 
bool_t rstatproc_stats();
 
void *
do_rstat(host)
char *host;
{
 CLIENT *rstat_clnt;
 statstime host_stat;
	bool_t rval;
	struct tm *tmp_time;
	struct tm host_time;
 struct tm host_uptime;
	char days_buf[16];
 char hours_buf[16];
 
	mutex_lock(&tty);
 printf("%s: starting¥n", host);
	mutex_unlock(&tty);
 
	/* rstat クライアントハンドル */
	rstat_clnt = clnt_create(host, RSTATPROG, RSTATVERS_TIME,
							"udp");
 if (rstat_clnt == NULL) {
		mutex_lock(&tty);				/* ty の制御権を取得 */
		clnt_pcreateerror(host);
		count++;
 	cond_signal(&cv_finish);
		mutex_unlock(&tty);	/* tty の制御権を解放 */
 
		thr_exit(0);
 
	}
 
	rval = rstatproc_stats(NULL, &host_stat, rstat_clnt);
	if (!rval) {
		mutex_lock(&tty);	/* tty の制御権を取得 */
 	clnt_perror(rstat_clnt, host);
		count++;
 	cond_signal(&cv_finish);
		mutex_unlock(&tty);	/* tty の制御権を解放 */
 
		thr_exit(0);
 
	}
 
	tmp_time = localtime_r(&host_stat.curtime.tv_sec,
                       &host_time);
 
	host_stat.curtime.tv_sec = host_stat.boottime.tv_sec;
 
	tmp_time = gmtime_r(&host_stat.curtime.tv_sec,
                    &host_uptime);
 
	if (host_uptime.tm_yday != 0)
		sprintf(days_buf, "%d day%s, ", host_uptime.tm_yday,
 	         (host_uptime.tm_yday > 1) ? "s" : "");
 else
		days_buf[0] = '¥0';
 
	if (host_uptime.tm_hour != 0)
		sprintf(hours_buf, "%2d:%02d,",
 		host_uptime.tm_hour, host_uptime.tm_min);
 
	else if (host_uptime.tm_min != 0)
		sprintf(hours_buf, "%2d mins,", host_uptime.tm_min);
	else
 
	 hours_buf[0] = '¥0';
 
	mutex_lock(&tty); /* tty の制御権を取得 */
	printf("%s: ", host);
 printf(" %2d:%02d%cm up %s%s load average: %.2f %.2f %.2f¥n",
 	(host_time.tm_hour > 12)  ? host_time.tm_hour - 12
 
	 : host_time.tm_hour,
		host_time.tm_min,
		(host_time.tm_hour >= 12) ? 'p'
		: 'a',
		days_buf,
 	hours_buf,
		(double)host_stat.avenrun[0]/FSCALE,
 	(double)host_stat.avenrun[1]/FSCALE,
		(double)host_stat.avenrun[2]/FSCALE);
 count++;
	cond_signal(&cv_finish);
	mutex_unlock(&tty); /* tty の制御権を解放 */
	clnt_destroy(rstat_clnt);
 
	sleep(10);
 thr_exit(0);
}
 
/* クライアント側の MT rstat プログラムの実行 */
/* clnt_control() を使用してデフォルトのタイムアウトを変更可能 */
static struct timeval TIMEOUT = { 25, 0 };
 
bool_t
rstatproc_stats(argp, clnt_resp, clnt)
	void *argp;
	statstime *clnt_resp;
 CLIENT *clnt;
{
 
	memset((char *)clnt_resp, 0, sizeof (statstime));
	if (clnt_call(clnt, RSTATPROC_STATS,
 	(xdrproc_t) xdr_void, (caddr_t) argp,
		(xdrproc_t) xdr_statstime, (caddr_t) clnt_resp,
		TIMEOUT) != RPC_SUCCESS) {
 	return (FALSE);
 
	}
	return (TRUE);
}
 

RPC_SVC_MTMODE_SET()

マルチスレッドモードの設定 

RPC_SVC_MTMODE_GET()

マルチスレッドの取得 

RPC_SVC_THRMAX_SET()

最大スレッド数の設定 

RPC_SVC_THRMAX_GET()

最大スレッド数の取得 

RPC_SVC_THRTOTAL_GET ()

現在アクティブなスレッドの合計数 

RPC_SVC_THRCREATES_GET()

RPC ライブラリ作成のスレッドの累積数 

RPC_SVC_THRERRORS_GET()

RPC ライブラリ内の thr_create エラー数

$ cc time_svc.c -lnsl -lthread

		#include <stdio.h>
		#include <rpc/rpc.h>
 	#include <synch.h>
		#include <thread.h>
 	#include "time_prot.h"
 
		void time_prog();
 
	 main(argc, argv)
		int argc;
		char *argv[];
 	{
		int transpnum;
		char *nettype;
 	int mode = RPC_SVC_MT_AUTO;
		int max = 20;      /* スレッド最大数を 20 に設定 */
 
	 if (argc > 2) {
	 	fprintf(stderr, "usage: %s [nettype]¥n", argv[0]);
			exit(1);
 	}
 
		if (argc == 2)
			nettype = argv[1];
		else
			nettype = "netpath";
 
	 if (!rpc_control(RPC_SVC_MTMODE_SET, &mode)) {
			printf("RPC_SVC_MTMODE_SET: failed¥n");
			exit(1);
		}
 	if (!rpc_control(RPC_SVC_THRMAX_SET, &max)) {
			printf("RPC_SVC_THRMAX_SET: failed¥n");
			exit(1);
		}
 	transpnum = svc_create( time_prog, TIME_PROG, TIME_VERS,
 			nettype);
 
		if (transpnum == 0) {
 		fprintf(stderr, "%s: cannot create %s service.¥n",
			argv[0], nettype);	
			exit(1);
		}
	 svc_run();
	}
 
	/*
  * サーバーのディスパッチプログラムです。RPC サーバーライブラリは、
	 * サーバーのディスパッチャルーチン time_prog () を実行するスレッドを
	 * 作成します。RPC ライブラリがスレッドを廃棄した後に行われます。
  */
 
	static void
	time_prog(rqstp, transp)
		struct svc_req *rqstp;
		SVCXPRT *transp;
 {
 
		switch (rqstp->rq_proc) {
 		case NULLPROC:
				svc_sendreply(transp, xdr_void, NULL);
 			return;
			case TIME_GET:
				dotime(transp);
 			break;
			default:
				svcerr_noproc(transp);
 			return;
		}
	}
	dotime(transp)
 SVCXPRT *transp;
	{
	
		struct timev rslt;
		time_t thetime;
	
 	thetime = time((time_t *)0);
		rslt.second = thetime % 60;
 	thetime /= 60;
		rslt.minute = thetime % 60;
 	thetime /= 60;
		rslt.hour = thetime % 24;
 	if (!svc_sendreply(transp, xdr_timev,(caddr_t) &rslt)) {
 		svcerr_systemerr(transp);
		}
	}

		include <rpc/types.h>
 
	 struct timev {
			int second;
			int minute;
 		int hour;
		};
 
		typedef struct timev timev;
		bool_t xdr_timev();
 
	 #define TIME_PROG 0x40000001
		#define TIME_VERS 1
 	#define TIME_GET 1
 

	#define	SVC2_PROG 0x30000002
	#define	SVC2_VERS 1
	#define SVC2_PROC_ADD 1)
	#define SVC2_PROC_MULT 2
 
	struct intpair {
		u_short a;
		u_short b;
	};
 
	typedef struct intpair intpair;
 
	struct svc2_add_args {
		int argument;
		SVCXPRT *transp;
 };
 
	struct svc2_mult_args {
	 intpair mult_argument;
		SVCXPRT *transp;
	};
 
	extern bool_t xdr_intpair();
 
	#define NTHREADS_CONST 500

#define	 _REENTRANT
#include <stdio.h>
#include <rpc/rpc.h>
#include <sys/uio.h>
#include <netconfig.h>
#include <netdb.h>
#include <rpc/nettype.h>
#include <thread.h>
#include "rpc_test.h"
void *doclient();
int NTHREADS;
struct thread_info {
	thread_t client_id;
	int client_status;
};
struct thread_info save_thread[NTHREADS_CONST];
main(argc, argv)
 int argc;
	char *argv[];
{
	int index, ret;
	int thread_status;
	thread_t departedid, client_id;
	char *hosts;
	if (argc < 3) {
 	printf("Usage: do_operation [n] host¥n");
		printf("¥twhere n is the number of threads¥n");
		exit(1);
 } else
		if (argc == 3) {
			NTHREADS = NTHREADS_CONST;
 		hosts = argv[1];  /* live_host */
		} else {
 		NTHREADS = atoi(argv[1]);
			hosts = argv[2];
 	}
	for (index = 0; index < NTHREADS; index++){
 	if (ret = thr_create(NULL, NULL, doclient,
		(void *)  hosts, THR_BOUND, &client_id)){
			printf("thr_create failed: return value %d", ret);
			printf(" for %dth thread¥n", index);
 		exit(1);
		}
		save_thread[index].client_id = client_id;
	}
	for (index = 0; index < NTHREADS; index++){
		if (thr_join(save_thread[index].client_id, &departedid,	
		(void *)
		&thread_status)){
 		printf("thr_join failed for thread %d¥n",
			save_thread[index].client_id);
 		exit(1);
		}
		save_thread[index].client_status = thread_status;
	}
}
	void *doclient(host)
 char *host;
{
	struct timeval tout;
 enum clnt_stat test;
	int result = 0;
	u_short mult_result = 0;
	int add_arg;
	int EXP_RSLT;
 intpair pair;
	CLIENT *clnt;
	if ((clnt = clnt_create(host, SVC2_PROG, SVC2_VERS, "udp"
==NULL) {
		clnt_pcreateerror("clnt_create error: ");
		thr_exit((void *) -1);
	}
 tout.tv_sec = 25;
	tout.tv_usec = 0;
	memset((char *) &result, 0, sizeof (result));
	memset((char *) &mult_result, 0, sizeof (mult_result));
	if (thr_self() % 2){
 	EXP_RSLT = thr_self() + 1;
		add_arg = thr_self();
 	test = clnt_call(clnt, SVC2_PROC_ADD, (xdrproc_t) xdr_int,
 	(caddr_t) &add_arg, (xdrproc_t) xdr_int, (caddr_t) &result,
 	tout);
	} else {
		pair.a = (u_short) thr_self();
 	pair.b = (u_short) 1;
		EXP_RSLT = pair.a * pair.b;
 	test = clnt_call(clnt, SVC2_PROC_MULT, (xdrproc_t)
		xdr_intpair,
 	(caddr_t) &pair, (xdrproc_t) xdr_u_short,
		(caddr_t) &mult_result, tout);
		result = mult_result;
 }
	if (test != RPC_SUCCESS) {
		printf("THREAD: %d clnt_call hav
		thr_exit((void *) -1);
	};
 thr_exit((void *) 0);
}
 

#define 	_REENTRANT
#include <stdio.h>
#include <rpc/rpc.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/uio.h>
#include <signal.h>
#include <thread.h>
#include "operations.h"
 
SVCXPRT *xprt;
void add_mult_prog();
void *svc2_add_worker();
void *svc2_mult_worker();
main(argc, argv)
	int argc;
 char **argv;
{
	int transpnum;
 char *nettype;
	int mode = RPC_SVC_MT_USER;
       if(rpc_control(RPC_SVC_MTMODE_SET,&mode) == FALSE){
                printf(" rpc_control is failed to set AUTO mode¥n");
                exit(0);
        }
	if (argc > 2) {
		fprintf(stderr, "usage: %s [nettype]¥n", argv[0]);
 	exit(1);
	}
	if (argc == 2)
 	nettype = argv[1];
	else
		nettype = "netpath";
 transpnum = svc_create(add_mult_prog, SVC2_PROG,
 	SVC2_VERS, nettype);
	if (transpnum == 0) {
		fprintf(stderr, "%s: cannot create %s service.¥n", argv[0],
 		nettype);
 	exit(1);
	}
	svc_run();
}
void add_mult_prog (rqstp, transp)
	struct svc_req *rqstp;
 SVCXPRT *transp;
{
	int argument;
 u_short mult_arg();
	intpair mult_argument;
 bool_t (*xdr_argument)();
	struct svc2_mult_args *sw_mult_data;
 struct svc2_add_args *sw_add_data;
	int ret;
 thread_t worker_id;
	switch (rqstp->rq_proc){
 	case NULLPROC:
			svc_sendreply(transp, xdr_void, (char *) 0);
			svc_done(transp);
			break;
 	case SVC2_PROC_ADD:
			xdr_argument = xdr_int;
 		(void) memset((char *) &argument, 0, sizeof (argument));
 		if (!svc_getargs(transp, xdr_argument,
			(char *) &argument)){
 			printf("problem with getargs¥n");
				svcerr_decode(transp);
 			exit(1);
			}
			sw_add_data = (struct svc2_add_args *)
			malloc(sizeof (struct svc2_add_args));
 		sw_add_data->transp = transp;
			sw_add_data->argument = argument;
			if (ret = thr_create(NULL, THR_MIN_STACK + 16 * 1024,
			svc2_add_worker, (void *) sw_add_data, THR_DETACHED,
 			printf("SERVER: thr_create failed:");
				printf(" return value %d", ret);
				printf(" for add thread¥n");
 			exit(1);
			}
			break;
	 case SVC2_PROC_MULT:
			xdr_argument = xdr_intpair;
 		(void) memset((char *) &mult_argument, 0,
			sizeof (mult_argument));
 		if (!svc_getargs(transp, xdr_argument,
			(char *) &mult_argument)){
 			printf("problem with getargs¥n");
				svcerr_decode(transp);
 			exit(1);
			}
			sw_mult_data = (struct svc2_mult_args *)
			malloc(sizeof (struct svc2_mult_args));
 		sw_mult_data->transp = transp;
			sw_mult_data->mult_argument.a = mult_argument.a;
			sw_mult_data->mult_argument.b = mult_argument.b;
 		if (ret = thr_create(NULL, THR_MIN_STACK + 16 * 1024,
		 svc2_mult_worker, (void *) sw_mult_data, THR_DETACHED,
			&worker_id)){
 			printf("SERVER: thr_create failed:");
				printf("return value %d", ret);
				printf("for multiply thread¥n");
 			exit(1);
 
			break;
		default:
 		svcerr_noproc(transp);
			svc_done(transp);
 		break;
	}
}
 
u_short mult_arg();
int add_one();
void *svc2_add_worker(add_arg)
struct svc2_add_args *add_arg;
{	int *result;
	bool_t (*xdr_result)();
	xdr_result = xdr_int;
	result = *malloc(sizeof (int));
	*result = add_one(add_arg->argument);
 if (!svc_sendreply(add_arg->transp, xdr_result,
	(caddr_t) result)){
		printf("sendreply failed¥n");
	 svcerr_systemerr(add_arg->transp);
		svc_done(add_arg->transp);
 	thr_exit((void *) -1);
	}
	svc_done(add_arg->transp);
 thr_exit((void *) 0);
}
void *svc2_mult_worker(m_arg)
struct svc2_mult_args *m_arg;
{
	u_short *result;
 bool_t (*xdr_result)();
	xdr_result = xdr_u_short;
 result = (u_short *) malloc(sizeof (u_short));
	*result = mult_arg(&m_arg->mult_argument);
 if (!svc_sendreply(m_arg->transp, xdr_result,
	(caddr_t) result)){
 	printf("sendreply failed¥n");
		svcerr_systemerr(m_arg->transp);
 	svc_done(m_arg->transp);
		thr_exit((void *) -1);
 }
	svc_done(m_arg->transp);
	thr_exit((void *) 0);
}
u_short mult_arg(pair)
 intpair *pair;
{
	u_short result;
 result = pair->a * pair->b;
	return (result);}
int add_one(arg)
	int arg;
{
	return (++arg);
}