refs #5189 Save last_xid in shutdown entry instead of increasing lsn and aliasing

git-svn-id: file:///svn/toku/tokudb@45271 c7de825b-a66e-492c-adef-691d508d4ae1

ft/ft_layout_version.h

@@ -25,7 +25,9 @@ enum ft_layout_version_e {
     FT_LAYOUT_VERSION_17 = 17, // Dr. No:  Add STAT64INFO_S to brt_header
     FT_LAYOUT_VERSION_18 = 18, // Dr. No:  Add HOT info to brt_header
     FT_LAYOUT_VERSION_19 = 19, // Doofenshmirtz: Add compression method, highest_unused_msn_for_upgrade
-    FT_LAYOUT_VERSION_20 = 20, // Clayface: Add compression method to log_fcreate, mgr_last_xid after begin checkpoint
+    FT_LAYOUT_VERSION_20 = 20, // Deadshot: Add compression method to log_fcreate,
+                               // mgr_last_xid after begin checkpoint,
+                               // last_xid to shutdown
     FT_NEXT_VERSION,           // the version after the current version
     FT_LAYOUT_VERSION   = FT_NEXT_VERSION-1, // A hack so I don't have to change this line.
     FT_LAYOUT_MIN_SUPPORTED_VERSION = FT_LAYOUT_VERSION_13, // Minimum version supported

ft/log_upgrade.c

@@ -24,13 +24,13 @@ toku_log_upgrade_get_footprint(void) {
 
 
 // return 0 if clean shutdown, TOKUDB_UPGRADE_FAILURE if not clean shutdown
-static int 
-verify_clean_shutdown_of_log_version_current(const char *log_dir, LSN * last_lsn) {
+static int
+verify_clean_shutdown_of_log_version_current(const char *log_dir, LSN * last_lsn, TXNID *last_xid) {
     int rval = TOKUDB_UPGRADE_FAILURE;
     TOKULOGCURSOR cursor = NULL;
     int r;
     FOOTPRINTSETUP(100);
-    
+
     FOOTPRINT(1);
 
     r = toku_logcursor_create(&cursor, log_dir);
@@ -41,8 +41,12 @@ verify_clean_shutdown_of_log_version_current(const char *log_dir, LSN * last_lsn
         FOOTPRINT(2);
         if (le->cmd==LT_shutdown) {
             LSN lsn = le->u.shutdown.lsn;
-            if (last_lsn)
-            *last_lsn = lsn;
+            if (last_lsn) {
+                *last_lsn = lsn;
+            }
+            if (last_xid) {
+                *last_xid = le->u.shutdown.last_xid;
+            }
             rval = 0;
         }
     }
@@ -54,12 +58,12 @@ verify_clean_shutdown_of_log_version_current(const char *log_dir, LSN * last_lsn
 
 
 // return 0 if clean shutdown, TOKUDB_UPGRADE_FAILURE if not clean shutdown
-static int 
-verify_clean_shutdown_of_log_version_old(const char *log_dir, LSN * last_lsn) {
+static int
+verify_clean_shutdown_of_log_version_old(const char *log_dir, LSN * last_lsn, TXNID *last_xid, uint32_t version) {
     int rval = TOKUDB_UPGRADE_FAILURE;
     int r;
     FOOTPRINTSETUP(10);
-    
+
     FOOTPRINT(1);
 
     int n_logfiles;
@@ -73,28 +77,51 @@ verify_clean_shutdown_of_log_version_old(const char *log_dir, LSN * last_lsn) {
     // Only look at newest log
     // basename points to first char after last / in file pathname
     basename = strrchr(logfiles[n_logfiles-1], '/') + 1;
-    int version;
+    uint32_t version_name;
     long long index = -1;
-    r = sscanf(basename, "log%lld.tokulog%d", &index, &version);
+    r = sscanf(basename, "log%lld.tokulog%u", &index, &version_name);
     assert(r==2);  // found index and version
+    invariant(version_name == version);
     assert(version>=TOKU_LOG_MIN_SUPPORTED_VERSION);
     assert(version< TOKU_LOG_VERSION); //Must be old
     // find last LSN
     r = toku_logcursor_create_for_file(&cursor, log_dir, basename);
-    if (r==0) {
-        r = toku_logcursor_last(cursor, &entry);
-        if (r == 0) {
-            FOOTPRINT(2);
-            if (entry->cmd==LT_shutdown) {
-                LSN lsn = entry->u.shutdown.lsn;
-                if (last_lsn)
-                    *last_lsn = lsn;
-                rval = 0;
+    if (r != 0) {
+        goto cleanup_no_logcursor;
+    }
+    r = toku_logcursor_last(cursor, &entry);
+    if (r != 0) {
+        goto cleanup;
+    }
+    FOOTPRINT(2);
+    //TODO: Remove this special case once FT_LAYOUT_VERSION_19 (and older) are not supported.
+    if (version <= FT_LAYOUT_VERSION_19) {
+        if (entry->cmd==LT_shutdown_up_to_19) {
+            LSN lsn = entry->u.shutdown_up_to_19.lsn;
+            if (last_lsn) {
+                *last_lsn = lsn;
             }
+            if (last_xid) {
+                // Use lsn as last_xid.
+                *last_xid = lsn.lsn;
+            }
+            rval = 0;
         }
-        r = toku_logcursor_destroy(&cursor);
-        assert(r == 0);
     }
+    else if (entry->cmd==LT_shutdown) {
+        LSN lsn = entry->u.shutdown.lsn;
+        if (last_lsn) {
+            *last_lsn = lsn;
+        }
+        if (last_xid) {
+            *last_xid = entry->u.shutdown.last_xid;
+        }
+        rval = 0;
+    }
+cleanup:
+    r = toku_logcursor_destroy(&cursor);
+    assert(r == 0);
+cleanup_no_logcursor:
     for(int i=0;i<n_logfiles;i++) {
         toku_free(logfiles[i]);
     }
@@ -105,14 +132,14 @@ verify_clean_shutdown_of_log_version_old(const char *log_dir, LSN * last_lsn) {
 
 
 static int
-verify_clean_shutdown_of_log_version(const char *log_dir, uint32_t version, LSN *last_lsn) {
+verify_clean_shutdown_of_log_version(const char *log_dir, uint32_t version, LSN *last_lsn, TXNID *last_xid) {
     // return 0 if clean shutdown, TOKUDB_UPGRADE_FAILURE if not clean shutdown
     int r = 0;
     FOOTPRINTSETUP(1000);
 
     if (version < TOKU_LOG_VERSION)  {
         FOOTPRINT(1);
-        r = verify_clean_shutdown_of_log_version_old(log_dir, last_lsn);
+        r = verify_clean_shutdown_of_log_version_old(log_dir, last_lsn, last_xid, version);
 	if (r != 0) {
 	    fprintf(stderr, "Cannot upgrade TokuDB version %d database.", version);
 	    fprintf(stderr, "  Previous improper shutdown detected.\n");
@@ -121,19 +148,19 @@ verify_clean_shutdown_of_log_version(const char *log_dir, uint32_t version, LSN
     else {
         FOOTPRINT(2);
         assert(version == TOKU_LOG_VERSION);
-        r = verify_clean_shutdown_of_log_version_current(log_dir, last_lsn);
+        r = verify_clean_shutdown_of_log_version_current(log_dir, last_lsn, last_xid);
     }
     FOOTPRINTCAPTURE;
     return r;
 }
-    
+
 
 // Actually create a log file of the current version, making the environment be of the current version.
 static int
-upgrade_log(const char *env_dir, const char *log_dir, LSN last_lsn) { // the real deal
+upgrade_log(const char *env_dir, const char *log_dir, LSN last_lsn, TXNID last_xid) { // the real deal
     int r;
     FOOTPRINTSETUP(10000);
-    
+
     LSN initial_lsn = last_lsn;
     initial_lsn.lsn++;
     CACHETABLE ct;
@@ -148,7 +175,7 @@ upgrade_log(const char *env_dir, const char *log_dir, LSN last_lsn) { // the rea
         r = toku_logger_create(&logger);
         assert(r == 0);
         toku_logger_set_cachetable(logger, ct);
-        r = toku_logger_open(log_dir, logger);
+        r = toku_logger_open_with_last_xid(log_dir, logger, last_xid);
         assert(r==0);
     }
     { //Checkpoint
@@ -156,7 +183,7 @@ upgrade_log(const char *env_dir, const char *log_dir, LSN last_lsn) { // the rea
         assert(r == 0);
     }
     { //Close cachetable and logger
-        r = toku_logger_shutdown(logger); 
+        r = toku_logger_shutdown(logger);
         assert(r==0);
         r = toku_cachetable_close(&ct);
         assert(r==0);
@@ -164,7 +191,7 @@ upgrade_log(const char *env_dir, const char *log_dir, LSN last_lsn) { // the rea
         assert(r==0);
     }
     {
-        r = verify_clean_shutdown_of_log_version(log_dir, TOKU_LOG_VERSION, NULL);
+        r = verify_clean_shutdown_of_log_version(log_dir, TOKU_LOG_VERSION, NULL, NULL);
         assert(r==0);
     }
     FOOTPRINTCAPTURE;
@@ -184,43 +211,49 @@ toku_maybe_upgrade_log(const char *env_dir, const char *log_dir, LSN * lsn_of_cl
 
     FOOTPRINT(1);
     r = toku_recover_lock(log_dir, &lockfd);
-    if (r == 0) {
+    if (r != 0) {
+        goto cleanup_no_lock;
+    }
     FOOTPRINT(2);
-        assert(log_dir);
-        assert(env_dir);
-
-        uint32_t version_of_logs_on_disk;
-        BOOL found_any_logs;
-        r = toku_get_version_of_logs_on_disk(log_dir, &found_any_logs, &version_of_logs_on_disk);
-        if (r==0) {
-        FOOTPRINT(3);
-            if (!found_any_logs)
-                r = 0; //No logs means no logs to upgrade.
-            else if (version_of_logs_on_disk > TOKU_LOG_VERSION)
-                r = TOKUDB_DICTIONARY_TOO_NEW;
-            else if (version_of_logs_on_disk < TOKU_LOG_MIN_SUPPORTED_VERSION)
-                r = TOKUDB_DICTIONARY_TOO_OLD;
-            else if (version_of_logs_on_disk == TOKU_LOG_VERSION)
-                r = 0; //Logs are up to date
-            else {
-                FOOTPRINT(4);
-                LSN last_lsn;
-                r = verify_clean_shutdown_of_log_version(log_dir, version_of_logs_on_disk, &last_lsn);
-                if (r==0) {
-                    FOOTPRINT(5);
-                    *lsn_of_clean_shutdown = last_lsn;
-                    *upgrade_in_progress = TRUE;
-                    r = upgrade_log(env_dir, log_dir, last_lsn);
-                }
-            }
-        }
-        {
-            //Clean up
-            int rc;
-            rc = toku_recover_unlock(lockfd);
-            if (r==0) r = rc;
+    assert(log_dir);
+    assert(env_dir);
+
+    uint32_t version_of_logs_on_disk;
+    BOOL found_any_logs;
+    r = toku_get_version_of_logs_on_disk(log_dir, &found_any_logs, &version_of_logs_on_disk);
+    if (r != 0) {
+        goto cleanup;
+    }
+    FOOTPRINT(3);
+    if (!found_any_logs)
+        r = 0; //No logs means no logs to upgrade.
+    else if (version_of_logs_on_disk > TOKU_LOG_VERSION)
+        r = TOKUDB_DICTIONARY_TOO_NEW;
+    else if (version_of_logs_on_disk < TOKU_LOG_MIN_SUPPORTED_VERSION)
+        r = TOKUDB_DICTIONARY_TOO_OLD;
+    else if (version_of_logs_on_disk == TOKU_LOG_VERSION)
+        r = 0; //Logs are up to date
+    else {
+        FOOTPRINT(4);
+        LSN last_lsn;
+        TXNID last_xid;
+        r = verify_clean_shutdown_of_log_version(log_dir, version_of_logs_on_disk, &last_lsn, &last_xid);
+        if (r != 0) {
+            goto cleanup;
         }
+        FOOTPRINT(5);
+        *lsn_of_clean_shutdown = last_lsn;
+        *upgrade_in_progress = TRUE;
+        r = upgrade_log(env_dir, log_dir, last_lsn, last_xid);
+    }
+cleanup:
+    {
+        //Clean up
+        int rc;
+        rc = toku_recover_unlock(lockfd);
+        if (r==0) r = rc;
     }
+cleanup_no_lock:
     FOOTPRINTCAPTURE;
     return r;
 }

ft/logfilemgr.c

@@ -44,8 +44,9 @@ int toku_logfilemgr_destroy(TOKULOGFILEMGR *lfm) {
     return r;
 }
 
-int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir) {
-    assert(lfm!=0);
+int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir, TXNID *last_xid_if_clean_shutdown) {
+    invariant_notnull(lfm);
+    invariant_notnull(last_xid_if_clean_shutdown);
 
     int r;
     int n_logfiles;
@@ -60,6 +61,7 @@ int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir) {
     long long index = -1;
     char *basename;
     LSN tmp_lsn = {0};
+    TXNID last_xid = TXNID_NONE;
     for(int i=0;i<n_logfiles;i++){
         lf_info = toku_xmalloc(sizeof(struct toku_logfile_info));
         // find the index
@@ -74,12 +76,20 @@ int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir) {
         lf_info->version = version;
         // find last LSN in logfile
         r = toku_logcursor_create_for_file(&cursor, log_dir, basename);
-        if (r!=0) 
+        if (r!=0) {
             return r;
+        }
         r = toku_logcursor_last(cursor, &entry);  // set "entry" to last log entry in logfile
-        if ( r == 0 ) {
+        if (r == 0) {
             lf_info->maxlsn = toku_log_entry_get_lsn(entry);
+
+            invariant(lf_info->maxlsn.lsn >= tmp_lsn.lsn);
             tmp_lsn = lf_info->maxlsn;
+            if (entry->cmd == LT_shutdown) {
+                last_xid = entry->u.shutdown.last_xid;
+            } else {
+                last_xid = TXNID_NONE;
+            }
         }
         else {
             lf_info->maxlsn = tmp_lsn; // handle empty logfile (no LSN in file) case
@@ -93,6 +103,7 @@ int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir) {
         toku_free(logfiles[i]);
     }
     toku_free(logfiles);
+    *last_xid_if_clean_shutdown = last_xid;
     return 0;
 }
 

ft/logfilemgr.h

@@ -27,7 +27,7 @@ typedef struct toku_logfilemgr *TOKULOGFILEMGR;
 int toku_logfilemgr_create(TOKULOGFILEMGR *lfm);
 int toku_logfilemgr_destroy(TOKULOGFILEMGR *lfm);
 
-int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir);
+int toku_logfilemgr_init(TOKULOGFILEMGR lfm, const char *log_dir, TXNID *last_xid_if_clean_shutdown);
 int toku_logfilemgr_num_logfiles(TOKULOGFILEMGR lfm);
 int toku_logfilemgr_add_logfile_info(TOKULOGFILEMGR lfm, TOKULOGFILEINFO lf_info);
 TOKULOGFILEINFO toku_logfilemgr_get_oldest_logfile_info(TOKULOGFILEMGR lfm);

ft/logformat.c

@@ -197,13 +197,22 @@ const struct logtype logtypes[] = {
     {"comment", 'T', FA{{"u_int64_t", "timestamp", 0},
                         {"BYTESTRING", "comment", 0},
                         NULLFIELD}, IGNORE_LOG_BEGIN},
-    // Note: Shutdown log entry is NOT ALLOWED TO BE CHANGED.
+    // Note: shutdown_up_to_19 log entry is NOT ALLOWED TO BE CHANGED.
     // Do not change the letter ('Q'), do not add fields,
     // do not remove fields.
+    // TODO: Kill this logentry entirely once we no longer support version 19.
+    {"shutdown_up_to_19", 'Q', FA{{"u_int64_t", "timestamp", 0},
+                         NULLFIELD}, IGNORE_LOG_BEGIN},
+    // Note: Shutdown log entry is NOT ALLOWED TO BE CHANGED.
+    // Do not change the letter ('0'), do not add fields,
+    // do not remove fields.
+    // You CAN leave this alone and add a new one, but then you have
+    // to deal with the upgrade mechanism again.
     // This is how we detect clean shutdowns from OLDER VERSIONS.
     // This log entry must always be readable for future versions.
     // If you DO change it, you need to write a separate log upgrade mechanism.
-    {"shutdown", 'Q', FA{{"u_int64_t", "timestamp", 0},
+    {"shutdown", '0', FA{{"u_int64_t", "timestamp", 0},
+                         {"TXNID", "last_xid", 0},
                          NULLFIELD}, IGNORE_LOG_BEGIN},
     {"load", 'l', FA{{"TXNID",      "xid", 0},
                      {"FILENUM",    "old_filenum", 0},

ft/logger.c

@@ -135,13 +135,15 @@ static int close_logdir(TOKULOGGER logger) {
     return closedir(logger->dir);
 }
 
-int toku_logger_open (const char *directory, TOKULOGGER logger) {
+int
+toku_logger_open_with_last_xid(const char *directory, TOKULOGGER logger, TXNID last_xid) {
     if (logger->is_open) return EINVAL;
     if (logger->is_panicked) return EINVAL;
 
     int r;
-    r = toku_logfilemgr_init(logger->logfilemgr, directory);
-    if ( r!=0 ) 
+    TXNID last_xid_if_clean_shutdown;
+    r = toku_logfilemgr_init(logger->logfilemgr, directory, &last_xid_if_clean_shutdown);
+    if ( r!=0 )
         return r;
     logger->lsn = toku_logfilemgr_get_last_lsn(logger->logfilemgr);
     logger->written_lsn = logger->lsn;
@@ -159,12 +161,19 @@ int toku_logger_open (const char *directory, TOKULOGGER logger) {
 
     logger->next_log_file_number = nexti;
     open_logfile(logger);
-    toku_txn_manager_set_last_xid_from_logger(logger->txn_manager, logger);
+    if (last_xid == TXNID_NONE) {
+        last_xid = last_xid_if_clean_shutdown;
+    }
+    toku_txn_manager_set_last_xid_from_logger(logger->txn_manager, last_xid);
 
     logger->is_open = TRUE;
     return 0;
 }
 
+int toku_logger_open (const char *directory, TOKULOGGER logger) {
+    return toku_logger_open_with_last_xid(directory, logger, TXNID_NONE);
+}
+
 bool toku_logger_rollback_is_open (TOKULOGGER logger) {
     return logger->rollback_cachefile != NULL;
 }
@@ -281,13 +290,7 @@ int toku_logger_shutdown(TOKULOGGER logger) {
         TXN_MANAGER mgr = logger->txn_manager;
         if (toku_txn_manager_num_live_txns(mgr) == 0) {
             TXNID last_xid = toku_txn_manager_get_last_xid(mgr);
-            // Increase the LSN of the shutdown log entry if it would be smaller
-            // than last_xid because we use the LSN of the shutdown log entry
-            // to seed the last_xid on bootup.
-            if (logger->lsn.lsn < last_xid) {
-                logger->lsn.lsn = last_xid;
-            }
-            r = toku_log_shutdown(logger, NULL, TRUE, 0);
+            r = toku_log_shutdown(logger, NULL, TRUE, 0, last_xid);
         }
     }
     return r;

ft/logger.h

@@ -25,6 +25,7 @@ enum {
 
 int toku_logger_create (TOKULOGGER *resultp);
 int toku_logger_open (const char *directory, TOKULOGGER logger);
+int toku_logger_open_with_last_xid(const char *directory, TOKULOGGER logger, TXNID last_xid);
 int toku_logger_shutdown(TOKULOGGER logger);
 int toku_logger_close(TOKULOGGER *loggerp);
 int toku_logger_open_rollback(TOKULOGGER logger, CACHETABLE cachetable, BOOL create);

ft/recover.c

@@ -1143,6 +1143,16 @@ static int toku_recover_backward_comment (struct logtype_comment *UU(l), RECOVER
     return 0;
 }
 
+static int toku_recover_shutdown_up_to_19 (struct logtype_shutdown_up_to_19 *UU(l), RECOVER_ENV UU(renv)) {
+    // nothing
+    return 0;
+}
+
+static int toku_recover_backward_shutdown_up_to_19 (struct logtype_shutdown_up_to_19 *UU(l), RECOVER_ENV UU(renv)) {
+    // nothing
+    return 0;
+}
+
 static int toku_recover_shutdown (struct logtype_shutdown *UU(l), RECOVER_ENV UU(renv)) {
     // nothing
     return 0;
@@ -1195,8 +1205,6 @@ static int toku_recover_backward_hot_index(struct logtype_hot_index *UU(l), RECO
 
 // Effects: If there are no log files, or if there is a clean "shutdown" at
 // the end of the log, then we don't need recovery to run.
-// Requires: the shutdown log entry does not change between tokudb versions,
-// must always remain readable.
 // Returns: TRUE if we need recovery, otherwise FALSE.
 int tokudb_needs_recovery(const char *log_dir, BOOL ignore_log_empty) {
     int needs_recovery;

ft/tests/xid_lsn_independent.c

@@ -133,7 +133,7 @@ static void test_xid_lsn_independent_crash_recovery(int N) {
 
     TXNID last_xid_after = logger_get_last_xid(logger);
 
-    invariant(last_xid_after >= last_xid_before);
+    invariant(last_xid_after == last_xid_before);
 
     shutdown_after_recovery(&logger, &ct);
 }
@@ -159,7 +159,7 @@ static void test_xid_lsn_independent_shutdown_recovery(int N) {
 
     TXNID last_xid_after = logger_get_last_xid(logger);
 
-    invariant(last_xid_after >= last_xid_before);
+    invariant(last_xid_after == last_xid_before);
 
     shutdown_after_recovery(&logger, &ct);
 }

ft/txn_manager.c

@@ -243,7 +243,7 @@ TXNID
 toku_txn_manager_get_oldest_living_xid(TXN_MANAGER txn_manager) {
     TXNID rval = TXNID_NONE_LIVING;
     toku_mutex_lock(&txn_manager->txn_manager_lock);
-    
+
     if (toku_omt_size(txn_manager->live_root_txns) > 0) {
         OMTVALUE txnidv;
         // We use live_root_txns because roots are always older than children,
@@ -831,10 +831,9 @@ void toku_txn_manager_resume(TXN_MANAGER txn_manager) {
 }
 
 void
-toku_txn_manager_set_last_xid_from_logger(TXN_MANAGER txn_manager, TOKULOGGER logger) {
+toku_txn_manager_set_last_xid_from_logger(TXN_MANAGER txn_manager, TXNID last_xid) {
     invariant(txn_manager->last_xid == TXNID_NONE);
-    LSN last_lsn = toku_logger_last_lsn(logger);
-    txn_manager->last_xid = last_lsn.lsn;
+    txn_manager->last_xid = last_xid;
 }
 
 void

ft/txn_manager.h

@@ -69,7 +69,7 @@ void toku_txn_manager_unpin_live_txn_unlocked(TXN_MANAGER txn_manager, TOKUTXN t
 void toku_txn_manager_suspend(TXN_MANAGER txn_manager);
 void toku_txn_manager_resume(TXN_MANAGER txn_manager);
 
-void toku_txn_manager_set_last_xid_from_logger(TXN_MANAGER txn_manager, TOKULOGGER logger);
+void toku_txn_manager_set_last_xid_from_logger(TXN_MANAGER txn_manager, TXNID last_xid);
 void toku_txn_manager_set_last_xid_from_recovered_checkpoint(TXN_MANAGER txn_manager, TXNID last_xid);
 TXNID toku_txn_manager_get_last_xid(TXN_MANAGER mgr);
 

toku_include/toku_assert.h

@@ -53,6 +53,8 @@ extern void (*do_assert_hook)(void); // Set this to a function you want called a
 #define lazy_assert(a)          assert(a)      // indicates code is incomplete 
 #define lazy_assert_zero(a)     assert_zero(a) // indicates code is incomplete 
 #define invariant(a)            assert(a)      // indicates a code invariant that must be true
+#define invariant_null(a)       assert_zero(a) // indicates a code invariant that must be true
+#define invariant_notnull(a)    assert(a)      // indicates a code invariant that must be true
 #define invariant_zero(a)       assert_zero(a) // indicates a code invariant that must be true
 #define resource_assert(a)      assert(a)      // indicates resource must be available, otherwise unrecoverable
 #define resource_assert_zero(a) assert_zero(a) // indicates resource must be available, otherwise unrecoverable