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 #include <stdio.h>
  41 #include <sys/types.h>
  42 #include <string.h>
  43
  44 #ifdef AFS_NT40_ENV
  45 #else
  46 #include <sys/param.h>
  47 #include <sys/file.h>
  48 #include <sys/ioctl.h>
  49 #include <sys/socket.h>
  50 #include <netinet/in.h>
  51 #include <arpa/inet.h>
  52 #include <netdb.h>
  53 #endif
  54
  55 #include <string.h>
  56 #include <fcntl.h>
  57 #ifdef AFS_NT40_ENV
  58 #include <io.h>
  59 #include <windows.h>
  60 #include <WINNT/afsevent.h>
  61 #else
  62 #include <pwd.h>
  63 #include <afs/venus.h>
  64 #include <sys/time.h>
  65 #include <netdb.h>
  66 #endif
  67 #include <afs/afsint.h>
  68 #define FSINT_COMMON_XG 1
  69 #include <sys/stat.h>
  70 #include <errno.h>
  71 #include <signal.h>
  72 #include <afs/vice.h>
  73 #include <afs/cmd.h>
  74 #include <afs/auth.h>
  75 #include <afs/cellconfig.h>
  76
  77 #include <afs/com_err.h>
  78 #ifdef HAVE_DIRENT_H
  79 #include <dirent.h>
  80 #endif
  81 #ifdef HAVE_DIRECT_H
  82 #include <direct.h>
  83 #endif
  84 #ifdef AFS_DARWIN_ENV
  85 #include <sys/malloc.h>
  86 #else
  87 #include <malloc.h>
  88 #endif
  89 #include <afs/errors.h>
  90 #include <afs/sys_prototypes.h>
  91 #include <rx/rx_prototypes.h>
  92 #ifdef AFS_PTHREAD_ENV
  93 #include <assert.h>
  94 #endif
  95
  96 extern const char *prog;
  97 const int ctx_key = 1;
  98
  99 #if 1
 100 #define RPC_TEST_GLOBAL_RX_INIT 1
 101 #else
 102 #undef RPC_TEST_GLOBAL_RX_INIT
 103 #endif
 104
 105 const afs_uint32 fs_port = 7000;
 106
 107 typedef struct rpc_test_pkg_params {
 108     pthread_mutex_t mtx;
 109     pthread_mutexattr_t mtx_attrs;
 110     afs_uint32 cb_next_port;
 111     afs_uint32 next_cno;
 112 } rpc_test_pkg_params;
 113 static rpc_test_pkg_params rpc_test_params;
 114
 115 afs_int32 rpc_test_PkgInit()
 116 {
 117     afs_int32 code = 0;
 118     static afs_uint32 rpc_test_initialized = 0; /* once */
 119
 120     if (!rpc_test_initialized) {
 121         rpc_test_initialized = 1;
 122     } else {
 123         printf("%s: rpc_test_PkgInit: package already initialized\n");
 124         exit(1);
 125     }
 126
 127 #ifndef AFS_NT40_ENV
 128     code = pthread_mutexattr_init(&rpc_test_params.mtx_attrs);
 129     if (code) {
 130         printf("%s: rpc_test_PkgInit: pthread_mutexattr_init failed\n", prog);
 131         exit(1);
 132     }
 133     code = pthread_mutex_init(&rpc_test_params.mtx, &rpc_test_params.mtx_attrs);
 134     if (code) {
 135         printf("%s: rpc_test_PkgInit: pthread_mutex_init failed\n", prog);
 136         exit(1);
 137     }
 138 #endif
 139
 140     /* start connection sequence */
 141     rpc_test_params.next_cno = 1;
 142
 143     /* set the starting port in sequence */
 144     rpc_test_params.cb_next_port = 7105;
 145
 146 #if defined(RPC_TEST_GLOBAL_RX_INIT)
 147     rx_Init(0);
 148 #endif
 149
 150     return (code);
 151
 152 }        /* rpc_test_PkgInit */
 153
 154 static void *
 155 init_callback_service_lwp(void *arg)
 156 {
 157     struct rx_securityClass *sc;
 158     struct rx_service *svc;
 159     afs_int32 code = 0;
 160
 161     rpc_test_request_ctx *ctx = (rpc_test_request_ctx *) arg;
 162
 163     printf("%s: init_callback_service_lwp: listen_addr: %s "
 164            "(%d) cb_port: %d\n",
 165            prog, ctx->cb_listen_addr_s, ctx->cb_listen_addr.addr_in[0],
 166            ctx->cb_port);
 167
 168     sc = (struct rx_securityClass *) rxnull_NewServerSecurityObject();
 169     if (!sc) {
 170         fprintf(stderr,"rxnull_NewServerSecurityObject failed for callback "
 171                 "service\n");
 172         exit(1);
 173     }
 174
 175 #if defined(RPC_TEST_GLOBAL_RX_INIT)
 176     svc = rx_NewServiceHost(htonl(INADDR_ANY), htons(ctx->cb_port), 1,
 177                             ctx->cb_svc_name, &sc, 1, RXAFSCB_ExecuteRequest);
 178 #else
 179     svc = rx_NewService(0, 1, ctx->cb_svc_name, &sc, 1, RXAFSCB_ExecuteRequest);
 180 #endif
 181     /* stash context */
 182     rx_SetServiceSpecific(svc, ctx_key, ctx);
 183
 184     if (!svc) {
 185         fprintf(stderr,"rx_NewServiceHost failed for callback service\n");
 186         exit(1);
 187     }
 188
 189     /* XXX stash service so we can hijack its rx_socket when inititiating
 190      * RPC calls */
 191     ctx->svc = svc;
 192
 193     /* release pkg mutex before entering rx processing loop */
 194     pthread_mutex_unlock(&rpc_test_params.mtx);
 195
 196     rx_StartServer(1);
 197
 198     printf("%s: init_callback_service_lwp: finished");
 199
 200     return (NULL);
 201
 202 }        /* callback_service_lwp */
 203
 204 afs_int32 init_callback_service(rpc_test_request_ctx *ctx)
 205 {
 206     pthread_t tid;
 207     pthread_attr_t tattr;
 208     afs_int32 code = 0;
 209
 210     afs_uuid_create(&(ctx->cb_listen_addr.uuid));
 211
 212 #if !defined(RPC_TEST_GLOBAL_RX_INIT)
 213 #if 0
 214     code = rx_InitHost(ctx->cb_listen_addr.addr_in[0],
 215                        (int) htons(ctx->cb_port));
 216 #else
 217     code = rx_Init((int) htons(ctx->cb_port));
 218 #endif
 219 #endif /* RPC_TEST_GLOBAL_RX_INIT */
 220
 221     assert(pthread_attr_init(&tattr) == 0);
 222     assert(pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_DETACHED) == 0);
 223     assert(pthread_create(&tid, &tattr, init_callback_service_lwp, ctx) == 0);
 224
 225     return (code);
 226
 227 }        /* init_callback_service */
 228
 229 afs_int32 init_fs_channel(rpc_test_request_ctx **octx, char *cb_if,
 230                           char *listen_addr_s, char *prefix, char *fs_addr_s,
 231                           afs_uint32 flags)
 232 {
 233     char cmd[512];
 234     rpc_test_request_ctx *ctx;
 235     afs_int32 code = 0;
 236 #ifdef AFS_NT40_ENV
 237     afs_int32 sslen = sizeof(struct sockaddr);
 238 #endif
 239
 240     ctx = *octx = (rpc_test_request_ctx *) malloc(sizeof(rpc_test_request_ctx));
 241     memset(ctx, 0, sizeof(rpc_test_request_ctx));
 242
 243     /* initialize a local mutex */
 244     code = pthread_mutex_init(&ctx->mtx, &rpc_test_params.mtx_attrs);
 245
 246     /* lock package before rx setup--which has global deps, atm */
 247     pthread_mutex_lock(&rpc_test_params.mtx);
 248
 249     ctx->cno = rpc_test_params.next_cno++;
 250     ctx->flags = flags;
 251
 252     /* afscbint (server) */
 253     sprintf(ctx->cb_svc_name, "cb_%d", ctx->cno);
 254     sprintf(ctx->cb_if_s, cb_if);
 255     sprintf(ctx->cb_listen_addr_s, listen_addr_s);
 256     sprintf(ctx->cb_prefix_s, prefix);
 257     sprintf(ctx->fs_addr_s, fs_addr_s);
 258
 259 #if defined(RPC_TEST_ADD_ADDRESSES)
 260 #if defined(AFS_LINUX26_ENV)
 261     sprintf(cmd, "ip addr add %s/%s dev %s label %s", listen_addr_s, prefix,
 262             cb_if, cb_if);
 263     code = system(cmd);
 264 #endif
 265 #endif /* RPC_TEST_ADD_ADDRESSES */
 266
 267     /* lock this */
 268     pthread_mutex_lock(&ctx->mtx);
 269
 270     /* set up rx */
 271     ctx->cb_port = rpc_test_params.cb_next_port++;
 272     ctx->cb_listen_addr.numberOfInterfaces = 1;
 273
 274 #ifdef AFS_NT40_ENV
 275     code = WSAStringToAddressA(listen_addr_s, AF_INET, NULL,
 276               (struct sockaddr*) &(ctx->cb_listen_addr), &sslen);
 277 #else
 278     code = inet_pton(AF_INET, listen_addr_s,
 279                      (void*) &(ctx->cb_listen_addr.addr_in[0]));
 280 #endif
 281
 282     code = init_callback_service(ctx /* LOCKED, && rpc_test_params->mtx LOCKED */);
 283
 284     /* fsint (client) */
 285
 286 #ifdef AFS_NT40_ENV
 287     code = WSAStringToAddressA(fs_addr_s, AF_INET, NULL,
 288               (struct sockaddr*) &(ctx->fs_addr.addr_in[0]), &sslen);
 289 #else
 290     code = inet_pton(AF_INET, fs_addr_s, (void*) &(ctx->fs_addr.addr_in[0]));
 291 #endif
 292     ctx->sc = rxnull_NewClientSecurityObject();
 293     ctx->sc_index = RX_SECIDX_NULL;
 294     ctx->conn = rx_NewConnection(ctx->fs_addr.addr_in[0], (int) htons(fs_port),
 295                                  1, ctx->sc, ctx->sc_index);
 296
 297     /* unlock this */
 298     pthread_mutex_unlock(&ctx->mtx);
 299
 300 out:
 301     return (code);
 302
 303 }        /* init_fs_channel */
 304
 305 /* XXX use the pkg lock to protect the state of rx_socket for
 306  * the duration of the call, switching it out for the stashed
 307  * rx_socket created by rx_NewService for this channel */
 308 #define RXCALL_WITH_SOCK(code, ctx, call) \
 309     do { \
 310         osi_socket prev_rx_socket; \
 311         pthread_mutex_lock(&rpc_test_params.mtx); \
 312         prev_rx_socket = rx_socket; \
 313         rx_socket = ctx->svc->socket; \
 314         code = call; \
 315         rx_socket = prev_rx_socket; \
 316         pthread_mutex_unlock(&rpc_test_params.mtx); \
 317 } while(0);
 318
 319 afs_int32
 320 rpc_test_afs_fetch_status(rpc_test_request_ctx *ctx, AFSFid *fid,
 321                               AFSFetchStatus *outstatus)
 322 {
 323     struct rx_call *tcall;
 324     struct AFSVolSync tsync;
 325     struct AFSCallBack tcb;
 326     afs_int32 code = 0;
 327
 328     RXCALL_WITH_SOCK(code, ctx,
 329        (RXAFS_FetchStatus(ctx->conn, fid, outstatus, &tcb, &tsync)));
 330
 331     return (code);
 332
 333 }        /* rpc_test_afs_fetch_status */
 334
 335 afs_int32
 336 rpc_test_afs_store_status(rpc_test_request_ctx *ctx, AFSFid *fid,
 337                     AFSStoreStatus *instatus, AFSFetchStatus *outstatus)
 338 {
 339     struct rx_call *tcall;
 340     struct AFSVolSync tsync;
 341     afs_int32 code = 0;
 342
 343     RXCALL_WITH_SOCK(code, ctx,
 344        (RXAFS_StoreStatus(ctx->conn, fid, instatus, outstatus, &tsync)));
 345
 346     return (code);
 347
 348 }        /* rpc_test_afs_fetch_status */
 349
 350 #if defined(AFS_BYTE_RANGE_FLOCKS)
 351 afs_int32 rpc_test_afs_set_byterangelock(rpc_test_request_ctx *ctx,
 352     AFSByteRangeLock * lock)
 353 {
 354     struct rx_call *tcall;
 355     afs_int32 code = 0;
 356
 357     RXCALL_WITH_SOCK(code, ctx,
 358        (RXAFS_SetByteRangeLock(ctx->conn, lock)));
 359
 360     return (code);
 361
 362 }        /* rpc_test_afs_set_byterangelock */
 363
 364 afs_int32 rpc_test_afs_release_byterangelock(rpc_test_request_ctx *ctx,
 365     AFSByteRangeLock * lock)
 366 {
 367     struct rx_call *tcall;
 368     afs_int32 code = 0;
 369
 370     RXCALL_WITH_SOCK(code, ctx,
 371        (RXAFS_ReleaseByteRangeLock(ctx->conn, lock)));
 372
 373     return (code);
 374
 375 }        /* rpc_test_afs_release_byterangelock */
 376
 377 afs_int32 rpc_test_afs_upgrade_byterangelock(rpc_test_request_ctx *ctx,
 378     AFSByteRangeLock * lock)
 379 {
 380     afs_int32 code = 0;
 381
 382     /* TODO:  implement */
 383
 384     return (code);
 385
 386 }        /* rpc_test_afs_upgrade_byterangelock */
 387
 388 afs_int32 rpc_test_afs_downgrade_byterangelock(rpc_test_request_ctx *ctx,
 389     AFSByteRangeLock * Lock)
 390 {
 391     afs_int32 code = 0;
 392
 393     /* TODO:  implement */
 394
 395     return (code);
 396
 397 }        /* rpc_test_afs_downgrade_byterangelock */
 398 #endif /* AFS_BYTE_RANGE_FLOCKS */
 399
 400 afs_int32
 401 destroy_fs_channel(rpc_test_request_ctx *ctx)
 402 {
 403     char cmd[512];
 404     afs_int32 code = 0;
 405 #if defined(RPC_TEST_ADD_ADDRESSES)
 406 #if defined(AFS_LINUX26_ENV)
 407     sprintf(cmd, "ip addr del %s/%s dev %s label %s", ctx->cb_listen_addr_s,
 408             ctx->cb_prefix_s, ctx->cb_if_s, ctx->cb_if_s);
 409     code = system(cmd);
 410 #endif
 411 #endif /* RPC_TEST_ADD_ADDRESSES */
 412     assert(ctx);
 413     free(ctx);
 414     return (code);
 415
 416 }        /* destroy_fs_channel */
 417
 418 void
 419 rpc_test_PkgShutdown()
 420 {
 421     afs_int32 code = 0;
 422
 423 }        /* rpc_test_PkgShutdown */