tests/rpctestlib/rpc_test_procs.c

   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 */