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