linux下rpc的多线程实现

solaris下rpc可以通过rpcgen -A让服务器根据处理新用户请求的需要自动创建线程。但是linux没有实现这个功能，原因据说是会产生非线程安全的代码。

本文给出了一种实现上述功能的方法。参考了
和

下面以unp2e2上的square3为例子，通过修改部分代码来实现

文件square.x

struct square_in{
    long arg1;
};
struct square_out{
    long res1;
};
program SQUARE_PROG{
    version SQARE_VERS{
        square_out SQUAREPROC(square_in) = 1;
    } = 2;

} = 0x31230000

接着执行：
rpcgen -aM square.x
该指令产生了文件：
square_clnt.c  square_server.c  Makefile.square  square_client.c  square.h       square_svc.c   square_xdr.c

着重要修改的是square_svc.c文件：

/* 被修改过的
 * Please do not edit this file.
 * It was generated using rpcgen.
 */

#include "square.h"
#include
#include
#include
#include
#include
#include
#include

#ifndef SIG_PF
#define SIG_PF void(*)(int)
#endif

pthread_t p_thread;
pthread_attr_t attr;

//static void
//square_prog_2(struct svc_req *rqstp, register SVCXPRT *transp)
void *
serv_request(void *data)
{    
    struct thr_data {
        struct svc_req *rqstp;
        SVCXPRT    *transp;
    } *ptr_data;

    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 *);

    ptr_data = (struct thr_data *)data;
    struct svc_req *rqstp = ptr_data->rqstp;
    register SVCXPRT *transp = ptr_data->transp;

    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;
    }
    retval = (bool_t) (*local)((char *)&argument, (void *)&result, rqstp);
    if (retval > 0 && !svc_sendreply(transp, (xdrproc_t), (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;
}

//new square_prog_2
static void
square_prog_2(struct svc_req *rqstp, register SVCXPRT *transp)
{
    struct data_str{
        struct svc_req *rqstp;
        SVCXPRT *transp;


    } *data_ptr=(struct data_str*)malloc(sizeof(struct));

    data_ptr->rqstp = rqstp;
    data_ptr->transp = transp;
    pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
    pthread_create(&p_thread,&attr,serv_request,(void*)data_ptr);
}

int
main (int argc, char **argv)
{
    register SVCXPRT *transp;

    pmap_unset (SQUARE_PROG, SQARE_VERS);

    transp = svcudp_create(RPC_ANYSOCK);
    if (transp == NULL) {
        fprintf (stderr, "%s", "cannot create udp service.");
        exit(1);
    }
    if (!svc_register(transp, SQUARE_PROG, SQARE_VERS, square_prog_2, IPPROTO_UDP)) {
        fprintf (stderr, "%s", "unable to register (SQUARE_PROG, SQARE_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, SQARE_VERS, square_prog_2, IPPROTO_TCP)) {
        fprintf (stderr, "%s", "unable to register (SQUARE_PROG, SQARE_VERS, tcp).");
        exit(1);
    }

    svc_run ();
    fprintf (stderr, "%s", "svc_run returned");
    exit (1);
    /* NOTREACHED */
}

蓝色部分是修改的地方，对比下面修改前的代码：

/* 修改前的文件
 * Please do not edit this file.
 * It was generated using rpcgen.
 */

#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

static void
square_prog_2(struct svc_req *rqstp, register SVCXPRT *transp)
{
    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;
    }
    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;

    pmap_unset (SQUARE_PROG, SQARE_VERS);

    transp = svcudp_create(RPC_ANYSOCK);
    if (transp == NULL) {
        fprintf (stderr, "%s", "cannot create udp service.");
        exit(1);
    }
    if (!svc_register(transp, SQUARE_PROG, SQARE_VERS, square_prog_2, IPPROTO_UDP)) {
        fprintf (stderr, "%s", "unable to register (SQUARE_PROG, SQARE_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, SQARE_VERS, square_prog_2, IPPROTO_TCP)) {
        fprintf (stderr, "%s", "unable to register (SQUARE_PROG, SQARE_VERS, tcp).");
        exit(1);
    }

    svc_run ();
    fprintf (stderr, "%s", "svc_run returned");
    exit (1);
    /* NOTREACHED */
}


可以看出，修改的思路就是把原square_prog_2函数功能改由serv_request来实现，让新的square_prog_2通过创建新线程来调用serv_request，serv_request则判断并调用相应的函数（在这个例子中只有SQUAREPROC一个函数）。实现的关键是要把struct svc_req *rqstp和 register SVCXPRT *transp这两个参数交给新线程函数serv_request处理，这样当serv_request返回时，便等价于原来的square_prog_2函数返回。

其它文件跟unp2e2上的例子一样


/* 文件square_server.c
 * This is sample code generated by rpcgen.
 * These are only templates and you can use them
 * as a guideline for developing your own functions.
 */

#include "square.h"

bool_t
squareproc_2_svc(square_in *inp, square_out *outp, struct svc_req *rqstp)
{
    bool_t retval;
    printf("Thread id = '%ld' started, arg = %d\n",pthread_self(),inp->arg1);
    sleep(5);
    outp->res1 = inp->arg1 * inp->arg1;
    printf("Thread id = '%ld' is done %d \n",pthread_self(),outp->res1);
    retval = TRUE;
    return retval;
}

int
square_prog_2_freeresult (SVCXPRT *transp, xdrproc_t xdr_result, caddr_t result)
{
    xdr_free (xdr_result, result);

    /*
     * Insert additional freeing code here, if needed
     */

    return 1;
}


/*

 *   该文件重写由rpcgen产生的原文件

 *   square_client.c


 */


#include "square.h"

int
main (int argc,char **argv)
{
    CLIENT *cl;
    square_in in;
    square_out out;
    if (argc != 3 ) {
        printf ("Usage : client \n");
          exit(1);
    }
    cl = clnt_create(argv[1], SQUARE_PROG, SQUARE_VERS, "tcp");
     if (cl == NULL) {
                clnt_sperror(cl, "call failed");
                exit (1);
        }
    in.arg1 = atol(argv[2]);
    if (squareproc_2(&in, &out, cl) != RPC_SUCCESS){
        printf("%s\n", clnt_sperror(cl, argv[1]));
           exit(1);
    }
    printf("result: %ld\n",out.res1);
    exit(0);
}


接着编译链接server端和client端程序

$ gcc -o client square_clnt.c square_xdr.c square_client.c
$ gcc -o server square_svc.c square_xdr.c square_server.c -lpthread


在一个终端下运行./server记得打开portmap
另一个终端下运行：
$ ./client localhost 1 & ./client localhost 2 & ./client localhost 3

便能在server端看到：

Thread id = '1082132816' started, arg = 3
Thread id = '1090525520' started, arg = 2
Thread id = '1098918224' started, arg = 1
Thread id = '1082132816' is done 9
Thread id = '1090525520' is done 4
Thread id = '1098918224' is done 1