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