src/budb/dbs_dump.c

   1 /*
   2  * Copyright 2000, International Business Machines Corporation and others.
   3  * All Rights Reserved.
   4  *
   5  * This software has been released under the terms of the IBM Public
   6  * License.  For details, see the LICENSE file in the top-level source
   7  * directory or online at http://www.openafs.org/dl/license10.html
   8  */
   9
  10 #include <afsconfig.h>
  11 #include <afs/param.h>
  12
  13
  14 #ifdef AFS_NT40_ENV
  15 #include <winsock2.h>
  16 #include <io.h>
  17 #include <fcntl.h>
  18 #else
  19 #include <netinet/in.h>
  20 #include <sys/time.h>
  21 #include <sys/resource.h>
  22 #include <sys/file.h>
  23 #endif
  24 #include <time.h>
  25 #include <sys/types.h>
  26 #include <afs/stds.h>
  27 #include <stdio.h>
  28 #include <lock.h>
  29 #include <ubik.h>
  30 #include <lwp.h>
  31 #include <rx/xdr.h>
  32 #include <rx/rx.h>
  33 #include <rx/rxkad.h>
  34 #include <string.h>
  35 #include <des.h>
  36 #include <afs/cellconfig.h>
  37 #include <errno.h>
  38 #include "budb.h"
  39 #include "budb_errs.h"
  40 #include "database.h"
  41 #include "budb_prototypes.h"
  42 #include "error_macros.h"
  43 #include "globals.h"
  44 #include "afs/audit.h"
  45
  46 afs_int32 DumpDB(struct rx_call *, int, afs_int32, charListT *, afs_int32 *);
  47 afs_int32 RestoreDbHeader(struct rx_call *, struct DbHeader *);
  48 void *dumpWatcher(void *);
  49
  50 /* dump ubik database - interface routines */
  51
  52 /* badEntry
  53  *      no checking for now.
  54  */
  55
  56 afs_int32
  57 badEntry(afs_uint32 dbAddr)
  58 {
  59     /* return entry ok */
  60     return (0);
  61 }
  62
  63 /* setupDbDump
  64  *      decode the arguments passed via LWP and dump the database.
  65  */
  66
  67 void *
  68 setupDbDump(void *param)
  69 {
  70     int writeFid = (int)param;
  71     afs_int32 code = 0;
  72
  73     code = InitRPC(&dumpSyncPtr->ut, LOCKREAD, 1);
  74     if (code)
  75         goto error_exit;
  76
  77     code = writeDatabase(dumpSyncPtr->ut, writeFid);
  78     if (code)
  79         LogError(code, "writeDatabase failed\n");
  80
  81     code = close(writeFid);
  82     if (code)
  83         LogError(code, "pipe writer close failed\n");
  84
  85     LogDebug(5, "writeDatabase complete\n");
  86
  87   error_exit:
  88     if (dumpSyncPtr->ut)
  89         ubik_EndTrans(dumpSyncPtr->ut);
  90     return (void *)(code);
  91 }
  92
  93
  94 afs_int32
  95 SBUDB_DumpDB(struct rx_call *call, int firstcall, afs_int32 maxLength,
  96              charListT *charListPtr, afs_int32 *done)
  97 {
  98     afs_int32 code;
  99
 100     code = DumpDB(call, firstcall, maxLength, charListPtr, done);
 101     osi_auditU(call, BUDB_DmpDBEvent, code, AUD_END);
 102     return code;
 103 }
 104
 105 afs_int32
 106 DumpDB(struct rx_call *call,
 107        int firstcall,           /* 1 - init.  0 - no init */
 108        afs_int32 maxLength,
 109        charListT *charListPtr,
 110        afs_int32 *done)
 111 {
 112 #ifdef AFS_PTHREAD_ENV
 113     pthread_t dumperPid, watcherPid;
 114     pthread_attr_t dumperPid_tattr;
 115     pthread_attr_t watcherPid_tattr;
 116 #else
 117     PROCESS dumperPid, watcherPid;
 118 #endif
 119     int readSize;
 120     afs_int32 code = 0;
 121
 122     if (callPermitted(call) == 0)
 123         ERROR(BUDB_NOTPERMITTED);
 124
 125     ObtainWriteLock(&dumpSyncPtr->ds_lock);
 126
 127     /* If asking for zero bytes, then this is a call to reset the timeToLive
 128      * timer. Reset it if there is a dump in progress.
 129      */
 130     if (maxLength == 0) {
 131         charListPtr->charListT_val = NULL;
 132         charListPtr->charListT_len = 0;
 133
 134         *done = ((dumpSyncPtr->statusFlags == 0) ? 1 : 0);
 135
 136         /* reset the clock on dump timeout */
 137         dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
 138         goto error_exit;
 139     }
 140
 141     if (dumpSyncPtr->statusFlags == 0) {
 142         if (!firstcall)
 143             ERROR(BUDB_DUMPFAILED);
 144
 145         LogDebug(5, "Setup dump\n");
 146
 147         /* no dump in progress - setup and retake lock */
 148         memset(dumpSyncPtr, 0, sizeof(*dumpSyncPtr));
 149 /*      ObtainWriteLock(&dumpSyncPtr->ds_lock); */
 150
 151         /* mark dump in progress */
 152         dumpSyncPtr->statusFlags = 1;
 153
 154         code = pipe(dumpSyncPtr->pipeFid);
 155         if (code)
 156             ERROR(errno);
 157
 158 #ifdef AFS_PTHREAD_ENV
 159         /* Initialize the condition variables and the mutexes we use
 160          * to signal and synchronize the reader and writer threads.
 161          */
 162         assert(pthread_cond_init(&dumpSyncPtr->ds_readerStatus_cond, (const pthread_condattr_t *)0) == 0);
 163         assert(pthread_cond_init(&dumpSyncPtr->ds_writerStatus_cond, (const pthread_condattr_t *)0) == 0);
 164         assert(pthread_mutex_init(&dumpSyncPtr->ds_readerStatus_mutex, (const pthread_mutexattr_t *)0) == 0);
 165         assert(pthread_mutex_init(&dumpSyncPtr->ds_writerStatus_mutex, (const pthread_mutexattr_t *)0) == 0);
 166
 167         /* Initialize the thread attributes and launch the thread */
 168
 169         assert(pthread_attr_init(&dumperPid_tattr) == 0);
 170         assert(pthread_attr_setdetachstate(&dumperPid_tattr, PTHREAD_CREATE_DETACHED) == 0);
 171         assert(pthread_create(&dumperPid, &dumperPid_tattr, (void *)setupDbDump, NULL) == 0);
 172
 173 #else
 174         code =
 175             LWP_CreateProcess(setupDbDump, 16384, 1,
 176                               (void *)dumpSyncPtr->pipeFid[1],
 177                               "Database Dumper", &dumperPid);
 178         if (code)
 179             goto error_exit;
 180 #endif
 181
 182         dumpSyncPtr->dumperPid = dumperPid;
 183         dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
 184
 185 #ifdef AFS_PTHREAD_ENV
 186         /* Initialize the thread attributes and launch the thread */
 187
 188         assert(pthread_attr_init(&watcherPid_tattr) == 0);
 189         assert(pthread_attr_setdetachstate(&watcherPid_tattr, PTHREAD_CREATE_DETACHED) == 0);
 190         assert(pthread_create(&watcherPid, &watcherPid_tattr, (void *)dumpWatcher, NULL) == 0);
 191 #else
 192         /* now create the watcher thread */
 193         code =
 194             LWP_CreateProcess(dumpWatcher, 16384, 1, 0,
 195                               "Database Dump Watchdog", &watcherPid);
 196 #endif
 197     } else if (firstcall)
 198         ERROR(BUDB_LOCKED);
 199
 200     /* now read the database and feed it to the rpc connection */
 201
 202     /* wait for data */
 203     while (dumpSyncPtr->ds_bytes == 0) {
 204         /* if no more data */
 205         if ((dumpSyncPtr->ds_writerStatus == DS_DONE)
 206             || (dumpSyncPtr->ds_writerStatus == DS_DONE_ERROR)) {
 207             break;
 208         }
 209
 210         if (dumpSyncPtr->ds_writerStatus == DS_WAITING) {
 211             LogDebug(6, "wakup writer\n");
 212             dumpSyncPtr->ds_writerStatus = 0;
 213 #ifdef AFS_PTHREAD_ENV
 214             assert(pthread_cond_broadcast(&dumpSyncPtr->ds_writerStatus_cond) == 0);
 215 #else
 216             code = LWP_SignalProcess(&dumpSyncPtr->ds_writerStatus);
 217             if (code)
 218                 LogError(code, "BUDB_DumpDB: signal delivery failed\n");
 219 #endif
 220         }
 221         LogDebug(6, "wait for writer\n");
 222         dumpSyncPtr->ds_readerStatus = DS_WAITING;
 223         ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 224 #ifdef AFS_PTHREAD_ENV
 225         assert(pthread_mutex_lock(&dumpSyncPtr->ds_readerStatus_mutex) == 0);
 226         assert(pthread_cond_wait(&dumpSyncPtr->ds_readerStatus_cond, &dumpSyncPtr->ds_readerStatus_mutex) == 0);
 227         assert(pthread_mutex_unlock(&dumpSyncPtr->ds_readerStatus_mutex) == 0);
 228 #else
 229         LWP_WaitProcess(&dumpSyncPtr->ds_readerStatus);
 230 #endif
 231         ObtainWriteLock(&dumpSyncPtr->ds_lock);
 232     }
 233
 234     charListPtr->charListT_val = (char *)malloc(maxLength);
 235     readSize =
 236         read(dumpSyncPtr->pipeFid[0], charListPtr->charListT_val, maxLength);
 237
 238     /* reset the clock on dump timeout */
 239     dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
 240
 241     LogDebug(4, "read of len %d returned %d\n", maxLength, readSize);
 242
 243     charListPtr->charListT_len = readSize;
 244
 245     if (readSize == 0) {        /* last chunk */
 246         *done = 1;
 247         close(dumpSyncPtr->pipeFid[0]);
 248         dumpSyncPtr->statusFlags = 0;
 249     } else
 250         *done = 0;
 251
 252     dumpSyncPtr->ds_bytes -= readSize;
 253     if (dumpSyncPtr->ds_writerStatus == DS_WAITING) {
 254         dumpSyncPtr->ds_writerStatus = 0;
 255 #ifdef AFS_PTHREAD_ENV
 256         assert(pthread_cond_broadcast(&dumpSyncPtr->ds_writerStatus_cond) == 0);
 257 #else
 258         code = LWP_SignalProcess(&dumpSyncPtr->ds_writerStatus);
 259         if (code)
 260             LogError(code, "BUDB_DumpDB: signal delivery failed\n");
 261 #endif
 262     }
 263
 264   error_exit:
 265     if (!code && (dumpSyncPtr->ds_writerStatus == DS_DONE_ERROR))
 266         code = -1;
 267     ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 268     return (code);
 269 }
 270
 271 afs_int32
 272 SBUDB_RestoreDbHeader(struct rx_call *call, struct DbHeader *header)
 273 {
 274     afs_int32 code;
 275
 276     code = RestoreDbHeader(call, header);
 277     osi_auditU(call, BUDB_RstDBHEvent, code, AUD_END);
 278     return code;
 279 }
 280
 281 afs_int32
 282 RestoreDbHeader(struct rx_call *call, struct DbHeader *header)
 283 {
 284     struct ubik_trans *ut = 0;
 285     afs_int32 code = 0;
 286
 287     extern struct memoryDB db;
 288
 289     if (callPermitted(call) == 0)
 290         ERROR(BUDB_NOTPERMITTED);
 291
 292     code = InitRPC(&ut, LOCKWRITE, 1);
 293     if (code)
 294         goto error_exit;
 295
 296     if (header->dbversion != ntohl(db.h.version))
 297         ERROR(BUDB_VERSIONMISMATCH);
 298
 299     /* merge rather than replace the header information */
 300     if (db.h.lastDumpId < htonl(header->lastDumpId))
 301         db.h.lastDumpId = htonl(header->lastDumpId);
 302
 303     if (db.h.lastTapeId < htonl(header->lastTapeId))
 304         db.h.lastTapeId = htonl(header->lastTapeId);
 305
 306     if (db.h.lastInstanceId < htonl(header->lastInstanceId))
 307         db.h.lastInstanceId = htonl(header->lastInstanceId);
 308
 309     code = dbwrite(ut, 0, (char *)&db.h, sizeof(db.h));
 310     if (code)
 311         code = BUDB_IO;
 312
 313   error_exit:
 314     if (ut)
 315         ubik_EndTrans(ut);
 316     return (code);
 317 }
 318
 319 /* dumpWatcher
 320  *      monitors the state of a database dump. If the dump calls do not
 321  *      reset the time to live value, the dump times out. In that case,
 322  *      we kill the database traversal thread and clean up all the other
 323  *      state. Most importantly, the database is unlocked so that other
 324  *      transactions can proceed.
 325  */
 326
 327 void *
 328 dumpWatcher(void *unused)
 329 {
 330     afs_int32 code;
 331
 332     while (1) {                 /*w */
 333
 334         /* printf("dumpWatcher\n"); */
 335         ObtainWriteLock(&dumpSyncPtr->ds_lock);
 336
 337         if (dumpSyncPtr->statusFlags == 0) {
 338             /* dump has finished */
 339             goto exit;
 340         }
 341
 342         /* check time to live */
 343         if (time(0) > dumpSyncPtr->timeToLive) {        /*i */
 344             /* dump has exceeded the allocated time - terminate it */
 345             LogError(0, "Database dump timeout exceeded: %s",
 346                      ctime(&dumpSyncPtr->timeToLive));
 347             LogError(0, "Terminating database dump\n");
 348
 349             close(dumpSyncPtr->pipeFid[0]);
 350             close(dumpSyncPtr->pipeFid[1]);
 351 #ifdef AFS_PTHREAD_ENV
 352             assert(pthread_cancel(dumpSyncPtr->dumperPid) == 0);
 353 #else
 354             code = LWP_DestroyProcess(dumpSyncPtr->dumperPid);
 355             if (code)
 356                 LogError(code, "dumpWatcher: failed to kill dump thread\n");
 357 #endif
 358
 359             if (dumpSyncPtr->ut) {
 360                 code = ubik_AbortTrans(dumpSyncPtr->ut);
 361                 if (code)
 362                     LogError(code, "Aborting dump transaction\n");
 363             }
 364
 365             memset(dumpSyncPtr, 0, sizeof(*dumpSyncPtr));
 366             goto exit;
 367         }
 368         /*i */
 369         ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 370 #ifdef AFS_PTHREAD_ENV
 371         sleep(5);
 372 #else
 373         IOMGR_Sleep(5);
 374 #endif
 375     }                           /*w */
 376
 377   exit:
 378     ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 379     /* printf("dumpWatcher exit\n"); */
 380     return (0);
 381 }