LINUX系统上使用多线程化的RPC

妖狐艹你老母 2022-07-12 13:50 288阅读 0赞

使用RPC时，在LINUX系统上使用rpcgen -M并不能生成具有多线程的RPC代理存根，只是生成的代码具有线程安全特性。在网络有一些网友讨论了一种实现方式，就是在接收到客户端的请求时，建立一个线程来处理客户端代码，这方

式一个外国的朋友给出了实现代码（[http://www.redhat.com/archives/open-source-now-list/2004-June/msg00000.html][http_www.redhat.com_archives_open-source-now-list_2004-June_msg00000.html] ），但在复制到我的LINUX（LINUX RHL5）时，并不能成功运行。在国内有一作者也写一个示例，是针对

UNIX网络编程第一卷求平方根的示例写的网址如下：[http://blog.chinaunix.net/u1/37472/showart\_726114.html][http_blog.chinaunix.net_u1_37472_showart_726114.html]  
我使用该示例仍然没能得到运行（RPC: Procedure unavailable），通过增加主线程的延时，得到了正确运行。  
但是通过增加延时只会加慢RPC服务器处理数据的时间，如果时间过长，会引起RCP调用超时，如果过短，而且RCP调用（新开的线程）还没有处理完成，也是会存在着问题的。我在本地通过延时一秒，并间隔一定时隙调用100次，发

现服务器出现段错误。  
我做过将使用指针修改为变量，出现错误：RPC: Server can't decode arguments，后来我发现RPC调用参数的结构体并不是一级指针，而是多级指针。也就是说该结构中还是指针变量成员。

为些我想到了使用一个线程锁来解决等待的问题，并且在产生100次连续调用之后，也没有发现问题，但并不代表没有问题，因为对于RPC的运行流程我只是在猜测，而且在返回给客户时，没有一些原始状态信息是否可以正常返回呢

？如果返回数据不是一个简单的数值，而是一个结构体数据，那会怎么样呢？所以希望能看到这时原朋友，如果使用时出现了问题，请告知我。我也会在有时间的时候来测试一下。

如下是服务端代码，是通过rpcgen生成之后再修改的.

/\* \* Please do not edit this file. \* It was generated using rpcgen. \*/ \#include <unistd.h> \#include "square.h" \#include <stdio.h> \#include <stdlib.h> \#include <rpc/pmap\_clnt.h> \#include <string.h> \#include <memory.h> \#include <sys/socket.h> \#include <netinet/in.h> \#ifndef SIG\_PF \#define SIG\_PF void(\*)(int) \#endif pthread\_t p\_thread; pthread\_attr\_t attr; \#ifdef USING\_MT\_RPC\_WITH\_LOCK pthread\_mutex\_t rpcmutex; \#endif void\* square\_prog\_thread\_func(void\* param); \#ifdef USING\_MT\_RPC\_WITH\_LOCK void square\_prog\_2\_thread\_func( struct svc\_req \*rqstp, register SVCXPRT \*transp, pthread\_mutex\_t\* pmutex); \#else void square\_prog\_2\_thread\_func( struct svc\_req \*rqstp, register SVCXPRT \*transp); \#endif static void square\_prog\_2(struct svc\_req \*rqstp, register SVCXPRT \*transp) \{ /\* square\_prog\_2\_thread\_func(rqstp, transp); return;\*/ struct data\_str \{ struct svc\_req \*rqstp; SVCXPRT \*transp; \#ifdef USING\_MT\_RPC\_WITH\_LOCK pthread\_mutex\_t \*pmutex; \#endif \} \*data\_ptr=(struct data\_str\*)malloc(sizeof(struct data\_str)); if (data\_ptr == NULL)\{ \#ifdef USING\_MT\_RPC\_WITH\_LOCK square\_prog\_2\_thread\_func(rqstp, transp, NULL); \#else square\_prog\_2\_thread\_func(rqstp, transp); \#endif return; \} data\_ptr-> rqstp = rqstp; data\_ptr-> transp = transp; \#ifdef USING\_MT\_RPC\_WITH\_LOCK data\_ptr-> pmutex = &rpcmutex; pthread\_mutex\_lock(&rpcmutex); \#endif /\* Create a new thread for client. \*/ pthread\_attr\_setdetachstate(&attr,PTHREAD\_CREATE\_DETACHED); pthread\_create(&p\_thread,&attr,square\_prog\_thread\_func ,(void \*)data\_ptr); \#ifdef USING\_MT\_RPC\_WITH\_LOCK pthread\_mutex\_lock(&rpcmutex); pthread\_mutex\_unlock(&rpcmutex); \#else sleep(1); \#endif \} void\* square\_prog\_thread\_func(void\* param) \{ struct thr\_data \{ struct svc\_req \*rqstp; SVCXPRT \*transp; \#ifdef USING\_MT\_RPC\_WITH\_LOCK pthread\_mutex\_t \*pmutex; \#endif \} \*ptr\_data; /\* recover param from thread param \*/ ptr\_data = (struct thr\_data \*)param; struct svc\_req \*rqstp = (ptr\_data-> rqstp); register SVCXPRT \*transp = (ptr\_data-> transp); /\* continue server processing ... \*/ \#ifdef USING\_MT\_RPC\_WITH\_LOCK square\_prog\_2\_thread\_func(rqstp, transp, ptr\_data->pmutex); \#else square\_prog\_2\_thread\_func(rqstp, transp); \#endif free(param); return NULL; \} \#ifdef USING\_MT\_RPC\_WITH\_LOCK void square\_prog\_2\_thread\_func( struct svc\_req \*rqstp, register SVCXPRT \*transp, pthread\_mutex\_t\* pmutex) \#else void square\_prog\_2\_thread\_func( struct svc\_req \*rqstp, register SVCXPRT \*transp) \#endif \{ union \{ square\_in squareproc\_2\_arg; \} argument; union \{ square\_out squareproc\_2\_res; \} result; bool\_t retval; xdrproc\_t \_xdr\_argument, \_xdr\_result; bool\_t (\*local)(char \*, void \*, struct svc\_req \*); switch (rqstp->rq\_proc) \{ case NULLPROC: (void) svc\_sendreply (transp, (xdrproc\_t) xdr\_void, (char \*)NULL); return; case SQUAREPROC: \_xdr\_argument = (xdrproc\_t) xdr\_square\_in; \_xdr\_result = (xdrproc\_t) xdr\_square\_out; local = (bool\_t (\*) (char \*, void \*, struct svc\_req \*))squareproc\_2\_svc; break; default: svcerr\_noproc (transp); return; \} memset ((char \*)&argument, 0, sizeof (argument)); if (!svc\_getargs (transp, (xdrproc\_t) \_xdr\_argument, (caddr\_t) &argument)) \{ svcerr\_decode (transp); return; \} /\* unlock rpc call thread \*/ \#ifdef USING\_MT\_RPC\_WITH\_LOCK if (pmutex) pthread\_mutex\_unlock(pmutex); \#endif /\* call user define function \*/ retval = (bool\_t) (\*local)((char \*)&argument, (void \*)&result, rqstp); if (retval > 0 && !svc\_sendreply(transp, (xdrproc\_t) \_xdr\_result, (char \*)&result)) \{ svcerr\_systemerr (transp); \} if (!svc\_freeargs (transp, (xdrproc\_t) \_xdr\_argument, (caddr\_t) &argument)) \{ fprintf (stderr, "%s", "unable to free arguments"); exit (1); \} if (!square\_prog\_2\_freeresult (transp, \_xdr\_result, (caddr\_t) &result)) fprintf (stderr, "%s", "unable to free results"); return; \} int main (int argc, char \*\*argv) \{ register SVCXPRT \*transp; \#ifdef USING\_MT\_RPC\_WITH\_LOCK /\* Create a mutex object for rpc calling \*/ pthread\_mutex\_init(&rpcmutex, NULL); \#endif pmap\_unset (SQUARE\_PROG, SQUARE\_VERS); transp = svcudp\_create(RPC\_ANYSOCK); if (transp == NULL) \{ fprintf (stderr, "%s", "cannot create udp service."); exit(1); \} if (!svc\_register(transp, SQUARE\_PROG, SQUARE\_VERS, square\_prog\_2, IPPROTO\_UDP)) \{ fprintf (stderr, "%s", "unable to register (SQUARE\_PROG, SQUARE\_VERS, udp)."); exit(1); \} transp = svctcp\_create(RPC\_ANYSOCK, 0, 0); if (transp == NULL) \{ fprintf (stderr, "%s", "cannot create tcp service."); exit(1); \} if (!svc\_register(transp, SQUARE\_PROG, SQUARE\_VERS, square\_prog\_2, IPPROTO\_TCP)) \{ fprintf (stderr, "%s", "unable to register (SQUARE\_PROG, SQUARE\_VERS, tcp)."); exit(1); \} svc\_run (); fprintf (stderr, "%s", "svc\_run returned"); exit (1); /\* NOTREACHED \*/ \}

在代码的\#70行处，可以看到在准备为客户求建立一个线程，在创建完线程之后，在\#73行开始等待锁，也就是在这个时候等待参数被RPC解析并到调用服务端处理代码期间。在\#152行，我释放了锁，这样在主线程就可以返回了，而辅助线程将进入自定义的服务器处理函数。这样就实现了一个比较合理的等待。

下面将程序代码以及MAKEFILE文件发到此处，希望需要这个方面信息的朋友可以测试一下，并提出更多的想法。

square.x

struct square\_in \{ long arg1; \}; struct square\_out \{ long res1; \}; program SQUARE\_PROG \{ version SQUARE\_VERS \{ square\_out SQUAREPROC(square\_in) = 1; /\* procedure number = 1 \*/ \} = 2; /\* version number = 2 \*/ \} = 0x31230000; /\* program number = 0x31230000 \*/

server.c 服务务用户处理函数所在地

\#include <unistd.h> \#include <pthread.h> \#include "square.h" bool\_t squareproc\_2\_svc(square\_in \*inp, square\_out \*outp, struct svc\_req \*rqstp) \{ printf("thread %ld started, arg = %ld/n", pthread\_self(), inp->arg1); sleep(5); outp->res1 = inp->arg1 \* inp->arg1; printf("thread %ld done/n", pthread\_self()); return(TRUE); \} int square\_prog\_2\_freeresult(SVCXPRT \*transp, xdrproc\_t xdr\_result, caddr\_t result) \{ xdr\_free(xdr\_result, result); return(1); \}

client.c 客户调用，在这里希望注意的就是clnt\_destroy函数的调用，尤其是在错误的时候没有调用时，将导致客户与服务器之间保持了一个连接（在LINUX上表示找开了一个文件），如果出现多次，将在服务器上相当于打开了许多的TCP连接（文件），将导致服务器程序因为受操作系统的进程打开文件数限制而出错（打开文件时）。所以请一定要注意！

\#include <rpc/rpc.h> \#include "square.h" int main(int argc, char \*\*argv) \{ CLIENT \*cl; square\_in in; square\_out out; if (argc != 3)\{ printf("usage: client <hostname> <integer-value>"); exit(0); \} cl = clnt\_create(argv\[1\], SQUARE\_PROG, SQUARE\_VERS, "tcp"); in.arg1 = atol(argv\[2\]); if (squareproc\_2(&in, &out, cl) != RPC\_SUCCESS)\{ printf("%s", clnt\_sperror(cl, argv\[1\])); clnt\_destroy(cl); exit(0); \} clnt\_destroy(cl); printf("result: %ld/n", out.res1); exit(0); \}

Makefile 编译文件

PROGS = client server \#cancele when you don't need thread lock to use mulitthread MACRO = -DUSING\_MT\_RPC\_WITH\_LOCK CFLAGS = -g -dH -Wall -DDEBUG $(MACRO) \#CFLAGS = -Wall LINKFLAGS = -lpthread CC = gcc all: $\{PROGS\} square.h square\_clnt.c square\_svc.c square\_xdr.c: square.x rpcgen -C -M square.x square\_clnt.o: square\_clnt.c square.h square\_svc.o: square\_svc.c square.h client: square.h client.o square\_clnt.o square\_xdr.o $\{CC\} $\{CFLAGS\} -o $@ client.o square\_clnt.o square\_xdr.o server: square.h server.o square\_svc.o square\_xdr.o $\{CC\} $\{CFLAGS\} -o $@ server.o square\_svc.o square\_xdr.o $(LINKFLAGS) clean: rm -f \*o

编译方法，先使用make命令编译程序，然后再将square\_svc.c文件覆盖，再编译，方能实现多线程。

最后，我发现，如果客户端过早终止，服务器将出错。问题出在square\_svc.c文件\#156行，在给TCP回复的时候出的问题：

Program received signal SIGSEGV, Segmentation fault.  
\[Switching to Thread -1208173680 (LWP 28672)\]  
0x00ac7ef5 in writetcp () from /lib/libc.so.6  
(gdb) where  
\#0  0x00ac7ef5 in writetcp () from /lib/libc.so.6  
\#1  0x00aca570 in xdrrec\_endofrecord\_internal () from /lib/libc.so.6  
\#2  0x00ac7e00 in svctcp\_reply () from /lib/libc.so.6  
\#3  0x00ac682c in svc\_sendreply\_internal () from /lib/libc.so.6  
\#4  0x08048b39 in square\_prog\_2\_thread\_func (rqstp=0xbfe9af18, transp=0x8f36708, pmutex=0x80491a4)  
    at square\_svc.c:156  
\#5  0x08048a3a in square\_prog\_thread\_func (param=0x8f33008) at square\_svc.c:99  
\#6  0x00b462db in start\_thread () from /lib/libpthread.so.0  
\#7  0x00aa014e in clone () from /lib/libc.so.6  
(gdb)  
是线程在完成服务端用户函数调用准备给启用返回结果时出现的。按理应该不是出现段错误的。

如果需要服务器工作稳定的话，需要增加去这个错误的处理。希望能解决的朋友帮忙解决一下。

[http_www.redhat.com_archives_open-source-now-list_2004-June_msg00000.html]: http://www.redhat.com/archives/open-source-now-list/2004-June/msg00000.html
[http_blog.chinaunix.net_u1_37472_showart_726114.html]: http://blog.chinaunix.net/u1/37472/showart_726114.html