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