[hcoop/debian/openafs.git] / tests / rpctestlib / rpc_test_procs.c

/*
 * Copyright (c) 2010, Linux Box Corporation.
 * All Rights Reserved.
 *
 * Portions Copyright (c) 2007, Hartmut Reuter,
 * RZG, Max-Planck-Institut f. Plasmaphysik.
 * All Rights Reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   1. Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *   2. Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in
 *      the documentation and/or other materials provided with the
 *      distribution.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
 * OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include <afsconfig.h>
#include <afs/param.h>

#include <roken.h>

#include <afs/stds.h>

#include "rpc_test_procs.h"

#ifdef AFS_NT40_ENV
#include <windows.h>
#include <WINNT/afsevent.h>
#else
#include <sys/file.h>
#include <afs/venus.h>
#include <sys/time.h>
#endif
#include <afs/afsint.h>
#define FSINT_COMMON_XG 1
#include <afs/vice.h>
#include <afs/cmd.h>
#include <afs/auth.h>
#include <afs/cellconfig.h>

#include <afs/com_err.h>
#ifdef HAVE_DIRECT_H
#include <direct.h>
#endif
#include <afs/errors.h>
#include <afs/sys_prototypes.h>
#include <rx/rx_prototypes.h>
#ifdef AFS_PTHREAD_ENV
#include <assert.h>
#endif

extern const char *prog;
const int ctx_key = 1;

#if 1
#define RPC_TEST_GLOBAL_RX_INIT 1
#else
#undef RPC_TEST_GLOBAL_RX_INIT
#endif

const afs_uint32 fs_port = 7000;

typedef struct rpc_test_pkg_params {
    pthread_mutex_t mtx;
    pthread_mutexattr_t mtx_attrs;
    afs_uint32 cb_next_port;
    afs_uint32 next_cno;
} rpc_test_pkg_params;
static rpc_test_pkg_params rpc_test_params;

afs_int32 rpc_test_PkgInit(void)
{
    afs_int32 code = 0;
    static afs_uint32 rpc_test_initialized = 0; /* once */

    if (!rpc_test_initialized) {
        rpc_test_initialized = 1;
    } else {
	printf("%s: rpc_test_PkgInit: package already initialized\n", prog);
        exit(1);
    }

#ifndef AFS_NT40_ENV
    code = pthread_mutexattr_init(&rpc_test_params.mtx_attrs);
    if (code) {
        printf("%s: rpc_test_PkgInit: pthread_mutexattr_init failed\n", prog);
        exit(1);
    }
    code = pthread_mutex_init(&rpc_test_params.mtx, &rpc_test_params.mtx_attrs);
    if (code) {
        printf("%s: rpc_test_PkgInit: pthread_mutex_init failed\n", prog);
        exit(1);
    }
#endif

    /* start connection sequence */
    rpc_test_params.next_cno = 1;

    /* set the starting port in sequence */
    rpc_test_params.cb_next_port = 7105;

#if defined(RPC_TEST_GLOBAL_RX_INIT)
    rx_Init(0);
#endif

    return (code);

}        /* rpc_test_PkgInit */

static void *
init_callback_service_lwp(void *arg)
{
    struct rx_securityClass *sc;
    struct rx_service *svc;

    rpc_test_request_ctx *ctx = (rpc_test_request_ctx *) arg;

    printf("%s: init_callback_service_lwp: listen_addr: %s "
	   "(%d) cb_port: %d\n",
	   prog, ctx->cb_listen_addr_s, ctx->cb_listen_addr.addr_in[0],
	   ctx->cb_port);

    sc = (struct rx_securityClass *) rxnull_NewServerSecurityObject();
    if (!sc) {
	fprintf(stderr,"rxnull_NewServerSecurityObject failed for callback "
                "service\n");
	exit(1);
    }

#if defined(RPC_TEST_GLOBAL_RX_INIT)
    svc = rx_NewServiceHost(htonl(INADDR_ANY), htons(ctx->cb_port), 1,
                            ctx->cb_svc_name, &sc, 1, RXAFSCB_ExecuteRequest);
#else
    svc = rx_NewService(0, 1, ctx->cb_svc_name, &sc, 1, RXAFSCB_ExecuteRequest);
#endif
    /* stash context */
    rx_SetServiceSpecific(svc, ctx_key, ctx);

    if (!svc) {
	fprintf(stderr,"rx_NewServiceHost failed for callback service\n");
	exit(1);
    }

    /* XXX stash service so we can hijack its rx_socket when inititiating
     * RPC calls */
    ctx->svc = svc;

    /* release pkg mutex before entering rx processing loop */
    pthread_mutex_unlock(&rpc_test_params.mtx);

    rx_StartServer(1);

    printf("%s: init_callback_service_lwp: finished", prog);

    return (NULL);

}        /* callback_service_lwp */

afs_int32 init_callback_service(rpc_test_request_ctx *ctx)
{
    pthread_t tid;
    pthread_attr_t tattr;
    afs_int32 code = 0;

    afs_uuid_create(&(ctx->cb_listen_addr.uuid));

#if !defined(RPC_TEST_GLOBAL_RX_INIT)
#if 0
    code = rx_InitHost(ctx->cb_listen_addr.addr_in[0],
                       (int) htons(ctx->cb_port));
#else
    code = rx_Init((int) htons(ctx->cb_port));
#endif
#endif /* RPC_TEST_GLOBAL_RX_INIT */

    assert(pthread_attr_init(&tattr) == 0);
    assert(pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED) == 0);
    assert(pthread_create(&tid, &tattr, init_callback_service_lwp, ctx) == 0);

    return (code);

}        /* init_callback_service */

afs_int32 init_fs_channel(rpc_test_request_ctx **octx, char *cb_if,
                          char *listen_addr_s, char *prefix, char *fs_addr_s,
                          afs_uint32 flags)
{
    rpc_test_request_ctx *ctx;
    afs_int32 code = 0;
#ifdef AFS_NT40_ENV
    afs_int32 sslen = sizeof(struct sockaddr);
#endif

    ctx = *octx = (rpc_test_request_ctx *) malloc(sizeof(rpc_test_request_ctx));
    memset(ctx, 0, sizeof(rpc_test_request_ctx));

    /* initialize a local mutex */
    code = pthread_mutex_init(&ctx->mtx, &rpc_test_params.mtx_attrs);

    /* lock package before rx setup--which has global deps, atm */
    pthread_mutex_lock(&rpc_test_params.mtx);

    ctx->cno = rpc_test_params.next_cno++;
    ctx->flags = flags;

    /* afscbint (server) */
    sprintf(ctx->cb_svc_name, "cb_%d", ctx->cno);
    sprintf(ctx->cb_if_s, "%s", cb_if);
    sprintf(ctx->cb_listen_addr_s, "%s", listen_addr_s);
    sprintf(ctx->cb_prefix_s, "%s", prefix);
    sprintf(ctx->fs_addr_s, "%s", fs_addr_s);

#if defined(RPC_TEST_ADD_ADDRESSES)
#if defined(AFS_LINUX26_ENV)
    sprintf(cmd, "ip addr add %s/%s dev %s label %s", listen_addr_s, prefix,
            cb_if, cb_if);
    code = system(cmd);
#endif
#endif /* RPC_TEST_ADD_ADDRESSES */

    /* lock this */
    pthread_mutex_lock(&ctx->mtx);

    /* set up rx */
    ctx->cb_port = rpc_test_params.cb_next_port++;
    ctx->cb_listen_addr.numberOfInterfaces = 1;

#ifdef AFS_NT40_ENV
    code = WSAStringToAddressA(listen_addr_s, AF_INET, NULL,
              (struct sockaddr*) &(ctx->cb_listen_addr), &sslen);
#else
    code = inet_pton(AF_INET, listen_addr_s,
                     (void*) &(ctx->cb_listen_addr.addr_in[0]));
#endif

    code = init_callback_service(ctx /* LOCKED, && rpc_test_params->mtx LOCKED */);

    /* fsint (client) */

#ifdef AFS_NT40_ENV
    code = WSAStringToAddressA(fs_addr_s, AF_INET, NULL,
              (struct sockaddr*) &(ctx->fs_addr.addr_in[0]), &sslen);
#else
    code = inet_pton(AF_INET, fs_addr_s, (void*) &(ctx->fs_addr.addr_in[0]));
#endif
    ctx->sc = rxnull_NewClientSecurityObject();
    ctx->sc_index = RX_SECIDX_NULL;
    ctx->conn = rx_NewConnection(ctx->fs_addr.addr_in[0], (int) htons(fs_port),
                                 1, ctx->sc, ctx->sc_index);

    /* unlock this */
    pthread_mutex_unlock(&ctx->mtx);

    return (code);

}        /* init_fs_channel */

/* XXX use the pkg lock to protect the state of rx_socket for
 * the duration of the call, switching it out for the stashed
 * rx_socket created by rx_NewService for this channel */
#define RXCALL_WITH_SOCK(code, ctx, call) \
    do { \
        osi_socket prev_rx_socket; \
        pthread_mutex_lock(&rpc_test_params.mtx); \
        prev_rx_socket = rx_socket; \
        rx_socket = ctx->svc->socket; \
        code = call; \
        rx_socket = prev_rx_socket; \
        pthread_mutex_unlock(&rpc_test_params.mtx); \
} while(0);

afs_int32
rpc_test_afs_fetch_status(rpc_test_request_ctx *ctx, AFSFid *fid,
                              AFSFetchStatus *outstatus)
{
    struct AFSVolSync tsync;
    struct AFSCallBack tcb;
    afs_int32 code = 0;

    RXCALL_WITH_SOCK(code, ctx,
       (RXAFS_FetchStatus(ctx->conn, fid, outstatus, &tcb, &tsync)));

    return (code);

}        /* rpc_test_afs_fetch_status */

afs_int32
rpc_test_afs_store_status(rpc_test_request_ctx *ctx, AFSFid *fid,
                    AFSStoreStatus *instatus, AFSFetchStatus *outstatus)
{
    struct AFSVolSync tsync;
    afs_int32 code = 0;

    RXCALL_WITH_SOCK(code, ctx,
       (RXAFS_StoreStatus(ctx->conn, fid, instatus, outstatus, &tsync)));

    return (code);

}        /* rpc_test_afs_fetch_status */

#if defined(AFS_BYTE_RANGE_FLOCKS)
afs_int32 rpc_test_afs_set_byterangelock(rpc_test_request_ctx *ctx,
    AFSByteRangeLock * lock)
{
    struct rx_call *tcall;
    afs_int32 code = 0;

    RXCALL_WITH_SOCK(code, ctx,
       (RXAFS_SetByteRangeLock(ctx->conn, lock)));

    return (code);

}        /* rpc_test_afs_set_byterangelock */

afs_int32 rpc_test_afs_release_byterangelock(rpc_test_request_ctx *ctx,
    AFSByteRangeLock * lock)
{
    struct rx_call *tcall;
    afs_int32 code = 0;

    RXCALL_WITH_SOCK(code, ctx,
       (RXAFS_ReleaseByteRangeLock(ctx->conn, lock)));

    return (code);

}        /* rpc_test_afs_release_byterangelock */

afs_int32 rpc_test_afs_upgrade_byterangelock(rpc_test_request_ctx *ctx,
    AFSByteRangeLock * lock)
{
    afs_int32 code = 0;

    /* TODO:  implement */

    return (code);

}        /* rpc_test_afs_upgrade_byterangelock */

afs_int32 rpc_test_afs_downgrade_byterangelock(rpc_test_request_ctx *ctx,
    AFSByteRangeLock * Lock)
{
    afs_int32 code = 0;

    /* TODO:  implement */

    return (code);

}        /* rpc_test_afs_downgrade_byterangelock */
#endif /* AFS_BYTE_RANGE_FLOCKS */

afs_int32
destroy_fs_channel(rpc_test_request_ctx *ctx)
{
    afs_int32 code = 0;
#if defined(RPC_TEST_ADD_ADDRESSES)
#if defined(AFS_LINUX26_ENV)
    sprintf(cmd, "ip addr del %s/%s dev %s label %s", ctx->cb_listen_addr_s,
            ctx->cb_prefix_s, ctx->cb_if_s, ctx->cb_if_s);
    code = system(cmd);
#endif
#endif /* RPC_TEST_ADD_ADDRESSES */
    assert(ctx);
    free(ctx);
    return (code);

}        /* destroy_fs_channel */

void
rpc_test_PkgShutdown(void)
{
}        /* rpc_test_PkgShutdown */
Commit	Line	Data
805e021f CE	1	/*
	2	* Copyright (c) 2010, Linux Box Corporation.
	3	* All Rights Reserved.
	4	*
	5	* Portions Copyright (c) 2007, Hartmut Reuter,
	6	* RZG, Max-Planck-Institut f. Plasmaphysik.
	7	* All Rights Reserved.
	8	*
	9	* Redistribution and use in source and binary forms, with or without
	10	* modification, are permitted provided that the following conditions are met:
	11	*
	12	* 1. Redistributions of source code must retain the above copyright
	13	* notice, this list of conditions and the following disclaimer.
	14	* 2. Redistributions in binary form must reproduce the above copyright
	15	* notice, this list of conditions and the following disclaimer in
	16	* the documentation and/or other materials provided with the
	17	* distribution.
	18	*
	19	* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES,
	20	* INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
	21	* AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
	22	* CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
	23	* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
	24	* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
	25	* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
	26	* OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
	27	* NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
	28	* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	29	*/
	30
	31	#include <afsconfig.h>
	32	#include <afs/param.h>
	33
	34	#include <roken.h>
	35
	36	#include <afs/stds.h>
	37
	38	#include "rpc_test_procs.h"
	39
	40	#ifdef AFS_NT40_ENV
	41	#include <windows.h>
	42	#include <WINNT/afsevent.h>
	43	#else
	44	#include <sys/file.h>
	45	#include <afs/venus.h>
	46	#include <sys/time.h>
	47	#endif
	48	#include <afs/afsint.h>
	49	#define FSINT_COMMON_XG 1
	50	#include <afs/vice.h>
	51	#include <afs/cmd.h>
	52	#include <afs/auth.h>
	53	#include <afs/cellconfig.h>
	54
	55	#include <afs/com_err.h>
	56	#ifdef HAVE_DIRECT_H
	57	#include <direct.h>
	58	#endif
	59	#include <afs/errors.h>
	60	#include <afs/sys_prototypes.h>
	61	#include <rx/rx_prototypes.h>
	62	#ifdef AFS_PTHREAD_ENV
	63	#include <assert.h>
	64	#endif
65
66	extern const char *prog;
67	const int ctx_key = 1;
68
69	#if 1
70	#define RPC_TEST_GLOBAL_RX_INIT 1
71	#else
72	#undef RPC_TEST_GLOBAL_RX_INIT
73	#endif
74
75	const afs_uint32 fs_port = 7000;
76
77	typedef struct rpc_test_pkg_params {
78	pthread_mutex_t mtx;
79	pthread_mutexattr_t mtx_attrs;
80	afs_uint32 cb_next_port;
81	afs_uint32 next_cno;
82	} rpc_test_pkg_params;
83	static rpc_test_pkg_params rpc_test_params;
84
85	afs_int32 rpc_test_PkgInit(void)
86	{
87	afs_int32 code = 0;
88	static afs_uint32 rpc_test_initialized = 0; /* once */
89
90	if (!rpc_test_initialized) {
91	rpc_test_initialized = 1;
92	} else {
93	printf("%s: rpc_test_PkgInit: package already initialized\n", prog);
94	exit(1);
95	}
96
97	#ifndef AFS_NT40_ENV
98	code = pthread_mutexattr_init(&rpc_test_params.mtx_attrs);
99	if (code) {
100	printf("%s: rpc_test_PkgInit: pthread_mutexattr_init failed\n", prog);
101	exit(1);
102	}
103	code = pthread_mutex_init(&rpc_test_params.mtx, &rpc_test_params.mtx_attrs);
104	if (code) {
105	printf("%s: rpc_test_PkgInit: pthread_mutex_init failed\n", prog);
106	exit(1);
107	}
108	#endif
109
110	/* start connection sequence */
111	rpc_test_params.next_cno = 1;
112
113	/* set the starting port in sequence */
114	rpc_test_params.cb_next_port = 7105;
115
116	#if defined(RPC_TEST_GLOBAL_RX_INIT)
117	rx_Init(0);
118	#endif
119
120	return (code);
121
122	} /* rpc_test_PkgInit */
123
124	static void *
125	init_callback_service_lwp(void *arg)
126	{
127	struct rx_securityClass *sc;
128	struct rx_service *svc;
129
130	rpc_test_request_ctx ctx = (rpc_test_request_ctx ) arg;
131
132	printf("%s: init_callback_service_lwp: listen_addr: %s "
133	"(%d) cb_port: %d\n",
134	prog, ctx->cb_listen_addr_s, ctx->cb_listen_addr.addr_in[0],
135	ctx->cb_port);
136
137	sc = (struct rx_securityClass *) rxnull_NewServerSecurityObject();
138	if (!sc) {
139	fprintf(stderr,"rxnull_NewServerSecurityObject failed for callback "
140	"service\n");
141	exit(1);
142	}
143
144	#if defined(RPC_TEST_GLOBAL_RX_INIT)
145	svc = rx_NewServiceHost(htonl(INADDR_ANY), htons(ctx->cb_port), 1,
146	ctx->cb_svc_name, &sc, 1, RXAFSCB_ExecuteRequest);
147	#else
148	svc = rx_NewService(0, 1, ctx->cb_svc_name, &sc, 1, RXAFSCB_ExecuteRequest);
149	#endif
150	/* stash context */
151	rx_SetServiceSpecific(svc, ctx_key, ctx);
152
153	if (!svc) {
154	fprintf(stderr,"rx_NewServiceHost failed for callback service\n");
155	exit(1);
156	}
157
158	/* XXX stash service so we can hijack its rx_socket when inititiating
159	* RPC calls */
160	ctx->svc = svc;
161
162	/* release pkg mutex before entering rx processing loop */
163	pthread_mutex_unlock(&rpc_test_params.mtx);
164
165	rx_StartServer(1);
166
167	printf("%s: init_callback_service_lwp: finished", prog);
168
169	return (NULL);
170
171	} /* callback_service_lwp */
172
173	afs_int32 init_callback_service(rpc_test_request_ctx *ctx)
174	{
175	pthread_t tid;
176	pthread_attr_t tattr;
177	afs_int32 code = 0;
178
179	afs_uuid_create(&(ctx->cb_listen_addr.uuid));
180
181	#if !defined(RPC_TEST_GLOBAL_RX_INIT)
182	#if 0
183	code = rx_InitHost(ctx->cb_listen_addr.addr_in[0],
184	(int) htons(ctx->cb_port));
185	#else
186	code = rx_Init((int) htons(ctx->cb_port));
187	#endif
188	#endif /* RPC_TEST_GLOBAL_RX_INIT */
189
190	assert(pthread_attr_init(&tattr) == 0);
191	assert(pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED) == 0);
192	assert(pthread_create(&tid, &tattr, init_callback_service_lwp, ctx) == 0);
193
194	return (code);
195
196	} /* init_callback_service */
197
198	afs_int32 init_fs_channel(rpc_test_request_ctx *octx, char cb_if,
199	char listen_addr_s, char prefix, char *fs_addr_s,
200	afs_uint32 flags)
201	{
202	rpc_test_request_ctx *ctx;
203	afs_int32 code = 0;
204	#ifdef AFS_NT40_ENV
205	afs_int32 sslen = sizeof(struct sockaddr);
206	#endif
207
208	ctx = octx = (rpc_test_request_ctx ) malloc(sizeof(rpc_test_request_ctx));
209	memset(ctx, 0, sizeof(rpc_test_request_ctx));
210
211	/* initialize a local mutex */
212	code = pthread_mutex_init(&ctx->mtx, &rpc_test_params.mtx_attrs);
213
214	/* lock package before rx setup--which has global deps, atm */
215	pthread_mutex_lock(&rpc_test_params.mtx);
216
217	ctx->cno = rpc_test_params.next_cno++;
218	ctx->flags = flags;
219
220	/* afscbint (server) */
221	sprintf(ctx->cb_svc_name, "cb_%d", ctx->cno);
222	sprintf(ctx->cb_if_s, "%s", cb_if);
223	sprintf(ctx->cb_listen_addr_s, "%s", listen_addr_s);
224	sprintf(ctx->cb_prefix_s, "%s", prefix);
225	sprintf(ctx->fs_addr_s, "%s", fs_addr_s);
226
227	#if defined(RPC_TEST_ADD_ADDRESSES)
228	#if defined(AFS_LINUX26_ENV)
229	sprintf(cmd, "ip addr add %s/%s dev %s label %s", listen_addr_s, prefix,
230	cb_if, cb_if);
231	code = system(cmd);
232	#endif
233	#endif /* RPC_TEST_ADD_ADDRESSES */
234
235	/* lock this */
236	pthread_mutex_lock(&ctx->mtx);
237
238	/* set up rx */
239	ctx->cb_port = rpc_test_params.cb_next_port++;
240	ctx->cb_listen_addr.numberOfInterfaces = 1;
241
242	#ifdef AFS_NT40_ENV
243	code = WSAStringToAddressA(listen_addr_s, AF_INET, NULL,
244	(struct sockaddr*) &(ctx->cb_listen_addr), &sslen);
245	#else
246	code = inet_pton(AF_INET, listen_addr_s,
247	(void*) &(ctx->cb_listen_addr.addr_in[0]));
248	#endif
249
250	code = init_callback_service(ctx /* LOCKED, && rpc_test_params->mtx LOCKED */);
251
252	/* fsint (client) */
253
254	#ifdef AFS_NT40_ENV
255	code = WSAStringToAddressA(fs_addr_s, AF_INET, NULL,
256	(struct sockaddr*) &(ctx->fs_addr.addr_in[0]), &sslen);
257	#else
258	code = inet_pton(AF_INET, fs_addr_s, (void*) &(ctx->fs_addr.addr_in[0]));
259	#endif
260	ctx->sc = rxnull_NewClientSecurityObject();
261	ctx->sc_index = RX_SECIDX_NULL;
262	ctx->conn = rx_NewConnection(ctx->fs_addr.addr_in[0], (int) htons(fs_port),
263	1, ctx->sc, ctx->sc_index);
264
265	/* unlock this */
266	pthread_mutex_unlock(&ctx->mtx);
267
268	return (code);
269
270	} /* init_fs_channel */
271
272	/* XXX use the pkg lock to protect the state of rx_socket for
273	* the duration of the call, switching it out for the stashed
274	* rx_socket created by rx_NewService for this channel */
275	#define RXCALL_WITH_SOCK(code, ctx, call) \
276	do { \
277	osi_socket prev_rx_socket; \
278	pthread_mutex_lock(&rpc_test_params.mtx); \
279	prev_rx_socket = rx_socket; \
280	rx_socket = ctx->svc->socket; \
281	code = call; \
282	rx_socket = prev_rx_socket; \
283	pthread_mutex_unlock(&rpc_test_params.mtx); \
284	} while(0);
285
286	afs_int32
287	rpc_test_afs_fetch_status(rpc_test_request_ctx ctx, AFSFid fid,
288	AFSFetchStatus *outstatus)
289	{
290	struct AFSVolSync tsync;
291	struct AFSCallBack tcb;
292	afs_int32 code = 0;
293
294	RXCALL_WITH_SOCK(code, ctx,
295	(RXAFS_FetchStatus(ctx->conn, fid, outstatus, &tcb, &tsync)));
296
297	return (code);
298
299	} /* rpc_test_afs_fetch_status */
300
301	afs_int32
302	rpc_test_afs_store_status(rpc_test_request_ctx ctx, AFSFid fid,
303	AFSStoreStatus instatus, AFSFetchStatus outstatus)
304	{
305	struct AFSVolSync tsync;
306	afs_int32 code = 0;
307
308	RXCALL_WITH_SOCK(code, ctx,
309	(RXAFS_StoreStatus(ctx->conn, fid, instatus, outstatus, &tsync)));
310
311	return (code);
312
313	} /* rpc_test_afs_fetch_status */
314
315	#if defined(AFS_BYTE_RANGE_FLOCKS)
316	afs_int32 rpc_test_afs_set_byterangelock(rpc_test_request_ctx *ctx,
317	AFSByteRangeLock * lock)
318	{
319	struct rx_call *tcall;
320	afs_int32 code = 0;
321
322	RXCALL_WITH_SOCK(code, ctx,
323	(RXAFS_SetByteRangeLock(ctx->conn, lock)));
324
325	return (code);
326
327	} /* rpc_test_afs_set_byterangelock */
328
329	afs_int32 rpc_test_afs_release_byterangelock(rpc_test_request_ctx *ctx,
330	AFSByteRangeLock * lock)
331	{
332	struct rx_call *tcall;
333	afs_int32 code = 0;
334
335	RXCALL_WITH_SOCK(code, ctx,
336	(RXAFS_ReleaseByteRangeLock(ctx->conn, lock)));
337
338	return (code);
339
340	} /* rpc_test_afs_release_byterangelock */
341
342	afs_int32 rpc_test_afs_upgrade_byterangelock(rpc_test_request_ctx *ctx,
343	AFSByteRangeLock * lock)
344	{
345	afs_int32 code = 0;
346
347	/* TODO: implement */
348
349	return (code);
350
351	} /* rpc_test_afs_upgrade_byterangelock */
352
353	afs_int32 rpc_test_afs_downgrade_byterangelock(rpc_test_request_ctx *ctx,
354	AFSByteRangeLock * Lock)
355	{
356	afs_int32 code = 0;
357
358	/* TODO: implement */
359
360	return (code);
361
362	} /* rpc_test_afs_downgrade_byterangelock */
363	#endif /* AFS_BYTE_RANGE_FLOCKS */
364
365	afs_int32
366	destroy_fs_channel(rpc_test_request_ctx *ctx)
367	{
368	afs_int32 code = 0;
369	#if defined(RPC_TEST_ADD_ADDRESSES)
370	#if defined(AFS_LINUX26_ENV)
371	sprintf(cmd, "ip addr del %s/%s dev %s label %s", ctx->cb_listen_addr_s,
372	ctx->cb_prefix_s, ctx->cb_if_s, ctx->cb_if_s);
373	code = system(cmd);
374	#endif
375	#endif /* RPC_TEST_ADD_ADDRESSES */
376	assert(ctx);
377	free(ctx);
378	return (code);
379
380	} /* destroy_fs_channel */
381
382	void
383	rpc_test_PkgShutdown(void)
384	{
385	} /* rpc_test_PkgShutdown */